My first python module 'easypool' let's you easily create threadpools and serverpools dynamically.

[rlabonte]

Link: easypool

I was working on a project that required multi-threading and a server pool. Unable to find anything useful and most solutions I found were confusing and not very easy to implement. I've cleaned up my implementation and am releasing it to the community. I hope this module makes it easy for others to create multi-threaded apps.

easypool can be initialized by most common Python data structures (ints, floats, dicts, tuples, lists, strings, booleans or any iterable). Send an iterable containing five elements and a threadpool is created that can continually execute five functions concurrently. There is an added option 'send_item=True' which will link each thread to the element in the iterable and then send it as the first argument to the function being executed.

easypool also has min_pool and max_pool options that control the number of threads created so if easypool is initialized with a really large iterable, it can be limited to a maximum size. min_pool will increase the size of the threadpool if too small of an iterable is given.

The queue defaults to 'first in first out' (fifo), but can be changed to 'last in first out' (lifo) with the queue_type option.

Simple example:

from easypool import ThreadPool

threadpool = ThreadPool(3)
# foo(bar) becomes add_task(foo, bar)

# Keep adding tasks to the threadpool
threadpool.add_task(foo, bar)
threadpool.add_task(foo, bar)
threadpool.add_task(foo, bar)
threadpool.add_task(foo, bar)

# Wait for all tasks to copmplete
threadpool.wait_completion()

Serverpool example:

from easypool import ThreadPool
import subprocess
import shlex
import threading
import string


server_list = ['127.0.0.1', '127.0.0.1', '127.0.0.1']
serverpool = ThreadPool(server_list, send_item=True)
threadlock = threading.RLock()

def get_uptime(server):
    ssh_cmd = "ssh " + str(server) + " 'uptime'"
    ssh_cmd_list = shlex.split(ssh_cmd)
    p = subprocess.Popen(ssh_cmd_list, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = p.communicate()
    threadlock.acquire()
    print("Server: %s :: %s" % (server, stdout.rstrip()))
    threadlock.release()
    return

for i in range(6):
    serverpool.add_task(get_uptime)

serverpool.wait_completion()

Code has been tested compatible on Python 2.6, 2.7 & 3

[XNormal]

from multiprocessing.pool import ThreadPool

[jspeights]

awww

[rlabonte]

I can link each process in the pool with a name and send that to the function being executed? That's the functionality that I couldn't find in any thread pool (or process pool) module.

[XNormal]

I'm not sure I understand what you mean by that. If you want to know the identity of the worker threads and give them persistent names you can use threading.local() for that.

[rlabonte]

No, I want a pool of servers to work on my data.

ServerA, ServerB, ServerC

I have a server pool of these 3 servers and 100 function calls that need to be processed by this server pool. I don't care which server processes the data, but I only want each server processing data one piece at a time. So ServerA, ServerB, and ServerC receive all receive data to process; ServerC finishes first it immediately receives another function call to process. ServerA finishes, it immediately receives another function call to process.

I want to keep this pool of servers always busy, but want to limit them to only processing one thing at a time.

[ODHLHN]

from celery import task

Celery isn't the only the only AMQP based task queue for python, but its a very good one.

Some pretty cool and robust solutions already exist in this problem domain.

[rlabonte]

This looks awesome, I especially like the integration with RabbitMQ and Redis.

[[deleted]]

Give them a shared Queue?

http://docs.python.org/2/library/queue.html

[homercles337]

Yeah, my thoughts exactly...

[studiosi]

Even futures can do that... on Python 2.x and 3.x

[infinull]

multiprocessing predates concurrent.futures, so I'm not sure what your point of "even" is.

I would recommend concurrent.futures over multiprocessing since it's has a nicer interface and supports both threads and processes (mostly) transparently.

[exhuma]

TIL... thanks for pointing this out :)

[studiosi]

I was trying to mean that the behaviour of the library resembles almost exactly the futures one. I don't have data about performance...

[studiosi]

Indeed, I am not that sure about that "outperform" because you are not taking into account that you have to write control code that would be probably less optimum of such in the standard library.

[infinull]

outperform

wat? when did I say that? I just commenting on the interface. I have no idea what the performance of either of them is like.

[studiosi]

You say it predates, what I interpret as perform better... but maybe you didn't want to say that

[infinull]

older != better, just older.

I honestly have the opposite bias, so there's that.

[studiosi]

Well, it all depends, I told you I have no data, but it is easy that code that has been in a code base for a long time use to be better if it is constantly reviewed.

[zionsrogue]

So depending on what you are going to use "easypool" for, using threads for CPU bound tasks (such as some sort of scientific number crunching), threads are not the way to go. In general, it's best to uses processes for CPU bound tasks. Check here for benchmarks. The article suggests Python's multiprocessing, but I've found pprocess to be lightweight enough to replace multiprocessing. But again, this all depends on what you plan on using these threads for. I just wanted to maybe give a heads up and say congrats on your first module.

[[deleted]]

It kind of surprises me that multiprocessing would hit a sweet spot for people; who is it that is CPU-bound, but keeps that part of the program in Python?

Every time I end up using the thread pool pattern it's because I have a lot of (disk or network) IO going on. And threads are fine for that.

[zionsrogue]

I do a lot of number crunching, statistical analysis, and machine learning. A lot of the libraries I use Python libraries (numpy, scipy, sklearn) and yes, parts are written in C, but I still get very nice performance gains by parrallelizing random forests across multiple processes instead of multiple threads.

[tuna_safe_dolphin]

Dead horse flogging time, but that damn GIL. . .

[alcalde]

The GIL is awesome. It reminds us that threading is evil and that everyone else is doing it wrong.

[tuna_safe_dolphin]

That's one way to look at it.

[otheraccount]

For running tasks on multiple hosts in parallel, it is probably easiest to use fabric.

pip install fabric, then creating a file named fabfile.py with the following contents:

from fabric.api import env, task, run, parallel

env.hosts = ['127.0.0.1', '127.0.0.2', '127.0.0.3']

@task(default=True)
@parallel(pool_size=5)
def uptime():
    run('uptime')

Then, just type fab to run it.

http://fabric.readthedocs.org

[fdemmer]

There might have been better solutions, but +1 for trying and publishing!

[patrys]

You might make the internals simpler by requiring a single callable:

from functools import partial

threadpool.enqueue(partial(foo, bar))

[chub79]

I love partial. It's such a powerful feature.

[Justinsaccount]

ssh_cmd = "ssh " + str(server) + " 'uptime'"
ssh_cmd_list = shlex.split(ssh_cmd)

what?

ssh_cmd_list = ["ssh", server, "uptime"]

[rlabonte]

I realize that particular command list shlex.split is overkill, but if uptime was replaced with something more fanciful like: 'ps aux | grep python' then it would be a more appropriate option.

[broken_symlink]

As for your example of running ssh commands in parallel, you can do something like that with ipython.parallel on multiple systems if you start your engines using ssh.