]>
Commit | Line | Data |
---|---|---|
8bbd1bbf RD |
1 | """ |
2 | This module supports the thread-safe, asynchronous transmission of data | |
3 | ('delayed results') from a worker (non-GUI) thread to the main thread. Ie you don't | |
4 | need to mutex lock any data, the worker thread doesn't wait (or even check) | |
5 | for the result to be received, and the main thread doesn't wait for the | |
6 | worker thread to send the result. Instead, the consumer will be called | |
7 | automatically by the wx app when the worker thread result is available. | |
8 | ||
9 | In most cases you just need to use startWorker() with the correct parameters | |
10 | (your worker function and your 'consumer' in the simplest of cases). The | |
11 | only requirement on consumer is that it must accept a DelayedResult instance | |
12 | as first arg. | |
13 | ||
14 | In the following example, this will call consumer(delayedResult) with the | |
15 | return value from workerFn:: | |
16 | ||
17 | from delayedresult import startWorker | |
18 | startWorker(consumer, workerFn) | |
19 | ||
20 | More advanced uses: | |
21 | ||
22 | - The other parameters to startWorker() | |
23 | - Derive from Producer to override _extraInfo (e.g. to provide traceback info) | |
24 | - Create your own worker-function-thread wrapper instead of using Producer | |
25 | - Create your own Handler-like wrapper to pre- or post-process the result | |
26 | (see PreProcessChain) | |
27 | - Derive from Sender to use your own way of making result hop over the | |
28 | "thread boundary" (from non-main thread to main thread), e.g. using Queue | |
29 | ||
30 | Thanks to Josiah Carlson for critical feedback/ideas that helped me | |
31 | improve this module. | |
32 | ||
33 | :Copyright: (c) 2006 by Oliver Schoenborn | |
34 | :License: wxWidgets license | |
35 | :Version: 1.0 | |
36 | ||
37 | """ | |
38 | ||
39 | __author__ = 'Oliver Schoenborn at utoronto dot ca' | |
40 | __version__ = '1.0' | |
41 | ||
42 | __all__ = ('Sender', 'SenderNoWx', 'SenderWxEvent', 'SenderCallAfter', | |
43 | 'Handler', 'DelayedResult', 'Producer', 'startWorker', 'PreProcessChain') | |
44 | ||
45 | ||
46 | import wx | |
47 | import threading | |
48 | ||
49 | ||
50 | class Struct: | |
51 | """ | |
52 | An object that has attributes built from the dictionary given in | |
53 | constructor. So ss=Struct(a=1, b='b') will satisfy assert ss.a == 1 | |
54 | and assert ss.b == 'b'. | |
55 | """ | |
56 | ||
57 | def __init__(self, **kwargs): | |
58 | self.__dict__.update( kwargs ) | |
59 | ||
60 | ||
61 | class Handler: | |
62 | """ | |
63 | Bind some of the arguments and keyword arguments of a callable ('listener'). | |
64 | Then when the Handler instance is called (e.g. handler(result, **kwargs)) | |
65 | the result is passed as first argument to callable, the kwargs is | |
66 | combined with those given at construction, and the args are those | |
67 | given at construction. Its return value is returned. | |
68 | """ | |
69 | def __init__(self, listener, *args, **kwargs ): | |
70 | """Bind args and kwargs to listener. """ | |
71 | self.__listener = listener | |
72 | self.__args = args | |
73 | self.__kwargs = kwargs | |
74 | ||
75 | def __call__(self, result, **moreKwargs): | |
76 | """Listener is assumed to take result as first arg, then *args, | |
77 | then the combination of moreKwargs and the kwargs given at construction.""" | |
78 | if moreKwargs: | |
79 | moreKwargs.update(self.__kwargs) | |
80 | else: | |
81 | moreKwargs = self.__kwargs | |
82 | return self.__listener(result, *self.__args, **moreKwargs) | |
83 | ||
84 | ||
85 | class Sender: | |
86 | """ | |
87 | Base class for various kinds of senders. A sender sends a result | |
88 | produced by a worker funtion to a result handler (listener). Note | |
89 | that each sender can be given a "job id". This can be anything | |
90 | (number, string, id, and object, etc) and is not used, it is | |
91 | simply added as attribute whenever a DelayedResult is created. | |
92 | This allows you to know, if desired, what result corresponds to | |
93 | which sender. Note that uniqueness is not necessary. | |
94 | ||
95 | Derive from this class if none of the existing derived classes | |
96 | are adequate, and override _sendImpl(). | |
97 | """ | |
98 | ||
99 | def __init__(self, jobID=None): | |
100 | """The optional jobID can be anything that you want to use to | |
101 | track which sender particular results come from. """ | |
102 | self.__jobID = jobID | |
103 | ||
104 | def getJobID(self): | |
105 | """Return the jobID given at construction""" | |
106 | return self.__jobID | |
107 | ||
108 | def sendResult(self, result): | |
109 | """This will send the result to handler, using whatever | |
110 | technique the derived class uses. """ | |
111 | delayedResult = DelayedResult(result, jobID=self.__jobID) | |
112 | self._sendImpl(delayedResult) | |
113 | ||
114 | def sendException(self, exception, extraInfo = None): | |
115 | """Use this when the worker function raised an exception. | |
116 | The *exception* is the instance of Exception caught. The extraInfo | |
117 | could be anything you want (e.g. locals or traceback etc), | |
118 | it will be added to the exception as attribute 'extraInfo'. The | |
119 | exception will be raised when DelayedResult.get() is called.""" | |
120 | assert exception is not None | |
121 | delayedResult = DelayedResult(extraInfo, | |
122 | exception=exception, jobID=self.__jobID) | |
123 | self._sendImpl(delayedResult) | |
124 | ||
125 | def _sendImpl(self, delayedResult): | |
126 | msg = '_sendImpl() must be implemented in %s' % self.__class__ | |
127 | raise NotImplementedError(msg) | |
128 | ||
129 | ||
130 | class SenderNoWx( Sender ): | |
131 | """ | |
132 | Sender that works without wx. The results are sent directly, ie | |
133 | the consumer will get them "in the worker thread". So it should | |
134 | only be used for testing. | |
135 | """ | |
136 | def __init__(self, consumer, jobID=None, args=(), kwargs={}): | |
137 | """The consumer can be any callable of the form | |
138 | callable(result, *args, **kwargs)""" | |
139 | Sender.__init__(self, jobID) | |
140 | if args or kwargs: | |
141 | self.__consumer = Handler(consumer, *args, **kwargs) | |
142 | else: | |
143 | self.__consumer = consumer | |
144 | ||
145 | def _sendImpl(self, delayedResult): | |
146 | self.__consumer(delayedResult) | |
147 | ||
148 | ||
149 | class SenderWxEvent( Sender ): | |
150 | """ | |
151 | This sender sends the delayed result produced in the worker thread | |
152 | to an event handler in the main thread, via a wx event of class | |
153 | *eventClass*. The result is an attribute of the event (default: | |
154 | "delayedResult". | |
155 | """ | |
156 | def __init__(self, handler, eventClass, resultAttr="delayedResult", | |
157 | jobID=None, **kwargs): | |
158 | """The handler must derive from wx.EvtHandler. The event class | |
159 | is typically the first item in the pair returned by | |
160 | wx.lib.newevent.NewEvent(). You can use the *resultAttr* | |
161 | to change the attribute name of the generated event's | |
162 | delayed result. """ | |
163 | Sender.__init__(self, jobID) | |
164 | if not isinstance(handler, wx.EvtHandler): | |
165 | msg = 'SenderWxEvent(handler=%s, ...) not allowed,' % type(handler) | |
166 | msg = '%s handler must derive from wx.EvtHandler' % msg | |
167 | raise ValueError(msg) | |
168 | self.__consumer = Struct(handler=handler, eventClass=eventClass, | |
169 | resultAttr=resultAttr, kwargs=kwargs) | |
170 | ||
171 | def _sendImpl(self, delayedResult): | |
172 | """Must not modify the consumer (that was created at construction) | |
173 | since might be shared by several senders, each sending from | |
174 | separate threads.""" | |
175 | consumer = self.__consumer | |
176 | kwargs = consumer.kwargs.copy() | |
177 | kwargs[ consumer.resultAttr ] = delayedResult | |
178 | event = consumer.eventClass(** kwargs) | |
179 | wx.PostEvent(consumer.handler, event) | |
180 | ||
181 | ||
182 | class SenderCallAfter( Sender ): | |
183 | """ | |
184 | This sender sends the delayed result produced in the worker thread | |
185 | to a callable in the main thread, via wx.CallAfter. | |
186 | """ | |
187 | def __init__(self, listener, jobID=None, args=(), kwargs={}): | |
188 | Sender.__init__(self, jobID) | |
189 | if args or kwargs: | |
190 | self.__consumer = Handler(listener, *args, **kwargs) | |
191 | else: | |
192 | self.__consumer = listener | |
193 | ||
194 | def _sendImpl(self, delayedResult): | |
195 | wx.CallAfter(self.__consumer, delayedResult) | |
196 | ||
197 | ||
198 | class DelayedResult: | |
199 | """ | |
200 | Represent the actual delayed result coming from the non-main thread. | |
201 | An instance of this is given to the result handler. This result is | |
202 | either a (reference to a) the value sent, or an exception. | |
203 | If the latter, the exception is raised when the get() method gets | |
204 | called. | |
205 | """ | |
206 | ||
207 | def __init__(self, result, jobID=None, exception = None): | |
208 | """You should never have to call this yourself. A DelayedResult | |
209 | is created by a concrete Sender for you.""" | |
210 | self.__result = result | |
211 | self.__exception = exception | |
212 | self.__jobID = jobID | |
213 | ||
214 | def getJobID(self): | |
215 | """Return the jobID given when Sender initialized, | |
216 | or None if none given. """ | |
217 | return self.__jobID | |
218 | ||
219 | def get(self): | |
220 | """Get the result. If an exception was sent instead of a result, | |
221 | (via Sender's sendExcept()), that **exception is raised**. | |
222 | Otherwise the result is simply returned. """ | |
223 | if self.__exception: # exception was raised! | |
224 | self.__exception.extraInfo = self.__result | |
225 | raise self.__exception | |
226 | ||
227 | return self.__result | |
228 | ||
229 | ||
45c0ea45 RD |
230 | class AbortedException(Exception): |
231 | """Raise this in your worker function so that the sender knows | |
232 | not to send a result to handler.""" | |
233 | pass | |
234 | ||
235 | ||
8bbd1bbf RD |
236 | class Producer(threading.Thread): |
237 | """ | |
238 | Represent the worker thread that produces delayed results. | |
239 | It causes the given function to run in a separate thread, | |
240 | and a sender to be used to send the return value of the function. | |
241 | As with any threading.Thread, instantiate and call start(). | |
45c0ea45 RD |
242 | Note that if the workerFn raises AbortedException, the result is not |
243 | sent and the thread terminates gracefully. | |
8bbd1bbf RD |
244 | """ |
245 | ||
246 | def __init__(self, sender, workerFn, args=(), kwargs={}, | |
247 | name=None, group=None, daemon=False, | |
248 | sendReturn=True, senderArg=None): | |
249 | """The sender will send the return value of | |
250 | workerFn(*args, **kwargs) to the main thread. The name and group | |
251 | are same as threading.Thread constructor parameters. Daemon causes | |
252 | setDaemon() to be called. If sendReturn is False, then the return | |
253 | value of workerFn() will not be sent. If senderArg is given, it | |
254 | must be the name of the keyword arg to use to pass the sender into | |
255 | the workerFn, so the function can send (typically many) results.""" | |
256 | if senderArg: | |
257 | kwargs[senderArg] = sender | |
258 | def wrapper(): | |
259 | try: | |
260 | result = workerFn(*args, **kwargs) | |
45c0ea45 RD |
261 | except AbortedException: |
262 | pass | |
8bbd1bbf RD |
263 | except Exception, exc: |
264 | extraInfo = self._extraInfo(exc) | |
265 | sender.sendException(exc, extraInfo) | |
266 | else: | |
267 | if sendReturn: | |
268 | sender.sendResult(result) | |
269 | ||
270 | threading.Thread.__init__(self, name=name, group=group, target=wrapper) | |
271 | if daemon: | |
272 | self.setDaemon(daemon) | |
273 | ||
274 | def _extraInfo(self, exception): | |
275 | """This method could be overridden in a derived class to provide | |
276 | extra information when an exception is being sent instead of a | |
277 | result. """ | |
278 | return None | |
279 | ||
280 | ||
45c0ea45 RD |
281 | class AbortEvent: |
282 | """ | |
283 | Convenience class that represents a kind of threading.Event that | |
284 | raises AbortedException when called (see the __call__ method, everything | |
285 | else is just to make it look like threading.Event). | |
286 | """ | |
287 | ||
288 | def __init__(self): | |
289 | self.__ev = threading.Event() | |
290 | ||
291 | def __call__(self, timeout=None): | |
292 | """See if event has been set (wait at most timeout if given). If so, | |
293 | raise AbortedException. Otherwise return None. Allows you to do | |
294 | 'while not event():' which will always succeed unless the event | |
295 | has been set (then AbortedException will cause while to exit).""" | |
296 | if timeout: | |
297 | self.__ev.wait(timeout) | |
298 | if self.__ev.isSet(): | |
299 | raise AbortedException() | |
300 | return None | |
301 | ||
302 | def __getattr__(self, name): | |
303 | """This allows us to be a kind of threading.Event.""" | |
304 | if name in ('set','clear','wait','isSet'): | |
305 | return getattr(self.__ev, name) | |
306 | ||
307 | ||
8bbd1bbf RD |
308 | def startWorker( |
309 | consumer, workerFn, | |
310 | cargs=(), ckwargs={}, | |
311 | wargs=(), wkwargs={}, | |
312 | jobID=None, group=None, daemon=False, | |
313 | sendReturn=True, senderArg=None): | |
314 | """ | |
315 | Convenience function to send data produced by workerFn(*wargs, **wkwargs) | |
316 | running in separate thread, to a consumer(*cargs, **ckwargs) running in | |
317 | the main thread. This function merely creates a SenderCallAfter (or a | |
318 | SenderWxEvent, if consumer derives from wx.EvtHandler), and a Producer, | |
319 | and returns immediately after starting the Producer thread. The jobID | |
320 | is used for the Sender and as name for the Producer thread. Returns the | |
321 | thread created, in case caller needs join/etc. | |
322 | """ | |
323 | ||
324 | if isinstance(consumer, wx.EvtHandler): | |
325 | eventClass = cargs[0] | |
326 | sender = SenderWxEvent(consumer, eventClass, jobID=jobID, **ckwargs) | |
327 | else: | |
328 | sender = SenderCallAfter(consumer, jobID, args=cargs, kwargs=ckwargs) | |
329 | ||
330 | thread = Producer( | |
331 | sender, workerFn, args=wargs, kwargs=wkwargs, | |
332 | name=jobID, group=group, daemon=daemon, | |
333 | senderArg=senderArg, sendReturn=sendReturn) | |
334 | ||
335 | thread.start() | |
336 | return thread | |
337 | ||
338 | ||
339 | class PreProcessChain: | |
340 | """ | |
341 | Represent a 'delayed result pre-processing chain', a kind of Handler. | |
342 | Useful when lower-level objects need to apply a sequence of transformations | |
343 | to the delayed result before handing it over to a final handler. | |
344 | This allows the starter of the worker function to not know | |
345 | anything about the lower-level objects. | |
346 | """ | |
347 | def __init__(self, handler, *args, **kwargs): | |
348 | """Wrap handler(result, *args, **kwargs) so that the result | |
349 | it receives has been transformed by us. """ | |
350 | if handler is None:# assume rhs is a chain | |
351 | self.__chain = args[0] | |
352 | else: | |
353 | if args or kwargs: | |
354 | handler = Handler(handler, *args, **kwargs) | |
355 | self.__chain = [handler] | |
356 | ||
357 | def addSub(self, callable, *args, **kwargs): | |
358 | """Add a sub-callable, ie a callable(result, *args, **kwargs) | |
359 | that returns a transformed result to the previously added | |
360 | sub-callable (or the handler given at construction, if this is | |
361 | the first call to addSub). """ | |
362 | self.__chain.append( Handler(callable, *args, **kwargs) ) | |
363 | ||
364 | def clone(self): | |
365 | """Clone the chain. Shallow only. Useful when several threads | |
366 | must be started but have different sub-callables. """ | |
367 | return PreProcessChain(None, self.__chain[:] ) | |
368 | ||
369 | def cloneAddSub(self, callable, *args, **kwargs): | |
370 | """Convenience method that first clones self, then calls addSub() | |
371 | on that clone with given arguments. """ | |
372 | cc = self.clone() | |
373 | cc.addSub(callable, *args, **kwargs) | |
374 | ||
375 | def count(self): | |
376 | """How many pre-processors in the chain""" | |
377 | return len(self.__chain) | |
378 | ||
379 | class Traverser: | |
380 | """ | |
381 | Traverses the chain of pre-processors it is given, transforming | |
382 | the original delayedResult along the way. The return value of each | |
383 | callable added via addSub() is given to the previous addSub() callable, | |
384 | until the handler is reached. | |
385 | """ | |
386 | def __init__(self, delayedResult, chain): | |
387 | self.__dr = delayedResult | |
388 | self.__chain = chain | |
389 | ||
390 | def get(self): | |
391 | """This makes handler think we are a delayedResult.""" | |
392 | if not self.__chain: | |
393 | return self.__dr.get() | |
394 | ||
395 | handler = self.__chain[0] | |
396 | del self.__chain[0] | |
397 | return handler(self) | |
398 | ||
399 | def getJobID(self): | |
400 | """Return the job id for the delayedResult we transform.""" | |
401 | return self.__dr.getJobID() | |
402 | ||
403 | ||
404 | def __call__(self, delayedResult): | |
405 | """This makes us a Handler. We just call handler(Traverser). The | |
406 | handler will think it is getting a delayed result, but in fact | |
407 | will be getting an instance of Traverser, which will take care | |
408 | of properly applying the chain of transformations to delayedResult.""" | |
409 | chainTrav = self.Traverser(delayedResult, self.__chain[1:]) | |
410 | handler = self.__chain[0] | |
411 | handler( chainTrav ) | |
412 |