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, request):
  38         pass
  39
  40     def add_request(self, request):
  41         self.queue.put(request)
  42
  43     def push_response(self, response):
  44         #print "response", response
  45         self.dispatcher.request_dispatcher.push_response(response)
  46
  47     def run(self):
  48         while not self.shared.stopped():
  49             request = self.queue.get(10000000000)
  50             try:
  51                 self.process(request)
  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.internal_ids = {}
  84         self.internal_id = 1
  85         self.lock = threading.Lock()
  86         self.sessions = []
  87         self.processors = {}
  88
  89     def push_response(self, item):
  90         self.response_queue.put(item)
  91
  92     def pop_response(self):
  93         return self.response_queue.get()
  94
  95     def push_request(self, session, item):
  96         self.request_queue.put((session, item))
  97
  98     def pop_request(self):
  99         return self.request_queue.get()
 100
 101     def get_session_by_address(self, address):
 102         for x in self.sessions:
 103             if x.address == address:
 104                 return x
 105
 106     def get_session_id(self, internal_id):
 107         with self.lock:
 108             return self.internal_ids.pop(internal_id)
 109
 110     def store_session_id(self, session, msgid):
 111         with self.lock:
 112             self.internal_ids[self.internal_id] = session, msgid
 113             r = self.internal_id
 114             self.internal_id += 1
 115             return r
 116
 117     def run(self):
 118         if self.shared is None:
 119             raise TypeError("self.shared not set in Processor")
 120         while not self.shared.stopped():
 121             session, request = self.pop_request()
 122             try:
 123                 self.do_dispatch(session, request)
 124             except:
 125                 traceback.print_exc(file=sys.stdout)
 126
 127         self.stop()
 128
 129     def stop(self):
 130         pass
 131
 132     def do_dispatch(self, session, request):
 133         """ dispatch request to the relevant processor """
 134
 135         method = request['method']
 136         params = request.get('params', [])
 137         suffix = method.split('.')[-1]
 138
 139         if session is not None:
 140             is_new = session.protocol_version >= 0.5
 141             if suffix == 'subscribe':
 142                 session.subscribe_to_service(method, params)
 143
 144         # store session and id locally
 145         request['id'] = self.store_session_id(session, request['id'])
 146
 147         prefix = request['method'].split('.')[0]
 148         try:
 149             p = self.processors[prefix]
 150         except:
 151             print_log("error: no processor for", prefix)
 152             return
 153
 154         p.add_request(request)
 155
 156         if method in ['server.version']:
 157             session.version = params[0]
 158             try:
 159                 session.protocol_version = float(params[1])
 160             except:
 161                 pass
 162
 163             #if session.protocol_version < 0.6:
 164             #    print_log("stopping session from old client", session.protocol_version)
 165             #    session.stop()
 166
 167     def get_sessions(self):
 168         with self.lock:
 169             r = self.sessions[:]
 170         return r
 171
 172     def add_session(self, session):
 173         with self.lock:
 174             self.sessions.append(session)
 175
 176     def collect_garbage(self):
 177         # Deep copy entire sessions list and blank it
 178         # This is done to minimize lock contention
 179         with self.lock:
 180             sessions = self.sessions[:]
 181
 182         active_sessions = []
 183         for session in sessions:
 184             if not session.stopped():
 185                 # If session is still alive then re-add it back
 186                 # to our internal register
 187                 active_sessions.append(session)
 188
 189         with self.lock:
 190             self.sessions = active_sessions[:]
 191
 192
 193
 194 class Session:
 195
 196     def __init__(self):
 197         self._stopped = False
 198         self.lock = threading.Lock()
 199         self.subscriptions = []
 200         self.address = ''
 201         self.name = ''
 202         self.version = 'unknown'
 203         self.protocol_version = 0.
 204         self.time = time.time()
 205         threading.Timer(2, self.info).start()
 206
 207     # Debugging method. Doesn't need to be threadsafe.
 208     def info(self):
 209         for sub in self.subscriptions:
 210             #print sub
 211             method = sub[0]
 212             if method == 'blockchain.address.subscribe':
 213                 addr = sub[1]
 214                 break
 215         else:
 216             addr = None
 217
 218         if self.subscriptions:
 219             print_log("%4s" % self.name,
 220                       "%15s" % self.address,
 221                       "%35s" % addr,
 222                       "%3d" % len(self.subscriptions),
 223                       self.version)
 224
 225     def stopped(self):
 226         with self.lock:
 227             return self._stopped
 228
 229     def subscribe_to_service(self, method, params):
 230         subdesc = self.build_subdesc(method, params)
 231         with self.lock:
 232             if subdesc is not None:
 233                 self.subscriptions.append(subdesc)
 234
 235     # subdesc = A subscription description
 236     @staticmethod
 237     def build_subdesc(method, params):
 238         if method == "blockchain.numblocks.subscribe":
 239             return method,
 240         elif method == "blockchain.headers.subscribe":
 241             return method,
 242         elif method in ["blockchain.address.subscribe"]:
 243             if not params:
 244                 return None
 245             else:
 246                 return method, params[0]
 247         else:
 248             return None
 249
 250     def contains_subscription(self, subdesc):
 251         with self.lock:
 252             return subdesc in self.subscriptions
 253
 254
 255 class ResponseDispatcher(threading.Thread):
 256
 257     def __init__(self, shared, request_dispatcher):
 258         self.shared = shared
 259         self.request_dispatcher = request_dispatcher
 260         threading.Thread.__init__(self)
 261         self.daemon = True
 262
 263     def run(self):
 264         while not self.shared.stopped():
 265             self.update()
 266
 267     def update(self):
 268         response = self.request_dispatcher.pop_response()
 269         #print "pop response", response
 270         internal_id = response.get('id')
 271         method = response.get('method')
 272         params = response.get('params')
 273
 274         # A notification
 275         if internal_id is None:  # and method is not None and params is not None:
 276             found = self.notification(method, params, response)
 277             if not found and method == 'blockchain.address.subscribe':
 278                 request = {
 279                     'id': None,
 280                     'method': method.replace('.subscribe', '.unsubscribe'),
 281                     'params': [self.shared.config.get('server', 'password')] + params,
 282                 }
 283
 284                 self.request_dispatcher.push_request(None, request)
 285         # A response
 286         elif internal_id is not None:
 287             self.send_response(internal_id, response)
 288         else:
 289             print_log("no method", response)
 290
 291     def notification(self, method, params, response):
 292         subdesc = Session.build_subdesc(method, params)
 293         found = False
 294         for session in self.request_dispatcher.sessions:
 295             if session.stopped():
 296                 continue
 297             if session.contains_subscription(subdesc):
 298                 session.send_response(response)
 299                 found = True
 300         # if not found: print_log("no subscriber for", subdesc)
 301         return found
 302
 303     def send_response(self, internal_id, response):
 304         session, message_id = self.request_dispatcher.get_session_id(internal_id)
 305         if session:
 306             response['id'] = message_id
 307             session.send_response(response)
 308         #else:
 309         #    print_log("send_response: no session", message_id, internal_id, response )