4d314682659557e77d1643ea6ca81f2c04cf10a9
[gc-dialer] / src / util / concurrent.py
1 #!/usr/bin/env python
2
3 from __future__ import with_statement
4
5 import os
6 import errno
7 import time
8 import functools
9 import contextlib
10 import logging
11
12 import misc
13
14
15 _moduleLogger = logging.getLogger(__name__)
16
17
18 class AsyncTaskQueue(object):
19
20         def __init__(self, taskPool):
21                 self._asyncs = []
22                 self._taskPool = taskPool
23
24         def add_async(self, func):
25                 self.flush()
26                 a = _AsyncGeneratorTask(self._taskPool, func)
27                 self._asyncs.append(a)
28                 return a
29
30         def flush(self):
31                 self._asyncs = [a for a in self._asyncs if not a.isDone]
32
33
34 class _AsyncGeneratorTask(object):
35
36         def __init__(self, pool, func):
37                 self._pool = pool
38                 self._func = func
39                 self._run = None
40                 self._isDone = False
41
42         @property
43         def isDone(self):
44                 return self._isDone
45
46         def start(self, *args, **kwds):
47                 assert self._run is None, "Task already started"
48                 self._run = self._func(*args, **kwds)
49                 trampoline, args, kwds = self._run.send(None) # priming the function
50                 self._pool.add_task(
51                         trampoline,
52                         args,
53                         kwds,
54                         self.on_success,
55                         self.on_error,
56                 )
57
58         @misc.log_exception(_moduleLogger)
59         def on_success(self, result):
60                 _moduleLogger.debug("Processing success for: %r", self._func)
61                 try:
62                         trampoline, args, kwds = self._run.send(result)
63                 except StopIteration, e:
64                         self._isDone = True
65                 else:
66                         self._pool.add_task(
67                                 trampoline,
68                                 args,
69                                 kwds,
70                                 self.on_success,
71                                 self.on_error,
72                         )
73
74         @misc.log_exception(_moduleLogger)
75         def on_error(self, error):
76                 _moduleLogger.debug("Processing error for: %r", self._func)
77                 try:
78                         trampoline, args, kwds = self._run.throw(error)
79                 except StopIteration, e:
80                         self._isDone = True
81                 else:
82                         self._pool.add_task(
83                                 trampoline,
84                                 args,
85                                 kwds,
86                                 self.on_success,
87                                 self.on_error,
88                         )
89
90         def __repr__(self):
91                 return "<async %s at 0x%x>" % (self._func.__name__, id(self))
92
93         def __hash__(self):
94                 return hash(self._func)
95
96         def __eq__(self, other):
97                 return self._func == other._func
98
99         def __ne__(self, other):
100                 return self._func != other._func
101
102
103 def synchronized(lock):
104         """
105         Synchronization decorator.
106
107         >>> import misc
108         >>> misc.validate_decorator(synchronized(object()))
109         """
110
111         def wrap(f):
112
113                 @functools.wraps(f)
114                 def newFunction(*args, **kw):
115                         lock.acquire()
116                         try:
117                                 return f(*args, **kw)
118                         finally:
119                                 lock.release()
120                 return newFunction
121         return wrap
122
123
124 @contextlib.contextmanager
125 def qlock(queue, gblock = True, gtimeout = None, pblock = True, ptimeout = None):
126         """
127         Locking with a queue, good for when you want to lock an item passed around
128
129         >>> import Queue
130         >>> item = 5
131         >>> lock = Queue.Queue()
132         >>> lock.put(item)
133         >>> with qlock(lock) as i:
134         ...     print i
135         5
136         """
137         item = queue.get(gblock, gtimeout)
138         try:
139                 yield item
140         finally:
141                 queue.put(item, pblock, ptimeout)
142
143
144 @contextlib.contextmanager
145 def flock(path, timeout=-1):
146         WAIT_FOREVER = -1
147         DELAY = 0.1
148         timeSpent = 0
149
150         acquired = False
151
152         while timeSpent <= timeout or timeout == WAIT_FOREVER:
153                 try:
154                         fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
155                         acquired = True
156                         break
157                 except OSError, e:
158                         if e.errno != errno.EEXIST:
159                                 raise
160                 time.sleep(DELAY)
161                 timeSpent += DELAY
162
163         assert acquired, "Failed to grab file-lock %s within timeout %d" % (path, timeout)
164
165         try:
166                 yield fd
167         finally:
168                 os.unlink(path)