Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Is it possible to execute function every x seconds in python, when it is performing pool.map?

I am running pool.map on big data array and i want to print report in console every minute. Is it possible? As i understand, python is synchronous language, it can't do this like nodejs.

Perhaps it can be done by threading.. or how?

finished = 0

def make_job():
   sleep(1)
   global finished
   finished += 1

# I want to call this function every minute
def display_status():
   print 'finished: ' + finished

def main():
    data = [...]
    pool = ThreadPool(45)
    results = pool.map(make_job, data)
    pool.close()
    pool.join()
like image 641
avasin Avatar asked Sep 17 '26 08:09

avasin


2 Answers

You can use a permanent threaded timer, like those from this question: Python threading.timer - repeat function every 'n' seconds

from threading import Timer,Event 

class perpetualTimer(object):

   # give it a cycle time (t) and a callback (hFunction) 
   def __init__(self,t,hFunction):
      self.t=t
      self.stop = Event()
      self.hFunction = hFunction
      self.thread = Timer(self.t,self.handle_function)

   def handle_function(self):
      self.hFunction()
      self.thread = Timer(self.t,self.handle_function)
      if not self.stop.is_set():
          self.thread.start()

   def start(self):
      self.stop.clear()
      self.thread.start()

   def cancel(self):
      self.stop.set()
      self.thread.cancel()

Basically this is just a wrapper for a Timer object that creates a new Timer object every time your desired function is called. Don't expect millisecond accuracy (or even close) from this, but for your purposes it should be ideal.

Using this your example would become:

finished = 0

def make_job():
   sleep(1)
   global finished
   finished += 1

def display_status():
   print 'finished: ' + finished

def main():
    data = [...]
    pool = ThreadPool(45)

    # set up the monitor to make run the function every minute
    monitor = PerpetualTimer(60,display_status)
    monitor.start()
    results = pool.map(make_job, data)
    pool.close()
    pool.join()
    monitor.cancel()

EDIT:

A cleaner solution may be (thanks to comments below):

from threading import Event,Thread 

class RepeatTimer(Thread):
    def __init__(self, t, callback, event):
        Thread.__init__(self)
        self.stop = event
        self.wait_time = t
        self.callback = callback
        self.daemon = True

    def run(self):
        while not self.stop.wait(self.wait_time):
            self.callback()

Then in your code:

def main():
    data = [...]
    pool = ThreadPool(45)
    stop_flag = Event()
    RepeatTimer(60,display_status,stop_flag).start()
    results = pool.map(make_job, data)
    pool.close()
    pool.join()
    stop_flag.set()
like image 150
ebarr Avatar answered Sep 18 '26 23:09

ebarr


One way to do this, is to use main thread as the monitoring one. Something like below should work:

def main():
   data = [...]
   results = []
   step = 0
   pool = ThreadPool(16)
   pool.map_async(make_job, data, callback=results.extend)
   pool.close()
   while True:
      if results:
          break
      step += 1
      sleep(1)
      if step % 60 == 0:
          print "status update" + ...

I've used .map() instead of .map_async() as the former is synchronous one. Also you probably will need to replace results.extend with something more efficient. And finally, due to GIL, speed improvement may be much smaller than expected.

BTW, it is little bit funny that you wrote that Python is synchronous in a question that asks about ThreadPool ;).

like image 27
RobertT Avatar answered Sep 18 '26 22:09

RobertT



Donate For Us

If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!