Merge branch 'master' of https://github.com/python/cpython into asyncio_mock

lisroach · lisroach · commit 45dddb708b25 · 2019-05-07T14:08:36.000-04:00
diff --git a/Doc/library/http.cookies.rst b/Doc/library/http.cookies.rst
@@ -55,8 +55,9 @@ in Cookie name (as :attr:`~Morsel.key`).
 .. class:: SimpleCookie([input])
 
    This class derives from :class:`BaseCookie` and overrides :meth:`value_decode`
-   and :meth:`value_encode` to be the identity and :func:`str` respectively.
-
+   and :meth:`value_encode`. SimpleCookie supports strings as cookie values.
+   When setting the value, SimpleCookie calls the builtin :func:`str()` to convert
+   the value to a string. Values received from HTTP are kept as strings.
 
 .. seealso::
 
@@ -76,15 +77,16 @@ Cookie Objects
 
 .. method:: BaseCookie.value_decode(val)
 
-   Return a decoded value from a string representation. Return value can be any
-   type. This method does nothing in :class:`BaseCookie` --- it exists so it can be
-   overridden.
+   Return a tuple ``(real_value, coded_value)`` from a string representation.
+   ``real_value`` can be any type. This method does no decoding in
+   :class:`BaseCookie` --- it exists so it can be overridden.
 
 
 .. method:: BaseCookie.value_encode(val)
 
-   Return an encoded value. *val* can be any type, but return value must be a
-   string. This method does nothing in :class:`BaseCookie` --- it exists so it can
+   Return a tuple ``(real_value, coded_value)``. *val* can be any type, but
+   ``coded_value`` will always be converted to a string.
+   This method does no encoding in :class:`BaseCookie` --- it exists so it can
    be overridden.
 
    In general, it should be the case that :meth:`value_encode` and
diff --git a/Doc/library/signal.rst b/Doc/library/signal.rst
@@ -16,7 +16,8 @@ The :func:`signal.signal` function allows defining custom handlers to be
 executed when a signal is received.  A small number of default handlers are
 installed: :const:`SIGPIPE` is ignored (so write errors on pipes and sockets
 can be reported as ordinary Python exceptions) and :const:`SIGINT` is
