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