X-Git-Url: https://git.novaco.in/?p=electrum-server.git;a=blobdiff_plain;f=transports%2Fstratum_tcp.py;h=aad4c17d87ffdafbd31c0e649c1a606bf2778f30;hp=17c2b9a2a244955a22f87c7b202414ae5eb6d01b;hb=4ce69b7ea24ead59ebbcc7ed335ea9762ae3724b;hpb=9c7a806d782ee4a3a8a49d2277c41f1513869dc5 diff --git a/transports/stratum_tcp.py b/transports/stratum_tcp.py index 17c2b9a..aad4c17 100644 --- a/transports/stratum_tcp.py +++ b/transports/stratum_tcp.py @@ -1,15 +1,19 @@ import json +import Queue as queue import socket import threading import time -import Queue as queue +import traceback, sys + +from processor import Session, Dispatcher +from utils import print_log -from processor import Session, Dispatcher, print_log class TcpSession(Session): - def __init__(self, connection, address, use_ssl, ssl_certfile, ssl_keyfile): - Session.__init__(self) + def __init__(self, dispatcher, connection, address, use_ssl, ssl_certfile, ssl_keyfile): + Session.__init__(self, dispatcher) + self.use_ssl = use_ssl if use_ssl: import ssl self._connection = ssl.wrap_socket( @@ -17,12 +21,20 @@ class TcpSession(Session): server_side=True, certfile=ssl_certfile, keyfile=ssl_keyfile, - ssl_version=ssl.PROTOCOL_SSLv23) + ssl_version=ssl.PROTOCOL_SSLv23, + do_handshake_on_connect=False) else: self._connection = connection - self.address = address[0] - self.name = "TCP" if not use_ssl else "TCP/SSL" + self.address = address[0] + ":%d"%address[1] + self.name = "TCP " if not use_ssl else "SSL " + self.timeout = 1000 + self.response_queue = queue.Queue() + self.dispatcher.add_session(self) + + def do_handshake(self): + if self.use_ssl: + self._connection.do_handshake() def connection(self): if self.stopped(): @@ -30,24 +42,39 @@ class TcpSession(Session): else: return self._connection - def stop(self): + def shutdown(self): + try: + self._connection.shutdown(socket.SHUT_RDWR) + except: + # print_log("problem shutting down", self.address) + # traceback.print_exc(file=sys.stdout) + pass + self._connection.close() - #print "Terminating connection:", self.address - with self.lock: - self._stopped = True def send_response(self, response): - data = json.dumps(response) + "\n" - # Possible race condition here by having session - # close connection? - # I assume Python connections are thread safe interfaces - try: - connection = self.connection() - while data: - l = connection.send(data) - data = data[l:] - except: - self.stop() + self.response_queue.put(response) + + +class TcpClientResponder(threading.Thread): + + def __init__(self, session): + self.session = session + threading.Thread.__init__(self) + + def run(self): + while not self.session.stopped(): + try: + response = self.session.response_queue.get(timeout=10) + except queue.Empty: + continue + data = json.dumps(response) + "\n" + try: + while data: + l = self.session.connection().send(data) + data = data[l:] + except: + self.session.stop() @@ -61,24 +88,25 @@ class TcpClientRequestor(threading.Thread): threading.Thread.__init__(self) def run(self): + try: + self.session.do_handshake() + except: + self.session.stop() + return + while not self.shared.stopped(): - if not self.update(): + + data = self.receive() + if not data: + self.session.stop() break + self.message += data self.session.time = time.time() while self.parse(): pass - def update(self): - data = self.receive() - if not data: - # close_session - self.session.stop() - return False - - self.message += data - return True def receive(self): try: @@ -93,7 +121,7 @@ class TcpClientRequestor(threading.Thread): raw_command = self.message[0:raw_buffer].strip() self.message = self.message[raw_buffer + 1:] - if raw_command == 'quit': + if raw_command == 'quit': self.session.stop() return False @@ -112,10 +140,13 @@ class TcpClientRequestor(threading.Thread): # Return an error JSON in response. self.dispatcher.push_response({"error": "syntax error", "request": raw_command}) else: - self.dispatcher.push_request(self.session,command) + self.dispatcher.push_request(self.session, command) + # sleep a bit to prevent a single session from DOSing the queue + time.sleep(0.01) return True + class TcpServer(threading.Thread): def __init__(self, dispatcher, host, port, use_ssl, ssl_certfile, ssl_keyfile): @@ -131,23 +162,33 @@ class TcpServer(threading.Thread): self.ssl_certfile = ssl_certfile def run(self): - if self.use_ssl: - print_log( "TCP/SSL server started.") - else: - print_log( "TCP server started.") + print_log( ("SSL" if self.use_ssl else "TCP") + " server started on port %d"%self.port) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.host, self.port)) - sock.listen(1) + sock.listen(5) + while not self.shared.stopped(): + + #if self.use_ssl: print_log("SSL: socket listening") try: - session = TcpSession(*sock.accept(), use_ssl=self.use_ssl, ssl_certfile=self.ssl_certfile, ssl_keyfile=self.ssl_keyfile) + connection, address = sock.accept() + except: + traceback.print_exc(file=sys.stdout) + time.sleep(0.1) + continue + + #if self.use_ssl: print_log("SSL: new session", address) + try: + session = TcpSession(self.dispatcher, connection, address, use_ssl=self.use_ssl, ssl_certfile=self.ssl_certfile, ssl_keyfile=self.ssl_keyfile) except BaseException, e: error = str(e) - print_log("cannot start TCP session", error) + print_log("cannot start TCP session", error, address) + connection.close() + time.sleep(0.1) continue - self.dispatcher.add_session(session) - self.dispatcher.collect_garbage() + client_req = TcpClientRequestor(self.dispatcher, session) client_req.start() - + responder = TcpClientResponder(session) + responder.start()