Class PipeConnection

Lib/multiprocessing/connection.py:268–361

Connection class based on a Windows named pipe. Overlapped I/O is used, so the handles must have been created with FILE_FLAG_OVERLAPPED.

if _winapi:

    class PipeConnection(_ConnectionBase):
        """
        Connection class based on a Windows named pipe.
        Overlapped I/O is used, so the handles must have been created
        with FILE_FLAG_OVERLAPPED.
        """
        _got_empty_message = False
        _send_ov = None

        def _close(self, _CloseHandle=_winapi.CloseHandle):
            ov = self._send_ov
            if ov is not None:
                # Interrupt WaitForMultipleObjects() in _send_bytes()
                ov.cancel()
            _CloseHandle(self._handle)

        def _send_bytes(self, buf):
            if self._send_ov is not None:
                # A connection should only be used by a single thread
                raise ValueError("concurrent send_bytes() calls "
                                 "are not supported")
            ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
            self._send_ov = ov
            try:
                if err == _winapi.ERROR_IO_PENDING:
                    waitres = _winapi.WaitForMultipleObjects(
                        [ov.event], False, INFINITE)
                    assert waitres == WAIT_OBJECT_0
            except:
                ov.cancel()
                raise
            finally:
                self._send_ov = None
                nwritten, err = ov.GetOverlappedResult(True)
            if err == _winapi.ERROR_OPERATION_ABORTED:
                # close() was called by another thread while
                # WaitForMultipleObjects() was waiting for the overlapped
                # operation.
                raise OSError(errno.EPIPE, "handle is closed")
            assert err == 0
            assert nwritten == len(buf)

        def _recv_bytes(self, maxsize=None):
            if self._got_empty_message:
                self._got_empty_message = False
                return io.BytesIO()
            else:
                bsize = 128 if maxsize is None else min(maxsize, 128)
                try:
                    ov, err = _winapi.ReadFile(self._handle, bsize,
                                                overlapped=True)
                    try:
                        if err == _winapi.ERROR_IO_PENDING:
                            waitres = _winapi.WaitForMultipleObjects(
                                [ov.event], False, INFINITE)
                            assert waitres == WAIT_OBJECT_0
                    except:
                        ov.cancel()