-translated into a :exc:`KeyboardInterrupt` exception.
+translated into a :exc:`KeyboardInterrupt` exception if the parent process
+has not changed it.
 
 A handler for a particular signal, once set, remains installed until it is
 explicitly reset (Python emulates the BSD style interface regardless of the
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
@@ -1306,7 +1306,8 @@ async def create_datagram_endpoint(self, protocol_factory,
                     if local_addr:
                         sock.bind(local_address)
                     if remote_addr:
-                        await self.sock_connect(sock, remote_address)
+                        if not allow_broadcast:
+                            await self.sock_connect(sock, remote_address)
                         r_addr = remote_address
                 except OSError as exc:
                     if sock is not None:
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
@@ -587,7 +587,10 @@ class _SelectorTransport(transports._FlowControlMixin,
     def __init__(self, loop, sock, protocol, extra=None, server=None):
         super().__init__(extra, loop)
         self._extra['socket'] = sock
-        self._extra['sockname'] = sock.getsockname()
+        try:
+            self._extra['sockname'] = sock.getsockname()
+        except OSError:
+            self._extra['sockname'] = None
         if 'peername' not in self._extra:
             try:
                 self._extra['peername'] = sock.getpeername()
@@ -976,9 +979,11 @@ def sendto(self, data, addr=None):
         if not data:
             return
 
-        if self._address and addr not in (None, self._address):
-            raise ValueError(
-                f'Invalid address: must be None or {self._address}')
+        if self._address:
+            if addr not in (None, self._address):
+                raise ValueError(
+                    f'Invalid address: must be None or {self._address}')
+            addr = self._address
 
         if self._conn_lost and self._address:
             if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
@@ -989,7 +994,7 @@ def sendto(self, data, addr=None):
         if not self._buffer:
             # Attempt to send it right away first.
             try:
-                if self._address:
+                if self._extra['peername']:
                     self._sock.send(data)
                 else:
                     self._sock.sendto(data, addr)
@@ -1012,7 +1017,7 @@ def _sendto_ready(self):
         while self._buffer:
             data, addr = self._buffer.popleft()
             try:
-                if self._address:
+                if self._extra['peername']:
                     self._sock.send(data)
                 else:
                     self._sock.sendto(data, addr)
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
@@ -51,7 +51,7 @@
 import queue
 from queue import Full
 import multiprocessing as mp
-from multiprocessing.connection import wait
+import multiprocessing.connection
 from multiprocessing.queues import Queue
 import threading
 import weakref
@@ -352,7 +352,7 @@ def shutdown_worker():
         # submitted, from the executor being shutdown/gc-ed, or from the
         # shutdown of the python interpreter.
         worker_sentinels = [p.sentinel for p in processes.values()]
-        ready = wait(readers + worker_sentinels)
+        ready = mp.connection.wait(readers + worker_sentinels)
 
         cause = None
         is_broken = True
diff --git a/Lib/logging/__init__.py b/Lib/logging/__init__.py
@@ -231,49 +231,38 @@ def _releaseLock():
 # Prevent a held logging lock from blocking a child from logging.
 
 if not hasattr(os, 'register_at_fork'):  # Windows and friends.
-    def _register_at_fork_acquire_release(instance):
+    def _register_at_fork_reinit_lock(instance):
         pass  # no-op when os.register_at_fork does not exist.
-else:  # The os.register_at_fork API exists
-    os.register_at_fork(before=_acquireLock,
-                        after_in_child=_releaseLock,
-                        after_in_parent=_releaseLock)
-
-    # A collection of instances with acquire and release methods (logging.Handler)
-    # to be called before and after fork.  The weakref avoids us keeping discarded
-    # Handler instances alive forever in case an odd program creates and destroys
-    # many over its lifetime.
-    _at_fork_acquire_release_weakset = weakref.WeakSet()
-
-
-    def _register_at_fork_acquire_release(instance):
-        # We put the instance itself in a single WeakSet as we MUST have only
-        # one atomic weak ref. used by both before and after atfork calls to
-        # guarantee matched pairs of acquire and release calls.
-        _at_fork_acquire_release_weakset.add(instance)
-
+else:
+    # A collection of instances with a createLock method (logging.Handler)
+    # to be called in the child after forking.  The weakref avoids us keeping
+    # discarded Handler instances alive.  A set is used to avoid accumulating
+    # duplicate registrations as createLock() is responsible for registering
+    # a new Handler instance with this set in the first place.
+    _at_fork_reinit_lock_weakset = weakref.WeakSet()
+
+    def _register_at_fork_reinit_lock(instance):
+        _acquireLock()
+        try:
+            _at_fork_reinit_lock_weakset.add(instance)
+        finally:
+            _releaseLock()
 
-    def _at_fork_weak_calls(method_name):
-        for instance in _at_fork_acquire_release_weakset:
-            method = getattr(instance, method_name)
+    def _after_at_fork_child_reinit_locks():
+        # _acquireLock() was called in the parent before forking.
+        for handler in _at_fork_reinit_lock_weakset:
             try:
-                method()
+                handler.createLock()
             except Exception as err:
                 # Similar to what PyErr_WriteUnraisable does.
                 print("Ignoring exception from logging atfork", instance,
-                      method_name, "method:", err, file=sys.stderr)
-
-
-    def _before_at_fork_weak_calls():
-        _at_fork_weak_calls('acquire')
+                      "._reinit_lock() method:", err, file=sys.stderr)
+        _releaseLock()  # Acquired by os.register_at_fork(before=.
 
 
-    def _after_at_fork_weak_calls():
-        _at_fork_weak_calls('release')
-
-
-    os.register_at_fork(before=_before_at_fork_weak_calls,
-                        after_in_child=_after_at_fork_weak_calls,
-                        after_in_parent=_after_at_fork_weak_calls)
+    os.register_at_fork(before=_acquireLock,
+                        after_in_child=_after_at_fork_child_reinit_locks,
+                        after_in_parent=_releaseLock)
 
 
 #---------------------------------------------------------------------------
@@ -900,7 +889,7 @@ def createLock(self):
         Acquire a thread lock for serializing access to the underlying I/O.
         """
         self.lock = threading.RLock()
-        _register_at_fork_acquire_release(self)
+        _register_at_fork_reinit_lock(self)
 
     def acquire(self):
         """
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
@@ -1586,6 +1586,23 @@ def test_create_datagram_endpoint_connect_err(self):
         self.assertRaises(
             OSError, self.loop.run_until_complete, coro)
 
+    def test_create_datagram_endpoint_allow_broadcast(self):
+        protocol = MyDatagramProto(create_future=True, loop=self.loop)
+        self.loop.sock_connect = sock_connect = mock.Mock()
+        sock_connect.return_value = []
+
+        coro = self.loop.create_datagram_endpoint(
+            lambda: protocol,
+            remote_addr=('127.0.0.1', 0),
+            allow_broadcast=True)
+
+        transport, _ = self.loop.run_until_complete(coro)
+        self.assertFalse(sock_connect.called)
+
+        transport.close()
+        self.loop.run_until_complete(protocol.done)
+        self.assertEqual('CLOSED', protocol.state)
+
     @patch_socket
     def test_create_datagram_endpoint_socket_err(self, m_socket):
         m_socket.getaddrinfo = socket.getaddrinfo
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py
@@ -1065,6 +1065,7 @@ def setUp(self):
         self.sock.fileno.return_value = 7
 
     def datagram_transport(self, address=None):
+        self.sock.getpeername.side_effect = None if address else OSError
         transport = _SelectorDatagramTransport(self.loop, self.sock,
                                                self.protocol,
                                                address=address)
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
@@ -668,10 +668,28 @@ def remove_loop(fname, tries):
     # register_at_fork mechanism is also present and used.
     @unittest.skipIf(not hasattr(os, 'fork'), 'Test requires os.fork().')
     def test_post_fork_child_no_deadlock(self):
-        """Ensure forked child logging locks are not held; bpo-6721."""
-        refed_h = logging.Handler()
+        """Ensure child logging locks are not held; bpo-6721 & bpo-36533."""
+        class _OurHandler(logging.Handler):
+            def __init__(self):
+                super().__init__()
+                self.sub_handler = logging.StreamHandler(
+                    stream=open('/dev/null', 'wt'))
+
+            def emit(self, record):
+                self.sub_handler.acquire()
+                try:
+                    self.sub_handler.emit(record)
+                finally:
+                    self.sub_handler.release()
+
+        self.assertEqual(len(logging._handlers), 0)
+        refed_h = _OurHandler()
         refed_h.name = 'because we need at least one for this test'
         self.assertGreater(len(logging._handlers), 0)
+        self.assertGreater(len(logging._at_fork_reinit_lock_weakset), 1)
+        test_logger = logging.getLogger('test_post_fork_child_no_deadlock')
+        test_logger.addHandler(refed_h)
+        test_logger.setLevel(logging.DEBUG)
 
         locks_held__ready_to_fork = threading.Event()
         fork_happened__release_locks_and_end_thread = threading.Event()
@@ -709,19 +727,24 @@ def lock_holder_thread_fn():
         locks_held__ready_to_fork.wait()
         pid = os.fork()
         if pid == 0:  # Child.
-            logging.error(r'Child process did not deadlock. \o/')
-            os._exit(0)
+            try:
+                test_logger.info(r'Child process did not deadlock. \o/')
+            finally:
+                os._exit(0)
         else:  # Parent.
+            test_logger.info(r'Parent process returned from fork. \o/')
             fork_happened__release_locks_and_end_thread.set()
             lock_holder_thread.join()
             start_time = time.monotonic()
             while True:
+                test_logger.debug('Waiting for child process.')
                 waited_pid, status = os.waitpid(pid, os.WNOHANG)
                 if waited_pid == pid:
                     break  # child process exited.
                 if time.monotonic() - start_time > 7:
                     break  # so long? implies child deadlock.
                 time.sleep(0.05)
+            test_logger.debug('Done waiting.')
             if waited_pid != pid:
                 os.kill(pid, signal.SIGKILL)
                 waited_pid, status = os.waitpid(pid, 0)
diff --git a/Misc/NEWS.d/next/Library/2018-05-30-01-05-50.bpo-31922.fobsXJ.rst b/Misc/NEWS.d/next/Library/2018-05-30-01-05-50.bpo-31922.fobsXJ.rst
@@ -0,0 +1,3 @@
+:meth:`asyncio.AbstractEventLoop.create_datagram_endpoint`:
+Do not connect UDP socket when broadcast is allowed.
+This allows to receive replies after a UDP broadcast.
diff --git a/Misc/NEWS.d/next/Library/2019-04-06-00-55-09.bpo-36533.kzMyRH.rst b/Misc/NEWS.d/next/Library/2019-04-06-00-55-09.bpo-36533.kzMyRH.rst
@@ -0,0 +1,6 @@
+Reinitialize logging.Handler locks in forked child processes instead of
+attempting to acquire them all in the parent before forking only to be
+released in the child process.  The acquire/release pattern was leading to
+deadlocks in code that has implemented any form of chained logging handlers
+that depend upon one another as the lock acquision order cannot be
+guaranteed.

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	+:meth:`asyncio.AbstractEventLoop.create_datagram_endpoint`:
	`2`	`+Do not connect UDP socket when broadcast is allowed.`
	`3`	`+This allows to receive replies after a UDP broadcast.`