Skip to content Skip to sidebar Skip to footer

Get A Unique Id For Worker In Python Multiprocessing Pool

Is there a way to assign each worker in a python multiprocessing pool a unique ID in a way that a job being run by a particular worker in the pool could know which worker is runnin

Solution 1:

It seems like what you want is simple: multiprocessing.current_process(). For example:

import multiprocessing

deff(x):
    print multiprocessing.current_process()
    return x * x

p = multiprocessing.Pool()
print p.map(f, range(6))

Output:

$ python foo.py 
<Process(PoolWorker-1, started daemon)>
<Process(PoolWorker-2, started daemon)>
<Process(PoolWorker-3, started daemon)>
<Process(PoolWorker-1, started daemon)>
<Process(PoolWorker-2, started daemon)>
<Process(PoolWorker-4, started daemon)>
[0, 1, 4, 9, 16, 25]

This returns the process object itself, so the process can be its own identity. You could also call id on it for a unique numerical id -- in cpython, this is the memory address of the process object, so I don't think there's any possibility of overlap. Finally, you can use the ident or the pid property of the process -- but that's only set once the process is started.

Furthermore, looking over the source, it seems to me very likely that autogenerated names (as exemplified by the first value in the Process repr strings above) are unique. multiprocessing maintains an itertools.counter object for every process, which is used to generate an _identity tuple for any child processes it spawns. So the top-level process produces child process with single-value ids, and they spawn process with two-value ids, and so on. Then, if no name is passed to the Process constructor, it simply autogenerates the name based on the _identity, using ':'.join(...). Then Poolalters the name of the process using replace, leaving the autogenerated id the same.

The upshot of all this is that although two Processes may have the same name, because you may assign the same name to them when you create them, they are unique if you don't touch the name parameter. Also, you could theoretically use _identity as a unique identifier; but I gather they made that variable private for a reason!

An example of the above in action:

import multiprocessing

deff(x):
    created = multiprocessing.Process()
    current = multiprocessing.current_process()
    print'running:', current.name, current._identity
    print'created:', created.name, created._identity
    return x * x

p = multiprocessing.Pool()
print p.map(f, range(6))

Output:

$ python foo.py 
running: PoolWorker-1 (1,)created: Process-1:1 (1, 1)running: PoolWorker-2 (2,)created: Process-2:1 (2, 1)running: PoolWorker-3 (3,)created: Process-3:1 (3, 1)running: PoolWorker-1 (1,)created: Process-1:2 (1, 2)running: PoolWorker-2 (2,)created: Process-2:2 (2, 2)running: PoolWorker-4 (4,)created: Process-4:1 (4, 1)
[0, 1, 4, 9, 16, 25]

Solution 2:

You can use multiprocessing.Queue to store the ids and then get the id at initialization of the pool process.

