Closed yarikoptic closed 1 year ago
@yarikoptic
initiated multiple
git rm
in parallel? may be we need thread locking there to prevent that?
We already acquire a lock on the AsyncDataset
object before calling git rm
. If you're referring to the fact that there are multiple lines of the form "Running: git ... rm ...", those are simply the messages logged before executing the commands; no message is logged after the commands return, so you cannot infer from the logs whether their runtimes overlapped.
initiated
rm
before precommit assuredaddurl --batch
to terminate
ds.repo.precommit()
isn't supposed to terminate the git-annex addurl
process. addurl
was only terminated because the backup process was shutting down after an error occurred.
no message is logged after the commands return, so you cannot infer from the logs whether their runtimes overlapped.
oh, ok. Please add logging on when any particular command finished -- it could help us to identify such overlaps in the future
ds.repo.precommit()
isn't supposed to terminate thegit-annex addurl
process.addurl
was only terminated because the backup process was shutting down after an error occurred.
ah! that must be it then
to be safe, we need to add termination of non-readonly registerurl
and addurl
to be done there alongside with ds.repo.precommit()
.
@yarikoptic Are you sure it's necessary to terminate registerurl
and addurl
before removing a file? I can only see that being a giant hassle.
@yarikoptic Are you sure it's necessary to terminate
registerurl
andaddurl
before removing a file? I can only see that being a giant hassle.
unless you see what else could hold the lock there: it seems to be needed here -- before we identified that running precommit, which stops all batch processes, resolved similar problem. We also do that in DataLad itself specifically for rm
: https://github.com/datalad/datalad/blob/HEAD/datalad/support/gitrepo.py#L1338 . So, unless we better identify what is holding the lock and how to mitigate -- I think this is the only way ATM.
@yarikoptic I've posted a question about this on the git-annex forum: https://git-annex.branchable.com/forum/Batch_process__44___git_rm__44___and_locking/
may be we could retry rm
a few times if hitting the lock?
Given the response from Joey the only other possibility (to interrupting addurls) I see is to add internal (thread?) locking to not run rm
while addurl
ing a file. (asked there about registerurl
just in case -- from description it shouldn't have the same locking need)
@yarikoptic
add internal (thread?) locking to not run
rm
whileaddurl
ing a file.
That's not how the program works. addurl
isn't called one invocation per file like rm
; it's a long-running batch process that is run while assets are being processed/downloaded. Moreover, the routine for processing an individual asset basically looks like this:
git rm
on the old assetregisterurl
addurl
Since we're processing multiple assets at a time, calls to git rm
are necessarily happening while addurl
is running.
I see two ways to separate the git rm
and addurl
calls:
addurl
and instead invoke a separate process for each URL so that git rm
and addurl
can be guarded by the same lock
addurl
would run at a time, in contrast to the current behavior where multiple URLs are downloaded in parallel.addurl
, store them in a list and run addurl
after all other asset processing is complete
@yarikoptic
add internal (thread?) locking to not run
rm
whileaddurl
ing a file.That's not how the program works.
addurl
isn't called one invocation per file likerm
; it's a long-running batch process that is run while assets are being processed/downloaded.
Correct, addurl
is batched process and long running. What I meant follows Joey's description:
With --batch, git-annex reads a line, handles it, outputs a response, and waits for the next line. It should not keep any lock held while it's waiting for the line.
so what I was suggesting is to lock internally for the duration of "handles it", which is (I guess) from moment in feed_addurl to the moment in read_addurl , or something like that, or may be making feed_addurl/read_addurl
less async and making into a single function to feed(lock)/read progress/finish(unlock). I do not think we want to de-batch addurl
since indeed it would considerably slow things down since git annex
startup time is notable etc.
Instead of immediately feeding text assets to
addurl
, store them in a list and runaddurl
after all other asset processing is complete
- This might increase the runtime a bit. Other than that, I'm unable to see any complications or drawbacks to this idea, but I wouldn't be surprised if there's something I'm missing.
I think something along these lines would work too! I.e. to "stage" operations: figure out what to be removed or added first, then perform removal and then addition as two separate stages. I think performance hit should be minimal.
@yarikoptic
so what I was suggesting is to lock internally for the duration of "handles it", which is (I guess) from moment in feed_addurl to the moment in read_addurl , or something like that, or may be making
feed_addurl/read_addurl
less async and making into a single function to feed(lock)/read progress/finish(unlock). I do not think we want to de-batchaddurl
since indeed it would considerably slow things down sincegit annex
startup time is notable etc.
I don't see a way to make either of these approaches work without eliminating the use of addurl
's parallelism, which would slow things down. Currently, addurl
runs a configurable number of download jobs in parallel (backups2datalad.cfg.yaml
currently sets this to 5 jobs). Adding locking so that either git rm
is running or up to 5 download jobs are running seems to be beyond the capabilities of the synchronization primitives provided by anyio.
Adding locking so that either
git rm
is running or up to 5 download jobs are running seems to be beyond the capabilities of the synchronization primitives provided by anyio.
I think we might be able to use need_index_change
lock which we need to acquire for entire rm
call or when manipulating (adding/removing) self.in_progress
so that lock is acquired only if in_progress
gets first entry, and released when the last entry pop'ed from in_progress
. Might need an auxiliary changing_in_progres
lock so only one thread does analysis over in_progress
ATM on either to acquire or release need_index_change
lock). WDYT (or I need to expand more on the idea)?
@yarikoptic Your "need_index_change
lock" link does not mention anything named need_index_change
, nor does that term appear anywhere in the anyio documentation. What are you referring to there?
BTW (didn't check the code) but hitting this lock and Joey's response made me think - are we certain to not collide by rm
and addurl
the same file, i.e. actually needing that clean separation of stages to start with or might end up with race condition over rm
and addurl
'ing the same path?
need_index_change
I just gave a suggestive name we would use for such a lock to describe its purpose -- to lock for operation requiring git index change (which both rm
and addurl
do)
@yarikoptic The code for processing an asset calls git rm
and addurl
on a given path in two separate steps, and it waits for the git rm
to complete before moving on. git rm
will never be called on a different asset path than the one being operated on.
@yarikoptic
I think we might be able to use
need_index_change
lock which we need to acquire for entirerm
call or when manipulating (adding/removing)self.in_progress
so that lock is acquired only ifin_progress
gets first entry, and released when the last entry pop'ed fromin_progress
.
We would have to use a semaphore instead of a lock, as adding entries to in_progress
happens in a different task than removing entries, and, as your link says, "only the task that acquired the lock is allowed to release it."
Might need an auxiliary
changing_in_progres
lock so only one thread does analysis overin_progress
ATM on either to acquire or releaseneed_index_change
lock).
I have to point this out to you at some point: asychronous programming in Python does not use threads*; it uses asynchronous tasks, all in a single thread, that are managed by an event loop that switches between them at await
points.
Either way, having a single task for managing in_progress
and need_index_change
would mean the tasks that currently add & remove items to/from in_progress
would need to communicate with this new task over a stream (or, if we were using threads, a synchronized queue; not sure where you're expecting a changing_in_progress
lock to fit in), and that would end up being more complicated than just using a semaphore.
* Technically, the async system has the ability to wait on blocking synchronous operations as though they were async by running them in a separate thread, and this is used to implement things like async file I/O, but the primary design of async is completely thread-free.
I thought about Semaphore but I don't think we need to count anything...
Might need an auxiliary
changing_in_progres
lock so only one thread does analysis overin_progress
ATM on either to acquire or releaseneed_index_change
lock).I have to point this out to you at some point: asychronous programming in Python does not use threads*; it uses asynchronous tasks, all in a single thread, that are managed by an event loop that switches between them at
await
points.
thanks (possibly again) for describing how async works -- eventually I will remember it by heart ;) If not threads, then I do not exactly grasp why we need to do some "communication" which you describe. And since we already use anyio.Lock
for "rm"
invocation, why something like this would not be sufficient?:
diff --git a/tools/backups2datalad/asyncer.py b/tools/backups2datalad/asyncer.py
index b8d6ab3..841fb60 100644
--- a/tools/backups2datalad/asyncer.py
+++ b/tools/backups2datalad/asyncer.py
@@ -125,6 +125,7 @@ class Downloader:
download_receiver: MemoryObjectReceiveStream[ToDownload] = field(init=False)
zarrs: dict[str, ZarrLink] = field(init=False, default_factory=dict)
need_add: list[str] = field(init=False, default_factory=list)
+ in_progress_lock: anyio.Lock = field(init=False, default_factory=anyio.Lock)
def __post_init__(self) -> None:
(
@@ -378,11 +379,21 @@ class Downloader:
async with self.addurl.p.stdin:
async with self.download_receiver:
async for td in self.download_receiver:
- self.in_progress[td.path] = td
+ with self.in_progress_lock:
+ if not self.in_progress: # first "active" addurl - need to acquire dataset wide lock
+ self.ds.lock.acquire()
+ self.in_progress[td.path] = td
self.log.info("%s: Downloading from %s", td.path, td.url)
await self.addurl.send(f"{td.url} {td.path}\n")
self.log.debug("Done feeding URLs to addurl")
+ def in_progress_pop(self, path):
+ with self.in_progress_lock:
+ val = self.in_progress.pop(path)
+ if self.in_progress: # if was last active - release acquire dataset wide lock
+ self.ds.lock.release()
+ return val
+
async def read_addurl(self) -> None:
async with aclosing(self.addurl) as lineiter:
async for line in lineiter:
@@ -399,7 +410,7 @@ class Downloader:
elif not data["success"]:
msg = format_errors(data["error-messages"])
self.log.error("%s: download failed:%s", data["file"], msg)
- self.in_progress.pop(data["file"])
+ self.in_progress_pop(data["file"])
if "exited 123" in msg:
self.log.info(
"Will try `git add`ing %s manually later", data["file"]
@@ -412,7 +423,7 @@ class Downloader:
key = data.get("key")
self.log.info("%s: Finished downloading (key = %s)", path, key)
self.report.downloaded += 1
- dl = self.in_progress.pop(path)
+ dl = self.in_progress_pop(path)
self.tracker.finish_asset(dl.path)
self.nursery.start_soon(
self.check_unannexed_hash,
@yarikoptic
I think you want that if self.in_progress
in in_progress_pop()
to be if not self.in_progress
, don't you?
feed_addurl()
and read_addurl()
run in two separate tasks, started here; hence, self.ds.lock
cannot be acquired by one of them and released by the other, as only the task that acquired the lock is allowed to release it. The way around this is to use a semaphore with a limit of 1, which is just like a lock but without the "only the task that acquired the lock is allowed to release it" restriction.
We could either change the type of AsyncDataset.lock
to a semaphore that is also acquired & released by {feed,read}_addurl()
as you describe, or we could add a separate semaphore on the Downloader
instance that is acquired & released by {feed,read}_addurl()
and also acquired & released by process_asset()
when calling self.ds.remove()
.
I believe in_progress_lock
is unnecessary in your suggested patch. In feed_addurl()
, the only await
point inside async with self.in_progress_lock
(You need to use "async with
" there) is at await self.ds.lock.acquire()
(You need to use "await
" there); thus, this is the only point at which execution can jump to a different task while holding the lock. The only task that can be jumped to that cares about self.in_progress
is read_addurl()
, and if self.in_progress
is empty (which is necessary to await acquire()
and possibly change task), then nothing is being downloaded and in_progress_pop()
should not be called (and if it is, we get a KeyError
, lock or no lock), and so nothing will be interferring with self.in_progress
and there's no reason to lock inside feed_addurl()
.
In read_addurl()
and in_progress_pop()
on the other hand, the call to self.ds.lock.release()
is instantaneous rather than asynchronous (If you have a lock, you don't need to wait for anything if you want to release it), so there are no context switches inside in_progress_pop()
and thus nothing can meddle with self.in_progress()
while the function is running.
Adding communication between tasks would only be necessary if we, for some reason, moved all the manipulation of in_progress
to a dedicated task that ran concurrently with {feed,read}_addurl()
; this dedicated task would receive instructions from the other tasks and send back any results.
@yarikoptic
- I think you want that
if self.in_progress
inin_progress_pop()
to beif not self.in_progress
, don't you?
correct! ;)
feed_addurl()
andread_addurl()
run in two separate tasks, started here; hence,self.ds.lock
cannot be acquired by one of them and released by the other, as only the task that acquired the lock is allowed to release it. The way around this is to use a semaphore with a limit of 1, which is just like a lock but without the "only the task that acquired the lock is allowed to release it" restriction.
ah, cool, so could be done!
We could either change the type of
AsyncDataset.lock
to a semaphore that is also acquired & released by{feed,read}_addurl()
as you describe, or we could add a separate semaphore on theDownloader
instance that is acquired & released by{feed,read}_addurl()
and also acquired & released byprocess_asset()
when callingself.ds.remove()
.
I think keeping the lock
(changed to semaphore) within AsyncDataset
makes most sense since it is not about Downloader per se, but rather locking on the underlying git operations.
- I believe
in_progress_lock
is unnecessary in your suggested patch. ...
Thanks for the analysis! Given better understanding (no threads!) indeed should not be needed.
Could you please craft complete PR based on this diff/discussion?
We had similar issue before (https://github.com/dandi/dandisets/issues/290) which was addressed by #292 invoking .repo.precommit to stop all background batch processes before invoking
git rm
. In the past day we had similar but on different files crash. Here is output from the run which I did manually after hard resetting and git cleaning 000026that log file has at the end
suggesting that we
git rm
in parallel? may be we need thread locking there to prevent that?rm
before precommit assuredaddurl --batch
to terminate, which might be what has held that lock file, although fuser was not really reporting any process to hold it. is logic inprecommit
flawed?situation seems to be quite reproducible and should be troubleshooted and addressed. meanwhile I will exclude 000026 from the cron job