| 3 | from eventhandler import EventHandler, event_loop |
| 4 | |
| 5 | class ThreadPoolHandler(EventHandler): |
| 6 | def __init__(self, nworkers): |
| 7 | self.signal_done_sock, self.done_sock = socket.socketpair() |
| 8 | self.pending = [] |
| 9 | self.pool = ThreadPoolExecutor(nworkers) |
| 10 | |
| 11 | def fileno(self): |
| 12 | return self.done_sock.fileno() |
| 13 | |
| 14 | # Callback that executes when the thread is done |
| 15 | def _complete(self, callback, r): |
| 16 | self.pending.append((callback, r.result())) |
| 17 | self.signal_done_sock.send(b'x') |
| 18 | |
| 19 | # Run a function in a thread pool |
| 20 | def run(self, func, args=(), kwargs={},*,callback): |
| 21 | r = self.pool.submit(func, *args, **kwargs) |
| 22 | r.add_done_callback(lambda r: self._complete(callback, r)) |
| 23 | |
| 24 | def wants_to_receive(self): |
| 25 | return True |
| 26 | |
| 27 | # Run callback functions of completed work |
| 28 | def handle_receive(self): |
| 29 | # Invoke all pending callback functions |
| 30 | for callback, result in self.pending: |
| 31 | callback(result) |
| 32 | self.done_sock.recv(1) |
| 33 | self.pending = [] |
| 34 | |
| 35 | # A really bad fibonacci implementation |
| 36 | def fib(n): |