@@ -87,6 +87,17 @@ def _path_importer_cache(cls, path):
8787import weakref
8888import zlib
8989
90+ if sys .version_info > (3 ,5 ):
91+ from os import get_blocking , set_blocking
92+ else :
93+ def get_blocking (fd ):
94+ return not fcntl .fcntl (fd , fcntl .F_GETFL ) & os .O_NONBLOCK
95+
96+ def set_blocking (fd , blocking ):
97+ fl = fcntl .fcntl (fd , fcntl .F_GETFL )
98+ if blocking : fcntl .fcntl (fd , fcntl .F_SETFL , fl & ~ os .O_NONBLOCK )
99+ else : fcntl .fcntl (fd , fcntl .F_SETFL , fl | os .O_NONBLOCK )
100+
90101try :
91102 # Python >= 3.4, PEP 451 ModuleSpec API
92103 import importlib .machinery
@@ -559,26 +570,6 @@ def set_cloexec(fd):
559570 fcntl .fcntl (fd , fcntl .F_SETFD , flags | fcntl .FD_CLOEXEC )
560571
561572
562- def set_nonblock (fd ):
563- """
564- Set the file descriptor `fd` to non-blocking mode. For most underlying file
565- types, this causes :func:`os.read` or :func:`os.write` to raise
566- :class:`OSError` with :data:`errno.EAGAIN` rather than block the thread
567- when the underlying kernel buffer is exhausted.
568- """
569- flags = fcntl .fcntl (fd , fcntl .F_GETFL )
570- fcntl .fcntl (fd , fcntl .F_SETFL , flags | os .O_NONBLOCK )
571-
572-
573- def set_block (fd ):
574- """
575- Inverse of :func:`set_nonblock`, i.e. cause `fd` to block the thread when
576- the underlying kernel buffer is exhausted.
577- """
578- flags = fcntl .fcntl (fd , fcntl .F_GETFL )
579- fcntl .fcntl (fd , fcntl .F_SETFL , flags & ~ os .O_NONBLOCK )
580-
581-
582573def io_op (func , * args ):
583574 """
584575 Wrap `func(*args)` that may raise :class:`select.error`, :class:`IOError`,
@@ -720,20 +711,30 @@ def import_module(modname):
720711 return __import__ (modname , None , None , ['' ])
721712
722713
723- def pipe ():
714+ def pipe (blocking = None ):
724715 """
725716 Create a UNIX pipe pair using :func:`os.pipe`, wrapping the returned
726717 descriptors in Python file objects in order to manage their lifetime and
727718 ensure they are closed when their last reference is discarded and they have
728719 not been closed explicitly.
729720 """
730721 rfd , wfd = os .pipe ()
722+ for fd in rfd , wfd :
723+ if blocking is not None : set_blocking (fd , blocking ) # noqa: E701
731724 return (
732725 os .fdopen (rfd , 'rb' , 0 ),
733726 os .fdopen (wfd , 'wb' , 0 )
734727 )
735728
736729
730+ def socketpair (blocking = None ):
731+ fp1 , fp2 = socket .socketpair ()
732+ for fp in fp1 , fp2 :
733+ fd = fp .fileno ()
734+ if blocking is not None : set_blocking (fd , blocking ) # noqa: E701
735+ return fp1 , fp2
736+
737+
737738def iter_split (buf , delim , func ):
738739 """
739740 Invoke `func(s)` for each `delim`-delimited chunk in the potentially large
@@ -1879,8 +1880,7 @@ def accept(self, rfp, wfp):
18791880 """
18801881 Attach a pair of file objects to :attr:`receive_side` and
18811882 :attr:`transmit_side`, after wrapping them in :class:`Side` instances.
1882- :class:`Side` will call :func:`set_nonblock` and :func:`set_cloexec`
1883- on the underlying file descriptors during construction.
1883+ :class:`Side` will call :func:`set_cloexec` on them.
18841884
18851885 The same file object may be used for both sides. The default
18861886 :meth:`on_disconnect` is handles the possibility that only one
@@ -2155,14 +2155,11 @@ class Side(object):
21552155 :param bool keep_alive:
21562156 If :data:`True`, the continued existence of this side will extend the
21572157 shutdown grace period until it has been unregistered from the broker.
2158- :param bool blocking:
2159- If :data:`False`, the descriptor has its :data:`os.O_NONBLOCK` flag
2160- enabled using :func:`fcntl.fcntl`.
21612158 """
21622159 _fork_refs = weakref .WeakValueDictionary ()
21632160 closed = False
21642161
2165- def __init__ (self , stream , fp , cloexec = True , keep_alive = True , blocking = False ):
2162+ def __init__ (self , stream , fp , cloexec = True , keep_alive = True ):
21662163 #: The :class:`Stream` for which this is a read or write side.
21672164 self .stream = stream
21682165 # File or socket object responsible for the lifetime of its underlying
@@ -2180,8 +2177,6 @@ def __init__(self, stream, fp, cloexec=True, keep_alive=True, blocking=False):
21802177 self ._fork_refs [id (self )] = self
21812178 if cloexec :
21822179 set_cloexec (self .fd )
2183- if not blocking :
2184- set_nonblock (self .fd )
21852180
21862181 def __repr__ (self ):
21872182 return '<Side of %s fd %s>' % (
@@ -2785,7 +2780,7 @@ def _get_socketpair(self):
27852780 try :
27862781 return self ._cls_idle_socketpairs .pop () # pop() must be atomic
27872782 except IndexError :
2788- rsock , wsock = socket . socketpair ()
2783+ rsock , wsock = socketpair ()
27892784 rsock .setblocking (False )
27902785 set_cloexec (rsock .fileno ())
27912786 set_cloexec (wsock .fileno ())
@@ -2958,7 +2953,8 @@ class Waker(Protocol):
29582953 @classmethod
29592954 def build_stream (cls , broker ):
29602955 stream = super (Waker , cls ).build_stream (broker )
2961- stream .accept (* pipe ())
2956+ rfp , wfp = pipe (blocking = False )
2957+ stream .accept (rfp , wfp )
29622958 return stream
29632959
29642960 def __init__ (self , broker ):
@@ -3056,7 +3052,8 @@ def build_stream(cls, name, dest_fd):
30563052 prevent break :meth:`on_shutdown` from calling :meth:`shutdown()
30573053 <socket.socket.shutdown>` on it.
30583054 """
3059- rsock , wsock = socket .socketpair ()
3055+ # Leave wsock & dest_fd blocking, so the subprocess will have sane stdio
3056+ rsock , wsock = socketpair ()
30603057 os .dup2 (wsock .fileno (), dest_fd )
30613058 stream = super (IoLoggerProtocol , cls ).build_stream (name )
30623059 stream .name = name
@@ -4038,6 +4035,9 @@ def _setup_master(self):
40384035 local_id = self .config ['context_id' ],
40394036 parent_ids = self .config ['parent_ids' ]
40404037 )
4038+ for f in in_fp , out_fp :
4039+ fd = f .fileno ()
4040+ set_blocking (fd , False )
40414041 self .stream .accept (in_fp , out_fp )
40424042 self .stream .name = 'parent'
40434043 self .stream .receive_side .keep_alive = False
0 commit comments