perf(precomputed): re-enable multithreading for local file reads/writes

AUTHORS

@@ -2,7 +2,7 @@ Forrest Collman <forrest.collman@gmail.com>
 Jingpeng Wu <jingpeng.wu@gmail.com>
 Manuel Castro <macastro@princeton.edu>
 Nicholas Turner <nturner.stanford@gmail.com>
-Nico Kemnitz <nkemnitz@princeton.edu>
+Nico Kemnitz <nico@zetta.ai>
 Pat Gunn <pgunn01@gmail.com>
 Shang Mu <smu@princeton.edu>
 Thomas Macrina <thomas.macrina@gmail.com>

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,

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(

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