Issue #12328: Fix multiprocessing's use of overlapped I/O on Windows.

Also, add a multiprocessing.connection.wait(rlist, timeout=None) function
for polling multiple objects at once.  Patch by sbt.

Complete changelist from sbt's patch:

* Adds a wait(rlist, timeout=None) function for polling multiple
  objects at once.  On Unix this is just a wrapper for
  select(rlist, [], [], timeout=None).

* Removes use of the SentinelReady exception and the sentinels argument
  to certain methods.  concurrent.futures.process has been changed to
  use wait() instead of SentinelReady.

* Fixes bugs concerning PipeConnection.poll() and messages of zero
  length.

* Fixes PipeListener.accept() to call ConnectNamedPipe() with
  overlapped=True.

* Fixes Queue.empty() and SimpleQueue.empty() so that they are
  threadsafe on Windows.

* Now PipeConnection.poll() and wait() will not modify the pipe except
  possibly by consuming a zero length message.  (Previously poll()
  could consume a partial message.)

* All of multiprocesing's pipe related blocking functions/methods are
  now interruptible by SIGINT on Windows.

Doc/library/multiprocessing.rst

@@ -415,13 +415,14 @@ The :mod:`multiprocessing` package mostly replicates the API of the
       A numeric handle of a system object which will become "ready" when
       the process ends.
 
+      You can use this value if you want to wait on several events at
+      once using :func:`multiprocessing.connection.wait`.  Otherwise
+      calling :meth:`join()` is simpler.
+
       On Windows, this is an OS handle usable with the ``WaitForSingleObject``
       and ``WaitForMultipleObjects`` family of API calls.  On Unix, this is
       a file descriptor usable with primitives from the :mod:`select` module.
 
-      You can use this value if you want to wait on several events at once.
-      Otherwise calling :meth:`join()` is simpler.
-
       .. versionadded:: 3.3
 
    .. method:: terminate()
@@ -785,6 +786,9 @@ Connection objects are usually created using :func:`Pipe` -- see also
       *timeout* is a number then this specifies the maximum time in seconds to
       block.  If *timeout* is ``None`` then an infinite timeout is used.
 
+      Note that multiple connection objects may be polled at once by
+      using :func:`multiprocessing.connection.wait`.
+
    .. method:: send_bytes(buffer[, offset[, size]])
 
       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
 
 However, the :mod:`multiprocessing.connection` module allows some extra
 flexibility.  It basically gives a high level message oriented API for dealing
-with sockets or Windows named pipes, and also has support for *digest
-authentication* using the :mod:`hmac` module.
+with sockets or Windows named pipes.  It also has support for *digest
+authentication* using the :mod:`hmac` module, and for polling
+multiple connections at the same time.
 
 
 .. function:: deliver_challenge(connection, authkey)
@@ -1878,6 +1883,38 @@ authentication* using the :mod:`hmac` module.
       The address from which the last accepted connection came.  If this is
       unavailable then it is ``None``.
 
+.. function:: 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.
+
+   For both Unix and Windows, an object can appear in *object_list* if
+   it is
+
+   * a readable :class:`~multiprocessing.Connection` object;
+   * a connected and readable :class:`socket.socket` object; or
+   * the :attr:`~multiprocessing.Process.sentinel` attribute of a
+     :class:`~multiprocessing.Process` object.
+
+   A connection or socket object is ready when there is data available
+   to be read from it, or the other end has been closed.
+
+   **Unix**: ``wait(object_list, timeout)`` almost equivalent
+   ``select.select(object_list, [], [], timeout)``.  The difference is
+   that, if :func:`select.select` is interrupted by a signal, it can
+   raise :exc:`OSError` with an error number of ``EINTR``, whereas
+   :func:`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 :meth:`fileno` method which returns a
+   socket handle or pipe handle.  (Note that pipe handles and socket
+   handles are **not** waitable handles.)
+
+   .. versionadded:: 3.3
 
 The module defines two exceptions:
 
@@ -1929,6 +1966,41 @@ server::
 
    conn.close()
 
+The following code uses :func:`~multiprocessing.connection.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)
+
 
 .. _multiprocessing-address-formats:
 

Lib/concurrent/futures/process.py

@@ -50,7 +50,8 @@ import os
 from concurrent.futures import _base
 import queue
 import multiprocessing
