Allow multiple ReceiveQueue threads

- defaults to 3

src/bitmessagemain.py

@@ -267,7 +267,8 @@ class Main:
             asyncoreThread = BMNetworkThread()
             asyncoreThread.daemon = True
             asyncoreThread.start()
-            receiveQueueThread = ReceiveQueueThread()
+            for i in range(BMConfigParser().getint("threads", "receive")):
+                receiveQueueThread = ReceiveQueueThread(i)
                 receiveQueueThread.daemon = True
                 receiveQueueThread.start()
             announceThread = AnnounceThread()

src/bmconfigparser.py

@@ -15,6 +15,9 @@ BMConfigDefaults = {
         "maxtotalconnections": 200,
         "maxuploadrate": 0,
     },
+    "threads": {
+        "receive": 3,
+    },
     "network": {
         "asyncore": True,
         "bind": None,

src/helper_threading.py

@@ -1,4 +1,6 @@
+from contextlib import contextmanager
 import threading
+
 try:
     import prctl
     def set_thread_name(name): prctl.set_name(name)
@@ -20,3 +22,16 @@ class StoppableThread(object):
     def stopThread(self):
         self._stopped = True
         self.stop.set()
+
+class BusyError(threading.ThreadError):
+    pass
+
+@contextmanager
+def nonBlocking(lock):
+    locked = lock.acquire(False)
+    if not locked:
+        raise BusyError
+    try:
+        yield
+    finally:
+        lock.release()

src/network/advanceddispatcher.py

@@ -6,6 +6,7 @@ import time
 
 import asyncore_pollchoose as asyncore
 from debug import logger
+from helper_threading import BusyError, nonBlocking
 
 class AdvancedDispatcher(asyncore.dispatcher):
     _buf_len = 2097152 # 2MB
@@ -22,6 +23,7 @@ class AdvancedDispatcher(asyncore.dispatcher):
         self.expectBytes = 0
         self.readLock = threading.RLock()
         self.writeLock = threading.RLock()
+        self.processingLock = threading.RLock()
 
     def append_write_buf(self, data):
         if data:
@@ -43,6 +45,7 @@ class AdvancedDispatcher(asyncore.dispatcher):
             return False
         while len(self.read_buf) >= self.expectBytes:
             try:
+                with nonBlocking(self.processingLock):
                     if getattr(self, "state_" + str(self.state))() is False:
                         break
             except AttributeError:

src/network/receivequeuethread.py

@@ -15,17 +15,16 @@ import protocol
 import state
 
 class ReceiveQueueThread(threading.Thread, StoppableThread):
-    def __init__(self):
+    def __init__(self, num=0):
-        threading.Thread.__init__(self, name="ReceiveQueueThread")
+        threading.Thread.__init__(self, name="ReceiveQueue_%i" %(num))
         self.initStop()
-        self.name = "ReceiveQueueThread"
+        self.name = "ReceiveQueue_%i" % (num)
-        logger.info("init receive queue thread")
+        logger.info("init receive queue thread %i", num)
 
     def run(self):
         while not self._stopped and state.shutdown == 0:
             try:
                 dest = receiveDataQueue.get(block=True, timeout=1)
-                receiveDataQueue.task_done()
             except Queue.Empty:
                 continue
 
@@ -44,3 +43,4 @@ class ReceiveQueueThread(threading.Thread, StoppableThread):
             # AttributeError = state isn't implemented
             except (KeyError, AttributeError):
                 pass
+            receiveDataQueue.task_done()