Added timeout and joining daemon thread

Author: prithviraj-chaudhuri

Lib/multiprocessing/connection.py

@@ -487,8 +487,10 @@ def __init__(self, address=None, family=None, backlog=1, authkey=None):
         self._authkey = authkey
 
     def settimeout(self, timeout):
-        if timeout:
-            self._listener.settimeout(timeout)
+        '''
+        Set timeout for the accept method.
+        '''
+        self._listener.settimeout(timeout)
 
     def accept(self):
         '''
@@ -645,7 +647,8 @@ def __init__(self, address, family, backlog=1):
             self._unlink = None
 
     def settimeout(self, timeout):
-        self._socket.settimeout(timeout)
+        if timeout:
+            self._socket.settimeout(timeout)
 
     def accept(self):
         s, self._last_accepted = self._socket.accept()
@@ -692,6 +695,7 @@ def __init__(self, address, backlog=None):
                 self, PipeListener._finalize_pipe_listener,
                 args=(self._handle_queue, self._address), exitpriority=0
                 )
+            self._timeout = _winapi.INFINITE
 
         def _new_handle(self, first=False):
             flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
@@ -706,7 +710,8 @@ def _new_handle(self, first=False):
                 )
 
         def settimeout(self, timeout):
-            pass
+            if timeout:
+                self._timeout = int(timeout * 1000)
 
         def accept(self):
             self._handle_queue.append(self._new_handle())
@@ -721,7 +726,9 @@ def accept(self):
             else:
                 try:
                     res = _winapi.WaitForMultipleObjects(
-                        [ov.event], False, INFINITE)
+                        [ov.event], False, self._timeout)
+                    if res == _winapi.WAIT_TIMEOUT:
+                        raise TimeoutError("PipeListener timed out")
                 except:
                     ov.cancel()
                     _winapi.CloseHandle(handle)

Lib/multiprocessing/managers.py

@@ -171,6 +171,7 @@ def serve_forever(self):
         '''
         self.stop_event = threading.Event()
         process.current_process()._manager_server = self
+        accepter = None
         try:
             accepter = threading.Thread(target=self.accepter)
             accepter.daemon = True
@@ -189,17 +190,19 @@ def serve_forever(self):
             sys.exit(0)
 
     def accepter(self):
-        while True:
+        handler_threads = []
+        while True and not self.stop_event.is_set():
             try:
-                self.listener.settimeout(20)
+                self.listener.settimeout(3)
                 c = self.listener.accept()
             except OSError:
                 continue
             t = threading.Thread(target=self.handle_request, args=(c,))
+            handler_threads.append(t)
             t.daemon = True
             t.start()
-            if self.stop_event.is_set():
-                break
+        for t in handler_threads:
+            t.join()
 
     def _handle_request(self, c):
         request = None

Lib/test/_test_multiprocessing.py

@@ -3399,7 +3399,7 @@ def test_mymanager(self):
         # bpo-30356: BaseManager._finalize_manager() sends SIGTERM
         # to the manager process if it takes longer than 1 second to stop,
         # which happens on slow buildbots.
-        self.assertIn(manager._process.exitcode, (0, -signal.SIGTERM))
+        self.assertEqual(manager._process.exitcode, 0)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_mymanager_context(self):