Coverage for /pythoncovmergedfiles/medio/medio/usr/lib/python3.9/socketserver.py: 37%
308 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:05 +0000
1"""Generic socket server classes.
3This module tries to capture the various aspects of defining a server:
5For socket-based servers:
7- address family:
8 - AF_INET{,6}: IP (Internet Protocol) sockets (default)
9 - AF_UNIX: Unix domain sockets
10 - others, e.g. AF_DECNET are conceivable (see <socket.h>
11- socket type:
12 - SOCK_STREAM (reliable stream, e.g. TCP)
13 - SOCK_DGRAM (datagrams, e.g. UDP)
15For request-based servers (including socket-based):
17- client address verification before further looking at the request
18 (This is actually a hook for any processing that needs to look
19 at the request before anything else, e.g. logging)
20- how to handle multiple requests:
21 - synchronous (one request is handled at a time)
22 - forking (each request is handled by a new process)
23 - threading (each request is handled by a new thread)
25The classes in this module favor the server type that is simplest to
26write: a synchronous TCP/IP server. This is bad class design, but
27saves some typing. (There's also the issue that a deep class hierarchy
28slows down method lookups.)
30There are five classes in an inheritance diagram, four of which represent
31synchronous servers of four types:
33 +------------+
34 | BaseServer |
35 +------------+
36 |
37 v
38 +-----------+ +------------------+
39 | TCPServer |------->| UnixStreamServer |
40 +-----------+ +------------------+
41 |
42 v
43 +-----------+ +--------------------+
44 | UDPServer |------->| UnixDatagramServer |
45 +-----------+ +--------------------+
47Note that UnixDatagramServer derives from UDPServer, not from
48UnixStreamServer -- the only difference between an IP and a Unix
49stream server is the address family, which is simply repeated in both
50unix server classes.
52Forking and threading versions of each type of server can be created
53using the ForkingMixIn and ThreadingMixIn mix-in classes. For
54instance, a threading UDP server class is created as follows:
56 class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
58The Mix-in class must come first, since it overrides a method defined
59in UDPServer! Setting the various member variables also changes
60the behavior of the underlying server mechanism.
62To implement a service, you must derive a class from
63BaseRequestHandler and redefine its handle() method. You can then run
64various versions of the service by combining one of the server classes
65with your request handler class.
67The request handler class must be different for datagram or stream
68services. This can be hidden by using the request handler
69subclasses StreamRequestHandler or DatagramRequestHandler.
71Of course, you still have to use your head!
73For instance, it makes no sense to use a forking server if the service
74contains state in memory that can be modified by requests (since the
75modifications in the child process would never reach the initial state
76kept in the parent process and passed to each child). In this case,
77you can use a threading server, but you will probably have to use
78locks to avoid two requests that come in nearly simultaneous to apply
79conflicting changes to the server state.
81On the other hand, if you are building e.g. an HTTP server, where all
82data is stored externally (e.g. in the file system), a synchronous
83class will essentially render the service "deaf" while one request is
84being handled -- which may be for a very long time if a client is slow
85to read all the data it has requested. Here a threading or forking
86server is appropriate.
88In some cases, it may be appropriate to process part of a request
89synchronously, but to finish processing in a forked child depending on
90the request data. This can be implemented by using a synchronous
91server and doing an explicit fork in the request handler class
92handle() method.
94Another approach to handling multiple simultaneous requests in an
95environment that supports neither threads nor fork (or where these are
96too expensive or inappropriate for the service) is to maintain an
97explicit table of partially finished requests and to use a selector to
98decide which request to work on next (or whether to handle a new
99incoming request). This is particularly important for stream services
100where each client can potentially be connected for a long time (if
101threads or subprocesses cannot be used).
103Future work:
104- Standard classes for Sun RPC (which uses either UDP or TCP)
105- Standard mix-in classes to implement various authentication
106 and encryption schemes
108XXX Open problems:
109- What to do with out-of-band data?
111BaseServer:
112- split generic "request" functionality out into BaseServer class.
113 Copyright (C) 2000 Luke Kenneth Casson Leighton <lkcl@samba.org>
115 example: read entries from a SQL database (requires overriding
116 get_request() to return a table entry from the database).
117 entry is processed by a RequestHandlerClass.
119"""
121# Author of the BaseServer patch: Luke Kenneth Casson Leighton
123__version__ = "0.4"
126import socket
127import selectors
128import os
129import sys
130import threading
131from io import BufferedIOBase
132from time import monotonic as time
134__all__ = ["BaseServer", "TCPServer", "UDPServer",
135 "ThreadingUDPServer", "ThreadingTCPServer",
136 "BaseRequestHandler", "StreamRequestHandler",
137 "DatagramRequestHandler", "ThreadingMixIn"]
138if hasattr(os, "fork"):
139 __all__.extend(["ForkingUDPServer","ForkingTCPServer", "ForkingMixIn"])
140if hasattr(socket, "AF_UNIX"):
141 __all__.extend(["UnixStreamServer","UnixDatagramServer",
142 "ThreadingUnixStreamServer",
143 "ThreadingUnixDatagramServer"])
145# poll/select have the advantage of not requiring any extra file descriptor,
146# contrarily to epoll/kqueue (also, they require a single syscall).
147if hasattr(selectors, 'PollSelector'):
148 _ServerSelector = selectors.PollSelector
149else:
150 _ServerSelector = selectors.SelectSelector
153class BaseServer:
155 """Base class for server classes.
157 Methods for the caller:
159 - __init__(server_address, RequestHandlerClass)
160 - serve_forever(poll_interval=0.5)
161 - shutdown()
162 - handle_request() # if you do not use serve_forever()
163 - fileno() -> int # for selector
165 Methods that may be overridden:
167 - server_bind()
168 - server_activate()
169 - get_request() -> request, client_address
170 - handle_timeout()
171 - verify_request(request, client_address)
172 - server_close()
173 - process_request(request, client_address)
174 - shutdown_request(request)
175 - close_request(request)
176 - service_actions()
177 - handle_error()
179 Methods for derived classes:
181 - finish_request(request, client_address)
183 Class variables that may be overridden by derived classes or
184 instances:
186 - timeout
187 - address_family
188 - socket_type
189 - allow_reuse_address
191 Instance variables:
193 - RequestHandlerClass
194 - socket
196 """
198 timeout = None
200 def __init__(self, server_address, RequestHandlerClass):
201 """Constructor. May be extended, do not override."""
202 self.server_address = server_address
203 self.RequestHandlerClass = RequestHandlerClass
204 self.__is_shut_down = threading.Event()
205 self.__shutdown_request = False
207 def server_activate(self):
208 """Called by constructor to activate the server.
210 May be overridden.
212 """
213 pass
215 def serve_forever(self, poll_interval=0.5):
216 """Handle one request at a time until shutdown.
218 Polls for shutdown every poll_interval seconds. Ignores
219 self.timeout. If you need to do periodic tasks, do them in
220 another thread.
221 """
222 self.__is_shut_down.clear()
223 try:
224 # XXX: Consider using another file descriptor or connecting to the
225 # socket to wake this up instead of polling. Polling reduces our
226 # responsiveness to a shutdown request and wastes cpu at all other
227 # times.
228 with _ServerSelector() as selector:
229 selector.register(self, selectors.EVENT_READ)
231 while not self.__shutdown_request:
232 ready = selector.select(poll_interval)
233 # bpo-35017: shutdown() called during select(), exit immediately.
234 if self.__shutdown_request:
235 break
236 if ready:
237 self._handle_request_noblock()
239 self.service_actions()
240 finally:
241 self.__shutdown_request = False
242 self.__is_shut_down.set()
244 def shutdown(self):
245 """Stops the serve_forever loop.
247 Blocks until the loop has finished. This must be called while
248 serve_forever() is running in another thread, or it will
249 deadlock.
250 """
251 self.__shutdown_request = True
252 self.__is_shut_down.wait()
254 def service_actions(self):
255 """Called by the serve_forever() loop.
257 May be overridden by a subclass / Mixin to implement any code that
258 needs to be run during the loop.
259 """
260 pass
262 # The distinction between handling, getting, processing and finishing a
263 # request is fairly arbitrary. Remember:
264 #
265 # - handle_request() is the top-level call. It calls selector.select(),
266 # get_request(), verify_request() and process_request()
267 # - get_request() is different for stream or datagram sockets
268 # - process_request() is the place that may fork a new process or create a
269 # new thread to finish the request
270 # - finish_request() instantiates the request handler class; this
271 # constructor will handle the request all by itself
273 def handle_request(self):
274 """Handle one request, possibly blocking.
276 Respects self.timeout.
277 """
278 # Support people who used socket.settimeout() to escape
279 # handle_request before self.timeout was available.
280 timeout = self.socket.gettimeout()
281 if timeout is None:
282 timeout = self.timeout
283 elif self.timeout is not None:
284 timeout = min(timeout, self.timeout)
285 if timeout is not None:
286 deadline = time() + timeout
288 # Wait until a request arrives or the timeout expires - the loop is
289 # necessary to accommodate early wakeups due to EINTR.
290 with _ServerSelector() as selector:
291 selector.register(self, selectors.EVENT_READ)
293 while True:
294 ready = selector.select(timeout)
295 if ready:
296 return self._handle_request_noblock()
297 else:
298 if timeout is not None:
299 timeout = deadline - time()
300 if timeout < 0:
301 return self.handle_timeout()
303 def _handle_request_noblock(self):
304 """Handle one request, without blocking.
306 I assume that selector.select() has returned that the socket is
307 readable before this function was called, so there should be no risk of
308 blocking in get_request().
309 """
310 try:
311 request, client_address = self.get_request()
312 except OSError:
313 return
314 if self.verify_request(request, client_address):
315 try:
316 self.process_request(request, client_address)
317 except Exception:
318 self.handle_error(request, client_address)
319 self.shutdown_request(request)
320 except:
321 self.shutdown_request(request)
322 raise
323 else:
324 self.shutdown_request(request)
326 def handle_timeout(self):
327 """Called if no new request arrives within self.timeout.
329 Overridden by ForkingMixIn.
330 """
331 pass
333 def verify_request(self, request, client_address):
334 """Verify the request. May be overridden.
336 Return True if we should proceed with this request.
338 """
339 return True
341 def process_request(self, request, client_address):
342 """Call finish_request.
344 Overridden by ForkingMixIn and ThreadingMixIn.
346 """
347 self.finish_request(request, client_address)
348 self.shutdown_request(request)
350 def server_close(self):
351 """Called to clean-up the server.
353 May be overridden.
355 """
356 pass
358 def finish_request(self, request, client_address):
359 """Finish one request by instantiating RequestHandlerClass."""
360 self.RequestHandlerClass(request, client_address, self)
362 def shutdown_request(self, request):
363 """Called to shutdown and close an individual request."""
364 self.close_request(request)
366 def close_request(self, request):
367 """Called to clean up an individual request."""
368 pass
370 def handle_error(self, request, client_address):
371 """Handle an error gracefully. May be overridden.
373 The default is to print a traceback and continue.
375 """
376 print('-'*40, file=sys.stderr)
377 print('Exception occurred during processing of request from',
378 client_address, file=sys.stderr)
379 import traceback
380 traceback.print_exc()
381 print('-'*40, file=sys.stderr)
383 def __enter__(self):
384 return self
386 def __exit__(self, *args):
387 self.server_close()
390class TCPServer(BaseServer):
392 """Base class for various socket-based server classes.
394 Defaults to synchronous IP stream (i.e., TCP).
396 Methods for the caller:
398 - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
399 - serve_forever(poll_interval=0.5)
400 - shutdown()
401 - handle_request() # if you don't use serve_forever()
402 - fileno() -> int # for selector
404 Methods that may be overridden:
406 - server_bind()
407 - server_activate()
408 - get_request() -> request, client_address
409 - handle_timeout()
410 - verify_request(request, client_address)
411 - process_request(request, client_address)
412 - shutdown_request(request)
413 - close_request(request)
414 - handle_error()
416 Methods for derived classes:
418 - finish_request(request, client_address)
420 Class variables that may be overridden by derived classes or
421 instances:
423 - timeout
424 - address_family
425 - socket_type
426 - request_queue_size (only for stream sockets)
427 - allow_reuse_address
429 Instance variables:
431 - server_address
432 - RequestHandlerClass
433 - socket
435 """
437 address_family = socket.AF_INET
439 socket_type = socket.SOCK_STREAM
441 request_queue_size = 5
443 allow_reuse_address = False
445 def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
446 """Constructor. May be extended, do not override."""
447 BaseServer.__init__(self, server_address, RequestHandlerClass)
448 self.socket = socket.socket(self.address_family,
449 self.socket_type)
450 if bind_and_activate:
451 try:
452 self.server_bind()
453 self.server_activate()
454 except:
455 self.server_close()
456 raise
458 def server_bind(self):
459 """Called by constructor to bind the socket.
461 May be overridden.
463 """
464 if self.allow_reuse_address:
465 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
466 self.socket.bind(self.server_address)
467 self.server_address = self.socket.getsockname()
469 def server_activate(self):
470 """Called by constructor to activate the server.
472 May be overridden.
474 """
475 self.socket.listen(self.request_queue_size)
477 def server_close(self):
478 """Called to clean-up the server.
480 May be overridden.
482 """
483 self.socket.close()
485 def fileno(self):
486 """Return socket file number.
488 Interface required by selector.
490 """
491 return self.socket.fileno()
493 def get_request(self):
494 """Get the request and client address from the socket.
496 May be overridden.
498 """
499 return self.socket.accept()
501 def shutdown_request(self, request):
502 """Called to shutdown and close an individual request."""
503 try:
504 #explicitly shutdown. socket.close() merely releases
505 #the socket and waits for GC to perform the actual close.
506 request.shutdown(socket.SHUT_WR)
507 except OSError:
508 pass #some platforms may raise ENOTCONN here
509 self.close_request(request)
511 def close_request(self, request):
512 """Called to clean up an individual request."""
513 request.close()
516class UDPServer(TCPServer):
518 """UDP server class."""
520 allow_reuse_address = False
522 socket_type = socket.SOCK_DGRAM
524 max_packet_size = 8192
526 def get_request(self):
527 data, client_addr = self.socket.recvfrom(self.max_packet_size)
528 return (data, self.socket), client_addr
530 def server_activate(self):
531 # No need to call listen() for UDP.
532 pass
534 def shutdown_request(self, request):
535 # No need to shutdown anything.
536 self.close_request(request)
538 def close_request(self, request):
539 # No need to close anything.
540 pass
542if hasattr(os, "fork"):
543 class ForkingMixIn:
544 """Mix-in class to handle each request in a new process."""
546 timeout = 300
547 active_children = None
548 max_children = 40
549 # If true, server_close() waits until all child processes complete.
550 block_on_close = True
552 def collect_children(self, *, blocking=False):
553 """Internal routine to wait for children that have exited."""
554 if self.active_children is None:
555 return
557 # If we're above the max number of children, wait and reap them until
558 # we go back below threshold. Note that we use waitpid(-1) below to be
559 # able to collect children in size(<defunct children>) syscalls instead
560 # of size(<children>): the downside is that this might reap children
561 # which we didn't spawn, which is why we only resort to this when we're
562 # above max_children.
563 while len(self.active_children) >= self.max_children:
564 try:
565 pid, _ = os.waitpid(-1, 0)
566 self.active_children.discard(pid)
567 except ChildProcessError:
568 # we don't have any children, we're done
569 self.active_children.clear()
570 except OSError:
571 break
573 # Now reap all defunct children.
574 for pid in self.active_children.copy():
575 try:
576 flags = 0 if blocking else os.WNOHANG
577 pid, _ = os.waitpid(pid, flags)
578 # if the child hasn't exited yet, pid will be 0 and ignored by
579 # discard() below
580 self.active_children.discard(pid)
581 except ChildProcessError:
582 # someone else reaped it
583 self.active_children.discard(pid)
584 except OSError:
585 pass
587 def handle_timeout(self):
588 """Wait for zombies after self.timeout seconds of inactivity.
590 May be extended, do not override.
591 """
592 self.collect_children()
594 def service_actions(self):
595 """Collect the zombie child processes regularly in the ForkingMixIn.
597 service_actions is called in the BaseServer's serve_forever loop.
598 """
599 self.collect_children()
601 def process_request(self, request, client_address):
602 """Fork a new subprocess to process the request."""
603 pid = os.fork()
604 if pid:
605 # Parent process
606 if self.active_children is None:
607 self.active_children = set()
608 self.active_children.add(pid)
609 self.close_request(request)
610 return
611 else:
612 # Child process.
613 # This must never return, hence os._exit()!
614 status = 1
615 try:
616 self.finish_request(request, client_address)
617 status = 0
618 except Exception:
619 self.handle_error(request, client_address)
620 finally:
621 try:
622 self.shutdown_request(request)
623 finally:
624 os._exit(status)
626 def server_close(self):
627 super().server_close()
628 self.collect_children(blocking=self.block_on_close)
631class _Threads(list):
632 """
633 Joinable list of all non-daemon threads.
634 """
635 def append(self, thread):
636 self.reap()
637 if thread.daemon:
638 return
639 super().append(thread)
641 def pop_all(self):
642 self[:], result = [], self[:]
643 return result
645 def join(self):
646 for thread in self.pop_all():
647 thread.join()
649 def reap(self):
650 self[:] = (thread for thread in self if thread.is_alive())
653class _NoThreads:
654 """
655 Degenerate version of _Threads.
656 """
657 def append(self, thread):
658 pass
660 def join(self):
661 pass
664class ThreadingMixIn:
665 """Mix-in class to handle each request in a new thread."""
667 # Decides how threads will act upon termination of the
668 # main process
669 daemon_threads = False
670 # If true, server_close() waits until all non-daemonic threads terminate.
671 block_on_close = True
672 # Threads object
673 # used by server_close() to wait for all threads completion.
674 _threads = _NoThreads()
676 def process_request_thread(self, request, client_address):
677 """Same as in BaseServer but as a thread.
679 In addition, exception handling is done here.
681 """
682 try:
683 self.finish_request(request, client_address)
684 except Exception:
685 self.handle_error(request, client_address)
686 finally:
687 self.shutdown_request(request)
689 def process_request(self, request, client_address):
690 """Start a new thread to process the request."""
691 if self.block_on_close:
692 vars(self).setdefault('_threads', _Threads())
693 t = threading.Thread(target = self.process_request_thread,
694 args = (request, client_address))
695 t.daemon = self.daemon_threads
696 self._threads.append(t)
697 t.start()
699 def server_close(self):
700 super().server_close()
701 self._threads.join()
704if hasattr(os, "fork"):
705 class ForkingUDPServer(ForkingMixIn, UDPServer): pass
706 class ForkingTCPServer(ForkingMixIn, TCPServer): pass
708class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
709class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
711if hasattr(socket, 'AF_UNIX'):
713 class UnixStreamServer(TCPServer):
714 address_family = socket.AF_UNIX
716 class UnixDatagramServer(UDPServer):
717 address_family = socket.AF_UNIX
719 class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): pass
721 class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): pass
723class BaseRequestHandler:
725 """Base class for request handler classes.
727 This class is instantiated for each request to be handled. The
728 constructor sets the instance variables request, client_address
729 and server, and then calls the handle() method. To implement a
730 specific service, all you need to do is to derive a class which
731 defines a handle() method.
733 The handle() method can find the request as self.request, the
734 client address as self.client_address, and the server (in case it
735 needs access to per-server information) as self.server. Since a
736 separate instance is created for each request, the handle() method
737 can define other arbitrary instance variables.
739 """
741 def __init__(self, request, client_address, server):
742 self.request = request
743 self.client_address = client_address
744 self.server = server
745 self.setup()
746 try:
747 self.handle()
748 finally:
749 self.finish()
751 def setup(self):
752 pass
754 def handle(self):
755 pass
757 def finish(self):
758 pass
761# The following two classes make it possible to use the same service
762# class for stream or datagram servers.
763# Each class sets up these instance variables:
764# - rfile: a file object from which receives the request is read
765# - wfile: a file object to which the reply is written
766# When the handle() method returns, wfile is flushed properly
769class StreamRequestHandler(BaseRequestHandler):
771 """Define self.rfile and self.wfile for stream sockets."""
773 # Default buffer sizes for rfile, wfile.
774 # We default rfile to buffered because otherwise it could be
775 # really slow for large data (a getc() call per byte); we make
776 # wfile unbuffered because (a) often after a write() we want to
777 # read and we need to flush the line; (b) big writes to unbuffered
778 # files are typically optimized by stdio even when big reads
779 # aren't.
780 rbufsize = -1
781 wbufsize = 0
783 # A timeout to apply to the request socket, if not None.
784 timeout = None
786 # Disable nagle algorithm for this socket, if True.
787 # Use only when wbufsize != 0, to avoid small packets.
788 disable_nagle_algorithm = False
790 def setup(self):
791 self.connection = self.request
792 if self.timeout is not None:
793 self.connection.settimeout(self.timeout)
794 if self.disable_nagle_algorithm:
795 self.connection.setsockopt(socket.IPPROTO_TCP,
796 socket.TCP_NODELAY, True)
797 self.rfile = self.connection.makefile('rb', self.rbufsize)
798 if self.wbufsize == 0:
799 self.wfile = _SocketWriter(self.connection)
800 else:
801 self.wfile = self.connection.makefile('wb', self.wbufsize)
803 def finish(self):
804 if not self.wfile.closed:
805 try:
806 self.wfile.flush()
807 except socket.error:
808 # A final socket error may have occurred here, such as
809 # the local error ECONNABORTED.
810 pass
811 self.wfile.close()
812 self.rfile.close()
814class _SocketWriter(BufferedIOBase):
815 """Simple writable BufferedIOBase implementation for a socket
817 Does not hold data in a buffer, avoiding any need to call flush()."""
819 def __init__(self, sock):
820 self._sock = sock
822 def writable(self):
823 return True
825 def write(self, b):
826 self._sock.sendall(b)
827 with memoryview(b) as view:
828 return view.nbytes
830 def fileno(self):
831 return self._sock.fileno()
833class DatagramRequestHandler(BaseRequestHandler):
835 """Define self.rfile and self.wfile for datagram sockets."""
837 def setup(self):
838 from io import BytesIO
839 self.packet, self.socket = self.request
840 self.rfile = BytesIO(self.packet)
841 self.wfile = BytesIO()
843 def finish(self):
844 self.socket.sendto(self.wfile.getvalue(), self.client_address)