use set instead of list in memorypool_update

[electrum-server.git] / transports / stratum_tcp.py

diff --git a/transports/stratum_tcp.py b/transports/stratum_tcp.py
index bc36a9e..aad4c17 100644 (file)
--- a/transports/stratum_tcp.py
+++ b/transports/stratum_tcp.py
@@ -3,6 +3,7 @@ import Queue as queue
 import socket
 import threading
 import time
+import traceback, sys
 
 from processor import Session, Dispatcher
 from utils import print_log
@@ -10,8 +11,9 @@ from utils import 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(
@@ -19,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.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():
@@ -32,24 +42,40 @@ 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()
+
 
 
 class TcpClientRequestor(threading.Thread):
@@ -62,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:
@@ -114,6 +141,8 @@ class TcpClientRequestor(threading.Thread):
             self.dispatcher.push_response({"error": "syntax error", "request": raw_command})
         else:
             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
 
@@ -133,22 +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()