Changeset 1af38d6 for fedd

Timestamp:

Aug 21, 2008 6:12:35 PM (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:

3441fe3

Parents:

6546868

Message:

multithreading for startup. Needs more testing

File:

• fedd/fedd_create_experiment.py (modified) (7 diffs)

fedd/fedd_create_experiment.py

@@ -16 +16 @@
 import tempfile
 import copy
+
+from threading import *
 
 from subprocess import *
@@ -35 +37 @@
             debug=False,
             muxmax=2,
+            nthreads=2,
             project_user = "faber",
             scp_exec="/usr/bin/scp",
@@ -50 +53 @@
             ):
         self.scripts = fedd_create_experiment_local.scripts
+        self.thread_with_rv = fedd_create_experiment_local.pooled_thread
+        self.thread_pool = fedd_create_experiment_local.thread_pool
 
         self.cert_file = cert_file
@@ -55 +60 @@
         self.debug = debug
         self.muxmax = muxmax
+        self.nthreads = nthreads
         self.project_user = project_user
         self.scp_exec = scp_exec
@@ -93 +99 @@
                 raise service_error(service_error.server_config,
                         "%s/%s not in local script dir" % (self.scripts_dir, s))
+    class thread_pool:
+        def __init__(self):
+            self.changed = Condition()
+            self.started = 0
+            self.terminated = 0
+
+        def acquire(self):
+            self.changed.acquire()
+
+        def release(self):
+            self.changed.release()
+
+        def wait(self, timeout = None):
+            self.changed.wait(timeout)
+
+        def start(self):
+            self.changed.acquire()
+            self.started += 1
+            self.changed.notifyAll()
+            self.changed.release()
+
+        def terminate(self):
+            self.changed.acquire()
+            self.terminated += 1
+            self.changed.notifyAll()
+            self.changed.release()
+
+        def clear(self):
+            self.changed.acquire()
+            self.started = 0
+            self.terminated =0
+            self.changed.notifyAll()
+            self.changed.release()
+
+
+
+    class pooled_thread(Thread):
+        def __init__(self, group=None, target=None, name=None, args=(), 
+                kwargs={}, pdata=None):
+            Thread.__init__(self, group, target, name, args, kwargs)
+            self.rv = None
+            self.exception = None
+            self.target=target
+            self.args = args
+            self.kwargs = kwargs
+            self.pdata = pdata
+        
+        def run(self):
+            if self.pdata:
+                self.pdata.start()
+
+            if self.target:
+                try:
+                    self.rv = self.target(*self.args, **self.kwargs)
+                except e:
+                    self.exception = e
+
+            print "%s done: %s" % (self.getName(), self.rv)
+            if self.pdata:
+                self.pdata.terminate()
 
     def copy_file(self, src, dest, size=1024):
@@ -686 +752 @@
                 active_end, tbparams, dtb, myname, desthost, type):
             """
-            Produce a gateway configuration file from a line of the gateways section
+            Produce a gateway configuration file from a gateways line. 
             """
 
@@ -1011 +1077 @@
             raise service_error(service_error.internal, 
                     "Cannot stage tarfile/rpm: %s" % e.strerror)
-        
-        # XXX: more parallelism
-        for tb in allocated.iterkeys():
-            if tb != master:
-                if self.start_segment(tb, eid, tbparams, tmpdir, 0):
-                    started[tb] = True
-                else:
-                    break
-
-        if len(started) == len(allocated)-1:
-            if self.start_segment(master, eid, tbparams, tmpdir):
-                started[master] = True
-        
-        if not fail_soft and  len(allocated) != len(started):
-            for tb in started.iterkeys():
+
+        thread_pool_info = self.thread_pool()
+        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
+            t  = self.pooled_thread(target=self.start_segment, 
+                    args=(tb, eid, tbparams, tmpdir, 0), name=tb,
+                    pdata=thread_pool_info)
+            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()
+
+        # If none failed, start the master
+        failed = [ t.getName() for t in threads if not t.rv ]
+
+        if len(failed) == 0:
+            if not self.start_segment(master, eid, tbparams, tmpdir):
+                failed.append(master)
+
+        # If one failed clean up
+        if not fail_soft and len(failed) > 0:
+            for tb in failed:
                 self.stop_segment(tb, eid, tbparams)
         else: