NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

SocketServer.py (8977B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\SocketServer.py
      2 __version__ = '0.4'
      3 import socket
      4 import sys
      5 import os
      6 try:
      7     import threading
      8 except ImportError:
      9     import dummy_threading as threading
     10 
     11 __all__ = ['TCPServer',
     12  'UDPServer',
     13  'ForkingUDPServer',
     14  'ForkingTCPServer',
     15  'ThreadingUDPServer',
     16  'ThreadingTCPServer',
     17  'BaseRequestHandler',
     18  'StreamRequestHandler',
     19  'DatagramRequestHandler',
     20  'ThreadingMixIn',
     21  'ForkingMixIn']
     22 if hasattr(socket, 'AF_UNIX'):
     23     __all__.extend(['UnixStreamServer',
     24      'UnixDatagramServer',
     25      'ThreadingUnixStreamServer',
     26      'ThreadingUnixDatagramServer'])
     27 
     28 class BaseServer:
     29     timeout = None
     30 
     31     def __init__(self, server_address, RequestHandlerClass):
     32         self.server_address = server_address
     33         self.RequestHandlerClass = RequestHandlerClass
     34         self.__is_shut_down = threading.Event()
     35         self.__shutdown_request = False
     36 
     37     def server_activate(self):
     38         pass
     39 
     40     def serve_forever(self, poll_interval = 0.5):
     41         self.__is_shut_down.clear()
     42         try:
     43             while not self.__shutdown_request:
     44                 if True:
     45                     self._handle_request_noblock()
     46 
     47         finally:
     48             self.__shutdown_request = False
     49             self.__is_shut_down.set()
     50 
     51     def shutdown(self):
     52         self.__shutdown_request = True
     53         self.__is_shut_down.wait()
     54 
     55     def handle_request(self):
     56         timeout = self.socket.gettimeout()
     57         if timeout is None:
     58             timeout = self.timeout
     59         elif self.timeout is not None:
     60             timeout = min(timeout, self.timeout)
     61         self._handle_request_noblock()
     62 
     63     def _handle_request_noblock(self):
     64         try:
     65             request, client_address = self.get_request()
     66         except socket.error:
     67             return
     68 
     69         if self.verify_request(request, client_address):
     70             try:
     71                 self.process_request(request, client_address)
     72             except:
     73                 self.handle_error(request, client_address)
     74                 self.shutdown_request(request)
     75 
     76     def handle_timeout(self):
     77         pass
     78 
     79     def verify_request(self, request, client_address):
     80         return True
     81 
     82     def process_request(self, request, client_address):
     83         self.finish_request(request, client_address)
     84         self.shutdown_request(request)
     85 
     86     def server_close(self):
     87         pass
     88 
     89     def finish_request(self, request, client_address):
     90         self.RequestHandlerClass(request, client_address, self)
     91 
     92     def shutdown_request(self, request):
     93         self.close_request(request)
     94 
     95     def close_request(self, request):
     96         pass
     97 
     98     def handle_error(self, request, client_address):
     99         print '-' * 40
    100         print 'Exception happened during processing of request from',
    101         print client_address
    102         import traceback
    103         traceback.print_exc()
    104         print '-' * 40
    105 
    106 
    107 class TCPServer(BaseServer):
    108     address_family = socket.AF_INET
    109     socket_type = socket.SOCK_STREAM
    110     request_queue_size = 5
    111     allow_reuse_address = False
    112 
    113     def __init__(self, server_address, RequestHandlerClass, bind_and_activate = True):
    114         BaseServer.__init__(self, server_address, RequestHandlerClass)
    115         self.socket = socket.socket(self.address_family, self.socket_type)
    116         if bind_and_activate:
    117             self.server_bind()
    118             self.server_activate()
    119 
    120     def server_bind(self):
    121         if self.allow_reuse_address:
    122             self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    123         self.socket.bind(self.server_address)
    124         self.server_address = self.socket.getsockname()
    125 
    126     def server_activate(self):
    127         self.socket.listen(self.request_queue_size)
    128 
    129     def server_close(self):
    130         self.socket.close()
    131 
    132     def fileno(self):
    133         return self.socket.fileno()
    134 
    135     def get_request(self):
    136         return self.socket.accept()
    137 
    138     def shutdown_request(self, request):
    139         try:
    140             request.shutdown(socket.SHUT_WR)
    141         except socket.error:
    142             pass
    143 
    144         self.close_request(request)
    145 
    146     def close_request(self, request):
    147         request.shutdown(socket.SHUT_WR)
    148         request.close()
    149 
    150 
    151 class UDPServer(TCPServer):
    152     allow_reuse_address = False
    153     socket_type = socket.SOCK_DGRAM
    154     max_packet_size = 8192
    155 
    156     def get_request(self):
    157         data, client_addr = self.socket.recvfrom(self.max_packet_size)
    158         return ((data, self.socket), client_addr)
    159 
    160     def server_activate(self):
    161         pass
    162 
    163     def shutdown_request(self, request):
    164         self.close_request(request)
    165 
    166     def close_request(self, request):
    167         pass
    168 
    169 
    170 class ForkingMixIn:
    171     timeout = 300
    172     active_children = None
    173     max_children = 40
    174 
    175     def collect_children(self):
    176         if self.active_children is None:
    177             return
    178         while len(self.active_children) >= self.max_children:
    179             try:
    180                 pid, status = os.waitpid(0, 0)
    181             except os.error:
    182                 pid = None
    183 
    184             if pid not in self.active_children:
    185                 continue
    186             self.active_children.remove(pid)
    187 
    188         for child in self.active_children:
    189             try:
    190                 pid, status = os.waitpid(child, os.WNOHANG)
    191             except os.error:
    192                 pid = None
    193 
    194             if not pid:
    195                 continue
    196             try:
    197                 self.active_children.remove(pid)
    198             except ValueError as e:
    199                 raise ValueError('%s. x=%d and list=%r' % (e.message, pid, self.active_children))
    200 
    201     def handle_timeout(self):
    202         self.collect_children()
    203 
    204     def process_request(self, request, client_address):
    205         self.collect_children()
    206         pid = os.fork()
    207         if pid:
    208             if self.active_children is None:
    209                 self.active_children = []
    210             self.active_children.append(pid)
    211             self.close_request(request)
    212             return
    213         try:
    214             self.finish_request(request, client_address)
    215             self.shutdown_request(request)
    216             os._exit(0)
    217         except:
    218             try:
    219                 self.handle_error(request, client_address)
    220                 self.shutdown_request(request)
    221             finally:
    222                 os._exit(1)
    223 
    224 
    225 class ThreadingMixIn:
    226     daemon_threads = False
    227 
    228     def process_request_thread(self, request, client_address):
    229         try:
    230             self.finish_request(request, client_address)
    231             self.shutdown_request(request)
    232         except:
    233             self.handle_error(request, client_address)
    234             self.shutdown_request(request)
    235 
    236     def process_request(self, request, client_address):
    237         t = threading.Thread(target=self.process_request_thread, args=(request, client_address))
    238         if self.daemon_threads:
    239             t.setDaemon(1)
    240         t.start()
    241 
    242 
    243 class ForkingUDPServer(ForkingMixIn, UDPServer):
    244     pass
    245 
    246 
    247 class ForkingTCPServer(ForkingMixIn, TCPServer):
    248     pass
    249 
    250 
    251 class ThreadingUDPServer(ThreadingMixIn, UDPServer):
    252     pass
    253 
    254 
    255 class ThreadingTCPServer(ThreadingMixIn, TCPServer):
    256     pass
    257 
    258 
    259 if hasattr(socket, 'AF_UNIX'):
    260 
    261     class UnixStreamServer(TCPServer):
    262         address_family = socket.AF_UNIX
    263 
    264 
    265     class UnixDatagramServer(UDPServer):
    266         address_family = socket.AF_UNIX
    267 
    268 
    269     class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer):
    270         pass
    271 
    272 
    273     class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer):
    274         pass
    275 
    276 
    277 class BaseRequestHandler:
    278 
    279     def __init__(self, request, client_address, server):
    280         self.request = request
    281         self.client_address = client_address
    282         self.server = server
    283         self.setup()
    284         try:
    285             self.handle()
    286         finally:
    287             self.finish()
    288 
    289     def setup(self):
    290         pass
    291 
    292     def handle(self):
    293         pass
    294 
    295     def finish(self):
    296         pass
    297 
    298 
    299 class StreamRequestHandler(BaseRequestHandler):
    300     rbufsize = -1
    301     wbufsize = 0
    302     timeout = None
    303     disable_nagle_algorithm = False
    304 
    305     def setup(self):
    306         self.connection = self.request
    307         if self.timeout is not None:
    308             self.connection.settimeout(self.timeout)
    309         if self.disable_nagle_algorithm:
    310             self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
    311         self.rfile = self.connection.makefile('rb', self.rbufsize)
    312         self.wfile = self.connection.makefile('wb', self.wbufsize)
    313 
    314     def finish(self):
    315         if not self.wfile.closed:
    316             self.wfile.flush()
    317         self.wfile.close()
    318         self.rfile.close()
    319 
    320 
    321 class DatagramRequestHandler(BaseRequestHandler):
    322 
    323     def setup(self):
    324         try:
    325             from cStringIO import StringIO
    326         except ImportError:
    327             from StringIO import StringIO
    328 
    329         self.packet, self.socket = self.request
    330         self.rfile = StringIO(self.packet)
    331         self.wfile = StringIO()
    332 
    333     def finish(self):
    334         self.socket.sendto(self.wfile.getvalue(), self.client_address)