Multiprocessing: How to use Pool.map on a function defined in a class?

by

I could not use the codes posted so far because the codes using “multiprocessing.Pool” do not work with lambda expressions and the codes not using “multiprocessing.Pool” spawn as many processes as there are work items.

I adapted the code s.t. it spawns a predefined amount of workers and only iterates through the input list if there exists an idle worker. I also enabled the “daemon” mode for the workers s.t. ctrl-c works as expected.

import multiprocessing


def fun(f, q_in, q_out):
    while True:
        i, x = q_in.get()
        if i is None:
            break
        q_out.put((i, f(x)))


def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]


if __name__ == '__main__':
    print(parmap(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8]))

More Related Contents:

  1. Python multiprocessing PicklingError: Can’t pickle
  2. Can’t pickle when using multiprocessing Pool.map()
  3. What can multiprocessing and dill do together?
  4. Why can I pass an instance method to multiprocessing.Process, but not a multiprocessing.Pool?
  5. Python Multiprocessing Lib Error (AttributeError: __exit__)
  6. multiprocessing.Pool – PicklingError: Can’t pickle : attribute lookup thread.lock failed
  7. Saving and loading multiple objects in pickle file?
  8. Why does multiprocessing use only a single core after I import numpy?
  9. Sharing a result queue among several processes
  10. Keyboard Interrupts with python’s multiprocessing Pool
  11. multiprocessing: Understanding logic behind `chunksize`
  12. multiprocessing.Pool: What’s the difference between map_async and imap?
  13. multiprocessing in python – sharing large object (e.g. pandas dataframe) between multiple processes
  14. Multiprocessing.Pool makes Numpy matrix multiplication slower
  15. Can I use a multiprocessing Queue in a function called by Pool.imap?
  16. Python pickling after changing a module’s directory
  17. Accessing an attribute of a multiprocessing Proxy of a class
  18. call multiprocessing in class method Python
  19. Leveraging “Copy-on-Write” to Copy Data to Multiprocessing.Pool() Worker Processes
  20. What causes the error “_pickle.UnpicklingError: invalid load key, ‘ ‘.”?
  21. How to get the return value of a function passed to multiprocessing.Process?
  22. How to share numpy random state of a parent process with child processes?
  23. multiprocessing vs multithreading vs asyncio
  24. pickle – putting more than 1 object in a file? [duplicate]
  25. Python multiprocessing – tracking the process of pool.map operation
  26. Python Pickling Dictionary EOFError
  27. Creating and updating nested dictionaries and lists inside multiprocessing.Manager object
  28. Why multiprocessing.Process behave differently on windows and linux for global object and function arguments
  29. How to use Python multiprocessing Pool.map to fill numpy array in a for loop
  30. Using python multiprocessing with different random seed for each process

Leave a Comment