Changeset a97394b for fedd

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

Location:

fedd

Files:

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

fedd/fedd.py

@@ -17 +17 @@
 from fedd_deter_impl import new_feddservice
 from service_error import *
+
+from threading import *
 
 # The SSL server here is based on the implementation described at
@@ -166 +168 @@
                 version="0.1")
 
-        self.set_defaults(host="localhost", port=23235, 
-                transport="soap",debug=0)
+        self.set_defaults(host="localhost", port=23235, transport="soap",
+                debug=0)
 
         self.add_option("-d", "--debug", action="count", dest="debug", 
@@ -177 +179 @@
         self.add_option("-p", "--port", action="store", type="int",
                 dest="port", help="Port to listen on (default %default)")
+        self.add_option("-s", "--service", action="append", type="string",
+                dest="services",
+                help="Service description: host:port:transport")
         self.add_option("-x","--transport", action="store", type="choice",
                 choices=("xmlrpc", "soap"),
@@ -182 +187 @@
         self.add_option("--trace", action="store_const", dest="tracefile", 
                 const=sys.stderr, help="Print SOAP exchange to stderr")
+
+def run_server(s):
+    if s: s.serve_forever()
+
+services = [ ]
+servers = [ ]
 
 opts, args = fedd_opts().parse_args()
@@ -209 +220 @@
             raise
 
-if opts.transport == "soap":
-    s = fedd_server(SOAP_port, fedd_soap_handler, ctx, impl)
-elif opts.transport == "xmlrpc":
-    s = fedd_server(SOAP_port, fedd_xmlrpc_handler, ctx, impl)
+if opts.services:
+    for s in opts.services:
+        h, p, t  = s.split(':')
+
+        if not h: h = opts.host
+        if not p: p = opts.port
+        if not t: h = opts.transport
+
+        p = int(p)
+
+        services.append((t, (h, p)))
 else:
-    s = None
-
-s.serve_forever()
+    services.append((opts.transport, (opts.host, opts.port)))
+
+for s in services:
+    if s[0] == "soap":
+        servers.append(fedd_server(s[1], fedd_soap_handler, ctx, impl))
+    elif s[0] == "xmlrpc":
+        servers.append(fedd_server(s[1], fedd_xmlrpc_handler, ctx, impl))
+    else: print >>sys.stderr, "Unknown transport: %s" % s[0]
+
+for s in servers:
+    t = Thread(target=run_server, args=(s,))
+    t.start()

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")