Hướng dẫn multiprocessing python return value

Question

For some reason, I couldn't find a general example of how to do this with Queue anywhere (even Python's doc examples don't spawn multiple processes), so here's what I got working after like 10 tries:

Nội dung chính Show

How to Return Value From a Process
Return Variable From Process with Value
Return Variable From Process with Pipe
Return Variable From Process with Queue
Example of Returning a Variable via a Value
Example of Returning a Variable via a Pipe
Example of Returning a Variable via a Queue
Further Reading

def add_helper(queue, arg1, arg2): # the func called in child processes
    ret = arg1 + arg2
    queue.put(ret)

def multi_add(): # spawns child processes
    q = Queue()
    processes = []
    rets = []
    for _ in range(0, 100):
        p = Process(target=add_helper, args=(q, 1, 2))
        processes.append(p)
        p.start()
    for p in processes:
        ret = q.get() # will block
        rets.append(ret)
    for p in processes:
        p.join()
    return rets

Queue is a blocking, thread-safe queue that you can use to store the return values from the child processes. So you have to pass the queue to each process. Something less obvious here is that you have to get() from the queue before you join the Processes or else the queue fills up and blocks everything.

Update for those who are object-oriented (tested in Python 3.4):

from multiprocessing import Process, Queue

class Multiprocessor():

    def __init__(self):
        self.processes = []
        self.queue = Queue()

    @staticmethod
    def _wrapper(func, queue, args, kwargs):
        ret = func(*args, **kwargs)
        queue.put(ret)

    def run(self, func, *args, **kwargs):
        args2 = [func, self.queue, args, kwargs]
        p = Process(target=self._wrapper, args=args2)
        self.processes.append(p)
        p.start()

    def wait(self):
        rets = []
        for p in self.processes:
            ret = self.queue.get()
            rets.append(ret)
        for p in self.processes:
            p.join()
        return rets

# tester
if __name__ == "__main__":
    mp = Multiprocessor()
    num_proc = 64
    for _ in range(num_proc): # queue up multiple tasks running `sum`
        mp.run(sum, [1, 2, 3, 4, 5])
    ret = mp.wait() # get all results
    print(ret)
    assert len(ret) == num_proc and all(r == 15 for r in ret)

You can return a variable from a child process using a multiprocessing.Value or a multiprocessing.Queue.

In this tutorial you will discover how to return a value from a process in Python.

Let’s get started.

Need to Return Value From Process
How to Return Value From a Process
- Return Variable From Process with Value
- Return Variable From Process with Pipe
- Return Variable From Process with Queue
Example of Returning a Variable via a Value
Example of Returning a Variable via a Pipe
Example of Returning a Variable via a Queue
Further Reading
Takeaways

A process is a running instance of a computer program.

Every Python program is executed in a Process, which is a new instance of the Python interpreter. This process has the name MainProcess and has one thread used to execute the program instructions called the MainThread. Both processes and threads are created and managed by the underlying operating system.

Sometimes we may need to create new child processes in our program in order to execute code concurrently.

Python provides the ability to create and manage new processes via the multiprocessing.Process class.

In multiprocessing programming we typically need to return a value from a process.

This is challenging as there are no direct methods for returning a value from a process to another calling process.

How can we return a value from a process?

How to Return Value From a Process

There are no direct methods to return a value from a process.

Instead, it can be achieved using indirect methods.

There are a number of indirect methods to choose from. The three most convenient and widely used methods to return a value from a process are as follows:

Use a multiprocessing.Value object.
Use a multiprocessing.Pipe object.
Use a multiprocessing.Queue object.

Other approaches might include:

Use a multiprocessing.Manager, e.g. processes interact with the same object via proxies.
Use multiprocessing.sharedctypes, e.g. same methods that underlie multiprocessing.Value.
Use multiprocessing.shared_memory.

Do you know of any other methods?
Let me know in the comments below.

Next, let’s take a closer look at some of these methods.

Return Variable From Process with Value

