add pbs_launchers

pipelet/launchers.py

@@ -161,36 +161,42 @@ def launch_ssh(pipe, host_list, address=None, authkey='secret'):
     for w in processlist:
         w.wait()
     
-def launch_pbs(pipe, host_list, address=None, authkey='secret'):
+def launch_pbs(pipe, n, address=None, authkey='secret', job_dir="/wrk/lejeune/pipelet/scripts", job_name="job_", log_dir="/wrk/lejeune/pipelet/logs", cpu_time="2:00:00", server=False, job_header="""
+#/bin/bash
+echo $PYTHONPATH
+""" ):
     """ Launch a bunch of distant workers through a PBS batch system.
-
-    """
-    pass
-#     s = scheduler.Scheduler(pipe)
     
-#     SchedulerManager.register('get_scheduler', callable=lambda:s)
+    """
+    s = scheduler.Scheduler(pipe)
     
-#     mgr = SchedulerManager(address=address, authkey=authkey)
-#     mgr.start()
-
-#     processlist = []
-#     for i, h in enumerate(host_list):
-#         jfilename="job_%d.py"%i
-#         with closing(file(jfilename,'w')) as f:
-#             f.write(_job_file%(str(address),str(authkey)))
-#         _scp(jfilename, h+':')
-#         w = subprocess.Popen(['ssh', h, "python %s"%jfilename])
-#         processlist.append(w)
+    SchedulerManager.register('get_scheduler', callable=lambda:s)
+    if server:
+        mgr = SchedulerManager(address=address, authkey=authkey)
+        mgr.start()
+            
+    processlist = []
+    for i in range(n):
+        jobfile = path.join(job_dir,"%s%d.py"%(job_name,i))
+        errfile = path.join(log_dir,"e_%s%d"%(job_name,i))
+        logfile = path.join(log_dir,"o_%s%d"%(job_name,i))
+        f = file(jobfile, 'w')
+        f.write (job_header+"\n")
+        f.write ("#PBS -o %s\n"%logfile)
+        f.write ("#PBS -e %s\n"%errfile)
+        f.write ("#PBS -N %s%d\n"%(job_name,i))
+        f.write ("#PBS -l select=1:ncpus=1,walltime=%s\n"%cpu_time)
+        f.write ("python -m pipelet.launchers -H %s -p %s -s %s"%(address[0],address[1],authkey))
+        f.close()
+
+        subprocess.Popen(['qsub',jobfile]).communicate()[0]
+        if server:
+            print 'launching the scheduler'
+            sched_proxy = mgr.get_scheduler()
+            sched_proxy.run()                   
+            
 
-#     sched_proxy = mgr.get_scheduler()
-#     sched_proxy.run()
-    
-#     # joining Zombie process
-#     for w in processlist:
-#         w.wait()
 
-#     for i, h in enumerate(host_list):
-#         subprocess.Popen(['ssh', h, "rm -f %s"%jfilename])
 
 _job_file = """
 #/bin/zsh