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.internal_ids = {}
  84         self.internal_id = 1
  85         self.lock = threading.Lock()
  86         self.idlock = threading.Lock()
  87         self.sessions = []
  88         self.processors = {}
  89
  90     def push_response(self, session, item):
  91         self.response_queue.put((session, item))
  92
  93     def pop_response(self):
  94         return self.response_queue.get()
  95
  96     def push_request(self, session, item):
  97         self.request_queue.put((session, item))
  98
  99     def pop_request(self):
 100         return self.request_queue.get()
 101
 102     def get_session_by_address(self, address):
 103         for x in self.sessions:
 104             if x.address == address:
 105                 return x
 106
 107     def get_session_id(self, internal_id):
 108         with self.idlock:
 109             return self.internal_ids.pop(internal_id)
 110
 111     def store_session_id(self, session, msgid):
 112         with self.idlock:
 113             self.internal_ids[self.internal_id] = session, msgid
 114             r = self.internal_id
 115             self.internal_id += 1
 116             return r
 117
 118     def run(self):
 119         if self.shared is None:
 120             raise TypeError("self.shared not set in Processor")
 121         while not self.shared.stopped():
 122             session, request = self.pop_request()
 123             try:
 124                 self.do_dispatch(session, request)
 125             except:
 126                 traceback.print_exc(file=sys.stdout)
 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[:]
 164         return r
 165
 166     def add_session(self, session):
 167         with self.lock:
 168             self.sessions.append(session)
 169
 170     def collect_garbage(self):
 171         # Deep copy entire sessions list and blank it
 172         # This is done to minimize lock contention
 173         with self.lock:
 174             sessions = self.sessions[:]
 175
 176         active_sessions = []
 177
 178         now = time.time()
 179         for session in sessions:
 180             if (now - session.time) > 1000:
 181                 session.stop()
 182
 183         bp = self.processors['blockchain']
 184
 185         for session in sessions:
 186             if not session.stopped():
 187                 # If session is still alive then re-add it back
 188                 # to our internal register
 189                 active_sessions.append(session)
 190             else:
 191                 session.stop_subscriptions(bp)
 192
 193         with self.lock:
 194             self.sessions = active_sessions[:]
 195
 196
 197
 198 class Session:
 199
 200     def __init__(self):
 201         self._stopped = False
 202         self.lock = threading.Lock()
 203         self.subscriptions = []
 204         self.address = ''
 205         self.name = ''
 206         self.version = 'unknown'
 207         self.protocol_version = 0.
 208         self.time = time.time()
 209         threading.Timer(2, self.info).start()
 210
 211
 212     # Debugging method. Doesn't need to be threadsafe.
 213     def info(self):
 214         for sub in self.subscriptions:
 215             #print sub
 216             method = sub[0]
 217             if method == 'blockchain.address.subscribe':
 218                 addr = sub[1]
 219                 break
 220         else:
 221             addr = None
 222
 223         if self.subscriptions:
 224             print_log("%4s" % self.name,
 225                       "%15s" % self.address,
 226                       "%35s" % addr,
 227                       "%3d" % len(self.subscriptions),
 228                       self.version)
 229
 230     def stopped(self):
 231         with self.lock:
 232             return self._stopped
 233
 234
 235     def subscribe_to_service(self, method, params):
 236         with self.lock:
 237             if (method, params) not in self.subscriptions:
 238                 self.subscriptions.append((method,params))
 239
 240
 241     def stop_subscriptions(self, bp):
 242         with self.lock:
 243             s = self.subscriptions[:]
 244
 245         for method, params in s:
 246             with bp.watch_lock:
 247                 if method == 'blockchain.numblocks.subscribe':
 248                     if self in bp.watch_blocks:
 249                         bp.watch_blocks.remove(self)
 250                 elif method == 'blockchain.headers.subscribe':
 251                     if self in bp.watch_headers:
 252                         bp.watch_headers.remove(self)
 253                 elif method == "blockchain.address.subscribe":
 254                     addr = params[0]
 255                     l = bp.watched_addresses.get(addr)
 256                     if not l:
 257                         continue
 258                     if self in l:
 259                         l.remove(self)
 260                     if l == []:
 261                         bp.watched_addresses.pop(addr)
 262
 263         with self.lock:
 264             self.subscriptions = []
 265
 266
 267 class ResponseDispatcher(threading.Thread):
 268
 269     def __init__(self, shared, request_dispatcher):
 270         self.shared = shared
 271         self.request_dispatcher = request_dispatcher
 272         threading.Thread.__init__(self)
 273         self.daemon = True
 274
 275     def run(self):
 276         while not self.shared.stopped():
 277             session, response = self.request_dispatcher.pop_response()
 278             session.send_response(response)