Index: fedd/fedd_create_experiment.py
===================================================================
--- fedd/fedd_create_experiment.py	(revision 6546868137c38bcdc7bdcdf452364e3d3c8f777a)
+++ fedd/fedd_create_experiment.py	(revision 1af38d6347a52bd6beaadc3f0d421a1e49f04001)
@@ -16,4 +16,6 @@
 import tempfile
 import copy
+
+from threading import *
 
 from subprocess import *
@@ -35,4 +37,5 @@
 	    debug=False,
 	    muxmax=2,
+	    nthreads=2,
 	    project_user = "faber",
 	    scp_exec="/usr/bin/scp",
@@ -50,4 +53,6 @@
 	    ):
 	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,4 +60,5 @@
 	self.debug = debug
 	self.muxmax = muxmax
+	self.nthreads = nthreads
 	self.project_user = project_user
 	self.scp_exec = scp_exec
@@ -93,4 +99,64 @@
 		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,5 +752,5 @@
 		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,19 +1077,42 @@
 	    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:
