清理 ¶

A proxy object uses a weakref callback so that when it gets garbage collected it deregisters itself from the manager which owns its referent.

A shared object gets deleted from the manager process when there are no longer any proxies referring to it.

进程池 ¶

可以创建用于履行提交给它的任务的一个进程池，采用 Pool 类。

class multiprocessing.pool. Pool ( [ processes [ , initializer [ , initargs [ , maxtasksperchild [ , context ] ] ] ] ] ) ¶

进程池对象，可以控制向其提交作业的工作者进程池。它支持带有超时和回调的异步结果，并拥有并行映射实现。

processes 是要使用的工作进程数。若 processes is None 那么，返回数通过 os.cpu_count() 被使用。

若 initializer 不是 None 那么，各工作者进程将调用 initializer(*initargs) 当它开始时。

maxtasksperchild 是工作进程在退出之前可以完成的任务数，然后以刷新工作者进程替换，释放未使用的资源。默认 maxtasksperchild is None ，这意味着工作者进程将活得与池一样长。

context 可用于指定为启动工作者进程的上下文。通常，创建池是使用函数 multiprocessing.Pool() 或 Pool() 方法为上下文对象。在这两种情况下， context 可适当设置。

注意：池对象的方法仅应由创建池的进程调用。

警告

multiprocessing.pool objects have internal resources that need to be properly managed (like any other resource) by using the pool as a context manager or by calling close() and terminate() manually. Failure to do this can lead to the process hanging on finalization.

Note that it is not correct to rely on the garbage collector to destroy the pool as CPython does not assure that the finalizer of the pool will be called (see object.__del__() 了解更多信息)。

3.2 版改变： 添加 maxtasksperchild 参数。

3.4 版改变： 添加 context 参数。

注意

工作者进程在 Pool typically live for the complete duration of the Pool’s work queue. A frequent pattern found in other systems (such as Apache, mod_wsgi, etc) to free resources held by workers is to allow a worker within a pool to complete only a set amount of work before being exiting, being cleaned up and a new process spawned to replace the old one. The maxtasksperchild 自变量到 Pool exposes this ability to the end user.

apply ( func [ , args [ , kwds ] ] ) ¶

调用 func 采用自变量 args 和关键词自变量 kwds 。它会阻塞，直到结果就绪为止。提供这种阻塞, apply_async() 更适合并行履行工作。此外， func 是池的唯一执行工作者。

apply_async ( func [ , args [ , kwds [ , callback [ , error_callback ] ] ] ] ) ¶

变体 apply() 方法返回 AsyncResult 对象。

若 callback is specified then it should be a callable which accepts a single argument. When the result becomes ready callback is applied to it, that is unless the call failed, in which case the error_callback is applied instead.

若 error_callback is specified then it should be a callable which accepts a single argument. If the target function fails, then the error_callback is called with the exception instance.

Callbacks should complete immediately since otherwise the thread which handles the results will get blocked.

map ( func , iterable [ , chunksize ] ) ¶

并行等效于 map() built-in function (it supports only one iterable argument though, for multiple iterables see starmap() ). It blocks until the result is ready.

This method chops the iterable into a number of chunks which it submits to the process pool as separate tasks. The (approximate) size of these chunks can be specified by setting chunksize to a positive integer.

Note that it may cause high memory usage for very long iterables. Consider using imap() or imap_unordered() with explicit chunksize option for better efficiency.

map_async ( func , iterable [ , chunksize [ , callback [ , error_callback ] ] ] ) ¶

变体 map() 方法返回 AsyncResult 对象。

若 callback is specified then it should be a callable which accepts a single argument. When the result becomes ready callback is applied to it, that is unless the call failed, in which case the error_callback is applied instead.

若 error_callback is specified then it should be a callable which accepts a single argument. If the target function fails, then the error_callback is called with the exception instance.

Callbacks should complete immediately since otherwise the thread which handles the results will get blocked.

imap ( func , iterable [ , chunksize ] ) ¶

惰性版本的 map() .

The chunksize argument is the same as the one used by the map() method. For very long iterables using a large value for chunksize can make the job complete much faster than using the default value of 1 .

Also if chunksize is 1 那么 next() method of the iterator returned by the imap() method has an optional timeout parameter: next(timeout) 会引发 multiprocessing.TimeoutError if the result cannot be returned within timeout 秒。

imap_unordered ( func , iterable [ , chunksize ] ) ¶

如同 imap() except that the ordering of the results from the returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be “correct”.)

starmap ( func , iterable [ , chunksize ] ) ¶

像 map() except that the elements of the iterable are expected to be iterables that are unpacked as arguments.

Hence an iterable of [(1,2), (3, 4)] 产生 [func(1,2), func(3,4)] .

Added in version 3.3.

starmap_async ( func , iterable [ , chunksize [ , callback [ , error_callback ] ] ] ) ¶

A combination of starmap() and map_async() that iterates over iterable of iterables and calls func with the iterables unpacked. Returns a result object.

Added in version 3.3.

close ( ) ¶

Prevents any more tasks from being submitted to the pool. Once all the tasks have been completed the worker processes will exit.

terminate ( ) ¶

Stops the worker processes immediately without completing outstanding work. When the pool object is garbage collected terminate() will be called immediately.

join ( ) ¶

等待工作者进程退出。必须调用 close() or terminate() before using join() .

3.3 版改变： Pool objects now support the context management protocol – see 上下文管理器类型 . __enter__() returns the pool object, and __exit__() 调用 terminate() .

class multiprocessing.pool. AsyncResult ¶

The class of the result returned by Pool.apply_async() and Pool.map_async() .

get ( [ timeout ] ) ¶

返回结果当它到达时。若 timeout 不是 None and the result does not arrive within timeout seconds then multiprocessing.TimeoutError is raised. If the remote call raised an exception then that exception will be reraised by get() .

wait ( [ timeout ] ) ¶

等待直到结果可用或直到 timeout 秒过去。

ready ( ) ¶

返回调用是否已完成。

successful ( ) ¶

Return whether the call completed without raising an exception. Will raise ValueError if the result is not ready.

3.7 版改变： If the result is not ready, ValueError is raised instead of AssertionError .

下列范例演示池的使用：

from multiprocessing import Pool
import time
def f(x):
    return x*x
if __name__ == '__main__':
    with Pool(processes=4) as pool:         # start 4 worker processes
        result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously in a single process
        print(result.get(timeout=1))        # prints "100" unless your computer is *very* slow
        print(pool.map(f, range(10)))       # prints "[0, 1, 4,..., 81]"
        it = pool.imap(f, range(10))
        print(next(it))                     # prints "0"
        print(next(it))                     # prints "1"
        print(it.next(timeout=1))           # prints "4" unless your computer is *very* slow
        result = pool.apply_async(time.sleep, (10,))
        print(result.get(timeout=1))        # raises multiprocessing.TimeoutError

Listener 和 Client ¶

Usually message passing between processes is done using queues or by using Connection objects returned by Pipe() .

不管怎样， multiprocessing.connection module allows some extra flexibility. It basically gives a high level message oriented API for dealing with sockets or Windows named pipes. It also has support for digest authentication 使用 hmac module, and for polling multiple connections at the same time.

multiprocessing.connection. deliver_challenge ( connection , authkey ) ¶

Send a randomly generated message to the other end of the connection and wait for a reply.

If the reply matches the digest of the message using authkey as the key then a welcome message is sent to the other end of the connection. Otherwise AuthenticationError 被引发。

multiprocessing.connection. answer_challenge ( connection , authkey ) ¶

Receive a message, calculate the digest of the message using authkey as the key, and then send the digest back.

If a welcome message is not received, then AuthenticationError 被引发。

multiprocessing.connection. Client ( address [ , 系列 [ , authkey ] ] ) ¶

Attempt to set up a connection to the listener which is using address address ，返回 Connection .

The type of the connection is determined by 系列 argument, but this can generally be omitted since it can usually be inferred from the format of address 。(见 地址格式 )

若 authkey 有给定且不是 None , it should be a byte string and will be used as the secret key for an HMAC-based authentication challenge. No authentication is done if authkey is None . AuthenticationError is raised if authentication fails. See 身份验证键 .

class multiprocessing.connection. Listener ( [ address [ , 系列 [ , backlog [ , authkey ] ] ] ] ) ¶

A wrapper for a bound socket or Windows named pipe which is ‘listening’ for connections.

address is the address to be used by the bound socket or named pipe of the listener object.

