mirrors/copyparty
1
0
Fork 0
mirror of https://github.com/9001/copyparty.git synced 2025-08-18 09:22:31 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix a spin

Browse source
This commit is contained in:
ed 2022-07-03 02:39:15 +02:00
parent 87bb49da36
commit 4792c2770d
1 changed files with 12 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
copyparty/up2k.py
View file

@ -106,6 +106,7 @@ class Up2k(object):
self.pending_tags: list[tuple[set[str], str, str, dict[str, Any]]] = []
self.hashq: Queue[tuple[str, str, str, str, float]] = Queue()
self.tagq: Queue[tuple[str, str, str, str]] = Queue()
self.tag_event = threading.Condition()
self.n_hashq = 0
self.n_tagq = 0
self.mpool_used = False
@ -958,6 +959,7 @@ class Up2k(object):
to_delete = {}
in_progress = {}
while True:
did_nothing = True
with self.mutex:
if gid != self.gid:
break
@ -979,6 +981,7 @@ class Up2k(object):
zq2 = cur.execute(q, (w,)).fetchall()
have: dict[str, Union[str, float]] = {x[0]: 1 for x in zq2}
did_nothing = False
parsers = self._get_parsers(ptop, have, abspath)
if not parsers:
to_delete[w] = True
@ -992,6 +995,7 @@ class Up2k(object):
done = self._flush_mpool(wcur)
for w in done:
to_delete[w] = True
did_nothing = False
in_progress.pop(w)
n_done += 1
@ -1004,6 +1008,10 @@ class Up2k(object):
if not warks:
break
if did_nothing:
with self.tag_event:
self.tag_event.wait(0.2)
if not jobs:
continue
@ -1132,6 +1140,10 @@ class Up2k(object):
except:
ex = traceback.format_exc()
self._log_tag_err(qe.mtp or self.mtag.backend, qe.abspath, ex)
finally:
if qe.mtp:
with self.tag_event:
self.tag_event.notify_all()
q.task_done()