processor.py

   1 import json
   2 import Queue as queue
   3 import socket
   4 import threading
   5 import time
   6 import traceback
   7 import sys
   8
   9 from utils import random_string, timestr, print_log
  10
  11
  12 class Shared:
  13
  14     def __init__(self, config):
  15         self.lock = threading.Lock()
  16         self._stopped = False
  17         self.config = config
  18
  19     def stop(self):
  20         print_log("Stopping Stratum")
  21         with self.lock:
  22             self._stopped = True
  23
  24     def stopped(self):
  25         with self.lock:
  26             return self._stopped
  27
  28
  29 class Processor(threading.Thread):
  30
  31     def __init__(self):
  32         threading.Thread.__init__(self)
  33         self.daemon = True
  34         self.dispatcher = None
  35         self.queue = queue.Queue()
  36
  37     def process(self, session, request):
  38         pass
  39
  40     def add_request(self, session, request):
  41         self.queue.put((session, request))
  42
  43     def push_response(self, session, response):
  44         #print "response", response
  45         self.dispatcher.request_dispatcher.push_response(session, response)
  46
  47     def close(self):
  48         pass
  49
  50     def run(self):
  51         while not self.shared.stopped():
  52             try:
  53                 request, session = self.queue.get(True, timeout=1)
  54             except:
  55                 continue
  56             try:
  57                 self.process(request, session)
  58             except:
  59                 traceback.print_exc(file=sys.stdout)
  60
  61         self.close()
  62
  63
  64 class Dispatcher:
  65
  66     def __init__(self, config):
  67         self.shared = Shared(config)
  68         self.request_dispatcher = RequestDispatcher(self.shared)
  69         self.request_dispatcher.start()
  70         self.response_dispatcher = \
  71             ResponseDispatcher(self.shared, self.request_dispatcher)
  72         self.response_dispatcher.start()
  73
  74     def register(self, prefix, processor):
  75         processor.dispatcher = self
  76         processor.shared = self.shared
  77         processor.start()
  78         self.request_dispatcher.processors[prefix] = processor
  79
  80
  81 class RequestDispatcher(threading.Thread):
  82
  83     def __init__(self, shared):
  84         self.shared = shared
  85         threading.Thread.__init__(self)
  86         self.daemon = True
  87         self.request_queue = queue.Queue()
  88         self.response_queue = queue.Queue()
  89         self.lock = threading.Lock()
  90         self.idlock = threading.Lock()
  91         self.sessions = {}
  92         self.processors = {}
  93
  94     def push_response(self, session, item):
  95         self.response_queue.put((session, item))
  96
  97     def pop_response(self):
  98         return self.response_queue.get()
  99
 100     def push_request(self, session, item):
 101         self.request_queue.put((session, item))
 102
 103     def pop_request(self):
 104         return self.request_queue.get()
 105
 106     def get_session_by_address(self, address):
 107         for x in self.sessions.values():
 108             if x.address == address:
 109                 return x
 110
 111     def run(self):
 112         if self.shared is None:
 113             raise TypeError("self.shared not set in Processor")
 114
 115         lastgc = 0
 116
 117         while not self.shared.stopped():
 118             session, request = self.pop_request()
 119             try:
 120                 self.do_dispatch(session, request)
 121             except:
 122                 traceback.print_exc(file=sys.stdout)
 123
 124             if time.time() - lastgc > 60.0:
 125                 self.collect_garbage()
 126                 lastgc = time.time()
 127
 128         self.stop()
 129
 130     def stop(self):
 131         pass
 132
 133     def do_dispatch(self, session, request):
 134         """ dispatch request to the relevant processor """
 135
 136         method = request['method']
 137         params = request.get('params', [])
 138         suffix = method.split('.')[-1]
 139
 140         if session is not None:
 141             if suffix == 'subscribe':
 142                 session.subscribe_to_service(method, params)
 143
 144         prefix = request['method'].split('.')[0]
 145         try:
 146             p = self.processors[prefix]
 147         except:
 148             print_log("error: no processor for", prefix)
 149             return
 150
 151         p.add_request(session, request)
 152
 153         if method in ['server.version']:
 154             session.version = params[0]
 155             try:
 156                 session.protocol_version = float(params[1])
 157             except:
 158                 pass
 159
 160
 161     def get_sessions(self):
 162         with self.lock:
 163             r = self.sessions.values()
 164         return r
 165
 166     def add_session(self, session):
 167         key = session.key()
 168         with self.lock:
 169             self.sessions[key] = session
 170
 171     def remove_session(self, session):
 172         key = session.key()
 173         with self.lock:
 174             self.sessions.pop(key)
 175
 176     def collect_garbage(self):
 177         now = time.time()
 178         for session in self.sessions.values():
 179             if (now - session.time) > session.timeout:
 180                 session.stop()
 181
 182
 183
 184 class Session:
 185
 186     def __init__(self, dispatcher):
 187         self.dispatcher = dispatcher
 188         self.bp = self.dispatcher.processors['blockchain']
 189         self._stopped = False
 190         self.lock = threading.Lock()
 191         self.subscriptions = []
 192         self.address = ''
 193         self.name = ''
 194         self.version = 'unknown'
 195         self.protocol_version = 0.
 196         self.time = time.time()
 197         threading.Timer(2, self.info).start()
 198
 199
 200     def key(self):
 201         return self.name + self.address
 202
 203
 204     # Debugging method. Doesn't need to be threadsafe.
 205     def info(self):
 206         for sub in self.subscriptions:
 207             #print sub
 208             method = sub[0]
 209             if method == 'blockchain.address.subscribe':
 210                 addr = sub[1]
 211                 break
 212         else:
 213             addr = None
 214
 215         if self.subscriptions:
 216             print_log("%4s" % self.name,
 217                       "%15s" % self.address,
 218                       "%35s" % addr,
 219                       "%3d" % len(self.subscriptions),
 220                       self.version)
 221
 222     def stop(self):
 223         with self.lock:
 224             if self._stopped:
 225                 return
 226             self._stopped = True
 227
 228         self.shutdown()
 229         self.dispatcher.remove_session(self)
 230         self.stop_subscriptions()
 231
 232
 233     def shutdown(self):
 234         pass
 235
 236
 237     def stopped(self):
 238         with self.lock:
 239             return self._stopped
 240
 241
 242     def subscribe_to_service(self, method, params):
 243         with self.lock:
 244             if self._stopped:
 245                 return
 246             if (method, params) not in self.subscriptions:
 247                 self.subscriptions.append((method,params))
 248         self.bp.do_subscribe(method, params, self)
 249
 250
 251     def stop_subscriptions(self):
 252         with self.lock:
 253             s = self.subscriptions[:]
 254         for method, params in s:
 255             self.bp.do_unsubscribe(method, params, self)
 256         with self.lock:
 257             self.subscriptions = []
 258
 259
 260 class ResponseDispatcher(threading.Thread):
 261
 262     def __init__(self, shared, request_dispatcher):
 263         self.shared = shared
 264         self.request_dispatcher = request_dispatcher
 265         threading.Thread.__init__(self)
 266         self.daemon = True
 267
 268     def run(self):
 269         while not self.shared.stopped():
 270             session, response = self.request_dispatcher.pop_response()
 271             session.send_response(response)