Merge branch 'release-1.8'

NEWS

@@ -1,3 +1,21 @@
+New in Version 1.8
+==================
+* major code restructuring on the server and executors. The executors now send
+  out a heartbeat signal to the server, which enables the server to notice 
+  an executor having died. Also, all executors now properly die when the server
+  is done (or gets killed).
+* pickling of the pipeline has been disabled, because it turned out to cause
+  huge communication delays (minutes in large pipelines: > 30000 stages)
+* command line tool added to query the server as to what's happening:
+  check_pipeline_status.py
+* logging is separated out again (each executor has its own log file)
+* Pyro4 environment variables should be set as:
+
+  PYRO_DETAILED_TRACEBACK=True
+  PYRO_SERVERTYPE=multiplex
+  PYRO_LOGLEVEL=INFO
+
+
 New in Version 1.7
 ==================
 * the communication between the server and the executors has been upgraded

pydpiper/application.py

@@ -16,7 +16,7 @@ def addApplicationOptionGroup(parser):
     group = OptionGroup(parser,  "General application options", "General options for all pydpiper applications.")
     group.add_option("--restart", dest="restart", 
                                action="store_true",
-                               help="Restart pipeline using backup files.")
+                               help="Restart pipeline using backup files. (Currently deprecated. Simply rerun the command you ran before)")
     group.add_option("--output-dir", dest="output_directory",
                                type="string", default=None,
                                help="Directory where output data and backups will be saved.")
@@ -126,8 +126,11 @@ def start(self):
             return 
 
         if self.options.restart:
-            logger.info("Restarting pipeline from pickled files.")
-            self.pipeline.restart()
+            print "\nThe restart option is deprecated (pipelines are not pickled anymore, because it takes too much time). Will restart based on which files exists already\n"
+            #logger.info("Restarting pipeline from pickled files.")
+            #self.pipeline.restart()
+            self.reconstructCommand()
+            self.run()
             self.pipeline.initialize()
             self.pipeline.printStages(self.appName)
         else:
@@ -158,8 +161,8 @@ def setup_appName(self):
     def setup_logger(self):
         """sets logging info specific to application"""
         FORMAT = '%(asctime)-15s %(name)s %(levelname)s %(process)d/%(threadName)s: %(message)s'
-        now = datetime.now()  
-        FILENAME = str(self.appName) + "-" + now.strftime("%Y%m%d-%H%M%S%f") + ".log"
+        now = datetime.now().strftime("%Y-%m-%d-at-%H:%M:%S")
+        FILENAME = str(self.appName) + "-" + now + '-pid-' + str(os.getpid())  + ".log"
         logging.basicConfig(filename=FILENAME, format=FORMAT, level=logging.DEBUG)
 
     def setup_options(self):

pydpiper/check_pipeline_status.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+
+import os
+from optparse import OptionParser
+
+# setup the log file name before importing the Pyro4 library
+
+import Pyro4
+
+""" check the status of a pydpiper pipeline by querying the server using its uri"""
+
+if __name__ == '__main__':
+    usage = "Usage: " + __file__ + " uri\n" + \
+            "   or: " + __file__ + "--help"
+
+    parser = OptionParser(usage)
+
+    options, args = parser.parse_args()
+
+    if len(args) != 1:
+        parser.error("please specify the uri file")
+
+    uri_file = args[0]
+
+    # find the server
+    try:
+        uf = open(uri_file)
+        serverURI = Pyro4.URI(uf.readline())
+        uf.close()
+    except:
+        print "There is a problem opening the specified uri file: %s" % uri_file
+        raise
+
+    proxyServer = Pyro4.Proxy(serverURI)
+
+    # total number of stages in the pipeline:
+    numStages = proxyServer.getTotalNumberOfStages()
+    processedStages = proxyServer.getNumberProcessedStages()
+    print "Total number of stages in the pipeline: ", numStages
+    print "Number of stages already processed:     ", processedStages, "\n"
+
+    # some info about executors
+    runningClients = proxyServer.getNumberOfRunningClients()
+    waitingClients = proxyServer.getNumberOfQueuedClients()
+    print "Number of active clients:               ", runningClients
+    print "Number of clients waiting in the queue: ", waitingClients, "\n"
+
+    # stages currently running:
+    runningStagesList = proxyServer.getCurrentlyRunningStages()
+    print "Currently running stages: "
+    for stage in runningStagesList:
+        print stage, " ", proxyServer.getStageCommand(stage)
+
+    # currently runnable jobs:
+    numberRunnableStages = proxyServer.getNumberRunnableStages()
+    print "\n\nNumber of runnable stages:               ", numberRunnableStages, "\n"
+
+
+
+