Changeset 8bc5754 for fedd/federation

Timestamp:

Jan 7, 2009 10:50:21 AM (16 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-1.30, version-2.00, version-3.01, version-3.02

Children:

b0c185f

Parents:

f9665d1

Message:

Parallelize testbed swapouts, including failed creations and terminate calls.
I also cleaned up the thread_pool interface some to aviod cutting and pasting
rather cryptic code throughout those functions. The interface is still a
little odd, but much more palatable.

File:

• fedd/federation/experiment_control.py (modified) (6 diffs)

fedd/federation/experiment_control.py

@@ -44 +44 @@
         task.  Manages the mutual exclusion of the states.
         """
-        def __init__(self):
+        def __init__(self, nthreads):
             """
             Start a pool.
@@ -51 +51 @@
             self.started = 0
             self.terminated = 0
+            self.nthreads = nthreads
 
         def acquire(self):
@@ -97 +98 @@
             self.changed.notifyAll()
             self.changed.release()
+
+        def wait_for_slot(self):
+            """
+            Wait until we have a free slot to start another pooled thread
+            """
+            self.acquire()
+            while self.started - self.terminated >= self.nthreads:
+                self.wait()
+            self.release()
+
+        def wait_for_all_done(self):
+            """
+            Wait until all active threads finish (and at least one has started)
+            """
+            self.acquire()
+            while self.started == 0 or self.started > self.terminated:
+                self.wait()
+            self.release()
 
     class pooled_thread(Thread):
@@ -1700 +1719 @@
             raise e
 
-        thread_pool_info = self.thread_pool()
+        thread_pool = self.thread_pool(self.nthreads)
         threads = [ ]
 
         for tb in [ k for k in allocated.keys() if k != master]:
-            # Wait until we have a free slot to start the next testbed load
-            thread_pool_info.acquire()
-            while thread_pool_info.started - \
-                    thread_pool_info.terminated >= self.nthreads:
-                thread_pool_info.wait()
-            thread_pool_info.release()
-
             # Create and start a thread to start the segment, and save it to
             # get the return value later
+            thread_pool.wait_for_slot()
             t  = self.pooled_thread(target=self.start_segment, 
                     args=(tb, eid, tbparams, tmpdir, 0), name=tb,
-                    pdata=thread_pool_info, trace_file=self.trace_file)
+                    pdata=thread_pool, trace_file=self.trace_file)
             threads.append(t)
             t.start()
 
-        # Wait until all finish (the first clause of the while is to make sure
-        # one starts)
-        thread_pool_info.acquire()
-        while thread_pool_info.started == 0 or \
-                thread_pool_info.started > thread_pool_info.terminated:
-            thread_pool_info.wait()
-        thread_pool_info.release()
+        # Wait until all finish 
+        thread_pool.wait_for_all_done()
 
         # If none failed, start the master
@@ -1738 +1746 @@
         if failed:
             if not fail_soft:
+                thread_pool.clear()
                 for tb in succeeded:
-                    self.stop_segment(tb, eid, tbparams)
+                    # Create and start a thread to stop the segment
+                    thread_pool.wait_for_slot()
+                    t  = self.pooled_thread(target=self.stop_segment, 
+                            args=(tb, eid, tbparams), name=tb,
+                            pdata=thread_pool, trace_file=self.trace_file)
+                    t.start()
+                # Wait until all finish
+                thread_pool.wait_for_all_done()
+
                 # release the allocations
                 for tb in tbparams.keys():
@@ -2031 +2048 @@
 
             # Stop everyone.
+            thread_pool = self.thread_pool(self.nthreads)
             for tb in tbparams.keys():
-                self.stop_segment(tb, tbparams[tb]['eid'], tbparams)
+                # Create and start a thread to stop the segment
+                thread_pool.wait_for_slot()
+                t  = self.pooled_thread(target=self.stop_segment, 
+                        args=(tb, tbparams[tb]['eid'], tbparams), name=tb,
+                        pdata=thread_pool, trace_file=self.trace_file)
+                t.start()
+            # Wait for completions
+            thread_pool.wait_for_all_done()
 
             # release the allocations