@@ -491,7 +491,7 @@ class Procs: queue = kwargs.get('queue') if not kwargs.has_key('tag'): kwargs['tag'] = string.join(cmdv) - if queue and self.queues.get(queue) != None: + if queue and self.queues.get(queue): # Will process later if Verbose > 1: print >>sys.stderr, "Queueing", cmdv, "to", queue, len(self.queues[queue]) self.queues[queue].append(kwargs) @@ -531,6 +531,17 @@ class Procs: return r + def wakeup(self, qname): + #Wake-up anything on this queue + if self.queues[qname]: + kwargs = self.queues[qname].pop(0) + #if not self.queues[qname]: + # If nothing left on queue, remove it so we won't block + #print "remove queue" + #del self.queues[qname] + if Verbose > 1: print >>sys.stderr, "Waking", kwargs + self.PopenNow(**kwargs) + def collect(self, ignoreErrors=False): """Execute any active or queued processes and return, a boolean success scalar and a list of the processes objects that ran.""" @@ -539,7 +550,7 @@ class Procs: total = 0 while True: p = self.waitpid(True) - if not p: break + if p == None: break results.append(p) total += 1 @@ -550,16 +561,6 @@ class Procs: err = p.stderr.read() p.stderr.close() if err: print >>sys.stderr, err - - #Wake-up anything on this queue - if self.queues[p.queue]: - kwargs = self.queues[p.queue].pop(0) - #if not self.queues[p.queue]: - # If nothing left on queue, remove it so we won't block - #print "remove queue" - #del self.queues[p.queue] - #if Verbose: print >>sys.stderr, "Waking", kwargs - self.PopenNow(**kwargs) for k,v in self.queues.items(): assert(not v) @@ -595,6 +596,8 @@ class Procs: del self.pids[p.pid] + self.wakeup(p.queue) + return p @@ -795,7 +798,7 @@ class JobScheduler: self.finalize(p) def finalize(self, p): - print >>p.syncfile, p.status, p.utime, p.stime + print >>p.syncfile, p.status, p.utime, p.stime, p.queue p.syncfile = None f = file(p.outdirname + "/." + p.outbasename + "/status", "w") fm edb488fe0201c59d6dda9f29f7a6177a83e81386 Michael E. Fisk mfisk@lanl.gov http://github.com/mfisk/filemap/commit/80062468209c4bd104613425a5776ea1cd278066 80062468209c4bd104613425a5776ea1cd278066 2009-11-06T05:51:46-08:00 2009-11-06T05:51:46-08:00 check for wakeups in more cases and fix bug that caused work to go on empty queues instead of running immediately. 2f934b41d24cefa187e3c75e12b4047507379e78 Michael E. Fisk mfisk@cfl-cs1.lanl.gov