Add lock to protect modifying connections dictionary in weioJSPYHandl…

…er() while iterating through it in weioRunner.py listener_thread()
nodesign · May 4, 2015 · 415a3cc · 415a3cc
1 parent 04e1dcb
commit 415a3cc
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 36 deletions.
diff --git a/handlers/weioJSPYHandler.py b/handlers/weioJSPYHandler.py
@@ -68,9 +68,14 @@ def __init__(self, *args, **kwargs):
 
     def on_open(self, request):
         # Add the connection to the connections dictionary
-        connUuid = uuid.uuid4()
-        weioRunnerGlobals.weioConnections[connUuid] = self
-        weioRunnerGlobals.weioConnUuids.append(connUuid)
+        # Do not forget the lock - this handler is called asynchronoisly,
+        # and the weioRunnerGlobals.weioConnections[] is accessed in the same time by
+        # weioRunner.py. This can lead to iteritems() in weioRunner while dictionary is modified,
+        # which will lead to exception in weioRunner.py listener_thread()
+        with weioRunnerGlobals.lockConn:
+            connUuid = uuid.uuid4()
+            weioRunnerGlobals.weioConnections[connUuid] = self
+            weioRunnerGlobals.weioConnUuids.append(connUuid)
 
         # collect client ip address and user machine info
         # print self.request to see all available info on user connection
@@ -135,14 +140,15 @@ def serve(self, data) :
 
 
     def on_close(self):
-        self.connection_closed = True
-        print "*SYSOUT* Client with IP address: " + self.ip + " disconnected from server"
-        # Remove client from the clients list and broadcast leave message
-        #self.connections.remove(self)
-        for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
-            if (conn == self):
-                weioRunnerGlobals.weioConnections.pop(connUuid)
-                weioRunnerGlobals.weioConnUuids.remove(connUuid)
+        with weioRunnerGlobals.lockConn:
+            self.connection_closed = True
+            print "*SYSOUT* Client with IP address: " + self.ip + " disconnected from server"
+            # Remove client from the clients list and broadcast leave message
+            #self.connections.remove(self)
+            for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
+                if (conn == self):
+                    weioRunnerGlobals.weioConnections.pop(connUuid)
+                    weioRunnerGlobals.weioConnUuids.remove(connUuid)
 
 
 ###
@@ -166,7 +172,7 @@ def on_message(self, data):
             """Parsing JSON data that is comming from browser into python object"""
             self.serve(json.loads(data))
 
-    def serve(self, data) :
+    def serve(self, data):
         # Create userAgentMessage and send it to the launcher process
         msg = weioRunnerGlobals.userAgentMessage()
         #print "DATA: ", data
@@ -185,8 +191,9 @@ def serve(self, data) :
             weioRunnerGlobals.QOUT.put(msg)
 
     def on_close(self):
-        self.remoteConn = None
+        with weioRunnerGlobals.lockConn:
+            self.remoteConn = None
 
-        # Remove client from the clients list and broadcast leave message
-        weioRunnerGlobals.weioConnections.pop(self.connUuid)
-        self.connUuid = None
+            # Remove client from the clients list and broadcast leave message
+            weioRunnerGlobals.weioConnections.pop(self.connUuid)
+            self.connUuid = None
diff --git a/weioLib/weioRunnerGlobals.py b/weioLib/weioRunnerGlobals.py
@@ -62,6 +62,11 @@
 # Initialize with multiprocessing.Manager().list in weioRunner
 weioConnUuids = None
 
+# Lock the connections dictionary not to mutate it during the iterations
+# I.e. prevent weioJSPYHandler to mdify it while weioRunner listener_thread()
+# is iterating it
+locakConn = None
+
 # Running flag
 running = multiprocessing.Value(ctypes.c_bool, False)
 

diff --git a/weioRunner.py b/weioRunner.py
@@ -106,7 +106,7 @@ def get(self):
             path = "www/signin.html"
         else :
             path = "www/userIndex.html"
-            
+
             #if (weioFiles.checkIfFileExists(confFile['last_opened_project'] + "/index.html")):
             #    path = "www/userIndex.html"
 
@@ -229,21 +229,21 @@ def userPlayer(self, fd, events):
 
     def launcher(self):
         #print "======>>> LAUNCHING..."
-        
-        # get configuration from file       
+
+        # get configuration from file
         confFile = weioConfig.getConfiguration()
         # get location of last opened project
         lp = confFile["last_opened_project"]
-        
+
          # check if main.py exists in current user project
         if (weioFiles.checkIfFileExists(lp+"/main.py")):
             # set the location of current project main.py
             projectModule = lp.replace('/', '.') + ".main"
-            
+
         else :
             # Use the location of default www/defaultMain/main.py
             projectModule = "www.defaultMain.main"
-              
+
         #print "CALL", projectModule
         # Init GPIO object for uper communication
         if (weioRunnerGlobals.WEIO_SERIAL_LINKED == False):
@@ -328,21 +328,27 @@ def listenerThread():
                 result = {}
 
                 if (msg.callbackJS is not None):
-                    result["serverPush"] = msg.callbackJS
-                    result["data"] = msg.res
-                    #print "RESULT",result
-                    if (weioRunnerGlobals.remoteConnected.value == True):
-                        if (msg.connUuid == "all"):
-                            for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
-                                weioRunnerGlobals.weioConnections[connUuid].send(json.dumps(result))
-                        else:
-                            weioRunnerGlobals.weioConnections[msg.connUuid].write_message(json.dumps(result))
-                    else:
-                        if (msg.connUuid == "all"):
-                            for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
-                                weioRunnerGlobals.weioConnections[connUuid].send(json.dumps(result))
+                    weioRunnerGlobals.lockConn.acquire()
+                    try:
+                        result["serverPush"] = msg.callbackJS
+                        result["data"] = msg.res
+                        #print "RESULT",result
+                        if (weioRunnerGlobals.remoteConnected.value == True):
+                            if (msg.connUuid == "all"):
+                                for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
+                                    weioRunnerGlobals.weioConnections[connUuid].send(json.dumps(result))
+                            else:
+                                weioRunnerGlobals.weioConnections[msg.connUuid].write_message(json.dumps(result))
                         else:
-                            weioRunnerGlobals.weioConnections[msg.connUuid].send(json.dumps(result))
+                            if (msg.connUuid == "all"):
+                                for connUuid, conn in weioRunnerGlobals.weioConnections.iteritems():
+                                    weioRunnerGlobals.weioConnections[connUuid].send(json.dumps(result))
+                            else:
+                                weioRunnerGlobals.weioConnections[msg.connUuid].send(json.dumps(result))
+                    except:
+                        print "FAILED!"
+                    finally:
+                        weioRunnerGlobals.lockConn.release()
 
 class WeioRemote():
     conn = None
@@ -460,6 +466,9 @@ def close(self):
     signal.signal(signal.SIGTERM, signalCallback)
     signal.signal(signal.SIGINT, signalCallback)
 
+    # Prepare lock for the thread
+    weioRunnerGlobals.lockConn = multiprocessing.Lock()
+
     # Start listener thread
     t = threading.Thread(target=listenerThread)
     t.daemon = True