agents/common/timertask.py

0001 #!/usr/bin/env python
0002 #
0003 # Licensed under the Apache License, Version 2.0 (the "License");
0004 # You may not use this file except in compliance with the License.
0005 # You may obtain a copy of the License at
0006 # http://www.apache.org/licenses/LICENSE-2.0OA
0007 #
0008 # Authors:
0009 # - Wen Guan, <wen.guan@cern.ch>, 2020
0010
0011
0012 import time
0013 import traceback
0014
0015
0016 class TimerTask(object):
0017     """
0018     The base class for Task which will be executed after some time
0019     """
0020
0021     def __init__(self, task_func, task_output_queue=None, task_args=tuple(), task_kwargs={}, delay_time=10, priority=1, logger=None):
0022         self.to_execute_time = time.time()
0023         self.delay_time = delay_time
0024         self.priority = priority
0025         self.task_func = task_func
0026         self.task_output_queue = task_output_queue
0027         self.task_args = task_args
0028         self.task_kwargs = task_kwargs
0029
0030         self.logger = logger
0031
0032     def __eq__(one, two):
0033         return (one.to_execute_time, one.priority) == (two.to_execute_time, two.priority)
0034
0035     def __lt__(one, two):
0036         return (one.to_execute_time, one.priority) < (two.to_execute_time, two.priority)
0037
0038     def __le__(one, two):
0039         return (one.to_execute_time, one.priority) <= (two.to_execute_time, two.priority)
0040
0041     def __gt__(one, two):
0042         return (one.to_execute_time, one.priority) > (two.to_execute_time, two.priority)
0043
0044     def __ge__(one, two):
0045         return (one.to_execute_time, one.priority) >= (two.to_execute_time, two.priority)
0046
0047     def is_ready(self):
0048         if self.to_execute_time <= time.time():
0049             return True
0050         return False
0051
0052     def execute(self):
0053         try:
0054             # set it to avoid an exception
0055             self.to_execute_time = time.time() + self.delay_time
0056
0057             ret = self.task_func(*self.task_args, **self.task_kwargs)
0058             if self.task_output_queue and ret is not None:
0059                 for ret_item in ret:
0060                     self.task_output_queue.put(ret_item)
0061
0062             # if there is no exception, this one is the correct one.
0063             self.to_execute_time = time.time() + self.delay_time
0064         except:
0065             if self.logger:
0066                 self.logger.error('Failed to execute task func: %s, %s' % (self.task_func, traceback.format_exc()))
0067             else:
0068                 print('Failed to execute task func: %s, %s' % (self.task_func, traceback.format_exc()))