Changeset a97394b for fedd/fedd_experiment_control.py

Timestamp:

Oct 7, 2008 11:03:50 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:

8ecfbad

Parents:

eee2b2e

Message:

beginnings of a real multithreaded server

File:

• fedd/fedd_experiment_control.py (modified) (12 diffs)

fedd/fedd_experiment_control.py

@@ -147 +147 @@
         self.state = { }
         self.state_filename = config.experiment_state_file
+        self.state_lock = Lock()
         self.tclsh = "/usr/local/bin/otclsh"
         self.tcl_splitter = "/usr/testbed/lib/ns2ir/parse.tcl"
@@ -252 +253 @@
         d.close()
 
+    # Call while holding self.state_lock
     def write_state(self):
         if os.access(self.state_filename, os.W_OK):
@@ -265 +267 @@
             print >>sys.stderr, "Pickling problem: %s" % e
 
+    # Call while holding self.state_lock
     def read_state(self):
         try:
@@ -1216 +1219 @@
                 req['experimentID'].has_key('localname'):
             eid = req['experimentID']['localname']
+            self.state_lock.acquire()
             while (self.state.has_key(eid)):
                 eid += random.choice(string.ascii_letters)
+            self.state[eid] = "placeholder"
+            self.state_lock.release()
         else:
             eid = self.exp_stem
             for i in range(0,5):
                 eid += random.choice(string.ascii_letters)
+            self.state_lock.acquire()
             while (self.state.has_key(eid)):
                 eid = self.exp_stem
                 for i in range(0,5):
                     eid += random.choice(string.ascii_letters)
+            self.state[eid] = "placeholder"
+            self.state_lock.release()
 
         try:
@@ -1351 +1360 @@
                 failed.append(master)
 
-        # If one failed clean up
-        if len(failed) > 0:
-            succeeded = [tb for tb in allocated.keys() if tb not in failed]
-            if fail_soft:
-                raise service_error(service_error.partial, \
-                        "Partial swap in on %s" % ",".join(succeeded))
-            else:
+        succeeded = [tb for tb in allocated.keys() if tb not in failed]
+        # If one failed clean up, unless fail_soft is set
+        if failed:
+            if not fail_soft:
                 for tb in succeeded:
                     self.stop_segment(tb, eid, tbparams)
+                # Remove the placeholder
+                self.state_lock.acquire()
+                del self.state[eid]
+                self.state_lock.release()
+
                 raise service_error(service_error.federant,
                     "Swap in failed on %s" % ",".join(failed))
@@ -1394 +1405 @@
                     'experimentAccess': { 'X509' : expcert },\
                 }
-        
+
+        self.state_lock.acquire()
         self.state[expid] = { 'federant' : [ tbparams[tb]['federant'] \
                 for tb in tbparams.keys() \
@@ -1406 +1418 @@
         self.state[eid] = self.state[expid]
         if self.state_filename: self.write_state()
-        return resp
+        self.state_lock.release()
+
+        if not failed:
+            return resp
+        else:
+            raise service_error(service_error.partial, \
+                    "Partial swap in on %s" % ",".join(succeeded))
+
 
     def get_vtopo(self, req, fid):
+        rv = None
 
         req = req.get('VtopoRequestBody', None)
@@ -1427 +1447 @@
             raise service_error(service_error.req, "No request?")
 
+        self.state_lock.acquire()
         if self.state.has_key(key):
-            return { 'experiment' : {keytype: key },\
+            rv = { 'experiment' : {keytype: key },\
                     'vtopo': self.state[key]['vtopo'],\
-                    }
-        else:
-            raise service_error(service_error.req, "No such experiment")
+                }
+        self.state_lock.release()
+
+        if rv: return rv
+        else: raise service_error(service_error.req, "No such experiment")
 
     def get_vis(self, req, fid):
+        rv = None
 
         req = req.get('VisRequestBody', None)
@@ -1453 +1477 @@
             raise service_error(service_error.req, "No request?")
 
+        self.state_lock.acquire()
         if self.state.has_key(key):
-            return { 'experiment' : {keytype: key },\
+            rv =  { 'experiment' : {keytype: key },\
                     'vis': self.state[key]['vis'],\
                     }
-        else:
-            raise service_error(service_error.req, "No such experiment")
+        self.state_lock.release()
+
+        if rv: return rv
+        else: raise service_error(service_error.req, "No such experiment")
 
     def get_info(self, req, fid):
+        rv = None
 
         req = req.get('InfoRequestBody', None)
@@ -1482 +1510 @@
         # get_info (e.g., encoded for XMLRPC transport) so send a copy of the
         # state.
+        self.state_lock.acquire()
         if self.state.has_key(key):
-            return copy.deepcopy(self.state[key])
-        else:
-            raise service_error(service_error.req, "No such experiment")
-
+            rv = copy.deepcopy(self.state[key])
+        self.state_lock.release()
+
+        if rv: return rv
+        else: raise service_error(service_error.req, "No such experiment")
 
 
@@ -1508 +1538 @@
             raise service_error(service_error.req, "No request?")
 
+        self.state_lock.acquire()
         fed_exp = self.state.get(key, None)
 
         if fed_exp:
+            # This branch of the conditional holds the lock to generate a
+            # consistent temporary tbparams variable to deallocate experiments.
+            # It releases the lock to do the deallocations and reacquires it to
+            # remove the experiment state when the termination is complete.
             ids = []
             #  experimentID is a list of dicts that are self-describing
@@ -1546 +1581 @@
                         'eid': eid,\
                     }
+            self.state_lock.release()
+
             # Stop everyone.
             for tb in tbparams.keys():
                 self.stop_segment(tb, tbparams[tb]['eid'], tbparams)
 
+            # Remove teh terminated experiment
+            self.state_lock.acquire()
             for id in ids:
                 if self.state.has_key(id): del self.state[id]
 
             if self.state_filename: self.write_state()
+            self.state_lock.release()
+
             return { 'experiment': exp }
         else:
+            # Don't forget to release the lock
+            self.state_lock.release()
             raise service_error(service_error.req, "No saved state")