-from multiprocessing.queues import SimpleQueue, SentinelReady, Full
+from multiprocessing.queues import SimpleQueue, Full
+from multiprocessing.connection import wait
 import threading
 import weakref
 
@@ -212,6 +213,8 @@ def _queue_management_worker(executor_reference,
         for p in processes.values():
             p.join()
 
+    reader = result_queue._reader
+
     while True:
         _add_call_item_to_queue(pending_work_items,
                                 work_ids_queue,
@@ -219,9 +222,10 @@ def _queue_management_worker(executor_reference,
 
         sentinels = [p.sentinel for p in processes.values()]
         assert sentinels
-        try:
-            result_item = result_queue.get(sentinels=sentinels)
-        except SentinelReady:
+        ready = wait([reader] + sentinels)
+        if reader in ready:
+            result_item = reader.recv()
+        else:
             # Mark the process pool broken so that submits fail right now.
             executor = executor_reference()
             if executor is not None:

Lib/multiprocessing/queues.py

@@ -44,7 +44,7 @@ import errno
 
 from queue import Empty, Full
 import _multiprocessing
-from multiprocessing.connection import Pipe, SentinelReady
+from multiprocessing.connection import Pipe
 from multiprocessing.synchronize import Lock, BoundedSemaphore, Semaphore, Condition
 from multiprocessing.util import debug, info, Finalize, register_after_fork
 from multiprocessing.forking import assert_spawning
@@ -360,6 +360,7 @@ class SimpleQueue(object):
     def __init__(self):
         self._reader, self._writer = Pipe(duplex=False)
         self._rlock = Lock()
+        self._poll = self._reader.poll
         if sys.platform == 'win32':
             self._wlock = None
         else:
@@ -367,7 +368,7 @@ class SimpleQueue(object):
         self._make_methods()
 
     def empty(self):
-        return not self._reader.poll()
+        return not self._poll()
 
     def __getstate__(self):
         assert_spawning(self)
@@ -380,10 +381,10 @@ class SimpleQueue(object):
     def _make_methods(self):
         recv = self._reader.recv
         racquire, rrelease = self._rlock.acquire, self._rlock.release
-        def get(*, sentinels=None):
+        def get():
             racquire()
             try:
-                return recv(sentinels)
+                return recv()
             finally:
                 rrelease()
         self.get = get

Lib/test/test_multiprocessing.py

@@ -1811,6 +1811,84 @@ class _TestListenerClient(BaseTestCase):
             p.join()
             l.close()
 
+class _TestPoll(unittest.TestCase):
+
+    ALLOWED_TYPES = ('processes', 'threads')
+
+    def test_empty_string(self):
+        a, b = self.Pipe()
+        self.assertEqual(a.poll(), False)
+        b.send_bytes(b'')
+        self.assertEqual(a.poll(), True)
+        self.assertEqual(a.poll(), True)
+
+    @classmethod
+    def _child_strings(cls, conn, strings):
+        for s in strings:
+            time.sleep(0.1)
+            conn.send_bytes(s)
+        conn.close()
+
+    def test_strings(self):
+        strings = (b'hello', b'', b'a', b'b', b'', b'bye', b'', b'lop')
+        a, b = self.Pipe()
+        p = self.Process(target=self._child_strings, args=(b, strings))
+        p.start()
+
+        for s in strings:
+            for i in range(200):
+                if a.poll(0.01):
+                    break
+            x = a.recv_bytes()
+            self.assertEqual(s, x)
+
+        p.join()
+
+    @classmethod
+    def _child_boundaries(cls, r):
+        # Polling may "pull" a message in to the child process, but we
+        # don't want it to pull only part of a message, as that would
+        # corrupt the pipe for any other processes which might later
+        # read from it.
+        r.poll(5)
+
+    def test_boundaries(self):
+        r, w = self.Pipe(False)
+        p = self.Process(target=self._child_boundaries, args=(r,))
+        p.start()
+        time.sleep(2)
+        L = [b"first", b"second"]
+        for obj in L:
+            w.send_bytes(obj)
+        w.close()
+        p.join()
+        self.assertIn(r.recv_bytes(), L)
+
+    @classmethod
+    def _child_dont_merge(cls, b):
+        b.send_bytes(b'a')
+        b.send_bytes(b'b')
+        b.send_bytes(b'cd')
+
+    def test_dont_merge(self):
+        a, b = self.Pipe()
+        self.assertEqual(a.poll(0.0), False)
+        self.assertEqual(a.poll(0.1), False)
+
+        p = self.Process(target=self._child_dont_merge, args=(b,))
+        p.start()
+
+        self.assertEqual(a.recv_bytes(), b'a')
+        self.assertEqual(a.poll(1.0), True)
+        self.assertEqual(a.poll(1.0), True)
+        self.assertEqual(a.recv_bytes(), b'b')
+        self.assertEqual(a.poll(1.0), True)
+        self.assertEqual(a.poll(1.0), True)
+        self.assertEqual(a.poll(0.0), True)
+        self.assertEqual(a.recv_bytes(), b'cd')
+
+        p.join()
+
 #
 # Test of sending connection and socket objects between processes
 #
@@ -2404,8 +2482,163 @@ class TestStdinBadfiledescriptor(unittest.TestCase):
         flike.flush()
         assert sio.getvalue() == 'foo'
 
+
+class TestWait(unittest.TestCase):
+
+    @classmethod
+    def _child_test_wait(cls, w, slow):
+        for i in range(10):
+            if slow:
+                time.sleep(random.random()*0.1)
+            w.send((i, os.getpid()))
+        w.close()
+
+    def test_wait(self, slow=False):
+        from multiprocessing import Pipe, Process
+        from multiprocessing.connection import wait
+        readers = []
+        procs = []
+        messages = []
+
+        for i in range(4):
+            r, w = Pipe(duplex=False)
+            p = Process(target=self._child_test_wait, args=(w, slow))
+            p.daemon = True
+            p.start()
+            w.close()
+            readers.append(r)
+            procs.append(p)
+
+        while readers:
+            for r in wait(readers):
+                try:
+                    msg = r.recv()
+                except EOFError:
+                    readers.remove(r)
+                    r.close()
+                else:
+                    messages.append(msg)
+
+        messages.sort()
+        expected = sorted((i, p.pid) for i in range(10) for p in procs)
+        self.assertEqual(messages, expected)
+
+    @classmethod
+    def _child_test_wait_socket(cls, address, slow):
+        s = socket.socket()
+        s.connect(address)
+        for i in range(10):
+            if slow:
+                time.sleep(random.random()*0.1)
+            s.sendall(('%s\n' % i).encode('ascii'))
+        s.close()
+
+    def test_wait_socket(self, slow=False):
+        from multiprocessing import Process
+        from multiprocessing.connection import wait
+        l = socket.socket()
+        l.bind(('', 0))
+        l.listen(4)
+        addr = ('localhost', l.getsockname()[1])
+        readers = []
+        procs = []
+        dic = {}
+
+        for i in range(4):
+            p = Process(target=self._child_test_wait_socket, args=(addr, slow))
+            p.daemon = True
+            p.start()
+            procs.append(p)
+
+        for i in range(4):
+            r, _ = l.accept()
+            readers.append(r)
+            dic[r] = []
+        l.close()
+
+        while readers:
+            for r in wait(readers):
+                msg = r.recv(32)
+                if not msg:
+                    readers.remove(r)
+                    r.close()
+                else:
+                    dic[r].append(msg)
+
+        expected = ''.join('%s\n' % i for i in range(10)).encode('ascii')
+        for v in dic.values():
+            self.assertEqual(b''.join(v), expected)
+
+    def test_wait_slow(self):
+        self.test_wait(True)
+
+    def test_wait_socket_slow(self):
+        self.test_wait(True)
+
+    def test_wait_timeout(self):
+        from multiprocessing.connection import wait
+
+        expected = 1
+        a, b = multiprocessing.Pipe()
+
+        start = time.time()
+        res = wait([a, b], 1)
+        delta = time.time() - start
+
+        self.assertEqual(res, [])
+        self.assertLess(delta, expected + 0.2)
+        self.assertGreater(delta, expected - 0.2)
+
+        b.send(None)
+
+        start = time.time()
+        res = wait([a, b], 1)
+        delta = time.time() - start
+
+        self.assertEqual(res, [a])
+        self.assertLess(delta, 0.2)
+
+    def test_wait_integer(self):
+        from multiprocessing.connection import wait
+
+        expected = 5
+        a, b = multiprocessing.Pipe()
+        p = multiprocessing.Process(target=time.sleep, args=(expected,))
+
+        p.start()
+        self.assertIsInstance(p.sentinel, int)
+
+        start = time.time()
+        res = wait([a, p.sentinel, b], expected + 20)
+        delta = time.time() - start
+
+        self.assertEqual(res, [p.sentinel])
+        self.assertLess(delta, expected + 1)
+        self.assertGreater(delta, expected - 1)
+
+        a.send(None)
+
+        start = time.time()
+        res = wait([a, p.sentinel, b], 20)
+        delta = time.time() - start
+
+        self.assertEqual(res, [p.sentinel, b])
+        self.assertLess(delta, 0.2)
+
+        b.send(None)
+
+        start = time.time()
+        res = wait([a, p.sentinel, b], 20)
+        delta = time.time() - start
+
+        self.assertEqual(res, [a, p.sentinel, b])
+        self.assertLess(delta, 0.2)
+
+        p.join()
+
+
 testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
-                   TestStdinBadfiledescriptor]
+                   TestStdinBadfiledescriptor, TestWait]
 
 #
 #

Misc/NEWS

@@ -13,6 +13,10 @@ Core and Builtins
 Library
 -------
 
+- Issue #12328: Fix multiprocessing's use of overlapped I/O on Windows.
+  Also, add a multiprocessing.connection.wait(rlist, timeout=None) function
+  for polling multiple objects at once.  Patch by sbt.
+
 - Issue #13719: Make the distutils and packaging upload commands aware of
   bdist_msi products.
 

Modules/_multiprocessing/win32_functions.c

@@ -60,16 +60,18 @@ typedef struct {
 static void
 overlapped_dealloc(OverlappedObject *self)
 {
+    DWORD bytes;
     int err = GetLastError();
     if (self->pending) {
-        if (check_CancelIoEx())
-            Py_CancelIoEx(self->handle, &self->overlapped);
-        else {
-            PyErr_SetString(PyExc_RuntimeError,
-                            "I/O operations still in flight while destroying "
-                            "Overlapped object, the process may crash");
-            PyErr_WriteUnraisable(NULL);
-        }
+        /* make it a programming error to deallocate while operation
+           is pending, even if we can safely cancel it */
+        if (check_CancelIoEx() &&
+                Py_CancelIoEx(self->handle, &self->overlapped))
+            GetOverlappedResult(self->handle, &self->overlapped, &bytes, TRUE);
+        PyErr_SetString(PyExc_RuntimeError,
+                        "I/O operations still in flight while destroying "
+                        "Overlapped object, the process may crash");
+        PyErr_WriteUnraisable(NULL);
     }
     CloseHandle(self->overlapped.hEvent);
     SetLastError(err);
@@ -85,6 +87,7 @@ overlapped_GetOverlappedResult(OverlappedObject *self, PyObject *waitobj)
     int wait;
     BOOL res;
     DWORD transferred = 0;
+    DWORD err;
 
     wait = PyObject_IsTrue(waitobj);
     if (wait < 0)
@@ -94,23 +97,27 @@ overlapped_GetOverlappedResult(OverlappedObject *self, PyObject *waitobj)
                               wait != 0);
     Py_END_ALLOW_THREADS
 
-    if (!res) {
-        int err = GetLastError();
-        if (err == ERROR_IO_INCOMPLETE)
-            Py_RETURN_NONE;
-        if (err != ERROR_MORE_DATA) {
+    err = res ? ERROR_SUCCESS : GetLastError();
+    switch (err) {
+        case ERROR_SUCCESS:
+        case ERROR_MORE_DATA:
+        case ERROR_OPERATION_ABORTED:
+            self->completed = 1;
+            self->pending = 0;
+            break;
+        case ERROR_IO_INCOMPLETE:
+            break;
+        default:
             self->pending = 0;
             return PyErr_SetExcFromWindowsErr(PyExc_IOError, err);
-        }
     }
-    self->pending = 0;
-    self->completed = 1;
-    if (self->read_buffer) {
+    if (self->completed && self->read_buffer != NULL) {
         assert(PyBytes_CheckExact(self->read_buffer));
-        if (_PyBytes_Resize(&self->read_buffer, transferred))
+        if (transferred != PyBytes_GET_SIZE(self->read_buffer) &&
+            _PyBytes_Resize(&self->read_buffer, transferred))
             return NULL;
     }
-    return Py_BuildValue("lN", (long) transferred, PyBool_FromLong(res));
+    return Py_BuildValue("II", (unsigned) transferred, (unsigned) err);
 }
 
 static PyObject *
@@ -522,9 +529,10 @@ win32_WriteFile(PyObject *self, PyObject *args, PyObject *kwds)
     HANDLE handle;
     Py_buffer _buf, *buf;
     PyObject *bufobj;
-    int written;
+    DWORD written;
     BOOL ret;
     int use_overlapped = 0;
+    DWORD err;
     OverlappedObject *overlapped = NULL;
     static char *kwlist[] = {"handle", "buffer", "overlapped", NULL};
 
@@ -553,8 +561,9 @@ win32_WriteFile(PyObject *self, PyObject *args, PyObject *kwds)
                     overlapped ? &overlapped->overlapped : NULL);
     Py_END_ALLOW_THREADS
 
+    err = ret ? 0 : GetLastError();
+
     if (overlapped) {
-        int err = GetLastError();
         if (!ret) {
             if (err == ERROR_IO_PENDING)
                 overlapped->pending = 1;
@@ -563,13 +572,13 @@ win32_WriteFile(PyObject *self, PyObject *args, PyObject *kwds)
                 return PyErr_SetExcFromWindowsErr(PyExc_IOError, 0);
             }
         }
-        return (PyObject *) overlapped;
+        return Py_BuildValue("NI", (PyObject *) overlapped, err);
     }
 
     PyBuffer_Release(buf);
     if (!ret)
         return PyErr_SetExcFromWindowsErr(PyExc_IOError, 0);
-    return PyLong_FromLong(written);
+    return Py_BuildValue("II", written, err);
 }
 
 static PyObject *
