Conversation
k-dominik
left a comment
k-dominik
left a comment
There was a problem hiding this comment.
Good to see fewer explicit locks! Really nice to rely on qeues for that.
lazyflow/request/threadPool.py
Outdated
| def _distribute_work(self): | ||
| while True: | ||
| try: | ||
| task = self.unassigned_tasks.get() |
There was a problem hiding this comment.
Shouldn't we stop the scheduler as well?
if task is STOP:
return
....
if worker is STOP:
return
....
def stop(...):
...
self.unassigned_tasks.put_nowait(STOP)
self.ready_workers.put_nowait(STOP)There was a problem hiding this comment.
It's a daemon thread and it doesn't do any actual work, so it will gracefully shut down by itself with app termination. Otherwise, we would need to add timeouts to both queues gets and introduce polling to properly terminate scheduler.
There was a problem hiding this comment.
Wouldn't putting STOP in both queues solve the problem?
Keeping a running thread after ThreadPool.stop() doesn't seem like a good behavior (it consumes resources and prevents GC for some objects), but this is probably not a big deal since we have 1 pool for the entire lifetime of a process anyway.
There was a problem hiding this comment.
And then we need to make STOP object comparable with objects in unassigned_tasks.
There was a problem hiding this comment.
It should be doable:
@functools.total_ordering
class Stop:
__lt__ = lambda _self, _other: True
__eq__ = lambda self, other: self.__class__ == other.__class__
STOP = Stop()And we also need to fix Request, which should be done anyway because right now it's __lt__ method is broken (works only with other request objects), and there is no total ordering:
@functools.total_ordering
class Request:
...
def __lt__(self, other):
if not hasattr(other, "_priority"):
return NotImplemented
return self._priority < other._priority
def __eq__(self, other):
if not hasattr(other, "_priority"):
return NotImplemented
return self._priority == other._priority This is probably even better because STOP will appear immediately as the next element in all queues, which is what we want.
There was a problem hiding this comment.
See updated revision
There was a problem hiding this comment.
Great! Explicitly passing the priority is an even better approach.
| # The ThreadPool._Worker loop has a local reference (next_task), | ||
| # so wait just a tic for the ThreadPool worker to cycle back to the top of its loop (and discard the reference) | ||
| time.sleep(0.1) | ||
| gc.collect() |
There was a problem hiding this comment.
Do we really need this in a test?
There was a problem hiding this comment.
It broke when I didn't free the reference :) More robust test would probably try to overload thread pool with work and than check if there was any memory leakage.
lazyflow/request/threadPool.py
Outdated
| __slots__ = ("obj", "exc", "priority") | ||
|
|
||
| def __init__(self, *, obj=None, exc=None, priority=None): | ||
| def __init__(self, *, obj: Any = None, exc: Exception = None, priority: Optional[List[int]] = None) -> None: |
There was a problem hiding this comment.
Maybe use a dataclass for this struct? Seems appropriate.
Also, could use TypeVar for obj in __init__ and unwrap:
There was a problem hiding this comment.
Don't really want to import dataclasses here, little benefit as I need my constructor for validation.
3 participants
No description provided.