Changeset 46e4682 for fedd/federation/experiment_control.py

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

File:

• fedd/federation/experiment_control.py (modified) (7 diffs)

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