diff --git a/AUTHORS b/AUTHORS index dadbba7b..42e3e39b 100644 --- a/AUTHORS +++ b/AUTHORS @@ -2,7 +2,7 @@ Forrest Collman Jingpeng Wu Manuel Castro Nicholas Turner -Nico Kemnitz +Nico Kemnitz Pat Gunn Shang Mu Thomas Macrina diff --git a/cloudvolume/datasource/precomputed/image/rx.py b/cloudvolume/datasource/precomputed/image/rx.py index ad04c2e1..5220560b 100644 --- a/cloudvolume/datasource/precomputed/image/rx.py +++ b/cloudvolume/datasource/precomputed/image/rx.py @@ -36,6 +36,9 @@ progress_queue = None # defined in common.initialize_synchronization fs_lock = None # defined in common.initialize_synchronization +# Below this many chunks, a threaded local disk read is slower than a serial one. +MIN_THREADED_DISK_READS = 16 + def download_sharded( requested_bbox, mip, meta, cache, lru, lru_encoding, spec, @@ -707,14 +710,23 @@ def process(cloudpath, filename, enable_cache, locking): progress = "Downloading" total = len(locations["local"]) + len(locations["remote"]) + + def disk_concurrency(num_files): + """Threads worth spending on num_files reads off local disk. Greenlets stay serial.""" + if green or are_all_lru_hits or num_files < MIN_THREADED_DISK_READS: + return 0 + return DEFAULT_THREADS + n_threads = DEFAULT_THREADS - if meta.path.protocol in ("file", "mem") or are_all_lru_hits: + if meta.path.protocol == "mem" or are_all_lru_hits: n_threads = 0 + elif meta.path.protocol == "file": + n_threads = disk_concurrency(total) with tqdm(desc=progress, total=total, disable=(not progress)) as pbar: schedule_jobs( - fns=local_downloads, - concurrency=0, + fns=local_downloads, + concurrency=disk_concurrency(len(locations['local'])), progress=pbar, total=len(locations['local']), green=green, diff --git a/cloudvolume/datasource/precomputed/image/tx.py b/cloudvolume/datasource/precomputed/image/tx.py index ba1b8e5b..cec07840 100644 --- a/cloudvolume/datasource/precomputed/image/tx.py +++ b/cloudvolume/datasource/precomputed/image/tx.py @@ -461,7 +461,8 @@ def process_and_update(i, *args, **kwargs): if callable(progress): progress() - if remote.protocol in ("file", "mem"): + # Greenlets cannot overlap C-level compression. + if remote.protocol == "mem" or (remote.protocol == "file" and green): n_threads = 0 schedule_jobs( diff --git a/cloudvolume/threaded_queue.py b/cloudvolume/threaded_queue.py index bca4e232..7ef152e2 100644 --- a/cloudvolume/threaded_queue.py +++ b/cloudvolume/threaded_queue.py @@ -18,6 +18,8 @@ def __init__(self, n_threads, queue_size=0, progress=None): self._terminate = threading.Event() self._processed_lock = threading.Lock() + self._done_cond = threading.Condition(self._processed_lock) + self._outstanding = 0 self.processed = 0 self._inserted = 0 @@ -45,6 +47,8 @@ def put(self, fn): Returns: self """ + with self._processed_lock: + self._outstanding += 1 self._inserted += 1 self._queue.put(fn, block=True) return self @@ -150,6 +154,8 @@ def _consume_queue(self, terminate_evt): self._consume_queue_execution(fn) except Exception as err: self._error_queue.put(err) + with self._done_cond: + self._done_cond.notify_all() self._close_interface(interface) @@ -178,7 +184,10 @@ def _consume_queue_execution(self, fn): finally: with self._processed_lock: self.processed += 1 + self._outstanding -= 1 self._queue.task_done() + if self._outstanding == 0: + self._done_cond.notify_all() def _check_errors(self): try: @@ -211,6 +220,16 @@ def wait(self, progress=None): if type(progress) is str: desc = progress + if not progress: + # Woken by the last task completing, or by a thread posting an error. + with self._done_cond: + while self._outstanding and self._error_queue.empty(): + self._done_cond.wait() + self._check_errors() + if self._queue.empty(): + self._inserted = 0 + return self + last = self._inserted with tqdm(total=self._inserted, disable=(not progress), desc=desc) as pbar: # Allow queue to consume, but check up on