import socket import socketserver import threading import time TIMEOUT = 1 _main_thread = threading.current_thread() class RequestHandler(socketserver.BaseRequestHandler): def handle(self): time.sleep(TIMEOUT) class TCPServer(socketserver.ThreadingTCPServer): daemon_threads = False def test_threads(daemon=False, daemon_threads=False): # Port 0 means to select an arbitrary unused port HOST, PORT = 'localhost', 0 # Tweak the class attribute for the demo TCPServer.daemon_threads = daemon_threads server = TCPServer((HOST, PORT), RequestHandler) # Start a thread with the server -- that thread will then start one # more thread for each request server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = daemon server_thread.start() time.sleep(.1) # Client connect, to spawn child thread sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(server.server_address) sock.close() # Now close server server.shutdown() if daemon: msg = " Daemon " else: msg = "Non-daemon " msg += ("server with %s.daemon_threads=%s spawns client:" % (TCPServer.__name__, TCPServer.daemon_threads)) print(msg) for thd in threading.enumerate(): if thd != _main_thread: print("\t%r" % (thd,)) while len(threading.enumerate()) > 1: time.sleep(.1) if __name__ == "__main__": # Server is non-daemon. Clients are OK. test_threads(daemon_threads=False) # Client thread is non-daemon test_threads(daemon_threads=True) # Client thread is daemon # This case is buggy: it should spawn non-daemon client thread. test_threads(daemon=True, daemon_threads=False) # This is OK too. test_threads(daemon=True, daemon_threads=True)