注意

If an address of ‘0.0.0.0’ is used, the address will not be a connectable end point on Windows. If you require a connectable end-point, you should use ‘127.0.0.1’.

系列 is the type of socket (or named pipe) to use. This can be one of the strings 'AF_INET' (for a TCP socket), 'AF_UNIX' (for a Unix domain socket) or 'AF_PIPE' (for a Windows named pipe). Of these only the first is guaranteed to be available. If 系列 is None then the family is inferred from the format of address 。若 address is also None then a default is chosen. This default is the family which is assumed to be the fastest available. See 地址格式 。注意，若 系列 is 'AF_UNIX' and address is None then the socket will be created in a private temporary directory created using tempfile.mkstemp() .

If the listener object uses a socket then backlog (1 by default) is passed to the listen() method of the socket once it has been bound.

若 authkey 有给定且不是 None , it should be a byte string and will be used as the secret key for an HMAC-based authentication challenge. No authentication is done if authkey is None . AuthenticationError is raised if authentication fails. See 身份验证键 .

accept ( ) ¶

Accept a connection on the bound socket or named pipe of the listener object and return a Connection object. If authentication is attempted and fails, then AuthenticationError 被引发。

close ( ) ¶

Close the bound socket or named pipe of the listener object. This is called automatically when the listener is garbage collected. However it is advisable to call it explicitly.

Listener 对象拥有以下只读特性：

address ¶

Listener 对象正在使用的地址。

last_accepted ¶

The address from which the last accepted connection came. If this is unavailable then it is None .

3.3 版改变： Listener objects now support the context management protocol – see 上下文管理器类型 . __enter__() returns the listener object, and __exit__() 调用 close() .

multiprocessing.connection. wait ( object_list , timeout = None ) ¶

Wait till an object in object_list is ready. Returns the list of those objects in object_list which are ready. If timeout is a float then the call blocks for at most that many seconds. If timeout is None then it will block for an unlimited period. A negative timeout is equivalent to a zero timeout.

For both POSIX and Windows, an object can appear in object_list if it is

• a readable Connection object;

• a connected and readable socket.socket object; or

• the sentinel attribute of a Process 对象。

A connection or socket object is ready when there is data available to be read from it, or the other end has been closed.

POSIX : wait(object_list, timeout) almost equivalent select.select(object_list, [], [], timeout) . The difference is that, if select.select() is interrupted by a signal, it can raise OSError with an error number of EINTR ，而 wait() will not.

Windows : An item in object_list must either be an integer handle which is waitable (according to the definition used by the documentation of the Win32 function WaitForMultipleObjects() ) or it can be an object with a fileno() method which returns a socket handle or pipe handle. (Note that pipe handles and socket handles are not waitable handles.)

Added in version 3.3.

范例

The following server code creates a listener which uses 'secret password' as an authentication key. It then waits for a connection and sends some data to the client:

from multiprocessing.connection import Listener
from array import array
address = ('localhost', 6000)     # family is deduced to be 'AF_INET'
with Listener(address, authkey=b'secret password') as listener:
    with listener.accept() as conn:
        print('connection accepted from', listener.last_accepted)
        conn.send([2.25, None, 'junk', float])
        conn.send_bytes(b'hello')
        conn.send_bytes(array('i', [42, 1729]))

The following code connects to the server and receives some data from the server:

from multiprocessing.connection import Client
from array import array
address = ('localhost', 6000)
with Client(address, authkey=b'secret password') as conn:
    print(conn.recv())                  # => [2.25, None, 'junk', float]
    print(conn.recv_bytes())            # => 'hello'
    arr = array('i', [0, 0, 0, 0, 0])
    print(conn.recv_bytes_into(arr))    # => 8
    print(arr)                          # => array('i', [42, 1729, 0, 0, 0])

The following code uses wait() to wait for messages from multiple processes at once:

import time, random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait
def foo(w):
    for i in range(10):
        w.send((i, current_process().name))
    w.close()
if __name__ == '__main__':
    readers = []
    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        # We close the writable end of the pipe now to be sure that
        # p is the only process which owns a handle for it.  This
        # ensures that when p closes its handle for the writable end,
        # wait() will promptly report the readable end as being ready.
        w.close()
    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                readers.remove(r)
            else:
                print(msg)