We can return a variable from a process using a multiprocessing.Value object.

These classes explicitly define data attributes designed to be shared between processes in a process-safe manner.

A process-safe manner means that only one process can read or access the variable at a time. Shared variables mean that changes made in one process are always propagated and made available to other processes.

An instance of the multiprocessing.Value can be defined in the constructor of a custom class as a shared instance variable.

The constructor of the multiprocessing.Value class requires that we specify the data type and an initial value.

The data type can be specified using ctype “type” or a typecode.

You can learn more about ctypes here:

ctypes — A foreign function library for Python

Typecodes are familiar and easy to use, for example ‘i’ for a signed integer or ‘f’ for a single floating-point value.

You can see a handy table of type codes here:

array — Efficient arrays of numeric values

For example, we can define a Value shared memory variable that holds a signed integer and is initialized to the value zero.

...

# initialize an integer shared variable

data=multiprocessing.Value('i',0)

This variable can then be initialized in a parent process and shared with a child process

We can change the value of the shared data variable via the “value” attribute.

For example:

...

# change the value of the shared variable

data.value=100

We can access the value of the shared data variable via the same “value” attribute.

For example:

...

# access the shared variable

value=data.value

The propagation of changes to the shared variable and mutual exclusion locking of the shared variable is all performed automatically behind the scenes.

Return Variable From Process with Pipe

We can return a variable from a process using the multiprocessing.Pipe class.

In multiprocessing, a pipe is a connection between two processes in Python.

It is used to send data from one process which is received by another process.

Under the covers, a pipe is implemented using a pair of connection objects, provided by the multiprocessing.connection.Connection class.

A pipe can be created by calling the constructor of the multiprocessing.Pipe class, which returns two multiprocessing.connection.Connection objects.

For example:

...

# create a pipe

conn1,conn2=multiprocessing.Pipe()

By default, the first connection (conn1) can only be used to receive data, whereas the second connection (conn2) can only be used to send data.

Objects can be shared between processes using the Pipe.

The Connection.send() function can be used to send objects from one process to another.

The objects sent must be picklable.

For example:

...

# send an object

conn2.send('Hello world')

The Connection.recv() function can be used to receive objects in one process sent by another.

The objects received will be automatically un-pickled.

For example:

...

# receive an object

data=conn1.recv()

The function call will block until an object is received.

You can learn more about pipes between processes in the tutorial:

Multiprocessing Pipe in Python

Return Variable From Process with Queue

We can return a variable from a process using the multiprocessing.Queue class.

A queue is a data structure on which items can be added by a call to put() and from which items can be retrieved by a call to get().

The multiprocessing.Queue provides a first-in, first-out FIFO queue, which means that the items are retrieved from the queue in the order they were added. The first items added to the queue will be the first items retrieved. This is opposed to other queue types such as last-in, first-out and priority queues.

The multiprocessing.Queue can be used by first creating an instance of the class. This will create an unbounded queue by default, that is, a queue with no size limit.

For example:

...

# created an unbounded queue

queue=multiprocessing.Queue()

Items can be added to the queue via a call to put(), for example:

...

# add an item to the queue

queue.put(item)

By default, the call to put() will block and will not use a timeout.

Items can be retrieved from the queue by calls to get().

For example:

...

# get an item from the queue

item=queue.get()

By default, the call to get() will block until an item is available to retrieve from the queue and will not use a timeout.

You can learn more about multiprocessing queues in the tutorial:

Multiprocessing Queue in Python

Now that we know how to return a variable from a process, let’s look at some worked examples.

Confused by the multiprocessing module API?
Download my FREE PDF cheat sheet

Example of Returning a Variable via a Value

We can return a variable from a process using a multiprocessing.Value.

In this example we will create a shared Value object, then create a new child process that will execute a custom function. The function will generate a random value and store it in the value. This value will then be “returned” to the parent process. Specifically, the parent process will access the shared value set by the child process.

First, we can define a function to execute in the child process.

We will name the function task() and it will take a shared multiprocessing.Value instance as an argument

# function to execute in a child process

def task(variable):

# ...

The function will then generate a random value between 0 and 1 using the random.random() function.

...

# generate some data

data=random()

The generated value will be reported and the process will block for a fraction of a second to simulate computational effort.

...

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

Finally, the child process will store the generated value in the shared variable so that the parent variable can access it.

This simulates a return value from the child process.

...

# return data via value

variable.value=data

Tying this together, the complete task() function is listed below.

# function to execute in a child process

def task(variable):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via value

variable.value =data

Next, in the main process we will first create the shared multiprocessing.Value instance.

We will configure it to hold a floating point value and initialize it with the value zero.

...

# create shared variable

variable=Value('f',0.0)

We can then create a new multiprocessing.Process instance and configure it to execute our task() function and pass the shared variable instance as an argument.

...

# create a child process process

process=Process(target=task,args=(variable,))

If you are new to executing a function in a new process, see the tutorial:

Run a Function in a Child Process

The process can then be started and the main process will block until the child process terminates using the join() function.

...

# start the process

process.start()

# wait for the process to finish

process.join()

If you are new to joining a child process, see the tutorial:

How to Join a Process in Python

Finally, the parent process will access the simulated “return” value from the child process via the shared multiprocessing.Value instance.

...

# report return value

print(f'Returned: {variable.value}')

Tying this together, the complete example is listed below.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

# SuperFastPython.com

# example of returning a variable from a process using a value

from random import random

from time import sleep

from multiprocessing import Value

from multiprocessing import Process

# function to execute in a child process

def task(variable):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via value

variable.value=data

# protect the entry point

if __name__=='__main__':

# create shared variable

variable=Value('f',0.0)

# create a child process process

process =Process(target=task,args=(variable,))

# start the process

process.start()

# wait for the process to finish

process.join()

# report return value

print(f'Returned: {variable.value}')

Running the example first creates the shared multiprocessing.Value instance.

The child process is configured and started and the main process blocks until the child process terminates.

The child process generates a random value and blocks for a moment. It then stores the generated value in the shared Value instance for the parent process to access.

The child process terminates and the main process unblocks.

The parent process then reports the number stored in the shared Value instance, simulating a return value.

We can see that the generated value in the child process matches the value accessed by the parent process, showing that the simulated return worked as expected.

Note, your specific results will differ given the use of random numbers.

Generated 0.9047471579259425

Returned: 0.9047471284866333

Need help with Python Multiprocessing?

Sign-up to my FREE 7-day email course. Discover how to use the Python multiprocessing module, including how to create and start child processes, how to use a mutex and semaphore, and much more!

Click the button below and enter your email address to sign-up and get the first lesson right now.

Start Your FREE Email Course Now!

Example of Returning a Variable via a Pipe

We can simulate returning a variable from a process using a multiprocessing.Pipe.

In this example we will create a pipe which will create two connection objects. One connection object will be passed to the child process and one will be held by the parent process. The child process will perform its work and send its result to the parent process via the pipe connection object.

Firstly, we can define a function to execute in a child process.

The function will take a connection object as an argument, generate a random value between 0 and 1, report it, block for a fraction of a second to simulate work and then send the result via the connection object.

The task() function below implements this.

# function to execute in a child process

def task(connection):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via pipe

connection.send(data)

The main process will first create the multiprocessing.Pipe which return two connection objects.

The first connection object is only used for receiving data from the child process and the second connection object is only used to send data from the child process.

...

# create the pipe

conn1,conn2=Pipe()

We can then configure a new multiprocessing.Process instance to execute our task() function and pass it the connection object for sending. Then the child process can then be started.

...

# create a child process process

process=Process(target=task,args=(conn2,))

# start the process

process.start()

The parent process will then call the recv() function on the connection and wait for a result to be “returned” or sent from the child process.

...

# wait for the return value

value=conn1.recv()

Once received, the result can be reported to confirm it matches the value that was generated.

...

# report return value

