Changeset 46e4682

Timestamp:

Jul 28, 2009 10:57:38 AM (15 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:

d15522f

Parents:

3c20a31

Message:

termination logs, a terminating state, and avoiding a deadlock

Location:

fedd

Files:

• fedd_client.py (modified) (3 diffs)
• federation/experiment_control.py (modified) (7 diffs)
• wsdl/fedd_types.xsd (modified) (1 diff)

fedd/fedd_client.py

@@ -163 +163 @@
                 action="store_true", default=False,
                 help="Force termination if experiment is in strange state")
+        self.add_option("--logfile", dest="logfile", default=None,
+                help="File to write log to")
+        self.add_option("--print_log", dest="print_log", default=False,
+                action="store_true",
+                help="Print deallocation log to standard output")
 
 class fedd_multi_exp_data_opts(fedd_client_opts):
@@ -1054 +1059 @@
         if opts.exp_name and opts.exp_certfile:
             sys.exit("Only one of --experiment_cert and " +\
-                    "--experiment_name permitted");
+                    "--experiment_name permitted")
+
+        if opts.print_log and opts.logfile:
+            sys.exit("Only one of --logfile and --print_log is permitted")
+        elif opts.print_log:
+            out = sys.stdout
+        elif opts.logfile:
+            try:
+                out = open(opts.logfile, "w")
+            except IOError,e:
+                sys.exit("Cannot open logfile: %s" %e)
+        else:
+            out = None
 
         exp_id = None
@@ -1082 +1099 @@
             sys.exit("Error processing RPC: %s" % e)
 
-        eid = resp_dict.get('experimentID', None)
-        if eid:
-            for id in eid:
-                for k in id.keys():
-                    if k == 'fedid': print "%s: %s" % (k,fedid(bits=id[k]))
-                    else: print "%s: %s" % (k, id[k])
+        if out:
+            log = resp_dict.get('deallocationLog', None)
+            if log:
+                print >>out, log
+                out.close()
+            else:
+                out.close()
+                sys.exit("No log returned")
 
 class create(fedd_rpc):

fedd/federation/experiment_control.py

@@ -2044 +2044 @@
         # Create a logger that logs to the experiment's state object as well as
         # to the main log file.
-
         alloc_log = logging.getLogger('fedd.experiment_control.%s' % eid)
         h = logging.StreamHandler(self.list_log(self.state[eid]['log']))
@@ -2053 +2052 @@
         alloc_log.addHandler(h)
         
-
-
-
-
         # Start a thread to do the resource allocation
         t  = Thread(target=self.allocate_resources,
@@ -2233 +2228 @@
             for f in rv.get('federant', []):
                 if f.has_key('allocID'): del f['allocID']
-
         return rv
 
@@ -2318 +2312 @@
         self.check_experiment_access(fid, key)
 
+        dealloc_list = [ ]
+
+
+        # Create a logger that logs to the dealloc_list as well as to the main
+        # log file.
+        dealloc_log = logging.getLogger('fedd.experiment_control.%s' % key)
+        h = logging.StreamHandler(self.list_log(dealloc_list))
+        # XXX: there should be a global one of these rather than repeating the
+        # code.
+        h.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s",
+                    '%d %b %y %H:%M:%S'))
+        dealloc_log.addHandler(h)
+
         self.state_lock.acquire()
         fed_exp = self.state.get(key, None)
@@ -2329 +2336 @@
             # First make sure that the experiment creation is complete.
             status = fed_exp.get('experimentStatus', None)
+
             if status:
-                if status == 'starting':
+                if status in ('starting', 'terminating'):
                     if not force:
                         self.state_lock.release()
                         raise service_error(service_error.partial, 
-                                'Experiment still being created')
+                                'Experiment still being created or destroyed')
                     else:
-                        self.log.warning('Experiment in starting state ' + \
-                                'being terminated by admin.')
+                        self.log.warning('Experiment in %s state ' % status + \
+                                'being terminated by force.')
             else:
                 # No status??? trouble
@@ -2381 +2389 @@
                         'aid': aid,\
                     }
+                print "%s %s" % (tb, tbparams[tb])
+            fed_exp['experimentStatus'] = 'terminating'
+            if self.state_filename: self.write_state()
             self.state_lock.release()
 
-            # Stop everyone.
-            thread_pool = self.thread_pool(self.nthreads)
-            for tb in tbparams.keys():
-                # Create and start a thread to stop the segment
-                thread_pool.wait_for_slot()
-                t  = self.pooled_thread(\
-                        target=self.stop_segment(log=self.log,
-                            keyfile=self.ssh_privkey_file, debug=self.debug), 
-                        args=(tb, tbparams[tb]['eid'], tbparams), name=tb,
-                        pdata=thread_pool, trace_file=self.trace_file)
-                t.start()
-            # Wait for completions
-            thread_pool.wait_for_all_done()
+            # Stop everyone.  NB, wait_for_all waits until a thread starts and
+            # then completes, so we can't wait if nothing starts.  So, no
+            # tbparams, no start.
+            if len(tbparams) > 0:
+                thread_pool = self.thread_pool(self.nthreads)
+                for tb in tbparams.keys():
+                    # Create and start a thread to stop the segment
+                    thread_pool.wait_for_slot()
+                    t  = self.pooled_thread(\
+                            target=self.stop_segment(log=dealloc_log,
+                                keyfile=self.ssh_privkey_file, debug=self.debug), 
+                            args=(tb, tbparams[tb]['eid'], tbparams), name=tb,
+                            pdata=thread_pool, trace_file=self.trace_file)
+                    t.start()
+                # Wait for completions
+                thread_pool.wait_for_all_done()
 
             # release the allocations (failed experiments have done this
@@ -2415 +2429 @@
             self.state_lock.release()
 
-            return { 'experiment': exp }
+            return { 
+                    'experiment': exp , 
+                    'deallocationLog': "".join(dealloc_list),
+                    }
         else:
             # Don't forget to release the lock

fedd/wsdl/fedd_types.xsd

@@ -607 +607 @@
     <xsd:sequence>
       <xsd:element name="experiment" type="tns:IDType"/>
+      <xsd:element name="deallocationLog" type="xsd:string" minOccurs="0"
+        maxOccurs="1"/>
     </xsd:sequence>
   </xsd:complexType>