Advantages:

  • You do not need to rely on internals.
  • If your use case is to manage resources/ devices then you can put in the device number directly. This will also ensure that no device is used twice: If you have more processes in your pool than devices, the additional processes will block on queue.get() and will not perform any work (This won't block your porgram, or at least it did not when I tested).

Disadvantages:

  • You have additional communication overhead and spawning the pool processes takes a tiny bit longer: Without the sleep(1) in the example all work might be performed by the first process, as others are not done initializing, yet.
  • You need a global (or at least I don't know a way around it)

Example:

import multiprocessing
fromtime import sleep

def init(queue):
    global idx
    idx = queue.get()

def f(x):
    global idx
    process = multiprocessing.current_process()
    sleep(1)
    return (idx, process.pid, x * x)

ids = [0, 1, 2, 3]
manager = multiprocessing.Manager()
idQueue = manager.Queue()

for i in ids:
    idQueue.put(i)

p = multiprocessing.Pool(8, init, (idQueue,))
print(p.map(f, range(8)))

Output:

[(0, 8289, 0), (1, 8290, 1), (2, 8294, 4), (3, 8291, 9), (0, 8289, 16), (1, 8290, 25), (2, 8294, 36), (3, 8291, 49)]

Note, that there are only 4 different pid, although the pool contains 8 processes and one idx is only used by one process.

Solution 3:

I did this with threading and ended up using a queue to handle job management. Here is the baseline. My complete version has a bunch of try-catches (particularly in the worker, to make sure that q.task_done() is called even on failure).

from threading import Thread
from queue import Queue
import time
import random


def run(idx, *args):
    time.sleep(random.random() * 1)
    print idx, ':', args


def run_jobs(jobs, workers=1):
    q = Queue()
    def worker(idx):
        while True:
            args = q.get()
            run(idx, *args)
            q.task_done()

    for job in jobs:
        q.put(job)

    for i in range(0, workers):
        t = Thread(target=worker, args=[i])
        t.daemon = True
        t.start()

    q.join()


if __name__ == "__main__":
    run_jobs([('job', i) for i in range(0,10)], workers=5)

I didn't need to use multiprocessing (my workers are just for calling an external process), but this could be extended. The API for multiprocessing changes it a touch, here's how you could adapt:

from multiprocessing import Process, Queue
from Queue import Empty
import time
import random

defrun(idx, *args):
    time.sleep(random.random() * i)
    print idx, ':', args


defrun_jobs(jobs, workers=1):
    q = Queue()
    defworker(idx):
        try:
            whileTrue:
                args = q.get(timeout=1)
                run(idx, *args)
        except Empty:
            returnfor job in jobs:
        q.put(job)

    processes = []
    for i inrange(0, workers):
        p = Process(target=worker, args=[i])
        p.daemon = True
        p.start()
        processes.append(p)

    for p in processes: 
        p.join()


if __name__ == "__main__":
    run_jobs([('job', i) for i inrange(0,10)], workers=5)

Both versions will output something like:

0 : ('job', 0)
1 : ('job', 2)
1 : ('job', 6)
3 : ('job', 3)
0 : ('job', 5)
1 : ('job', 7)
2 : ('job', 1)
4 : ('job', 4)
3 : ('job', 8)
0 : ('job', 9)

Solution 4:

I'm not sure how it would work with Pool, but printing Process gives some unique output:

x = Process(target=time.sleep, args=[20])
x.start()
print(x)  # <Processname='Process-5'pid=97121parent=95732started>

Solution 5:

I managed to map to a class method by getting the function handle using getattr, then using a wrapper to pack and unpack as many arguments as I wanted to pass to the method being mapped. In my case I was passing methods from the same class where the pool was being launched, but you can also pass an object as well to map to different classes.

This is the code:

import multiprocessing
from multiprocessing import Pool


defwarp(args):
    func = args[0]
    frame = args[1]
    left_over = args[2:]
    func(frame, *left_over)


classMyClass:

    def__init__(self):
        self.my_flag = 5defexec_method(self, method, int_list, *args):
        obj = getattr(self, method.__name__)

        packed = list()
        for i in int_list:
            pack = list()
            pack.append(obj)
            pack.append(i)
            for arg in args:
                pack.append(arg)
            packed.append(pack)

        print("Start")
        pool = Pool(processes=multiprocessing.cpu_count())
        pool.map(warp, packed)
        print("End")

    defmethod1(self, my_str):
        print(self.my_flag, my_str)

    defmethod2(self, i, print_str, bool_flat):
        print(multiprocessing.current_process(), self.my_flag, i, print_str, str(bool_flat))


cls: MyClass = MyClass()
cls.my_flag = 58
cls.exec_method(cls.method2, [1, 5, 10, 20, 30], "this is a string", True)

This is the output:

Start
<ForkProcess(ForkPoolWorker-1, started daemon)> 581 this is a stringTrue
<ForkProcess(ForkPoolWorker-2, started daemon)> 585 this is a stringTrue
<ForkProcess(ForkPoolWorker-4, started daemon)> 5820 this is a stringTrue
<ForkProcess(ForkPoolWorker-5, started daemon)> 5830 this is a stringTrue
<ForkProcess(ForkPoolWorker-3, started daemon)> 5810 this is a stringTrueEnd

Post a Comment for "Get A Unique Id For Worker In Python Multiprocessing Pool"