Lines Matching refs:self
90 def __init__(self, nworkers=0, name="Pool"): argument
95 self._closed = False
96 self._tasks = task_group()
97 self._pool = [None,]*default_num_threads() # Dask asks for len(_pool)
99 def apply(self, func, args=(), kwds=dict()): argument
102 return self.apply_async(func, args, kwds).get()
104 def map(self, func, iterable, chunksize=None): argument
112 return self.map_async(func, iterable, chunksize).get()
114 def imap(self, func, iterable, chunksize=1): argument
129 self._create_sequences(func, iterable, chunksize, collector)
132 def imap_unordered(self, func, iterable, chunksize=1): argument
138 self._create_sequences(func, iterable, chunksize, collector)
141 def apply_async(self, func, args=(), kwds=dict(), callback=None): argument
150 assert not self._closed # No lock here. We assume it's atomic...
153 self._tasks.run(job)
156 def map_async(self, func, iterable, chunksize=None, callback=None): argument
167 if not self._create_sequences(func, iterable, chunksize, collector):
171 def imap_async(self, func, iterable, chunksize=None, callback=None): argument
183 if not self._create_sequences(func, iterable, chunksize, collector):
187 def imap_unordered_async(self, func, iterable, chunksize=None, argument
200 if not self._create_sequences(func, iterable, chunksize, collector):
204 def close(self): argument
209 self._closed = True
211 def terminate(self): argument
215 self.close()
216 self._tasks.cancel()
218 def join(self): argument
221 self._tasks.wait()
223 def __enter__(self): argument
224 return self
226 def __exit__(self, exc_type, exc_value, traceback): argument
227 self.join()
229 def __del__(self): argument
230 self.terminate()
231 self.join()
233 def _create_sequences(self, func, iterable, chunksize, collector): argument
244 assert not self._closed # No lock here. We assume it's atomic...
262 self._tasks.run(t)
269 def __init__(self, func, args, kwds, apply_result): argument
275 self._func = func
276 self._args = args
277 self._kwds = kwds
278 self._result = apply_result
280 def __call__(self): argument
287 result = self._func(*self._args, **self._kwds)
289 self._result._set_exception()
291 self._result._set_value(result)
298 def __init__(self, jobs): argument
299 self._jobs = jobs
301 def __call__(self): argument
305 for job in self._jobs:
318 def __init__(self, collector=None, callback=None): argument
327 self._success = False
328 self._event = threading.Event()
329 self._data = None
330 self._collector = None
331 self._callback = callback
334 collector.register_result(self)
335 self._collector = collector
337 def get(self, timeout=None): argument
344 if not self.wait(timeout):
346 if self._success:
347 return self._data
348 raise self._data[0](self._data[1]).with_traceback(self._data[2])
350 def wait(self, timeout=None): argument
353 self._event.wait(timeout)
354 return self._event.isSet()
356 def ready(self): argument
358 return self._event.isSet()
360 def successful(self): argument
364 assert self.ready()
365 return self._success
367 def _set_value(self, value): argument
372 assert not self.ready()
373 self._data = value
374 self._success = True
375 self._event.set()
376 if self._collector is not None:
377 self._collector.notify_ready(self)
378 if self._callback is not None:
380 self._callback(value)
384 def _set_exception(self): argument
390 assert not self.ready()
391 self._data = sys.exc_info()
392 self._success = False
393 self._event.set()
394 if self._collector is not None:
395 self._collector.notify_ready(self)
405 def __init__(self, to_notify): argument
410 self._to_notify = to_notify
412 def register_result(self, apply_result): argument
421 def notify_ready(self, apply_result): argument
430 def _get_result(self, idx, timeout=None): argument
441 def __iter__(self): argument
443 return CollectorIterator(self)
452 def __init__(self, collector): argument
454 self._collector = collector
455 self._idx = 0
457 def __iter__(self): argument
458 return self
460 def next(self, timeout=None): argument
465 apply_result = self._collector._get_result(self._idx, timeout)
468 self._idx = 0
471 self._idx = 0
473 self._idx += 1
477 def __next__(self): argument
478 return self.next()
487 def __init__(self, to_notify=None): argument
492 AbstractResultCollector.__init__(self, to_notify)
493 self._cond = threading.Condition()
494 self._collection = []
495 self._expected = 0
497 def register_result(self, apply_result): argument
504 self._expected += 1
506 def _get_result(self, idx, timeout=None): argument
515 self._cond.acquire()
517 if idx >= self._expected:
519 elif idx < len(self._collection):
520 return self._collection[idx]
521 elif idx != len(self._collection):
525 self._cond.wait(timeout=timeout)
527 return self._collection[idx]
532 self._cond.release()
534 def notify_ready(self, apply_result=None): argument
542 self._cond.acquire()
544 self._collection.append(apply_result)
545 first_item = (len(self._collection) == 1)
547 self._cond.notifyAll()
549 self._cond.release()
551 if first_item and self._to_notify is not None:
552 self._to_notify._set_value(iter(self))
561 def __init__(self, to_notify=None, as_iterator=True): argument
570 AbstractResultCollector.__init__(self, to_notify)
571 self._results = []
572 self._lock = threading.Lock()
573 self._remaining = 0
574 self._as_iterator = as_iterator
576 def register_result(self, apply_result): argument
583 self._results.append(apply_result)
584 self._remaining += 1
586 def _get_result(self, idx, timeout=None): argument
595 res = self._results[idx]
599 def notify_ready(self, apply_result): argument
608 self._lock.acquire()
610 assert self._remaining > 0
611 got_first = (len(self._results) == self._remaining)
612 self._remaining -= 1
613 got_last = (self._remaining == 0)
615 self._lock.release()
617 if self._to_notify is not None:
618 if self._as_iterator and got_first:
619 self._to_notify._set_value(iter(self))
620 elif not self._as_iterator and got_last:
622 lst = [r.get(0) for r in self._results]
624 self._to_notify._set_exception()
626 self._to_notify._set_value(lst)