Mark Slee | c98d050 | 2006-09-06 02:42:25 +0000 | [diff] [blame] | 1 | import sys |
| 2 | import traceback |
Mark Slee | 3c4d7fd | 2006-10-02 17:53:20 +0000 | [diff] [blame] | 3 | import threading |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 4 | import Queue |
Mark Slee | c98d050 | 2006-09-06 02:42:25 +0000 | [diff] [blame] | 5 | |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 6 | from thrift.Thrift import TProcessor |
| 7 | from thrift.transport import TTransport |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 8 | from thrift.protocol import TBinaryProtocol |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 9 | |
| 10 | class TServer: |
| 11 | |
Mark Slee | 794993d | 2006-09-20 01:56:10 +0000 | [diff] [blame] | 12 | """Base interface for a server, which must have a serve method.""" |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 13 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 14 | def __init__(self, processor, serverTransport, transportFactory=None, protocolFactory=None): |
Mark Slee | d788b2e | 2006-09-07 01:26:35 +0000 | [diff] [blame] | 15 | self.processor = processor |
| 16 | self.serverTransport = serverTransport |
| 17 | if transportFactory == None: |
| 18 | self.transportFactory = TTransport.TTransportFactoryBase() |
| 19 | else: |
| 20 | self.transportFactory = transportFactory |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 21 | if protocolFactory == None: |
| 22 | self.protocolFactory = TBinaryProtocol.TBinaryProtocolFactory() |
| 23 | else: |
| 24 | self.protocolFactory = protocolFactory |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 25 | |
Mark Slee | 794993d | 2006-09-20 01:56:10 +0000 | [diff] [blame] | 26 | def serve(self): |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 27 | pass |
| 28 | |
| 29 | class TSimpleServer(TServer): |
| 30 | |
| 31 | """Simple single-threaded server that just pumps around one transport.""" |
| 32 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 33 | def __init__(self, processor, serverTransport, transportFactory=None, protocolFactory=None): |
| 34 | TServer.__init__(self, processor, serverTransport, transportFactory, protocolFactory) |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 35 | |
Mark Slee | 794993d | 2006-09-20 01:56:10 +0000 | [diff] [blame] | 36 | def serve(self): |
Mark Slee | d788b2e | 2006-09-07 01:26:35 +0000 | [diff] [blame] | 37 | self.serverTransport.listen() |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 38 | while True: |
Mark Slee | d788b2e | 2006-09-07 01:26:35 +0000 | [diff] [blame] | 39 | client = self.serverTransport.accept() |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 40 | (itrans, otrans) = self.transportFactory.getIOTransports(client) |
| 41 | (iprot, oprot) = self.protocolFactory.getIOProtocols(itrans, otrans) |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 42 | try: |
| 43 | while True: |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 44 | self.processor.process(iprot, oprot) |
Mark Slee | 4f0fed6 | 2006-10-02 17:50:08 +0000 | [diff] [blame] | 45 | except TTransport.TTransportException, tx: |
| 46 | pass |
Mark Slee | c967656 | 2006-09-05 17:34:52 +0000 | [diff] [blame] | 47 | except Exception, x: |
Mark Slee | c98d050 | 2006-09-06 02:42:25 +0000 | [diff] [blame] | 48 | print '%s, %s, %s' % (type(x), x, traceback.format_exc()) |
Mark Slee | d788b2e | 2006-09-07 01:26:35 +0000 | [diff] [blame] | 49 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 50 | itrans.close() |
| 51 | otrans.close() |
Mark Slee | 4f0fed6 | 2006-10-02 17:50:08 +0000 | [diff] [blame] | 52 | |
| 53 | class TThreadedServer(TServer): |
| 54 | |
| 55 | """Threaded server that spawns a new thread per each connection.""" |
| 56 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 57 | def __init__(self, processor, serverTransport, transportFactory=None, protocolFactory=None): |
| 58 | TServer.__init__(self, processor, serverTransport, transportFactory, protocolFactory) |
Mark Slee | 4f0fed6 | 2006-10-02 17:50:08 +0000 | [diff] [blame] | 59 | |
| 60 | def serve(self): |
| 61 | self.serverTransport.listen() |
| 62 | while True: |
| 63 | try: |
| 64 | client = self.serverTransport.accept() |
| 65 | t = threading.Thread(target = self.handle, args=(client,)) |
| 66 | t.start() |
| 67 | except Exception, x: |
| 68 | print '%s, %s, %s,' % (type(x), x, traceback.format_exc()) |
| 69 | |
| 70 | def handle(self, client): |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 71 | (itrans, otrans) = self.transportFactory.getIOTransports(client) |
| 72 | (iprot, oprot) = self.protocolFactory.getIOProtocols(itrans, otrans) |
Mark Slee | 4f0fed6 | 2006-10-02 17:50:08 +0000 | [diff] [blame] | 73 | try: |
| 74 | while True: |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 75 | self.processor.process(iprot, oprot) |
Mark Slee | 4f0fed6 | 2006-10-02 17:50:08 +0000 | [diff] [blame] | 76 | except TTransport.TTransportException, tx: |
| 77 | pass |
| 78 | except Exception, x: |
| 79 | print '%s, %s, %s' % (type(x), x, traceback.format_exc()) |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 80 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 81 | itrans.close() |
| 82 | otrans.close() |
| 83 | |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 84 | class TThreadPoolServer(TServer): |
| 85 | |
| 86 | """Server with a fixed size pool of threads which service requests.""" |
| 87 | |
| 88 | def __init__(self, processor, serverTransport, transportFactory=None): |
| 89 | TServer.__init__(self, processor, serverTransport, transportFactory) |
| 90 | self.clients = Queue.Queue() |
| 91 | self.threads = 10 |
| 92 | |
Mark Slee | 4ce787f | 2006-10-24 18:54:06 +0000 | [diff] [blame] | 93 | def setNumThreads(self, num): |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 94 | """Set the number of worker threads that should be created""" |
| 95 | self.threads = num |
| 96 | |
| 97 | def serveThread(self): |
| 98 | """Loop around getting clients from the shared queue and process them.""" |
| 99 | while True: |
| 100 | try: |
Mark Slee | 9a695ba | 2006-10-24 18:55:36 +0000 | [diff] [blame] | 101 | client = self.clients.get() |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 102 | self.serveClient(client) |
| 103 | except Exception, x: |
| 104 | print '%s, %s, %s' % (type(x), x, traceback.format_exc()) |
| 105 | |
| 106 | def serveClient(self, client): |
| 107 | """Process input/output from a client for as long as possible""" |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 108 | (itrans, otrans) = self.transportFactory.getIOTransports(client) |
| 109 | (iprot, oprot) = self.protocolFactory.getIOProtocols(itrans, otrans) |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 110 | try: |
| 111 | while True: |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 112 | self.processor.process(iprot, oprot) |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 113 | except TTransport.TTransportException, tx: |
| 114 | pass |
| 115 | except Exception, x: |
| 116 | print '%s, %s, %s' % (type(x), x, traceback.format_exc()) |
| 117 | |
Mark Slee | 4ac459f | 2006-10-25 21:39:01 +0000 | [diff] [blame^] | 118 | itrans.close() |
| 119 | otrans.close() |
| 120 | |
Mark Slee | b90aa7c | 2006-10-24 18:49:45 +0000 | [diff] [blame] | 121 | def serve(self): |
| 122 | """Start a fixed number of worker threads and put client into a queue""" |
| 123 | for i in range(self.threads): |
| 124 | try: |
| 125 | t = threading.Thread(target = self.serveThread) |
| 126 | t.start() |
| 127 | except Exception, x: |
| 128 | print '%s, %s, %s,' % (type(x), x, traceback.format_exc()) |
| 129 | |
| 130 | # Pump the socket for clients |
| 131 | self.serverTransport.listen() |
| 132 | while True: |
| 133 | try: |
| 134 | client = self.serverTransport.accept() |
| 135 | self.clients.put(client) |
| 136 | except Exception, x: |
| 137 | print '%s, %s, %s' % (type(x), x, traceback.format_exc()) |