mirror of
				https://github.com/python/cpython.git
				synced 2025-11-03 23:21:29 +00:00 
			
		
		
		
	[3.9] bpo-37193: Remove thread objects which finished process its request (GH-23127) (GH-24750)
This reverts commitaca67da4fe. (cherry picked from commitb5711c940f) Co-authored-by: Jason R. Coombs <jaraco@jaraco.com> Automerge-Triggered-By: GH:jaraco
This commit is contained in:
		
							parent
							
								
									ff6a021c25
								
							
						
					
					
						commit
						0e76157b0c
					
				
					 3 changed files with 64 additions and 12 deletions
				
			
		| 
						 | 
					@ -628,6 +628,39 @@ def server_close(self):
 | 
				
			||||||
            self.collect_children(blocking=self.block_on_close)
 | 
					            self.collect_children(blocking=self.block_on_close)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					class _Threads(list):
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    Joinable list of all non-daemon threads.
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    def append(self, thread):
 | 
				
			||||||
 | 
					        self.reap()
 | 
				
			||||||
 | 
					        if thread.daemon:
 | 
				
			||||||
 | 
					            return
 | 
				
			||||||
 | 
					        super().append(thread)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def pop_all(self):
 | 
				
			||||||
 | 
					        self[:], result = [], self[:]
 | 
				
			||||||
 | 
					        return result
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def join(self):
 | 
				
			||||||
 | 
					        for thread in self.pop_all():
 | 
				
			||||||
 | 
					            thread.join()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def reap(self):
 | 
				
			||||||
 | 
					        self[:] = (thread for thread in self if thread.is_alive())
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					class _NoThreads:
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    Degenerate version of _Threads.
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    def append(self, thread):
 | 
				
			||||||
 | 
					        pass
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def join(self):
 | 
				
			||||||
 | 
					        pass
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
class ThreadingMixIn:
 | 
					class ThreadingMixIn:
 | 
				
			||||||
    """Mix-in class to handle each request in a new thread."""
 | 
					    """Mix-in class to handle each request in a new thread."""
 | 
				
			||||||
 | 
					
 | 
				
			||||||
| 
						 | 
					@ -636,9 +669,9 @@ class ThreadingMixIn:
 | 
				
			||||||
    daemon_threads = False
 | 
					    daemon_threads = False
 | 
				
			||||||
    # If true, server_close() waits until all non-daemonic threads terminate.
 | 
					    # If true, server_close() waits until all non-daemonic threads terminate.
 | 
				
			||||||
    block_on_close = True
 | 
					    block_on_close = True
 | 
				
			||||||
    # For non-daemonic threads, list of threading.Threading objects
 | 
					    # Threads object
 | 
				
			||||||
    # used by server_close() to wait for all threads completion.
 | 
					    # used by server_close() to wait for all threads completion.
 | 
				
			||||||
    _threads = None
 | 
					    _threads = _NoThreads()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def process_request_thread(self, request, client_address):
 | 
					    def process_request_thread(self, request, client_address):
 | 
				
			||||||
        """Same as in BaseServer but as a thread.
 | 
					        """Same as in BaseServer but as a thread.
 | 
				
			||||||
| 
						 | 
					@ -655,23 +688,17 @@ def process_request_thread(self, request, client_address):
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def process_request(self, request, client_address):
 | 
					    def process_request(self, request, client_address):
 | 
				
			||||||
        """Start a new thread to process the request."""
 | 
					        """Start a new thread to process the request."""
 | 
				
			||||||
 | 
					        if self.block_on_close:
 | 
				
			||||||
 | 
					            vars(self).setdefault('_threads', _Threads())
 | 
				
			||||||
        t = threading.Thread(target = self.process_request_thread,
 | 
					        t = threading.Thread(target = self.process_request_thread,
 | 
				
			||||||
                             args = (request, client_address))
 | 
					                             args = (request, client_address))
 | 
				
			||||||
        t.daemon = self.daemon_threads
 | 
					        t.daemon = self.daemon_threads
 | 
				
			||||||
        if not t.daemon and self.block_on_close:
 | 
					 | 
				
			||||||
            if self._threads is None:
 | 
					 | 
				
			||||||
                self._threads = []
 | 
					 | 
				
			||||||
        self._threads.append(t)
 | 
					        self._threads.append(t)
 | 
				
			||||||
        t.start()
 | 
					        t.start()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def server_close(self):
 | 
					    def server_close(self):
 | 
				
			||||||
        super().server_close()
 | 
					        super().server_close()
 | 
				
			||||||
        if self.block_on_close:
 | 
					        self._threads.join()
 | 
				
			||||||
            threads = self._threads
 | 
					 | 
				
			||||||
            self._threads = None
 | 
					 | 
				
			||||||
            if threads:
 | 
					 | 
				
			||||||
                for thread in threads:
 | 
					 | 
				
			||||||
                    thread.join()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
if hasattr(os, "fork"):
 | 
					if hasattr(os, "fork"):
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
| 
						 | 
					@ -275,6 +275,13 @@ class MyHandler(socketserver.StreamRequestHandler):
 | 
				
			||||||
            t.join()
 | 
					            t.join()
 | 
				
			||||||
            s.server_close()
 | 
					            s.server_close()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def test_close_immediately(self):
 | 
				
			||||||
 | 
					        class MyServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
 | 
				
			||||||
 | 
					            pass
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        server = MyServer((HOST, 0), lambda: None)
 | 
				
			||||||
 | 
					        server.server_close()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_tcpserver_bind_leak(self):
 | 
					    def test_tcpserver_bind_leak(self):
 | 
				
			||||||
        # Issue #22435: the server socket wouldn't be closed if bind()/listen()
 | 
					        # Issue #22435: the server socket wouldn't be closed if bind()/listen()
 | 
				
			||||||
        # failed.
 | 
					        # failed.
 | 
				
			||||||
| 
						 | 
					@ -489,6 +496,22 @@ def shutdown_request(self, request):
 | 
				
			||||||
        self.assertEqual(server.shutdown_called, 1)
 | 
					        self.assertEqual(server.shutdown_called, 1)
 | 
				
			||||||
        server.server_close()
 | 
					        server.server_close()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def test_threads_reaped(self):
 | 
				
			||||||
 | 
					        """
 | 
				
			||||||
 | 
					        In #37193, users reported a memory leak
 | 
				
			||||||
 | 
					        due to the saving of every request thread. Ensure that
 | 
				
			||||||
 | 
					        not all threads are kept forever.
 | 
				
			||||||
 | 
					        """
 | 
				
			||||||
 | 
					        class MyServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
 | 
				
			||||||
 | 
					            pass
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        server = MyServer((HOST, 0), socketserver.StreamRequestHandler)
 | 
				
			||||||
 | 
					        for n in range(10):
 | 
				
			||||||
 | 
					            with socket.create_connection(server.server_address):
 | 
				
			||||||
 | 
					                server.handle_request()
 | 
				
			||||||
 | 
					        self.assertLess(len(server._threads), 10)
 | 
				
			||||||
 | 
					        server.server_close()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
if __name__ == "__main__":
 | 
					if __name__ == "__main__":
 | 
				
			||||||
    unittest.main()
 | 
					    unittest.main()
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
| 
						 | 
					@ -0,0 +1,2 @@
 | 
				
			||||||
 | 
					Fixed memory leak in ``socketserver.ThreadingMixIn`` introduced in Python
 | 
				
			||||||
 | 
					3.7.
 | 
				
			||||||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue