X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=stratum.py;h=ce39acad751f9571b0713615bbb05f159d27e098;hb=cb1ebe091b58cb616458c1afd1b97251d61eabed;hp=d91b7d17e1b842394c6f336b860038ef6e760821;hpb=4792b79d87e45deee06c806f390fbfbed9c95da5;p=electrum-server.git diff --git a/stratum.py b/stratum.py index d91b7d1..ce39aca 100644 --- a/stratum.py +++ b/stratum.py @@ -4,83 +4,15 @@ import threading import time import Queue as queue -class Processor(threading.Thread): +from processor import Session, Dispatcher, Shared - def __init__(self): - self.shared = None - self.lock = threading.Lock() - self.sessions = [] - threading.Thread.__init__(self) - self.daemon = True - self.request_queue = queue.Queue() - self.response_queue = queue.Queue() - - def add_session(self, session): - with self.lock: - self.sessions.append(session) - - def push_response(self, session, item): - self.response_queue.put((session,item)) - - def pop_response(self): - return self.response_queue.get() - - def push_request(self, session, item): - self.request_queue.put((session,item)) - - def pop_request(self): - return self.request_queue.get() - - def run(self): - if self.shared is None: - raise TypeError("self.shared not set in Processor") - while not self.shared.stopped(): - # Deep copy entire sessions list and blank it - # This is done to minimise lock contention - with self.lock: - sessions = self.sessions[:] - self.sessions = [] - for session in sessions: - if not session.stopped(): - # If session is still alive then re-add it back - # to our internal register - self.add_session(session) - - session, request = self.pop_request() - self.process(session, request) - - self.stop() - - def stop(self): - pass - - def process(self, session, request): - print "New request", request - # Do stuff... - # response = request - # When ready, you call - # self.push_response(session,response) - -class Session: +class TcpSession(Session): def __init__(self, connection, address): self._connection = connection self.address = address - self._stopped = False - self.lock = threading.Lock() - self.numblocks_sub = None - self.addresses_sub = {} - print "new session", address - - def stop(self): - self._connection.close() - print "Terminating connection:", self.address[0] - with self.lock: - self._stopped = True - - def stopped(self): - with self.lock: - return self._stopped + Session.__init__(self) + print "New session", address def connection(self): if self.stopped(): @@ -88,34 +20,24 @@ class Session: else: return self._connection - def subscribe_to_numblocks(self,message_id): - with self.lock: - self.numblocks_sub = message_id - - def subscribe_to_address(self,address,message_id,status): + def stop(self): + self._connection.close() + print "Terminating connection:", self.address[0] with self.lock: - self.addresses_sub[address] = message_id,status + self._stopped = True + def send_response(self, response): + raw_response = json.dumps(response) + # Possible race condition here by having session + # close connection? + # I assume Python connections are thread safe interfaces + try: + connection = self.connection() + connection.send(raw_response + "\n") + except: + self.stop() -class TcpResponder(threading.Thread): - def __init__(self, shared, processor): - self.shared = shared - self.processor = processor - threading.Thread.__init__(self) - - def run(self): - while not self.shared.stopped(): - session,response = self.processor.pop_response() - raw_response = json.dumps(response) - # Possible race condition here by having session - # close connection? - # I assume Python connections are thread safe interfaces - try: - connection = session.connection() - connection.send(raw_response + "\n") - except: - session.stop() class TcpClientRequestor(threading.Thread): @@ -147,7 +69,7 @@ class TcpClientRequestor(threading.Thread): def receive(self): try: return self.session.connection().recv(1024) - except socket.error: + except: return '' def parse(self): @@ -164,8 +86,7 @@ class TcpClientRequestor(threading.Thread): try: command = json.loads(raw_command) except: - self.processor.push_response(self.session, - {"error": "bad JSON", "request": raw_command}) + self.processor.push_response({"error": "bad JSON", "request": raw_command}) return True try: @@ -175,8 +96,7 @@ class TcpClientRequestor(threading.Thread): method = command['method'] except KeyError: # Return an error JSON in response. - self.processor.push_response(self.session, - {"error": "syntax error", "request": raw_command}) + self.processor.push_response({"error": "syntax error", "request": raw_command}) else: self.processor.push_request(self.session,command) @@ -187,11 +107,11 @@ class TcpServer(threading.Thread): def __init__(self, shared, processor, host, port): self.shared = shared self.processor = processor - self.clients = [] threading.Thread.__init__(self) self.daemon = True self.host = host self.port = port + self.lock = threading.Lock() def run(self): print "TCP server started." @@ -199,47 +119,13 @@ class TcpServer(threading.Thread): sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.host, self.port)) sock.listen(1) - responder = TcpResponder(self.shared, self.processor) - responder.start() while not self.shared.stopped(): - session = Session(*sock.accept()) + session = TcpSession(*sock.accept()) client_req = TcpClientRequestor(self.shared, self.processor, session) client_req.start() self.processor.add_session(session) + self.processor.collect_garbage() -class Shared: - def __init__(self): - self.lock = threading.Lock() - self._stopped = False - def stop(self): - print "Stopping Stratum" - with self.lock: - self._stopped = True - - def stopped(self): - with self.lock: - return self._stopped - -class Stratum: - - def start(self, processor): - shared = Shared() - # Bind shared to processor since constructor is user defined - processor.shared = shared - processor.start() - # Create various transports we need - transports = TcpServer(shared, processor, "176.31.24.241", 50001), - for server in transports: - server.start() - while not shared.stopped(): - if raw_input() == "quit": - shared.stop() - time.sleep(1) - -if __name__ == "__main__": - processor = Processor() - app = Stratum() - app.start(processor)