AsyncBuffer #88
AsyncBuffer #88
Conversation
crusaderky
commented
Mar 21, 2023
•
crusaderky
commented
- Partially closes [WIP] Asynchronous SpillBuffer distributed#7686
crusaderky
force-pushed
the
async_buffer2
branch
27 times, most recently
from
bd957ed
to
2bcd214
Compare
crusaderky
force-pushed
the
async_buffer2
branch
from
2bcd214
to
bda0501
Compare
crusaderky
force-pushed
the
async_buffer2
branch
from
bda0501
to
426ca0b
Compare
crusaderky
force-pushed
the
async_buffer2
branch
5 times, most recently
from
1c6518d
to
c009687
Compare
crusaderky
force-pushed
the
async_buffer2
branch
from
f66d9f8
to
67d1ec2
Compare
zict/async_buffer.py
Outdated
| if self.evicting is not None: | ||
| weight = min(weight, self.evicting) | ||
| if weight <= n: | ||
| return | ||
|
|
||
| self.evicting = n | ||
|
|
||
| def done(_: asyncio.Future) -> None: | ||
| self.evicting = None | ||
|
|
||
| # Note: this can get cancelled by LRU.close(), which in turn is | ||
| # triggered by Buffer.close() | ||
| future = self._offload(self.evict_until_below_target, n) | ||
| future.add_done_callback(done) |
There was a problem hiding this comment.
IIUC this is possibly called multiple times (even if it is locked) but the done callbacks of concurrently running evict_untiL_below_target would overwrite / first one would already reset evicting. Is this intentional?
There was a problem hiding this comment.
You're right, fixed
zict/async_buffer.py
Outdated
| if self.executor is None: | ||
| self.executor = ThreadPoolExecutor( | ||
| 1, thread_name_prefix="zict.Buffer offloader" | ||
| ) |
There was a problem hiding this comment.
nit: From an API POV I would suggest to allow the init to take an TPE and if none is provided, we'll initialize it during initialization of the class
| def async_get( | ||
| self, keys: Collection[KT], missing: Literal["raise", "omit"] = "raise" | ||
| ) -> asyncio.Future[dict[KT, VT]]: |
There was a problem hiding this comment.
Mostly curious, I don't think we should change this now but why is the API accepting multiple keys? is this somehow a requirement or an optimization?
The reason why I'm asking is because I think this drives some complexity and I would expect missing not to be required otherwise. I'm not entirely sure about this.
There was a problem hiding this comment.
It's an optimization to avoid having an extremely long queue to the thread pool, with each individual key incurring in thread synchronization overhead.
| # Do not pull keys towards the top of the LRU unless they are all available. | ||
| # This matters when there is a very long queue of async_get futures. | ||
| d = self.fast.get_all_or_nothing(keys) |
There was a problem hiding this comment.
Strictly speaking, fast doesn't need to be a LRU, does it? In this case, I suspect we'll get an AttributeError here, won't we?
There was a problem hiding this comment.
Buffer.fast is always an LRU:
Lines 79 to 85 in b96afc4
|
@fjetter all review comments have been addressed |
|
Ouch. That was a merge without squash. I force-pushed to fix it; could we force squashing in the repo like we already did in distributed? |
