Changeset faea607 for fedd

Timestamp:

Dec 3, 2010 3:01:34 PM (14 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master

Children:

1d73342

Parents:

6fd2b29

Message:

Move thread pools into their own package. Starting on #10

Location:

fedd/federation

Files:

• experiment_control.py (modified) (11 diffs)
• thread_pool.py (added)

fedd/federation/experiment_control.py

@@ -32 +32 @@
 from experiment_partition import experiment_partition
 from authorizer import abac_authorizer
+from thread_pool import thread_pool, pooled_thread
 
 import topdl
@@ -81 +82 @@
     class ssh_cmd_timeout(RuntimeError): pass
     
-    class thread_pool:
-        """
-        A class to keep track of a set of threads all invoked for the same
-        task.  Manages the mutual exclusion of the states.
-        """
-        def __init__(self, nthreads):
-            """
-            Start a pool.
-            """
-            self.changed = Condition()
-            self.started = 0
-            self.terminated = 0
-            self.nthreads = nthreads
-
-        def acquire(self):
-            """
-            Get the pool's lock.
-            """
-            self.changed.acquire()
-
-        def release(self):
-            """
-            Release the pool's lock.
-            """
-            self.changed.release()
-
-        def wait(self, timeout = None):
-            """
-            Wait for a pool thread to start or stop.
-            """
-            self.changed.wait(timeout)
-
-        def start(self):
-            """
-            Called by a pool thread to report starting.
-            """
-            self.changed.acquire()
-            self.started += 1
-            self.changed.notifyAll()
-            self.changed.release()
-
-        def terminate(self):
-            """
-            Called by a pool thread to report finishing.
-            """
-            self.changed.acquire()
-            self.terminated += 1
-            self.changed.notifyAll()
-            self.changed.release()
-
-        def clear(self):
-            """
-            Clear all pool data.
-            """
-            self.changed.acquire()
-            self.started = 0
-            self.terminated =0
-            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, timeout=None):
-            """
-            Wait until all active threads finish (and at least one has
-            started).  If a timeout is given, return after waiting that long
-            for termination.  If all threads are done (and one has started in
-            the since the last clear()) return True, otherwise False.
-            """
-            if timeout:
-                deadline = time.time() + timeout
-            self.acquire()
-            while self.started == 0 or self.started > self.terminated:
-                self.wait(timeout)
-                if timeout:
-                    if time.time() > deadline:
-                        break
-                    timeout = deadline - time.time()
-            self.release()
-            return not (self.started == 0 or self.started > self.terminated)
-
-    class pooled_thread(Thread):
-        """
-        One of a set of threads dedicated to a specific task.  Uses the
-        thread_pool class above for coordination.
-        """
-        def __init__(self, group=None, target=None, name=None, args=(), 
-                kwargs={}, pdata=None, trace_file=None):
-            Thread.__init__(self, group, target, name, args, kwargs)
-            self.rv = None          # Return value of the ops in this thread
-            self.exception = None   # Exception that terminated this thread
-            self.target=target      # Target function to run on start()
-            self.args = args        # Args to pass to target
-            self.kwargs = kwargs    # Additional kw args
-            self.pdata = pdata      # thread_pool for this class
-            # Logger for this thread
-            self.log = logging.getLogger("fedd.experiment_control")
-        
-        def run(self):
-            """
-            Emulate Thread.run, except add pool data manipulation and error
-            logging.
-            """
-            if self.pdata:
-                self.pdata.start()
-
-            if self.target:
-                try:
-                    self.rv = self.target(*self.args, **self.kwargs)
-                except service_error, s:
-                    self.exception = s
-                    self.log.error("Thread exception: %s %s" % \
-                            (s.code_string(), s.desc))
-                except:
-                    self.exception = sys.exc_info()[1]
-                    self.log.error(("Unexpected thread exception: %s" +\
-                            "Trace %s") % (self.exception,\
-                                traceback.format_exc()))
-            if self.pdata:
-                self.pdata.terminate()
-
     call_RequestAccess = service_caller('RequestAccess')
     call_ReleaseAccess = service_caller('ReleaseAccess')
@@ -234 +107 @@
             return rv
 
-        self.thread_with_rv = experiment_control_local.pooled_thread
-        self.thread_pool = experiment_control_local.thread_pool
         self.list_log = list_log.list_log
 
@@ -1141 +1012 @@
         log = alloc_log or self.log
 
-        thread_pool = self.thread_pool(self.nthreads)
+        tp = thread_pool(self.nthreads)
         threads = [ ]
         starters = [ ]
@@ -1156 +1027 @@
             # to get the return value later
             tb_attrs = copy.copy(attrs)
-            thread_pool.wait_for_slot()
+            tp.wait_for_slot()
             uri = tbparams[tb].get('uri', tbmap.get(testbed_base(tb), None))
             base, suffix = split_testbed(tb)
@@ -1181 +1052 @@
                     log_collector=log_collector)
             starters.append(s)
-            t  = self.pooled_thread(\
+            t  = pooled_thread(\
                     target=s, name=tb,
                     args=(uri, aid, topo[tb], masters, tb_attrs, connInfo[tb]),
-                    pdata=thread_pool, trace_file=self.trace_file)
+                    pdata=tp, trace_file=self.trace_file)
             threads.append(t)
             t.start()
@@ -1191 +1062 @@
         mins = 0
         revoked = False
-        while not thread_pool.wait_for_all_done(60.0):
+        while not tp.wait_for_all_done(60.0):
             mins += 1
             alloc_log.info("Waiting for sub threads (it has been %d mins)" \
@@ -1214 +1085 @@
         if failed:
             if not fail_soft:
-                thread_pool.clear()
+                tp.clear()
                 for tb in succeeded:
                     # Create and start a thread to stop the segment
-                    thread_pool.wait_for_slot()
+                    tp.wait_for_slot()
                     uri = tbparams[tb]['uri']
-                    t  = self.pooled_thread(\
+                    t  = pooled_thread(\
                             target=self.terminate_segment(log=log,
                                 testbed=tb,
@@ -1228 +1099 @@
                             args=(uri, tbparams[tb]['federant']['allocID']),
                             name=tb,
-                            pdata=thread_pool, trace_file=self.trace_file)
+                            pdata=tp, trace_file=self.trace_file)
                     t.start()
                 # Wait until all finish (if any are being stopped)
                 if succeeded:
-                    thread_pool.wait_for_all_done()
+                    tp.wait_for_all_done()
 
                 # release the allocations
@@ -2508 +2379 @@
                 # no tbparams, no start.
                 if len(tbparams) > 0:
-                    thread_pool = self.thread_pool(self.nthreads)
+                    tp = thread_pool(self.nthreads)
                     for k in tbparams.keys():
                         # Create and start a thread to stop the segment
-                        thread_pool.wait_for_slot()
+                        tp.wait_for_slot()
                         uri, aid = tbparams[k]
-                        t  = self.pooled_thread(\
+                        t  = pooled_thread(\
                                 target=self.terminate_segment(log=dealloc_log,
                                     testbed=uri,
@@ -2521 +2392 @@
                                     caller=self.call_TerminateSegment),
                                 args=(uri, aid), name=k,
-                                pdata=thread_pool, trace_file=self.trace_file)
+                                pdata=tp, trace_file=self.trace_file)
                         t.start()
                     # Wait for completions
-                    thread_pool.wait_for_all_done()
+                    tp.wait_for_all_done()
 
                 # release the allocations (failed experiments have done this