objects.transfer: minor refactoring, move lazier taskset inside custom executor #6591
Conversation
skshetry
added
refactoring
dvc/utils/threadpool.py
Outdated
| class ThreadPoolExecutor(futures.ThreadPoolExecutor): | ||
| @property | ||
| def max_workers(self) -> int: | ||
| return self._max_workers | ||
|
|
||
| def imap_unordered( | ||
| self, fn: Callable[..., _T], *iterables: Iterable[Any] | ||
| ) -> Iterator[_T]: | ||
| """Lazier version of map that does not preserve ordering of results. | ||
|
|
||
| It does not create all the futures at once to reduce memory usage. | ||
| """ | ||
|
|
||
| def create_taskset(n: int) -> Set[futures.Future]: | ||
| return {self.submit(fn, *args) for args in islice(it, n)} | ||
|
|
||
| it = zip(*iterables) | ||
| tasks = create_taskset(self.max_workers * 5) | ||
| while tasks: | ||
| done, tasks = futures.wait( | ||
| tasks, return_when=futures.FIRST_COMPLETED | ||
| ) | ||
| for fut in done: | ||
| yield fut.result() | ||
| tasks.update(create_taskset(len(done))) |
There was a problem hiding this comment.
Btw, there was a plan to try using async here e.g. with fsspec's put/get methods. Maybe it would be reasonable to do that instead now?
There was a problem hiding this comment.
Could you please elaborate? I am not sure I understand.
There was a problem hiding this comment.
So we use thread pool executor right now to parallelize uploading/downloading of objects, but fsspec provides put/get methods that accept batches and async filesystems do that with async, effectively managing the coroutines themselves, instead of us manually using threads.
There was a problem hiding this comment.
So what you are suggesting is to have something like TransferManager that handles multithreaded and async transfer as well right?
There was a problem hiding this comment.
@skshetry Ah, no, I meant that we possibly won't need transfer manager at all, and could just fs.put(obj_list) and let it do its thing. So we would be effectively delegating transfer management to particular filesystem's put/get methods.
There was a problem hiding this comment.
@skshetry Btw, happy to jump on a call to talk about it a bit more.
There was a problem hiding this comment.
I think we should wait for fsspec changes to be done, this will likely require us to have colored APIs too (odb.add_async, fs.utils.transfer_async etc.). And, there's a problem of running fsspec's sync FS in async mode too.
As this change is a very minor restructuring, I'd prefer to merge this as-is.
skshetry
force-pushed
the
transfer-minor-refactoring
branch
from
a0eb98f
to
d4207c0
Compare
| break | ||
| except (FileNotFoundError, ObjectFormatError): | ||
| pass | ||
| dir_obj = find_tree_by_obj_id([cache_odb, src], dir_hash) |
There was a problem hiding this comment.
Why do we call it dir_obj instead of tree?
There was a problem hiding this comment.
It's a terminology mess (+ .dir objects as we call them, and not .tree). Also driving me insane for 3.0 🙂
There was a problem hiding this comment.
.dir is a physical repr thing similar to how it's json and kept in some structure, it's fine till 3.0. Though I'd say that we s/dir_obj/tree in all of these abstractions right away.
There was a problem hiding this comment.
It's a lot of leftovers from when remote was migrated to odb, and yeah it would be good to just drop the old terminology entirely whenever possible
No description provided.