bpo-37193: Remove thread objects which finished process its request (GH-23127)

This reverts commit aca67da4fe.
(cherry picked from commit b5711c940f)

Co-authored-by: Jason R. Coombs <jaraco@jaraco.com>
This commit is contained in:
Jason R. Coombs 2020-12-31 15:19:30 -05:00 committed by Miss Islington (bot)
parent 82f24ff344
commit 5072d61390
3 changed files with 64 additions and 12 deletions

View File

@ -628,6 +628,39 @@ if hasattr(os, "fork"):
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 @@ class ThreadingMixIn:
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: self._threads.append(t)
if self._threads is None:
self._threads = []
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"):

View File

@ -276,6 +276,13 @@ class SocketServerTest(unittest.TestCase):
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.
@ -490,6 +497,22 @@ class MiscTestCase(unittest.TestCase):
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()

View File

@ -0,0 +1,2 @@
Fixed memory leak in ``socketserver.ThreadingMixIn`` introduced in Python
3.7.