@@ -581,6 +590,7 @@ win32_ReadFile(PyObject *self, PyObject *args, PyObject *kwds)
     PyObject *buf;
     BOOL ret;
     int use_overlapped = 0;
+    DWORD err;
     OverlappedObject *overlapped = NULL;
     static char *kwlist[] = {"handle", "size", "overlapped", NULL};
 
@@ -607,8 +617,9 @@ win32_ReadFile(PyObject *self, PyObject *args, PyObject *kwds)
                    overlapped ? &overlapped->overlapped : NULL);
     Py_END_ALLOW_THREADS
 
+    err = ret ? 0 : GetLastError();
+
     if (overlapped) {
-        int err = GetLastError();
         if (!ret) {
             if (err == ERROR_IO_PENDING)
                 overlapped->pending = 1;
@@ -617,16 +628,16 @@ win32_ReadFile(PyObject *self, PyObject *args, PyObject *kwds)
                 return PyErr_SetExcFromWindowsErr(PyExc_IOError, 0);
             }
         }
-        return (PyObject *) overlapped;
+        return Py_BuildValue("NI", (PyObject *) overlapped, err);
     }
 
-    if (!ret && GetLastError() != ERROR_MORE_DATA) {
+    if (!ret && err != ERROR_MORE_DATA) {
         Py_DECREF(buf);
         return PyErr_SetExcFromWindowsErr(PyExc_IOError, 0);
     }
     if (_PyBytes_Resize(&buf, nread))
         return NULL;
-    return Py_BuildValue("NN", buf, PyBool_FromLong(ret));
+    return Py_BuildValue("NI", buf, err);
 }
 
 static PyObject *
@@ -783,7 +794,11 @@ create_win32_namespace(void)
 
     WIN32_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS);
     WIN32_CONSTANT(F_DWORD, ERROR_BROKEN_PIPE);
+    WIN32_CONSTANT(F_DWORD, ERROR_IO_PENDING);
+    WIN32_CONSTANT(F_DWORD, ERROR_MORE_DATA);
+    WIN32_CONSTANT(F_DWORD, ERROR_NETNAME_DELETED);
     WIN32_CONSTANT(F_DWORD, ERROR_NO_SYSTEM_RESOURCES);
+    WIN32_CONSTANT(F_DWORD, ERROR_OPERATION_ABORTED);
     WIN32_CONSTANT(F_DWORD, ERROR_PIPE_BUSY);
     WIN32_CONSTANT(F_DWORD, ERROR_PIPE_CONNECTED);
     WIN32_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT);