Changeset 429

Timestamp:

08/03/2011 05:44:27 PM (12 years ago)

Author:

marcus

Message:

inter-thread communication tests with 140 parallel threads

File:

• SHX/trunk/sandbox/pubsubthread.py (modified) (5 diffs)

SHX/trunk/sandbox/pubsubthread.py

@@ -9 +9 @@
 import wx
 from random import randint
-from SeismicHandler.utils.pubsub import pub as msgs
+from SeismicHandler.utils.pubsub import pub
 from SeismicHandler.utils.wxAnyThread import anythread
+
+#threadinput = ['one', 'two', 'three', 'four', 'five', 'six', 'seven']
+#threadinput = map(chr, range(65, 91) + range(97, 123))
+#threadinput = map(chr, range(65, 91))
+threadinput = map(chr, range(97, 103) + range(65, 71))
+threadnames = []
+for i in threadinput:
+    for j in threadinput:
+        threadnames.append(i+j)
 
 class thready(threading.Thread):
@@ -21 +30 @@
         self.name = "thread.%s" % name
         self.main = main
+        self.stopped = True
 
-        msgs.subscribe(self.monitor, "thread.%s" % name)
+        pub.subscribe(self.monitor, "thread.%s" % name)
 
     def run(self):
@@ -38 +48 @@
             # random update of main threads status bar
             if randint(0, 20) == 10:
-                msgs.sendMessage("ui.change.statustext", text=self.name)
-                msgs.sendMessage(self.name, content="self called")
+                pub.sendMessage("ui.change.statustext", text=self.name)
+                pub.sendMessage(self.name, content="self called")
 
             time.sleep(.1)
 
     def monitor(self, content):
+        if self.stopped:
+            return
+
         print time.strftime("%H:%M:%S", time.localtime()), self.name, \
                                                            "received:", content
-        if not self.name.endswith("two"):
-            msgs.sendMessage("thread.two", content="test from "+self.name)
+
+#        return
+        call = threadnames[randint(1, len(threadnames))-1]
+        if not self.name.endswith(call) and randint(0, 1):
+            try:
+                pub.sendMessage("thread.%s" % call,
+                                         content="via monitor from "+self.name)
+            except RuntimeError, e:
+                print e
+                self.stop()
 
     def stop(self):
+        print time.strftime("%H:%M:%S", time.localtime()), self.name, "got stop"
         self.stopped = True
+        pub.unsubscribe(self.monitor, self.name)
 
 
@@ -62 +85 @@
         self.Bind(wx.EVT_CLOSE, self.OnQuit)
 
-        tlist = ['one', 'two', 'three', 'four', 'five', 'six', 'seven']
+        threads = {}
+        for name in threadnames:
+            threads[name] = thready(name, self)
+            print time.strftime("%H:%M:%S"), "start", name
+            threads[name].start()
+        self.threads = threads
 
-        for i, name in enumerate(tlist):
-            tlist[i] = thready(name, self)
-            print time.strftime("%H:%M:%S"), "start", name
-            tlist[i].start()
-        self.tlist = tlist
-
-        msgs.subscribe(self.ListenUpdateStatusText, "ui.change.statustext")
+        pub.subscribe(self.ListenUpdateStatusText, "ui.change.statustext")
 
         self.Show()
 
     def OnQuit(self, evt):
-        for t in self.tlist:
-            print time.strftime("%H:%M:%S"), "stopping", t.name
-            t.stop()
-            t.join()
+        for name in self.threads:
+            print time.strftime("%H:%M:%S"), "stopping", name
+            if self.threads[name].isAlive():
+                self.threads[name].stop()
+
+        final = False
+        threads = self.threads.values()
+        tries = 0
+        while not final:
+            final = True
+            
+            for i, t in enumerate(threads[:]):
+                print time.strftime("%H:%M:%S"), "joining", t.name
+                if t.isAlive():
+                    t.join(1.1)
+                    
+                    if not t.isAlive():
+                        del threads[i]
+                        
+            if threads and tries < 20:
+                print "%u threads left" % len(threads)
+                time.sleep(1)
+                tries += 1
+            else:
+                final = True
+                
+            print tries
         evt.Skip()
 
@@ -96 +141 @@
 
 
-def monitorGlobal(topic=msgs.AUTO_TOPIC, **kwargs):
-    print time.strftime("%H:%M:%S", time.localtime()), "monitorGlobal", topic, kwargs
-msgs.subscribe(monitorGlobal, msgs.ALL_TOPICS)
+def monitorGlobal(topic=pub.AUTO_TOPIC, **kwargs):
+    print time.strftime("%H:%M:%S", time.localtime()), "monitorGlobal", \
+                                                                  topic, kwargs
+pub.subscribe(monitorGlobal, pub.ALL_TOPICS)