Index: fedd/federation/experiment_control.py
===================================================================
--- fedd/federation/experiment_control.py	(revision 696a68980dd7bfda4b92aaf2233a53ad10948d48)
+++ fedd/federation/experiment_control.py	(revision faea6075d7ea0e3bb7fb580a4381f8557abe1a3e)
@@ -32,4 +32,5 @@
 from experiment_partition import experiment_partition
 from authorizer import abac_authorizer
+from thread_pool import thread_pool, pooled_thread
 
 import topdl
@@ -81,132 +82,4 @@
     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,6 +107,4 @@
             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,5 +1012,5 @@
         log = alloc_log or self.log
 
-        thread_pool = self.thread_pool(self.nthreads)
+        tp = thread_pool(self.nthreads)
         threads = [ ]
 	starters = [ ]
@@ -1156,5 +1027,5 @@
 	    # 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,8 +1052,8 @@
 		    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,5 +1062,5 @@
 	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,10 +1085,10 @@
         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,9 +1099,9 @@
                             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,10 +2379,10 @@
 		# 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,8 +2392,8 @@
 				    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
