git.maemo.org Git - watersofshiloah/blob - src/util/go_utils.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import with_statement
   4
   5 import time
   6 import functools
   7 import threading
   8 import Queue
   9 import logging
  10
  11 import gobject
  12
  13 import algorithms
  14 import misc
  15
  16
  17 _moduleLogger = logging.getLogger(__name__)
  18
  19
  20 def make_idler(func):
  21         """
  22         Decorator that makes a generator-function into a function that will continue execution on next call
  23         """
  24         a = []
  25
  26         @functools.wraps(func)
  27         def decorated_func(*args, **kwds):
  28                 if not a:
  29                         a.append(func(*args, **kwds))
  30                 try:
  31                         a[0].next()
  32                         return True
  33                 except StopIteration:
  34                         del a[:]
  35                         return False
  36
  37         return decorated_func
  38
  39
  40 def async(func):
  41         """
  42         Make a function mainloop friendly. the function will be called at the
  43         next mainloop idle state.
  44
  45         >>> import misc
  46         >>> misc.validate_decorator(async)
  47         """
  48
  49         @functools.wraps(func)
  50         def new_function(*args, **kwargs):
  51
  52                 def async_function():
  53                         func(*args, **kwargs)
  54                         return False
  55
  56                 gobject.idle_add(async_function)
  57
  58         return new_function
  59
  60
  61 class Async(object):
  62
  63         def __init__(self, func, once = True):
  64                 self.__func = func
  65                 self.__idleId = None
  66                 self.__once = once
  67
  68         def start(self):
  69                 assert self.__idleId is None
  70                 if self.__once:
  71                         self.__idleId = gobject.idle_add(self._on_once)
  72                 else:
  73                         self.__idleId = gobject.idle_add(self.__func)
  74
  75         def is_running(self):
  76                 return self.__idleId is not None
  77
  78         def cancel(self):
  79                 if self.__idleId is not None:
  80                         gobject.source_remove(self.__idleId)
  81                         self.__idleId = None
  82
  83         def __call__(self):
  84                 return self.start()
  85
  86         @misc.log_exception(_moduleLogger)
  87         def _on_once(self):
  88                 self.cancel()
  89                 try:
  90                         self.__func()
  91                 except Exception:
  92                         pass
  93                 return False
  94
  95
  96 class Timeout(object):
  97
  98         def __init__(self, func):
  99                 self.__func = func
 100                 self.__timeoutId = None
 101
 102         def start(self, **kwds):
 103                 assert self.__timeoutId is None
 104
 105                 assert len(kwds) == 1
 106                 timeoutInSeconds = kwds["seconds"]
 107                 assert 0 <= timeoutInSeconds
 108                 if timeoutInSeconds == 0:
 109                         self.__timeoutId = gobject.idle_add(self._on_once)
 110                 else:
 111                         self.__timeoutId = timeout_add_seconds(timeoutInSeconds, self._on_once)
 112
 113         def is_running(self):
 114                 return self.__timeoutId is not None
 115
 116         def cancel(self):
 117                 if self.__timeoutId is not None:
 118                         gobject.source_remove(self.__timeoutId)
 119                         self.__timeoutId = None
 120
 121         def __call__(self, **kwds):
 122                 return self.start(**kwds)
 123
 124         @misc.log_exception(_moduleLogger)
 125         def _on_once(self):
 126                 self.cancel()
 127                 try:
 128                         self.__func()
 129                 except Exception:
 130                         pass
 131                 return False
 132
 133
 134 _QUEUE_EMPTY = object()
 135
 136
 137 class AsyncPool(object):
 138
 139         def __init__(self):
 140                 self.__workQueue = Queue.Queue()
 141                 self.__thread = threading.Thread(
 142                         name = type(self).__name__,
 143                         target = self.__consume_queue,
 144                 )
 145                 self.__isRunning = True
 146
 147         def start(self):
 148                 self.__thread.start()
 149
 150         def stop(self):
 151                 self.__isRunning = False
 152                 for _ in algorithms.itr_available(self.__workQueue):
 153                         pass # eat up queue to cut down dumb work
 154                 self.__workQueue.put(_QUEUE_EMPTY)
 155
 156         def clear_tasks(self):
 157                 for _ in algorithms.itr_available(self.__workQueue):
 158                         pass # eat up queue to cut down dumb work
 159
 160         def add_task(self, func, args, kwds, on_success, on_error):
 161                 task = func, args, kwds, on_success, on_error
 162                 self.__workQueue.put(task)
 163
 164         @misc.log_exception(_moduleLogger)
 165         def __trampoline_callback(self, on_success, on_error, isError, result):
 166                 if not self.__isRunning:
 167                         if isError:
 168                                 _moduleLogger.error("Masking: %s" % (result, ))
 169                         isError = True
 170                         result = StopIteration("Cancelling all callbacks")
 171                 callback = on_success if not isError else on_error
 172                 try:
 173                         callback(result)
 174                 except Exception:
 175                         _moduleLogger.exception("Callback errored")
 176                 return False
 177
 178         @misc.log_exception(_moduleLogger)
 179         def __consume_queue(self):
 180                 while True:
 181                         task = self.__workQueue.get()
 182                         if task is _QUEUE_EMPTY:
 183                                 break
 184                         func, args, kwds, on_success, on_error = task
 185
 186                         try:
 187                                 result = func(*args, **kwds)
 188                                 isError = False
 189                         except Exception, e:
 190                                 _moduleLogger.error("Error, passing it back to the main thread")
 191                                 result = e
 192                                 isError = True
 193                         self.__workQueue.task_done()
 194
 195                         gobject.idle_add(self.__trampoline_callback, on_success, on_error, isError, result)
 196                 _moduleLogger.debug("Shutting down worker thread")
 197
 198
 199 class AsyncLinearExecution(object):
 200
 201         def __init__(self, pool, func):
 202                 self._pool = pool
 203                 self._func = func
 204                 self._run = None
 205
 206         def start(self, *args, **kwds):
 207                 assert self._run is None
 208                 self._run = self._func(*args, **kwds)
 209                 trampoline, args, kwds = self._run.send(None) # priming the function
 210                 self._pool.add_task(
 211                         trampoline,
 212                         args,
 213                         kwds,
 214                         self.on_success,
 215                         self.on_error,
 216                 )
 217
 218         @misc.log_exception(_moduleLogger)
 219         def on_success(self, result):
 220                 _moduleLogger.debug("Processing success for: %r", self._func)
 221                 try:
 222                         trampoline, args, kwds = self._run.send(result)
 223                 except StopIteration, e:
 224                         pass
 225                 else:
 226                         self._pool.add_task(
 227                                 trampoline,
 228                                 args,
 229                                 kwds,
 230                                 self.on_success,
 231                                 self.on_error,
 232                         )
 233
 234         @misc.log_exception(_moduleLogger)
 235         def on_error(self, error):
 236                 _moduleLogger.debug("Processing error for: %r", self._func)
 237                 try:
 238                         trampoline, args, kwds = self._run.throw(error)
 239                 except StopIteration, e:
 240                         pass
 241                 else:
 242                         self._pool.add_task(
 243                                 trampoline,
 244                                 args,
 245                                 kwds,
 246                                 self.on_success,
 247                                 self.on_error,
 248                         )
 249
 250
 251 def throttled(minDelay, queue):
 252         """
 253         Throttle the calls to a function by queueing all the calls that happen
 254         before the minimum delay
 255
 256         >>> import misc
 257         >>> import Queue
 258         >>> misc.validate_decorator(throttled(0, Queue.Queue()))
 259         """
 260
 261         def actual_decorator(func):
 262
 263                 lastCallTime = [None]
 264
 265                 def process_queue():
 266                         if 0 < len(queue):
 267                                 func, args, kwargs = queue.pop(0)
 268                                 lastCallTime[0] = time.time() * 1000
 269                                 func(*args, **kwargs)
 270                         return False
 271
 272                 @functools.wraps(func)
 273                 def new_function(*args, **kwargs):
 274                         now = time.time() * 1000
 275                         if (
 276                                 lastCallTime[0] is None or
 277                                 (now - lastCallTime >= minDelay)
 278                         ):
 279                                 lastCallTime[0] = now
 280                                 func(*args, **kwargs)
 281                         else:
 282                                 queue.append((func, args, kwargs))
 283                                 lastCallDelta = now - lastCallTime[0]
 284                                 processQueueTimeout = int(minDelay * len(queue) - lastCallDelta)
 285                                 gobject.timeout_add(processQueueTimeout, process_queue)
 286
 287                 return new_function
 288
 289         return actual_decorator
 290
 291
 292 def _old_timeout_add_seconds(timeout, callback):
 293         return gobject.timeout_add(timeout * 1000, callback)
 294
 295
 296 def _timeout_add_seconds(timeout, callback):
 297         return gobject.timeout_add_seconds(timeout, callback)
 298
 299
 300 try:
 301         gobject.timeout_add_seconds
 302         timeout_add_seconds = _timeout_add_seconds
 303 except AttributeError:
 304         timeout_add_seconds = _old_timeout_add_seconds