| #!/usr/bin/env python |
| |
| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import glib |
| import logging |
| import random |
| |
| DEFAULT_MAX_RANDOM_DELAY_MS = 10000 |
| |
| _instance = None |
| |
| def get_instance(): |
| """ |
| Return the singleton instance of the TaskLoop class. |
| |
| """ |
| global _instance |
| if _instance is None: |
| _instance = TaskLoop() |
| return _instance |
| |
| |
| class TaskLoop(object): |
| """ |
| The context to place asynchronous calls. |
| |
| This is a wrapper around the GLIB mainloop interface, exposing methods to |
| place (delayed) asynchronous calls. In addition to wrapping around the GLIB |
| API, this provides switches to control how delays are incorporated in method |
| calls globally. |
| |
| This class is meant to be a singleton. |
| Do not create an instance directly, use the module level function |
| get_instance() instead. |
| |
| Running the TaskLoop is blocking for the caller. So use this class like so: |
| |
| tl = task_loop.get_instance() |
| # Setup other things. |
| # Add initial tasks to tl to do stuff, post more tasks, and make the world a |
| # better place. |
| tl.start() |
| # This thread is now blocked. Some task should eventually call tl.stop() to |
| continue here. |
| |
| @var ignore_delays: Flag to control if delayed tasks are posted immediately. |
| |
| @var random_delays: Flag to control if arbitrary delays are inserted between |
| posted tasks. |
| |
| @var max_random_delay_ms: When random_delays is True, the maximum delay |
| inserted between posted tasks. |
| |
| """ |
| |
| |
| def __init__(self): |
| self._logger = logging.getLogger(__name__) |
| |
| # Initialize properties |
| self._ignore_delays = False |
| self._random_delays = False |
| self._max_random_delay_ms = DEFAULT_MAX_RANDOM_DELAY_MS |
| |
| # Get the mainloop so that tasks can be posted even before running the |
| # task loop. |
| self._mainloop = glib.MainLoop() |
| |
| # Initialize dictionary to track posted tasks. |
| self._next_post_id = 0 |
| self._posted_tasks = {} |
| |
| |
| @property |
| def ignore_delays(self): |
| """ |
| Boolean flag to control if delayed tasks are posted immediately. |
| |
| If True, all tasks posted henceforth are immediately marked active |
| ignoring any delay requested. With this switch, all other delay related |
| switches are ignored. |
| |
| """ |
| return self._ignore_delays |
| |
| |
| @ignore_delays.setter |
| def ignore_delays(self, value): |
| """ |
| Set |ignore_delays|. |
| |
| @param value: Boolean value for the |ignore_delays| flag |
| |
| """ |
| self._logger.debug('Turning %s delays ignored mode.', ('on' if value |
| else 'off')) |
| self._ignore_delays = value |
| |
| |
| @property |
| def random_delays(self): |
| """ |
| Boolean flag to control if random delays are inserted in posted tasks. |
| |
| If True, arbitrary delays in range [0, |max_random_delay_ms|] are |
| inserted in all posted tasks henceforth, ignoring the actual delay |
| requested. |
| |
| """ |
| return self._random_delays |
| |
| |
| @random_delays.setter |
| def random_delays(self, value): |
| """ |
| Set |random_delays|. |
| |
| @param value: Boolean value for the random_delays flag. |
| |
| """ |
| self._logger.debug('Turning %s random delays.', ('on' if value else |
| 'off')) |
| self._random_delays = value |
| |
| |
| @property |
| def max_random_delay_ms(self): |
| """ |
| The maximum arbitrary delay inserted in posted tasks in milliseconds. |
| Type: int |
| |
| """ |
| return self._max_random_delay_ms |
| |
| |
| @max_random_delay_ms.setter |
| def max_random_delay_ms(self, value): |
| """ |
| Set |max_random_delay_ms|. |
| |
| @param value: Non-negative int value for |max_random_delay_ms|. Negative |
| values are clamped to 0. |
| |
| """ |
| if value < 0: |
| self._logger.warning( |
| 'Can not set max_random_delay_ms to negative value %s. ' |
| 'Setting to 0 instead.', |
| value) |
| value = 0 |
| self._logger.debug('Set max random delay to %d. Random delay is %s', |
| value, ('on' if self.random_delays else 'off')) |
| self._max_random_delay_ms = value |
| |
| |
| def start(self): |
| """ |
| Run the task loop. |
| |
| This call is blocking. The thread that calls TaskLoop.start(...) becomes |
| the task loop itself and is blocked as such till TaskLoop.stop(...) is |
| called. |
| |
| """ |
| self._logger.info('Task Loop is now processing tasks...') |
| self._mainloop.run() |
| |
| |
| def stop(self): |
| """ |
| Stop the task loop. |
| |
| """ |
| self._logger.info('Task Loop quitting.') |
| self._mainloop.quit() |
| |
| |
| def post_repeated_task(self, callback, delay_ms=0): |
| """ |
| Post the given callback repeatedly forever until cancelled. |
| |
| The posted callback must not expect any arguments. It likely does not |
| make sense to provide fixed data parameters to a repeated task. Use the |
| object reference to provide context. |
| |
| In the |ignore_delays| mode, the task is reposted immediately after |
| dispatch. |
| In the |random_delays| mode, a new arbitrary delay is inserted before |
| each call to |callback|. |
| |
| @param callback: The function to call repeatedly. |callback| must expect |
| an object reference as the only argument. The return value from |
| |callback| is ignored. |
| |
| @param delay_ms: The delay between repeated calls to |callback|. The |
| first call is also delayed by this amount. Default: 0 |
| |
| @return: An integer ID that can be used to cancel the posted task. |
| |
| """ |
| assert callback is not None |
| |
| post_id = self._next_post_id |
| self._next_post_id += 1 |
| |
| next_delay_ms = self._next_delay_ms(delay_ms) |
| self._posted_tasks[post_id] = glib.timeout_add( |
| next_delay_ms, |
| TaskLoop._execute_repeated_task, |
| self, |
| post_id, |
| callback, |
| delay_ms) |
| return post_id |
| |
| |
| def post_task_after_delay(self, callback, delay_ms, *args, **kwargs): |
| """ |
| Post the given callback once to be dispatched after |delay_ms|. |
| |
| @param callback: The function to call. The function may expect arbitrary |
| number of arguments, passed in as |*args| and |**kwargs|. The |
| return value from |callback| is ignored. |
| |
| @param delay_ms: The delay before the call to |callback|. Default: 0 |
| |
| @return: An integer ID that can be used to cancel the posted task. |
| |
| """ |
| assert callback is not None |
| post_id = self._next_post_id |
| self._next_post_id = self._next_post_id + 1 |
| delay_ms = self._next_delay_ms(delay_ms) |
| self._posted_tasks[post_id] = glib.timeout_add(delay_ms, callback, |
| *args, **kwargs) |
| return post_id |
| |
| |
| def post_task(self, callback, *args, **kwargs): |
| """ |
| Post the given callback once. |
| |
| In |random_delays| mode, this function is equivalent to |
| |post_task_after_delay|. |
| |
| @param callback: The function to call. The function may expect arbitrary |
| number of arguments, passed in as |*args| and |**kwargs|. The |
| return value from |callback| is ignored. |
| |
| @return: An integer ID that can be used to cancel the posted task. |
| |
| """ |
| self._logger.debug('Task posted: %s', repr(callback)) |
| self._logger.debug('Arguments: %s, Keyword arguments: %s', |
| repr(args), repr(kwargs)) |
| return self.post_task_after_delay(callback, 0, *args, **kwargs) |
| |
| |
| def cancel_posted_task(self, post_id): |
| """ |
| Cancels a previously posted task that is yet to be dispatched. |
| |
| @param post_id: The |post_id| of the task to cancel, as returned by one |
| of the functions that post a task. |
| |
| @return: True if the posted task was removed. |
| |
| """ |
| if post_id in self._posted_tasks: |
| retval = glib.source_remove(self._posted_tasks[post_id]) |
| if retval: |
| del self._posted_tasks[post_id] |
| return retval |
| else: |
| return False |
| |
| |
| def _next_delay_ms(self, user_delay_ms): |
| """ |
| Determine the actual delay to post the next task. |
| |
| The actual delay posted may be different from the user requested delay |
| based on what mode we're in. |
| |
| @param user_delay_ms: The delay requested by the user. |
| |
| @return The actual delay to be posted. |
| |
| """ |
| next_delay_ms = user_delay_ms |
| if self.ignore_delays: |
| next_delay_ms = 0 |
| elif self.random_delays: |
| next_delay_ms = random.randint(0, self.max_random_delay_ms) |
| return next_delay_ms |
| |
| |
| def _execute_repeated_task(self, post_id, callback, delay_ms): |
| """ |
| A wrapper to repost an executed task, and return False. |
| |
| We need this to be able to repost the task at arbitrary intervals. |
| |
| @param post_id: The private post_id tracking this repeated task. |
| |
| @param callback: The user callback that must be called. |
| |
| @param delay_ms: The user requested delay between calls. |
| |
| """ |
| retval = callback() |
| self._logger.debug('Ignored return value from repeated task: %s', |
| repr(retval)) |
| |
| next_delay_ms = self._next_delay_ms(delay_ms) |
| self._posted_tasks[post_id] = glib.timeout_add( |
| next_delay_ms, |
| TaskLoop._execute_repeated_task, |
| self, |
| post_id, |
| callback, |
| delay_ms) |
| return False |