print(f'Returned: {value}')

Tying this together, the complete example is listed below.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

# SuperFastPython.com

# example of returning a variable from a process using a pipe

from random import random

from time import sleep

from multiprocessing import Pipe

from multiprocessing import Process

# function to execute in a child process

def task(connection):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via pipe

connection.send(data)

# protect the entry point

if __name__=='__main__':

# create the pipe

conn1,conn2=Pipe()

# create a child process process

process =Process(target=task,args=(conn2,))

# start the process

process.start()

# wait for the return value

value=conn1.recv()

# report return value

print(f'Returned: {value}')

Running the example first creates the Pipe, which returns two connection objects.

The child process is then configured and started.

The parent process then blocks, waiting on the pipe for the result to be sent by the child process.

The child process generates a number, reports it, then sends it to the parent process via the pipe.

The parent process receives value, then reports it.

We can see that the value that was reported by the parent process matches the value generated by the child process.

Note, your specific results will differ given the use of random numbers.

Generated 0.7498487341429124

Returned: 0.7498487341429124

Example of Returning a Variable via a Queue

We can simulate returning a variable from a process using a multiprocessing.Queue.

In this example we will create a queue which will be shared between the two processes. The child process will prepare some data and put it on the shared queue. The parent process will block and wait for data to arrive on the queue from the child process.

Firstly, we can define a function to execute in a child process.

The function will take the shared queue object as an argument, generate a random value between 0 and 1, report it, block for a fraction of a second to simulate work and then put the result on the queue.

The task() function below implements this.

# function to execute in a child process

def task(queue):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via queue

queue.put(data)

The main process will first create the multiprocessing.Queue which will be shared between the processes.

...

# create the queue

queue=Queue()

We can then configure a new multiprocessing.Process instance to execute our task() function and pass it the shared Queue object. Then the child process can then be started.

...

# create a child process process

process=Process(target=task,args=(queue,))

# start the process

process.start()

The parent process will then call the get() function on the Queue and wait for a result to be “returned” or sent from the child process.

...

# wait for the return value

value=queue.get()

Once received, the result can be reported to confirm it matches the value that was generated.

...

# report return value

print(f'Returned: {value}')

Tying this together, the complete example is listed below.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

# SuperFastPython.com

# example of returning a variable from a process using a queue

from random import random

from time import sleep

from multiprocessing import Queue

from multiprocessing import Process

# function to execute in a child process

def task(queue):

# generate some data

data=random()

# block, to simulate computational effort

print(f'Generated {data}',flush=True)

sleep(data)

# return data via queue

queue.put(data)

# protect the entry point

if __name__=='__main__':

# create the queue

queue=Queue()

# create a child process process

process =Process(target=task,args=(queue,))

# start the process

process.start()

# wait for the return value

value=queue.get()

# report return value

print(f'Returned: {value}')

Running the example first creates the shared Queue.

The child process is then configured and started.

The parent process then blocks, waiting in the queue for the result to arrive.

The child process generates a number, reports it, then puts it on the queue to simulate returning a value.

The parent process retrieves a value from the queue and then reports it.

We can see that the value that was reported by the parent process matches the value generated by the child process.

Note, your specific results will differ given the use of random numbers.

Generated 0.3732253790840232

Returned: 0.3732253790840232

Takeaways

You now know how to return a value from a process.

Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.

Photo by Sean Chen on Unsplash

programming python Process in Python Parallel Python

Hướng dẫn multiprocessing python return value

How to Return Value From a Process

Return Variable From Process with Value

Return Variable From Process with Pipe

Return Variable From Process with Queue

Example of Returning a Variable via a Value

Example of Returning a Variable via a Pipe

Example of Returning a Variable via a Queue

Further Reading

Takeaways

Bài Viết Liên Quan

Quảng Cáo

Có thể bạn quan tâm

Toplist được quan tâm

Quảng cáo

Xem Nhiều

Quảng cáo

Chúng tôi

Điều khoản

Trợ giúp

Mạng xã hội