[LNT] r255963 - Fixups for background jobs

[Chris Matthews via llvm-commits]

Author: cmatthews
Date: Thu Dec 17 19:44:55 2015
New Revision: 255963

URL: http://llvm.org/viewvc/llvm-project?rev=255963&view=rev
Log:
Fixups for background jobs

Modified:
    lnt/trunk/lnt/util/ImportData.py
    lnt/trunk/lnt/util/async_ops.py

Modified: lnt/trunk/lnt/util/ImportData.py
URL: http://llvm.org/viewvc/llvm-project/lnt/trunk/lnt/util/ImportData.py?rev=255963&r1=255962&r2=255963&view=diff
==============================================================================
--- lnt/trunk/lnt/util/ImportData.py (original)
+++ lnt/trunk/lnt/util/ImportData.py Thu Dec 17 19:44:55 2015
@@ -95,9 +95,6 @@ def import_and_report(config, db_name, d
         report_url = "%s/db_%s/" % (config.zorgURL, db_name)
     else:
         report_url = "localhost"
-    # Add a handy relative link to the submitted run.
-    ts_name = data['Run']['Info'].get('tag')
-    result['result_url'] = "db_{}/v4/{}/{}".format(db_name, ts_name, run.id)
 
     if not disable_report:
         NTEmailReport.emailReport(result, db, run, report_url,
@@ -114,7 +111,9 @@ def import_and_report(config, db_name, d
         db.commit()
     else:
         db.rollback()
-
+    # Add a handy relative link to the submitted run.
+    ts_name = data['Run']['Info'].get('tag')
+    result['result_url'] = "db_{}/v4/{}/{}".format(db_name, ts_name, run.id)
     result['report_time'] = time.time() - importStartTime
     result['total_time'] = time.time() - startTime
 

Modified: lnt/trunk/lnt/util/async_ops.py
URL: http://llvm.org/viewvc/llvm-project/lnt/trunk/lnt/util/async_ops.py?rev=255963&r1=255962&r2=255963&view=diff
==============================================================================
--- lnt/trunk/lnt/util/async_ops.py (original)
+++ lnt/trunk/lnt/util/async_ops.py Thu Dec 17 19:44:55 2015
@@ -15,8 +15,8 @@ import lnt.server.db.fieldchange as fiel
 import lnt.server.db.v4db
 import traceback
 import multiprocessing
-from multiprocessing import Pool
-
+from multiprocessing import Pool, TimeoutError
+from lnt.testing.util.commands import note
 NUM_WORKERS = 2  # The number of subprocesses to spawn per LNT process.
 WORKERS = None  # The worker pool.
 
@@ -27,6 +27,7 @@ def launch_workers():
     if not WORKERS:
         logger = multiprocessing.log_to_stderr()
         logger.setLevel(logging.INFO)
+        note("Starting workers")
         WORKERS = Pool(NUM_WORKERS)
 
 
@@ -43,24 +44,43 @@ def async_fieldchange_calc(ts, run):
 def async_run_job(job, ts, func_args):
     """Send a job to the async wrapper in the subprocess."""
     # If the run is not in the database, we can't do anything more.
-    print "Queuing background job to process fieldchanges"
+    note("Queuing background job to process fieldchanges")
     args = {'tsname': ts.name,
             'db': ts.v4db.settings()}
     launch_workers()
-    job = WORKERS.apply_async(async_wrapper, [job, args, func_args])
-
+    job = WORKERS.apply_async(async_wrapper,
+                              [job, args, func_args],
+                              callback=async_job_finished)
+    # Lets see if we crash right away?
+    try:
+        job.get(timeout=1)
+    except TimeoutError:
+        pass
 
 def async_wrapper(job, ts_args, func_args):
     """Setup test-suite in this subprocess and run something."""
     try:
-        print >> sys.stderr, "Running async wrapper"
-        logging.info(str(job))
+        print >>sys.stderr,"Test"
+        h = logging.handlers.MemoryHandler(1024 * 1024)
+        h.setLevel(logging.DEBUG)
+        logging.getLogger('LNT').addHandler(h)
+        note("Running async wrapper")
+        note(str(job))
         _v4db = lnt.server.db.v4db.V4DB(**ts_args['db'])
         ts = _v4db.testsuite[ts_args['tsname']]
-        logging.info("Calculating field changes for ")
+        note("Calculating field changes for ")
         job(ts, **func_args)
-        logging.info("Done calculating field changes")
+        note("Done calculating field changes")
     except:
         # Put all exception text into an exception and raise that for our
         # parent process.
-        raise Exception("".join(traceback.format_exception(*sys.exc_info())))
+        return Exception("".join(traceback.format_exception(*sys.exc_info())))
+    return h.buffer
+
+def async_job_finished(arg):
+    if isinstance(arg, Exception):
+        raise arg
+    if isinstance(arg, list):
+        for log_entry in arg:
+            logging.getLogger('LNT').handle(log_entry)
+