Changeset 32e7d93

Timestamp:

Sep 9, 2009 1:40:11 PM (15 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-2.00, version-3.01, version-3.02

Children:

e2a7a413

Parents:

3c6dbec

Message:

Incremental logging and correct failures.

File:

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

fedd/federation/experiment_control.py

@@ -120 +120 @@
             self.release()
 
-        def wait_for_all_done(self):
-            """
-            Wait until all active threads finish (and at least one has started)
-            """
+        def wait_for_all_done(self, timeout=None):
+            """
+            Wait until all active threads finish (and at least one has
+            started).  If a timeout is given, return after waiting that long
+            for termination.  If all threads are done (and one has started in
+            the since the last clear()) return True, otherwise False.
+            """
+            if timeout:
+                deadline = time.time() + timeout
             self.acquire()
             while self.started == 0 or self.started > self.terminated:
-                self.wait()
+                self.wait(timeout)
+                if timeout:
+                    if time.time() > deadline:
+                        break
+                    timeout = deadline - time.time()
             self.release()
+            return not (self.started == 0 or self.started > self.terminated)
 
     class pooled_thread(Thread):
@@ -994 +1004 @@
             t.start()
 
-        # Wait until all finish 
-        thread_pool.wait_for_all_done()
+        # Wait until all finish (keep pinging the log, though)
+        mins = 0
+        while not thread_pool.wait_for_all_done(60.0):
+            mins += 1
+            alloc_log.info("Waiting for sub threads (it has been %d mins)" \
+                    % mins)
+
+        thread_pool.clear()
 
         # If none failed, start the master
@@ -1012 +1028 @@
                 raise service_error(service_error.internal, 
                     "No alloc id for testbed %s !?" % master)
-            starter = self.start_segment(log=log, debug=self.debug,
-                    testbed=master, cert_file=self.cert_file,
-                    cert_pwd=self.cert_pwd, trusted_certs=self.trusted_certs,
-                    caller=self.call_StartSegment,
-                    log_collector=log_collector)
-            if not starter(uri, aid, topo[master], True, attrs):
-                failed.append(master)
+            t = self.pooled_thread(
+                    target=self.start_segment(log=log, debug=self.debug, 
+                        testbed=master, cert_file=self.cert_file, 
+                        cert_pwd=self.cert_pwd, 
+                        trusted_certs=self.trusted_certs,
+                        caller=self.call_StartSegment,
+                        log_collector=log_collector),
+                    args =(uri, aid, topo[master], True, attrs),
+                    name=master, pdata=thread_pool, trace_file=self.trace_file)
+            threads.append(t)
+            t.start()
+            # Wait until the master finishes (keep pinging the log, though)
+            mins = 0
+            while not thread_pool.wait_for_all_done(60.0):
+                mins += 1
+                alloc_log.info("Waiting for master (it has been %d mins)" \
+                        % mins)
+            # update failed to include the master, if it failed
+            failed = [ t.getName() for t in threads if not t.rv ]
 
         succeeded = [tb for tb in allocated.keys() if tb not in failed]
         # If one failed clean up, unless fail_soft is set
-        if failed and False:
+        if failed:
             if not fail_soft:
                 thread_pool.clear()
@@ -1028 +1056 @@
                     # Create and start a thread to stop the segment
                     thread_pool.wait_for_slot()
+                    uri = self.tbmap.get(tb, None)
                     t  = self.pooled_thread(\
-                            target=self.stop_segment(log=log,
+                            target=self.terminate_segment(log=log,
                                 testbed=tb,
-                                keyfile=self.ssh_privkey_file,
-                                debug=self.debug), 
-                            args=(tb, eid, tbparams), name=tb,
+                                cert_file=self.cert_file, 
+                                cert_pwd=self.cert_pwd,
+                                trusted_certs=self.trusted_certs,
+                                caller=self.call_TerminateSegment),
+                            args=(uri, tbparams[tb]['federant']['allocID']),
+                            name=tb,
                             pdata=thread_pool, trace_file=self.trace_file)
                     t.start()