Make UPnP into a thread

- UPnP is now a separate thread that will continue to setup UPnP
- shutdown waits for threads that shutdown correctly (Addresses
Bitmessage#549)

src/bitmessagemain.py

@@ -149,9 +149,6 @@ class Main:
         # is the application already running?  If yes then exit.
         thisapp = singleton.singleinstance("", daemon)
 
-        import upnp
-        upnp.createPortMapping()
-
         # get curses flag
         curses = False
         if '-c' in sys.argv:
@@ -211,6 +208,11 @@ class Main:
         singleListenerThread.daemon = True  # close the main program even if there are threads left
         singleListenerThread.start()
         
+        if shared.safeConfigGetBoolean('bitmessagesettings','upnp'):
+            import upnp
+            upnpThread = upnp.uPnPThread()
+            upnpThread.start()
+
         if daemon == False and shared.safeConfigGetBoolean('bitmessagesettings', 'daemon') == False:
             if curses == False:
                 if not depends.check_pyqt():

src/shared.py

@@ -384,8 +384,6 @@ def doCleanShutdown():
         'Flushing inventory in memory out to disk. This should normally only take a second...'))
     flushInventory()
     
-    import upnp
-    upnp.deletePortMapping()
     # Verify that the objectProcessor has finished exiting. It should have incremented the 
     # shutdown variable from 1 to 2. This must finish before we command the sqlThread to exit.
     while shutdown == 1:
@@ -401,6 +399,11 @@ def doCleanShutdown():
     # shutdown variable and exit. If the main thread closes before they do then they won't stop.
     time.sleep(.25)
     
+    for thread in threading.enumerate():
+        if thread.name == "uPnPThread" or "outgoingSynSender" in thread.name:
+            logger.debug("Waiting for thread %s", thread.name)
+            thread.join()
+
     if safeConfigGetBoolean('bitmessagesettings','daemon'):
         logger.info('Clean shutdown complete.')
         os._exit(0)

src/upnp.py

@@ -1,45 +1,10 @@
 # A simple upnp module to forward port for BitMessage
 # Reference: http://mattscodecave.com/posts/using-python-and-upnp-to-forward-a-port
 import socket
-import httplib
+from struct import unpack, pack
-from shared import config
+import threading
-
+import time
-routers = []
+import shared
-
-def searchRouter():
-    SSDP_ADDR = "239.255.255.250"
-    SSDP_PORT = 1900
-    SSDP_MX = 2
-    SSDP_ST = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
-
-    ssdpRequest = "M-SEARCH * HTTP/1.1\r\n" + \
-                    "HOST: %s:%d\r\n" % (SSDP_ADDR, SSDP_PORT) + \
-                    "MAN: \"ssdp:discover\"\r\n" + \
-                    "MX: %d\r\n" % (SSDP_MX, ) + \
-                    "ST: %s\r\n" % (SSDP_ST, ) + "\r\n"
-
-    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    sock.sendto(ssdpRequest, (SSDP_ADDR, SSDP_PORT))
-    routers = []
-    sock.settimeout(0.5)
-    try:
-        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
-        sock.settimeout(2)
-        logger.debug("Sending UPnP query")
-        sock.sendto(ssdpRequest, (SSDP_ADDR, SSDP_PORT))
-    except:
-        logger.exception("UPnP sock failed")
-    try:
-        while True:
-            resp,(ip,port) = sock.recvfrom(1000)
-            if resp is None:
-                continue
-            routers.append(Router(resp, ip))
-            resp,(ip,port) = sock.recvfrom(1000)
-    except:pass
-        
-    return routers
 
 def createRequestXML(service, action, arguments=[]):
     from xml.dom.minidom import Document
@@ -98,6 +63,8 @@ class Router:
     path = ""
     address = None
     routerPath = None
+    extPort = None
+    
     def __init__(self, ssdpResponse, address):
         import urllib2
         from xml.dom.minidom import parseString
@@ -143,6 +110,7 @@ class Router:
             pass
 
     def AddPortMapping(self, externalPort, internalPort, internalClient, protocol, description, leaseDuration = 0, enabled = 1):
+        from debug import logger
         resp = self.soapRequest('WANIPConnection:1', 'AddPortMapping', [
                 ('NewExternalPort', str(externalPort)),
                 ('NewProtocol', protocol),
@@ -153,6 +121,7 @@ class Router:
                 ('NewLeaseDuration', str(leaseDuration))
             ])
         self.extPort = externalPort
