• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Generic socket server classes.
2
3This module tries to capture the various aspects of defining a server:
4
5For socket-based servers:
6
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)
14
15For request-based servers (including socket-based):
16
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)
24
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
27save some typing.  (There's also the issue that a deep class hierarchy
28slows down method lookups.)
29
30There are five classes in an inheritance diagram, four of which represent
31synchronous servers of four types:
32
33        +------------+
34        | BaseServer |
35        +------------+
36              |
37              v
38        +-----------+        +------------------+
39        | TCPServer |------->| UnixStreamServer |
40        +-----------+        +------------------+
41              |
42              v
43        +-----------+        +--------------------+
44        | UDPServer |------->| UnixDatagramServer |
45        +-----------+        +--------------------+
46
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.
51
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:
55
56        class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
57
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.
61
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.
66
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.
70
71Of course, you still have to use your head!
72
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.
80
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.
87
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.
93
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 select() 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).
102
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
107- Standard framework for select-based multiplexing
108
109XXX Open problems:
110- What to do with out-of-band data?
111
112BaseServer:
113- split generic "request" functionality out into BaseServer class.
114  Copyright (C) 2000  Luke Kenneth Casson Leighton <lkcl@samba.org>
115
116  example: read entries from a SQL database (requires overriding
117  get_request() to return a table entry from the database).
118  entry is processed by a RequestHandlerClass.
119
120"""
121
122# Author of the BaseServer patch: Luke Kenneth Casson Leighton
123
124__version__ = "0.4"
125
126
127import socket
128import select
129import sys
130import os
131import errno
132try:
133    import threading
134except ImportError:
135    import dummy_threading as threading
136
137__all__ = ["TCPServer","UDPServer","ForkingUDPServer","ForkingTCPServer",
138           "ThreadingUDPServer","ThreadingTCPServer","BaseRequestHandler",
139           "StreamRequestHandler","DatagramRequestHandler",
140           "ThreadingMixIn", "ForkingMixIn"]
141if hasattr(socket, "AF_UNIX"):
142    __all__.extend(["UnixStreamServer","UnixDatagramServer",
143                    "ThreadingUnixStreamServer",
144                    "ThreadingUnixDatagramServer"])
145
146def _eintr_retry(func, *args):
147    """restart a system call interrupted by EINTR"""
148    while True:
149        try:
150            return func(*args)
151        except (OSError, select.error) as e:
152            if e.args[0] != errno.EINTR:
153                raise
154
155class BaseServer:
156
157    """Base class for server classes.
158
159    Methods for the caller:
160
161    - __init__(server_address, RequestHandlerClass)
162    - serve_forever(poll_interval=0.5)
163    - shutdown()
164    - handle_request()  # if you do not use serve_forever()
165    - fileno() -> int   # for select()
166
167    Methods that may be overridden:
168
169    - server_bind()
170    - server_activate()
171    - get_request() -> request, client_address
172    - handle_timeout()
173    - verify_request(request, client_address)
174    - server_close()
175    - process_request(request, client_address)
176    - shutdown_request(request)
177    - close_request(request)
178    - handle_error()
179
180    Methods for derived classes:
181
182    - finish_request(request, client_address)
183
184    Class variables that may be overridden by derived classes or
185    instances:
186
187    - timeout
188    - address_family
189    - socket_type
190    - allow_reuse_address
191
192    Instance variables:
193
194    - RequestHandlerClass
195    - socket
196
197    """
198
199    timeout = None
200
201    def __init__(self, server_address, RequestHandlerClass):
202        """Constructor.  May be extended, do not override."""
203        self.server_address = server_address
204        self.RequestHandlerClass = RequestHandlerClass
205        self.__is_shut_down = threading.Event()
206        self.__shutdown_request = False
207
208    def server_activate(self):
209        """Called by constructor to activate the server.
210
211        May be overridden.
212
213        """
214        pass
215
216    def serve_forever(self, poll_interval=0.5):
217        """Handle one request at a time until shutdown.
218
219        Polls for shutdown every poll_interval seconds. Ignores
220        self.timeout. If you need to do periodic tasks, do them in
221        another thread.
222        """
223        self.__is_shut_down.clear()
224        try:
225            while not self.__shutdown_request:
226                # XXX: Consider using another file descriptor or
227                # connecting to the socket to wake this up instead of
228                # polling. Polling reduces our responsiveness to a
229                # shutdown request and wastes cpu at all other times.
230                r, w, e = _eintr_retry(select.select, [self], [], [],
231                                       poll_interval)
232                if self in r:
233                    self._handle_request_noblock()
234        finally:
235            self.__shutdown_request = False
236            self.__is_shut_down.set()
237
238    def shutdown(self):
239        """Stops the serve_forever loop.
240
241        Blocks until the loop has finished. This must be called while
242        serve_forever() is running in another thread, or it will
243        deadlock.
244        """
245        self.__shutdown_request = True
246        self.__is_shut_down.wait()
247
248    # The distinction between handling, getting, processing and
249    # finishing a request is fairly arbitrary.  Remember:
250    #
251    # - handle_request() is the top-level call.  It calls
252    #   select, get_request(), verify_request() and process_request()
253    # - get_request() is different for stream or datagram sockets
254    # - process_request() is the place that may fork a new process
255    #   or create a new thread to finish the request
256    # - finish_request() instantiates the request handler class;
257    #   this constructor will handle the request all by itself
258
259    def handle_request(self):
260        """Handle one request, possibly blocking.
261
262        Respects self.timeout.
263        """
264        # Support people who used socket.settimeout() to escape
265        # handle_request before self.timeout was available.
266        timeout = self.socket.gettimeout()
267        if timeout is None:
268            timeout = self.timeout
269        elif self.timeout is not None:
270            timeout = min(timeout, self.timeout)
271        fd_sets = _eintr_retry(select.select, [self], [], [], timeout)
272        if not fd_sets[0]:
273            self.handle_timeout()
274            return
275        self._handle_request_noblock()
276
277    def _handle_request_noblock(self):
278        """Handle one request, without blocking.
279
280        I assume that select.select has returned that the socket is
281        readable before this function was called, so there should be
282        no risk of blocking in get_request().
283        """
284        try:
285            request, client_address = self.get_request()
286        except socket.error:
287            return
288        if self.verify_request(request, client_address):
289            try:
290                self.process_request(request, client_address)
291            except:
292                self.handle_error(request, client_address)
293                self.shutdown_request(request)
294        else:
295            self.shutdown_request(request)
296
297    def handle_timeout(self):
298        """Called if no new request arrives within self.timeout.
299
300        Overridden by ForkingMixIn.
301        """
302        pass
303
304    def verify_request(self, request, client_address):
305        """Verify the request.  May be overridden.
306
307        Return True if we should proceed with this request.
308
309        """
310        return True
311
312    def process_request(self, request, client_address):
313        """Call finish_request.
314
315        Overridden by ForkingMixIn and ThreadingMixIn.
316
317        """
318        self.finish_request(request, client_address)
319        self.shutdown_request(request)
320
321    def server_close(self):
322        """Called to clean-up the server.
323
324        May be overridden.
325
326        """
327        pass
328
329    def finish_request(self, request, client_address):
330        """Finish one request by instantiating RequestHandlerClass."""
331        self.RequestHandlerClass(request, client_address, self)
332
333    def shutdown_request(self, request):
334        """Called to shutdown and close an individual request."""
335        self.close_request(request)
336
337    def close_request(self, request):
338        """Called to clean up an individual request."""
339        pass
340
341    def handle_error(self, request, client_address):
342        """Handle an error gracefully.  May be overridden.
343
344        The default is to print a traceback and continue.
345
346        """
347        print '-'*40
348        print 'Exception happened during processing of request from',
349        print client_address
350        import traceback
351        traceback.print_exc() # XXX But this goes to stderr!
352        print '-'*40
353
354
355class TCPServer(BaseServer):
356
357    """Base class for various socket-based server classes.
358
359    Defaults to synchronous IP stream (i.e., TCP).
360
361    Methods for the caller:
362
363    - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
364    - serve_forever(poll_interval=0.5)
365    - shutdown()
366    - handle_request()  # if you don't use serve_forever()
367    - fileno() -> int   # for select()
368
369    Methods that may be overridden:
370
371    - server_bind()
372    - server_activate()
373    - get_request() -> request, client_address
374    - handle_timeout()
375    - verify_request(request, client_address)
376    - process_request(request, client_address)
377    - shutdown_request(request)
378    - close_request(request)
379    - handle_error()
380
381    Methods for derived classes:
382
383    - finish_request(request, client_address)
384
385    Class variables that may be overridden by derived classes or
386    instances:
387
388    - timeout
389    - address_family
390    - socket_type
391    - request_queue_size (only for stream sockets)
392    - allow_reuse_address
393
394    Instance variables:
395
396    - server_address
397    - RequestHandlerClass
398    - socket
399
400    """
401
402    address_family = socket.AF_INET
403
404    socket_type = socket.SOCK_STREAM
405
406    request_queue_size = 5
407
408    allow_reuse_address = False
409
410    def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
411        """Constructor.  May be extended, do not override."""
412        BaseServer.__init__(self, server_address, RequestHandlerClass)
413        self.socket = socket.socket(self.address_family,
414                                    self.socket_type)
415        if bind_and_activate:
416            try:
417                self.server_bind()
418                self.server_activate()
419            except:
420                self.server_close()
421                raise
422
423    def server_bind(self):
424        """Called by constructor to bind the socket.
425
426        May be overridden.
427
428        """
429        if self.allow_reuse_address:
430            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
431        self.socket.bind(self.server_address)
432        self.server_address = self.socket.getsockname()
433
434    def server_activate(self):
435        """Called by constructor to activate the server.
436
437        May be overridden.
438
439        """
440        self.socket.listen(self.request_queue_size)
441
442    def server_close(self):
443        """Called to clean-up the server.
444
445        May be overridden.
446
447        """
448        self.socket.close()
449
450    def fileno(self):
451        """Return socket file number.
452
453        Interface required by select().
454
455        """
456        return self.socket.fileno()
457
458    def get_request(self):
459        """Get the request and client address from the socket.
460
461        May be overridden.
462
463        """
464        return self.socket.accept()
465
466    def shutdown_request(self, request):
467        """Called to shutdown and close an individual request."""
468        try:
469            #explicitly shutdown.  socket.close() merely releases
470            #the socket and waits for GC to perform the actual close.
471            request.shutdown(socket.SHUT_WR)
472        except socket.error:
473            pass #some platforms may raise ENOTCONN here
474        self.close_request(request)
475
476    def close_request(self, request):
477        """Called to clean up an individual request."""
478        request.close()
479
480
481class UDPServer(TCPServer):
482
483    """UDP server class."""
484
485    allow_reuse_address = False
486
487    socket_type = socket.SOCK_DGRAM
488
489    max_packet_size = 8192
490
491    def get_request(self):
492        data, client_addr = self.socket.recvfrom(self.max_packet_size)
493        return (data, self.socket), client_addr
494
495    def server_activate(self):
496        # No need to call listen() for UDP.
497        pass
498
499    def shutdown_request(self, request):
500        # No need to shutdown anything.
501        self.close_request(request)
502
503    def close_request(self, request):
504        # No need to close anything.
505        pass
506
507class ForkingMixIn:
508
509    """Mix-in class to handle each request in a new process."""
510
511    timeout = 300
512    active_children = None
513    max_children = 40
514
515    def collect_children(self):
516        """Internal routine to wait for children that have exited."""
517        if self.active_children is None:
518            return
519
520        # If we're above the max number of children, wait and reap them until
521        # we go back below threshold. Note that we use waitpid(-1) below to be
522        # able to collect children in size(<defunct children>) syscalls instead
523        # of size(<children>): the downside is that this might reap children
524        # which we didn't spawn, which is why we only resort to this when we're
525        # above max_children.
526        while len(self.active_children) >= self.max_children:
527            try:
528                pid, _ = os.waitpid(-1, 0)
529                self.active_children.discard(pid)
530            except OSError as e:
531                if e.errno == errno.ECHILD:
532                    # we don't have any children, we're done
533                    self.active_children.clear()
534                elif e.errno != errno.EINTR:
535                    break
536
537        # Now reap all defunct children.
538        for pid in self.active_children.copy():
539            try:
540                pid, _ = os.waitpid(pid, os.WNOHANG)
541                # if the child hasn't exited yet, pid will be 0 and ignored by
542                # discard() below
543                self.active_children.discard(pid)
544            except OSError as e:
545                if e.errno == errno.ECHILD:
546                    # someone else reaped it
547                    self.active_children.discard(pid)
548
549    def handle_timeout(self):
550        """Wait for zombies after self.timeout seconds of inactivity.
551
552        May be extended, do not override.
553        """
554        self.collect_children()
555
556    def process_request(self, request, client_address):
557        """Fork a new subprocess to process the request."""
558        self.collect_children()
559        pid = os.fork()
560        if pid:
561            # Parent process
562            if self.active_children is None:
563                self.active_children = set()
564            self.active_children.add(pid)
565            self.close_request(request) #close handle in parent process
566            return
567        else:
568            # Child process.
569            # This must never return, hence os._exit()!
570            try:
571                self.finish_request(request, client_address)
572                self.shutdown_request(request)
573                os._exit(0)
574            except:
575                try:
576                    self.handle_error(request, client_address)
577                    self.shutdown_request(request)
578                finally:
579                    os._exit(1)
580
581
582class ThreadingMixIn:
583    """Mix-in class to handle each request in a new thread."""
584
585    # Decides how threads will act upon termination of the
586    # main process
587    daemon_threads = False
588
589    def process_request_thread(self, request, client_address):
590        """Same as in BaseServer but as a thread.
591
592        In addition, exception handling is done here.
593
594        """
595        try:
596            self.finish_request(request, client_address)
597            self.shutdown_request(request)
598        except:
599            self.handle_error(request, client_address)
600            self.shutdown_request(request)
601
602    def process_request(self, request, client_address):
603        """Start a new thread to process the request."""
604        t = threading.Thread(target = self.process_request_thread,
605                             args = (request, client_address))
606        t.daemon = self.daemon_threads
607        t.start()
608
609
610class ForkingUDPServer(ForkingMixIn, UDPServer): pass
611class ForkingTCPServer(ForkingMixIn, TCPServer): pass
612
613class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass
614class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
615
616if hasattr(socket, 'AF_UNIX'):
617
618    class UnixStreamServer(TCPServer):
619        address_family = socket.AF_UNIX
620
621    class UnixDatagramServer(UDPServer):
622        address_family = socket.AF_UNIX
623
624    class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): pass
625
626    class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): pass
627
628class BaseRequestHandler:
629
630    """Base class for request handler classes.
631
632    This class is instantiated for each request to be handled.  The
633    constructor sets the instance variables request, client_address
634    and server, and then calls the handle() method.  To implement a
635    specific service, all you need to do is to derive a class which
636    defines a handle() method.
637
638    The handle() method can find the request as self.request, the
639    client address as self.client_address, and the server (in case it
640    needs access to per-server information) as self.server.  Since a
641    separate instance is created for each request, the handle() method
642    can define other arbitrary instance variables.
643
644    """
645
646    def __init__(self, request, client_address, server):
647        self.request = request
648        self.client_address = client_address
649        self.server = server
650        self.setup()
651        try:
652            self.handle()
653        finally:
654            self.finish()
655
656    def setup(self):
657        pass
658
659    def handle(self):
660        pass
661
662    def finish(self):
663        pass
664
665
666# The following two classes make it possible to use the same service
667# class for stream or datagram servers.
668# Each class sets up these instance variables:
669# - rfile: a file object from which receives the request is read
670# - wfile: a file object to which the reply is written
671# When the handle() method returns, wfile is flushed properly
672
673
674class StreamRequestHandler(BaseRequestHandler):
675
676    """Define self.rfile and self.wfile for stream sockets."""
677
678    # Default buffer sizes for rfile, wfile.
679    # We default rfile to buffered because otherwise it could be
680    # really slow for large data (a getc() call per byte); we make
681    # wfile unbuffered because (a) often after a write() we want to
682    # read and we need to flush the line; (b) big writes to unbuffered
683    # files are typically optimized by stdio even when big reads
684    # aren't.
685    rbufsize = -1
686    wbufsize = 0
687
688    # A timeout to apply to the request socket, if not None.
689    timeout = None
690
691    # Disable nagle algorithm for this socket, if True.
692    # Use only when wbufsize != 0, to avoid small packets.
693    disable_nagle_algorithm = False
694
695    def setup(self):
696        self.connection = self.request
697        if self.timeout is not None:
698            self.connection.settimeout(self.timeout)
699        if self.disable_nagle_algorithm:
700            self.connection.setsockopt(socket.IPPROTO_TCP,
701                                       socket.TCP_NODELAY, True)
702        self.rfile = self.connection.makefile('rb', self.rbufsize)
703        self.wfile = self.connection.makefile('wb', self.wbufsize)
704
705    def finish(self):
706        if not self.wfile.closed:
707            try:
708                self.wfile.flush()
709            except socket.error:
710                # A final socket error may have occurred here, such as
711                # the local error ECONNABORTED.
712                pass
713        self.wfile.close()
714        self.rfile.close()
715
716
717class DatagramRequestHandler(BaseRequestHandler):
718
719    """Define self.rfile and self.wfile for datagram sockets."""
720
721    def setup(self):
722        try:
723            from cStringIO import StringIO
724        except ImportError:
725            from StringIO import StringIO
726        self.packet, self.socket = self.request
727        self.rfile = StringIO(self.packet)
728        self.wfile = StringIO()
729
730    def finish(self):
731        self.socket.sendto(self.wfile.getvalue(), self.client_address)
732