| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent 1e88f3f commit bdb1cf1
7 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -415,13 +415,14 @@ The :mod:`multiprocessing` package mostly replicates the API of the | |||
| 415 | 415 | A numeric handle of a system object which will become "ready" when | |
| 416 | 416 | the process ends. | |
| 417 | 417 | ||
| 418 | + You can use this value if you want to wait on several events at | ||
| 419 | + once using :func:`multiprocessing.connection.wait`. Otherwise | ||
| 420 | + calling :meth:`join()` is simpler. | ||
| 421 | + | ||
| 418 | 422 | On Windows, this is an OS handle usable with the ``WaitForSingleObject`` | |
| 419 | 423 | and ``WaitForMultipleObjects`` family of API calls. On Unix, this is | |
| 420 | 424 | a file descriptor usable with primitives from the :mod:`select` module. | |
| 421 | 425 | ||
| 422 | - You can use this value if you want to wait on several events at once. | ||
| 423 | - Otherwise calling :meth:`join()` is simpler. | ||
| 424 | - | ||
| 425 | 426 | .. versionadded:: 3.3 | |
| 426 | 427 | ||
| 427 | 428 | .. method:: terminate() | |
@@ -785,6 +786,9 @@ Connection objects are usually created using :func:`Pipe` -- see also | |||
| 785 | 786 | *timeout* is a number then this specifies the maximum time in seconds to | |
| 786 | 787 | block. If *timeout* is ``None`` then an infinite timeout is used. | |
| 787 | 788 | ||
| 789 | + Note that multiple connection objects may be polled at once by | ||
| 790 | + using :func:`multiprocessing.connection.wait`. | ||
| 791 | + | ||
| 788 | 792 | .. method:: send_bytes(buffer[, offset[, size]]) | |
| 789 | 793 | ||
| 790 | 794 | Send byte data from an object supporting the buffer interface as a | |
@@ -1779,8 +1783,9 @@ Usually message passing between processes is done using queues or by using | |||
| 1779 | 1783 | ||
| 1780 | 1784 | However, the :mod:`multiprocessing.connection` module allows some extra | |
| 1781 | 1785 | flexibility. It basically gives a high level message oriented API for dealing | |
| 1782 | - with sockets or Windows named pipes, and also has support for *digest | ||
| 1783 | - authentication* using the :mod:`hmac` module. | ||
| 1786 | + with sockets or Windows named pipes. It also has support for *digest | ||
| 1787 | + authentication* using the :mod:`hmac` module, and for polling | ||
| 1788 | + multiple connections at the same time. | ||
| 1784 | 1789 | ||
| 1785 | 1790 | ||
| 1786 | 1791 | .. function:: deliver_challenge(connection, authkey) | |
@@ -1878,6 +1883,38 @@ authentication* using the :mod:`hmac` module. | |||
| 1878 | 1883 | The address from which the last accepted connection came. If this is | |
| 1879 | 1884 | unavailable then it is ``None``. | |
| 1880 | 1885 | ||
| 1886 | + .. function:: wait(object_list, timeout=None) | ||
| 1887 | + | ||
| 1888 | + Wait till an object in *object_list* is ready. Returns the list of | ||
| 1889 | + those objects in *object_list* which are ready. If *timeout* is a | ||
| 1890 | + float then the call blocks for at most that many seconds. If | ||
| 1891 | + *timeout* is ``None`` then it will block for an unlimited period. | ||
| 1892 | + | ||
| 1893 | + For both Unix and Windows, an object can appear in *object_list* if | ||
| 1894 | + it is | ||
| 1895 | + | ||
| 1896 | + * a readable :class:`~multiprocessing.Connection` object; | ||
| 1897 | + * a connected and readable :class:`socket.socket` object; or | ||
| 1898 | + * the :attr:`~multiprocessing.Process.sentinel` attribute of a | ||
| 1899 | + :class:`~multiprocessing.Process` object. | ||
| 1900 | + | ||
| 1901 | + A connection or socket object is ready when there is data available | ||
| 1902 | + to be read from it, or the other end has been closed. | ||
| 1903 | + | ||
| 1904 | + **Unix**: ``wait(object_list, timeout)`` almost equivalent | ||
| 1905 | + ``select.select(object_list, [], [], timeout)``. The difference is | ||
| 1906 | + that, if :func:`select.select` is interrupted by a signal, it can | ||
| 1907 | + raise :exc:`OSError` with an error number of ``EINTR``, whereas | ||
| 1908 | + :func:`wait` will not. | ||
| 1909 | + | ||
| 1910 | + **Windows**: An item in *object_list* must either be an integer | ||
| 1911 | + handle which is waitable (according to the definition used by the | ||
| 1912 | + documentation of the Win32 function ``WaitForMultipleObjects()``) | ||
| 1913 | + or it can be an object with a :meth:`fileno` method which returns a | ||
| 1914 | + socket handle or pipe handle. (Note that pipe handles and socket | ||
| 1915 | + handles are **not** waitable handles.) | ||
| 1916 | + | ||
| 1917 | + .. versionadded:: 3.3 | ||
| 1881 | 1918 | ||
| 1882 | 1919 | The module defines two exceptions: | |
| 1883 | 1920 | ||
@@ -1929,6 +1966,41 @@ server:: | |||
| 1929 | 1966 | ||
| 1930 | 1967 | conn.close() | |
| 1931 | 1968 | ||
| 1969 | + The following code uses :func:`~multiprocessing.connection.wait` to | ||
| 1970 | + wait for messages from multiple processes at once:: | ||
| 1971 | + | ||
| 1972 | + import time, random | ||
| 1973 | + from multiprocessing import Process, Pipe, current_process | ||
| 1974 | + from multiprocessing.connection import wait | ||
| 1975 | + | ||
| 1976 | + def foo(w): | ||
| 1977 | + for i in range(10): | ||
| 1978 | + w.send((i, current_process().name)) | ||
| 1979 | + w.close() | ||
| 1980 | + | ||
| 1981 | + if __name__ == '__main__': | ||
| 1982 | + readers = [] | ||
| 1983 | + | ||
| 1984 | + for i in range(4): | ||
| 1985 | + r, w = Pipe(duplex=False) | ||
| 1986 | + readers.append(r) | ||
| 1987 | + p = Process(target=foo, args=(w,)) | ||
| 1988 | + p.start() | ||
| 1989 | + # We close the writable end of the pipe now to be sure that | ||
| 1990 | + # p is the only process which owns a handle for it. This | ||
| 1991 | + # ensures that when p closes its handle for the writable end, | ||
| 1992 | + # wait() will promptly report the readable end as being ready. | ||
| 1993 | + w.close() | ||
| 1994 | + | ||
| 1995 | + while readers: | ||
| 1996 | + for r in wait(readers): | ||
| 1997 | + try: | ||
| 1998 | + msg = r.recv() | ||
| 1999 | + except EOFError: | ||
| 2000 | + readers.remove(r) | ||
| 2001 | + else: | ||
| 2002 | + print(msg) | ||
| 2003 | + | ||
| 1932 | 2004 | ||
| 1933 | 2005 | .. _multiprocessing-address-formats: | |
| 1934 | 2006 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -50,7 +50,8 @@ | |||
| 50 | 50 | from concurrent.futures import _base | |
| 51 | 51 | import queue | |
| 52 | 52 | import multiprocessing | |
| 53 | - from multiprocessing.queues import SimpleQueue, SentinelReady, Full | ||
| 53 | + from multiprocessing.queues import SimpleQueue, Full | ||
| 54 | + from multiprocessing.connection import wait | ||
| 54 | 55 | import threading | |
| 55 | 56 | import weakref | |
| 56 | 57 | ||
@@ -212,16 +213,19 @@ def shutdown_worker(): | |||
| 212 | 213 | for p in processes.values(): | |
| 213 | 214 | p.join() | |
| 214 | 215 | ||
| 216 | + reader = result_queue._reader | ||
| 217 | + | ||
| 215 | 218 | while True: | |
| 216 | 219 | _add_call_item_to_queue(pending_work_items, | |
| 217 | 220 | work_ids_queue, | |
| 218 | 221 | call_queue) | |
| 219 | 222 | ||
| 220 | 223 | sentinels = [p.sentinel for p in processes.values()] | |
| 221 | 224 | assert sentinels | |
| 222 | - try: | ||
| 223 | - result_item = result_queue.get(sentinels=sentinels) | ||
| 224 | - except SentinelReady: | ||
| 225 | + ready = wait([reader] + sentinels) | ||
| 226 | + if reader in ready: | ||
| 227 | + result_item = reader.recv() | ||
| 228 | + else: | ||
| 225 | 229 | # Mark the process pool broken so that submits fail right now. | |
| 226 | 230 | executor = executor_reference() | |
| 227 | 231 | if executor is not None: | |
| Back | FazBrowse Home | New Git URL |
0 commit comments