+        logger.info("Successfully established UPnP mapping for %s:%i on external port %i", internalClient, internalPort, externalPort)
         return resp
 
     def DeletePortMapping(self, externalPort, protocol):
@@ -191,38 +160,90 @@ class Router:
             raise UPnPError(errinfo[0].childNodes[0].data) 
         return resp
 
-def createPortMappingInternal(router):
+class uPnPThread(threading.Thread):
+    def __init__ (self):
+        threading.Thread.__init__(self, name="uPnPThread")
+        self.localPort = shared.config.getint('bitmessagesettings', 'port')
+        self.extPort = None
+        self.routers = []
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
+        self.sock.settimeout(10)
+        self.sendSleep = 60
+
+    def run(self):
         from debug import logger
         
-    for i in range(0, 50):
+        logger.debug("Starting UPnP thread")
+        lastSent = 0
+        while shared.shutdown == 0:
+            if time.time() - lastSent > self.sendSleep and len(self.routers) == 0:
+                self.sendSearchRouter()
+                lastSent = time.time()
+            try:
+                while shared.shutdown == 0:
+                    resp,(ip,port) = self.sock.recvfrom(1000)
+                    if resp is None:
+                        continue
+                    newRouter = Router(resp, ip)
+                    for router in self.routers:
+                        if router.location == newRouter.location:
+                            break
+                    else:
+                        logger.debug("Found UPnP router at %s", ip)
+                        self.routers.append(newRouter)
+                        self.createPortMapping(newRouter)
+                        break
+            except socket.timeout as e:
+                pass
+            except:
+                logger.error("Failure running UPnP router search.", exc_info=True)
+            for router in self.routers:
+                if router.extPort is None:
+                    self.createPortMapping(router)
+        for router in self.routers:
+            if router.extPort is not None:
+                self.deletePortMapping(router)
+        logger.debug("UPnP thread done")
+
+    def sendSearchRouter(self):
+        from debug import logger
+        SSDP_ADDR = "239.255.255.250"
+        SSDP_PORT = 1900
+        SSDP_MX = 2
+        SSDP_ST = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
+        ssdpRequest = "M-SEARCH * HTTP/1.1\r\n" + \
+                    "HOST: %s:%d\r\n" % (SSDP_ADDR, SSDP_PORT) + \
+                    "MAN: \"ssdp:discover\"\r\n" + \
+                    "MX: %d\r\n" % (SSDP_MX, ) + \
+                    "ST: %s\r\n" % (SSDP_ST, ) + "\r\n"
+
+        try:
+            logger.debug("Sending UPnP query")
+            self.sock.sendto(ssdpRequest, (SSDP_ADDR, SSDP_PORT))
+        except:
+            logger.exception("UPnP send query failed")
+
+    def createPortMapping(self, router):
+        from debug import logger
+
+        for i in range(50):
             try:
                 routerIP, = unpack('>I', socket.inet_aton(router.address))
                 localIP = router.localAddress
-            localPort = shared.config.getint('bitmessagesettings', 'port')
                 if i == 0:
-                extPort = localPort # try same port first
+                    extPort = self.localPort # try same port first
                 else:
                     extPort = randint(32767, 65535)
-            logger.debug("Requesting UPnP mapping for %s:%i on external port %i", localIP, localPort,  extPort)
+                logger.debug("Requesting UPnP mapping for %s:%i on external port %i", localIP, self.localPort,  extPort)
-            router.AddPortMapping(extPort, localPort, localIP, 'TCP', 'BitMessage')
+                router.AddPortMapping(extPort, self.localPort, localIP, 'TCP', 'BitMessage')
-            logger.info("Successfully established UPnP mapping for %s:%i on external port %i", localIP, localPort, extPort)
                 shared.extPort = extPort
                 break
             except UPnPError:
                 logger.debug("UPnP error: ", exc_info=True)
 
-def createPortMapping():
+    def deletePortMapping(self, router):
-    from debug import logger
-    global routers
-
-    routers = searchRouter()
-    logger.debug("Found %i UPnP routers", len(routers))
-
-    for router in routers:
-        createPortMappingInternal(router)
-
-def deletePortMapping():
-    localPort = config.getint('bitmessagesettings', 'port')
-    for router in routers:
-        if hasattr(router, "extPort"):
         router.DeletePortMapping(router.extPort, 'TCP')
+
+
+