use set instead of list in memorypool_update
[electrum-server.git] / 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 (method, params) not in self.subscriptions:
239                 self.subscriptions.append((method,params))
240
241
242     def stop_subscriptions(self):
243         bp = self.bp
244
245         with self.lock:
246             s = self.subscriptions[:]
247
248         for method, params in s:
249             with bp.watch_lock:
250                 if method == 'blockchain.numblocks.subscribe':
251                     if self in bp.watch_blocks:
252                         bp.watch_blocks.remove(self)
253                 elif method == 'blockchain.headers.subscribe':
254                     if self in bp.watch_headers:
255                         bp.watch_headers.remove(self)
256                 elif method == "blockchain.address.subscribe":
257                     addr = params[0]
258                     l = bp.watched_addresses.get(addr)
259                     if not l:
260                         continue
261                     if self in l:
262                         l.remove(self)
263                     if self in l:
264                         print "error rc!!"
265                         bp.shared.stop()
266
267                     if l == []:
268                         bp.watched_addresses.pop(addr)
269
270         with self.lock:
271             self.subscriptions = []
272
273
274 class ResponseDispatcher(threading.Thread):
275
276     def __init__(self, shared, request_dispatcher):
277         self.shared = shared
278         self.request_dispatcher = request_dispatcher
279         threading.Thread.__init__(self)
280         self.daemon = True
281
282     def run(self):
283         while not self.shared.stopped():
284             session, response = self.request_dispatcher.pop_response()
285             session.send_response(response)