Source Code for Module gozerbot.runner

  1  # gozerbot/runner.py 
  2  # 
  3  # 
  4   
  5  """ threads management to run jobs """ 
  6   
  7  __copyright__ = 'this file is in the public domain' 
  8   
  9  from gozerbot.thr import start_new_thread 
 10  from gozerbot.generic import handle_exception, rlog, lockdec 
 11  from gozerbot.threadloop import ThreadLoop 
 12  from gozerbot.periodical import minutely 
 13  import Queue, time, thread, random 
 14   
 15  runlock = thread.allocate_lock() 
 16  locked = lockdec(runlock) 
 17   
 18 -class Runner(ThreadLoop): 
 19   
 20 -    def __init__(self, name=""): 
 21          ThreadLoop.__init__(self, name) 
 22          self.working = False 
 23          self.starttime = time.time() 
 24          self.elapsed = self.starttime 
 25          self.finished = time.time() 
 26   
 27 -    def handle(self, descr, func, *args, **kwargs): 
 28          self.working = True 
 29          self.name = descr 
 30          try: 
 31              rlog(0, 'runner', 'running %s: %s' % (descr, str(func))) 
 32              self.starttime = time.time() 
 33              func(*args, **kwargs) 
 34              self.finished = time.time() 
 35              self.elapsed = self.finished - self.starttime 
 36              if self.elapsed > 3: 
 37                  rlog(10, 'runner', 'ALERT %s %s job taking too long: %s \ 
 38  seconds' % (descr, str(func), self.elapsed)) 
 39          except Exception, ex: 
 40              handle_exception() 
 41          self.working = False 
 42   
 43 -class CommandRunner(Runner): 
 44   
 45 -    def handle(self, descr, func, bot, ievent, *args, **kwargs): 
 46          self.working = True 
 47          try: 
 48              self.name = descr 
 49              rlog(10, 'runner', '%s (%s) running %s: %s at speed %s' % \ 
 50  (ievent.nick, ievent.userhost, descr, str(func), ievent.speed)) 
 51              self.starttime = time.time() 
 52              func(bot, ievent, *args, **kwargs) 
 53              for queue in ievent.queues: 
 54                  queue.put_nowait(None) 
 55              self.finished = time.time() 
 56              self.elapsed = self.finished - self.starttime 
 57              if self.elapsed > 3: 
 58                  rlog(10, 'runner', 'ALERT %s %s job taking too long: %s \ 
 59  seconds' % (descr, str(func), self.elapsed)) 
 60          except Exception, ex: 
 61              handle_exception(ievent) 
 62          self.working = False 
 63   
 64 -class Runners(object): 
 65   
 66 -    def __init__(self, max=100, runnertype=Runner): 
 67          self.max = max 
 68          self.runners = [] 
 69          self.cleanup() 
 70          self.runnertype=runnertype 
 71   
 72 -    def runnersizes(self): 
 73          result = [] 
 74          for runner in self.runners: 
 75              result.append(runner.queue.qsize()) 
 76          return result 
 77   
 78 -    def stop(self): 
 79          for runner in self.runners: 
 80              runner.stop() 
 81          self.cleanup() 
 82   
 83 -    def start(self): 
 84          pass 
 85    
 86 -    def put(self, *data): 
 87          launched = False 
 88          for runner in self.runners: 
 89              if not runner.working and runner.running: 
 90                  runner.put(*data) 
 91                  launched = True 
 92                  return 
 93          runner = self.makenew() 
 94          runner.put(*data) 
 95   
 96 -    def running(self): 
 97          result = [] 
 98          for runner in self.runners: 
 99              if runner.working: 
100                  result.append(runner.nowrunning) 
101          return result 
102   
103 -    def makenew(self): 
104          runner = None 
105          if len(self.runners) < self.max: 
106              runner = self.runnertype() 
107              runner.start() 
108              self.runners.append(runner) 
109          else: 
110              runner = random.choice(self.runners) 
111          return runner 
112   
113      @minutely 
114 -    def cleanup(self): 
115          rlog(-10, 'runners', 'running cleanup') 
116          for runner in self.runners: 
117              if not runner.working: 
118                  runner.stop()  
119                  self.runners.remove(runner) 
120   
121  cbrunners = [Runners(10-i) for i in range(10)] 
122  cmndrunners = [Runners(12-i, CommandRunner) for i in range(10)] 
123   
124 -def runners_start(): 
125      for runner in cbrunners: 
126          runner.start() 
127      for runner in cmndrunners: 
128          runner.start() 
129   
130 -def runners_stop(): 
131      for runner in cbrunners: 
132          runner.stop() 
133      for runner in cmndrunners: 
134          runner.stop() 
135