Repository: seung-lab/cloud-volume
Pull request #342 (2020-05-07 23:01:10)

perf: parallelized ShardReader.exists
Need to group the minishard requests more effectively.

@akhileshh needs this to make the graphene server more responsive.

codecov-io: # [Codecov](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=h1) Report
> Merging [#342](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=desc) into [master](https://codecov.io/gh/seung-lab/cloud-volume/commit/f11719eee73d571017dbb86e9fc43577e60ca4ec&el=desc) will **decrease** coverage by `0.45%`.
> The diff coverage is `33.87%`.

[![Impacted file tree graph](https://codecov.io/gh/seung-lab/cloud-volume/pull/342/graphs/tree.svg?width=650&height=150&src=pr&token=7WwqdI3qkW)](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #342      +/-   ##
==========================================
- Coverage   78.91%   78.46%   -0.46%     
==========================================
  Files          75       75              
  Lines        9079     9157      +78     
==========================================
+ Hits         7165     7185      +20     
- Misses       1914     1972      +58     
```


| [Impacted Files](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [cloudvolume/cacheservice.py](https://codecov.io/gh/seung-lab/cloud-volume/pull/342/diff?src=pr&el=tree#diff-Y2xvdWR2b2x1bWUvY2FjaGVzZXJ2aWNlLnB5) | `77.46% <25.00%> (-2.32%)` | :arrow_down: |
| [cloudvolume/datasource/precomputed/sharding.py](https://codecov.io/gh/seung-lab/cloud-volume/pull/342/diff?src=pr&el=tree#diff-Y2xvdWR2b2x1bWUvZGF0YXNvdXJjZS9wcmVjb21wdXRlZC9zaGFyZGluZy5weQ==) | `72.20% <28.86%> (-8.49%)` | :arrow_down: |
| [cloudvolume/scheduler.py](https://codecov.io/gh/seung-lab/cloud-volume/pull/342/diff?src=pr&el=tree#diff-Y2xvdWR2b2x1bWUvc2NoZWR1bGVyLnB5) | `82.97% <73.33%> (-6.22%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=footer). Last update [f11719e...2592ba6](https://codecov.io/gh/seung-lab/cloud-volume/pull/342?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/cloudvolume/cacheservice.py b/cloudvolume/cacheservice.py
index b8dcd0a..09c80ae 100644
--- a/cloudvolume/cacheservice.py
+++ b/cloudvolume/cacheservice.py
@@ -1,8 +1,11 @@
+from functools import partial
 import json
 import os
 import posixpath
 import shutil
 
+from . import scheduler
+
 from .provenance import DataLayerProvenance
 from .storage import SimpleStorage, Storage, GreenStorage
 from .storage.storage_interfaces import COMPRESSION_EXTENSIONS
@@ -375,6 +378,78 @@ class CacheService(object):
 
     return filedata
 
+  def download_as(self, requests, compress=None, progress=None):
+    """
+    Works with byte ranges.
+
+    requests: [{
+      'path': ...,
+      'local_alias': ...,
+      'start': ...,
+      'end': ...,
+    }]
+    """
+    # import pdb; pdb.set_trace()
+    if len(requests) == 0:
+      return {}
+
+    progress = progress if progress is not None else self.config.progress
+
+    aliases = [ req['local_alias'] for req in requests ]
+    alias_tuples = { req['local_alias']: (req['path'], req['start'], req['end']) for req in requests }
+    alias_to_path = { req['local_alias']: req['path'] for req in requests }
+    path_to_alias = { v:k for k,v in alias_tuples.items() }
+
+    if None in alias_to_path:
+      del alias_to_path[None]
+      del alias_tuples[None]
+
+    locs = self.compute_data_locations(aliases)
+
+    fragments = {}
+    if self.enabled:
+      fragments = self.get(locs['local'], progress=progress)
+      keys = list(fragments.keys())
+      for key in keys:
+        return_key = (alias_to_path[key], alias_tuples[key][1], alias_tuples[key][2])
+        fragments[return_key] = fragments[key]
+        del fragments[key]
+      for alias in locs['local']:
+        del alias_tuples[alias]
+
+    remote_path_tuples = list(alias_tuples.values())
+
+    remote_paths = ( path for path, start, end in remote_path_tuples )
+    starts = ( start for path, start, end in remote_path_tuples )
+    ends = ( end for path, start, end in remote_path_tuples )
+
+    StorageClass = self.pick_storage_class(remote_path_tuples)
+    with StorageClass(self.meta.cloudpath, progress=progress) as stor:
+      remote_fragments = stor.get_files(remote_paths, starts=starts, ends=ends)
+
+    for frag in remote_fragments:
+      if frag['error'] is not None:
+        raise frag['error']
+
+    remote_fragments = { 
+      (res['filename'], res['byte_range'][0], res['byte_range'][1]): res['content'] \
+      for res in remote_fragments 
+    }
+
+    if self.enabled:
+      self.put(
+        [ 
+          (path_to_alias[file_bytes_tuple], content) \
+          for file_bytes_tuple, content in remote_fragments.items() \
+          if content is not None 
+        ],
+        compress=compress,
+        progress=progress
+      )
+
+    fragments.update(remote_fragments)
+    return fragments
+
   def download(self, paths, compress=None, progress=None):
     """
     Download the provided paths, but grab them from cache first
@@ -452,6 +527,7 @@ class CacheService(object):
     StorageClass = self.pick_storage_class(files)
 
     save_location = 'file://' + self.path
+    progress = 'to Cache' if progress else None
     with StorageClass(save_location, progress=progress) as stor:
       stor.put_files(
         [ (name, content) for name, content in files ],
diff --git a/cloudvolume/datasource/precomputed/sharding.py b/cloudvolume/datasource/precomputed/sharding.py
index 3f71a01..991e2c8 100644
--- a/cloudvolume/datasource/precomputed/sharding.py
+++ b/cloudvolume/datasource/precomputed/sharding.py
@@ -3,6 +3,7 @@ from __future__ import print_function
 from collections import namedtuple, defaultdict
 import copy
 import json
+from os.path import basename
 import struct
 
 import numpy as np
@@ -273,6 +274,39 @@ class ShardReader(object):
     self.shard_index_cache[filename] = index
     return index
 
+  def get_indices(self, filenames, path="", progress=None):
+    filenames = toiter(filenames)
+    filenames = [ self.meta.join(path, fname) for fname in filenames ]
+
+    fufilled = { 
+      fname: self.shard_index_cache[fname] \
+      for fname in filenames \
+      if fname in self.shard_index_cache  
+    }
+    requests = []
+    for fname in filenames:
+      if fname in fufilled:
+        continue
+      requests.append({
+        'path': fname, 
+        'local_alias': fname + '.index', 
+        'start': 0, 
+        'end': self.spec.index_length(),
+      })
+
+    progress = 'Shard Indices' if progress else False
+    binaries = self.cache.download_as(requests, progress=progress)
+    for (fname, start, end), content in binaries.items():
+      try:
+        index = self.decode_index(content, fname)
+        self.shard_index_cache[fname] = index
+        fufilled[fname] = index
+      except EmptyFileException:
+        self.shard_index_cache[fname] = None
+        fufilled[fname] = None
+
+    return fufilled
+
   def decode_index(self, binary, filename='Shard'):
     if binary is None or len(binary) == 0:
       raise EmptyFileException(filename + " was zero bytes.")
@@ -306,17 +340,87 @@ class ShardReader(object):
     return minishard_index 
 
   def get_minishard_index(self, filename, index, minishard_no, path=""):
+    """
+    Retrieves the minishard index for a given minishard number.
+
+    Returns: uint64 Nx3 array with multiple rows of [segid, byte start, byte end]
+    """
     res = self.get_minishard_indices(filename, index, minishard_no, path)
     return res[minishard_no]
 
   def get_minishard_indices(self, filename, index, minishard_nos, path=""):
     """
-    Retrieves the minishard index for a given minishard number.
+    Retrieves the minishard indices for a set of minishard numbers.
 
-    Returns: uint64 Nx3 array with multiple rows of [segid, byte start, byte end]
+    Returns: { minishard_no: uint64 Nx3 array of [segid, byte start, byte end], ... }
+    """
+    res = self.get_minishard_indices_for_files(( (filename, index, minishard_nos), ), path)
+    return res[filename]
+
+  def get_minishard_indices_for_files(self, requests, path="", progress=None):
+    """
+    Fetches the specified minishard indices for all the specified files
+    at once. This is required to get high performance as opposed to fetching
+    the all minishard indices for a single file.
+
+    requests: iterable of tuples
+      [  (filename, index, minishard_numbers), ... ]
+
+    Returns: map of filename -> minishard numbers -> minishard indices
+
+    e.g. 
+    {
+      filename_1: {
+          0: uint64 Nx3 array of [segid, byte start, byte end],
+          1: ...,
+      }
+      filename_2: ...
+    }
+    """
+    fufilled_by_filename = defaultdict(dict)
+    msn_map = {}
+
+    download_requests = []
+    for filename, index, minishard_nos in requests:
+      fufilled_requests, pending_requests = self.compute_minishard_index_requests(
+        filename, index, minishard_nos, path
+      ) 
+      fufilled_by_filename[filename] = fufilled_requests
+      for msn, start, end in pending_requests:
+        msn_map[(filename, start, end)] = msn
+
+        filepath = self.meta.join(path, filename)
+
+        download_requests.append({
+          'path': filepath,
+          'local_alias': '{}-{}.msi'.format(filepath, msn),
+          'start': start,
+          'end': end,
+        })
+
+    progress = 'Minishard Indices' if progress else False
+    results = self.cache.download_as(download_requests, progress=progress)
+  
+    for (filename, start, end), content in results.items():
+      filename = basename(filename)
+      cache_key = (filename, start, end)
+      msn = msn_map[cache_key]
+      minishard_index = self.decode_minishard_index(content, filename)
+      self.minishard_index_cache[cache_key] = minishard_index
+      fufilled_by_filename[filename][msn] = minishard_index
+
+    return fufilled_by_filename
+
+  def compute_minishard_index_requests(self, filename, index, minishard_nos, path=""):
+    """
+    Helper method for get_minishard_indices_for_files. 
+    Computes which requests must be made over the network vs can be fufilled from LRU cache.
     """
     minishard_nos = toiter(minishard_nos)
 
+    if index is None:
+      return ({ msn: None for msn in minishard_nos }, [])
+
     fufilled_requests = {}
 
     byte_ranges = {}
@@ -339,30 +443,11 @@ class ShardReader(object):
       if cache_key in self.minishard_index_cache:
         fufilled_requests[msn] = self.minishard_index_cache[cache_key]
       else:
-        pending_requests.append((bytes_start, bytes_end))
-
-    StorageClass = SimpleStorage if len(pending_requests) == 1 else Storage
-
-    with StorageClass(full_path) as stor:
-      filenames = ( filename for _ in pending_requests )
-      starts = ( start for (start, end) in pending_requests )
-      ends = ( end for (start, end) in pending_requests )
-      results = stor.get_files(filenames, starts, ends)
-  
-    del pending_requests
+        pending_requests.append((msn, bytes_start, bytes_end))
 
-    byte_ranges = { v: k for k, v in byte_ranges.items() }
-    for res in results:
-      start, end = res['byte_range']
-      msn = byte_ranges[(start, end)]
-      cache_key = (filename, start, end)
-      minishard_index = self.decode_minishard_index(res['content'], filename)
-      self.minishard_index_cache[cache_key] = minishard_index
-      fufilled_requests[msn] = minishard_index
-
-    return fufilled_requests
+    return (fufilled_requests, pending_requests)
 
-  def exists(self, labels, path="", return_byte_range=False):
+  def exists(self, labels, path="", return_byte_range=False, progress=None):
     """
     Checks a shard's minishard index for whether a file exists.
 
@@ -384,45 +469,46 @@ class ShardReader(object):
     except TypeError:
       return_one = True
 
-    results = {}
+    to_labels = defaultdict(list)
+    to_all_labels = defaultdict(list)
+    filename_to_minishard_num = defaultdict(list)
+
     for label in set(toiter(labels)):
       filename, minishard_number = self.compute_shard_location(label)
-      
-      filepath = self.meta.join(path, filename)
-
-      if self.cache.enabled:
-        cached = self.cache.has(self.meta.join(path, str(label)), progress=False)
-        if cached is not None:
-          results[label] = filepath
-          continue
+      to_labels[(filename, minishard_number)].append(label)
+      to_all_labels[filename].append(label)
+      filename_to_minishard_num[filename].append(minishard_number)
 
-      try:
-        index = self.get_index(filename, path)
-      except EmptyFileException:
-        results[label] = None
-        continue
+    indices = self.get_indices(to_all_labels.keys(), path, progress=progress)
 
-      minishard_index = self.get_minishard_index(
-        filename, index, 
-        minishard_number, path
-      )
+    all_minishards = self.get_minishard_indices_for_files([ 
+      (basename(filepath), index, filename_to_minishard_num[basename(filepath)]) \
+      for filepath, index in indices.items()
+    ], path, progress=progress)
 
-      if minishard_index is None:
-        results[label] = None
-        continue
-
-      idx = np.where(minishard_index[:,0] == label)[0]
-      if len(idx) == 0:
-        results[label] = None
-      else:
-        if return_byte_range:
-          _, offset, size = minishard_index[idx,:][0]
-          results[label] = [ filepath, int(offset), int(size) ]
-        else:
-          results[label] = filepath
+    results = {}
+    for filename, file_minishards in all_minishards.items():
+      filepath = self.meta.join(path, filename)
+      for mini_no, msi in file_minishards.items():
+        labels = to_labels[(filename, mini_no)]
+
+        for label in labels:
+          if msi is None:
+            results[label] = None
+            continue
+
+          idx = np.where(msi[:,0] == label)[0]
+          if len(idx) == 0:
+            results[label] = None
+          else:
+            if return_byte_range:
+              _, offset, size = msi[idx,:][0]
+              results[label] = [ filepath, int(offset), int(size) ]
+            else:
+              results[label] = filepath
 
     if return_one:
-      return results[label]
+      return next(results.values())
     return results
 
   def disassemble_shard(self, shard):
diff --git a/cloudvolume/lib.py b/cloudvolume/lib.py
index 09963f3..cfb65be 100644
--- a/cloudvolume/lib.py
+++ b/cloudvolume/lib.py
@@ -58,6 +58,15 @@ def toiter(obj, is_iter=False):
       return [ obj ], False
     return [ obj ]
 
+def duplicates(lst):
+  dupes = []
+  seen = set()
+  for elem in lst:
+    if elem in seen:
+      dupes.append(elem)
+    seen.add(elem)
+  return set(dupes)
+
 def jsonify(obj, **kwargs):
   return json.dumps(obj, cls=NumpyEncoder, **kwargs)
 
diff --git a/cloudvolume/lru.py b/cloudvolume/lru.py
index 161ccfb..df80ede 100644
--- a/cloudvolume/lru.py
+++ b/cloudvolume/lru.py
@@ -100,6 +100,11 @@ class DoublyLinkedList(object):
   def delete(self, node):
     nxt, prev = node.next, node.prev
 
+    if self.head == node:
+      self.head = nxt
+    if self.tail == node:
+      self.tail = prev
+
     if prev is not None:
       prev.next = nxt
 
@@ -133,11 +138,12 @@ class DoublyLinkedList(object):
       self.head = ListNode(val, None, None)
       self.tail = self.head
     elif self.head is None:
-      self.head = ListNode(val, self.tail, None)
+      self.head = ListNode(val, next=self.tail, prev=None)
       self.tail.prev = self.head
     else:
-      self.head = ListNode(val, self.head, None)
-      self.head.next.prev = self.head
+      prev_head = self.head
+      self.head = ListNode(val, next=prev_head, prev=None)
+      prev_head.prev = self.head
 
     self.size += 1
 
@@ -170,6 +176,15 @@ class LRU(object):
   def __len__(self):
     return self.queue.size
 
+  def keys(self):
+    return self.hash.keys()
+
+  def values(self):
+    return ( node.val for val in self.queue )
+
+  def items(self):
+    return ( (key, node.val) for key, node in self.hash.items() )
+
   def clear(self):
     self.queue = DoublyLinkedList()
     self.hash = {}
@@ -190,19 +205,26 @@ class LRU(object):
       (key,val) = self.queue.delete_tail()
       del self.hash[key]
 
-  def __contains__(self, key):
-    return key in self.hash
-
-  def __getitem__(self, key):
+  def delete(self, key):
     if key not in self.hash:
       raise KeyError("{} not in cache.".format(key))
 
+    node = self.hash[key]
+    self.queue.delete(node)
+    del self.hash[key]
+
+  def get(self, key, default=None):
+    if key not in self.hash:
+      if default is None:
+        raise KeyError("{} not in cache.".format(key))
+      return default
+
     node = self.hash[key]
     self.queue.promote_to_head(node)
 
     return node.val[1]
 
-  def __setitem__(self, key, val):
+  def set(self, key, val):
     if self.size == 0:
       return
 
@@ -218,7 +240,19 @@ class LRU(object):
 
     while len(self.queue) > self.size:
       (tkey,tval) = self.queue.delete_tail()
-      del self.hash[tkey]      
+      del self.hash[tkey]     
+
+  def __contains__(self, key):
+    return key in self.hash
+
+  def __getitem__(self, key):
+    return self.get(key)
+
+  def __setitem__(self, key, val):
+    return self.set(key, val)
+
+  def __delitem__(self, key):
+    return self.delete(key)
 
   def __str__(self):
     return str(self.queue)
diff --git a/cloudvolume/scheduler.py b/cloudvolume/scheduler.py
index 5552438..f3565bb 100644
--- a/cloudvolume/scheduler.py
+++ b/cloudvolume/scheduler.py
@@ -9,14 +9,27 @@ def schedule_threaded_jobs(
     progress=None, total=None
   ):
 
-  def wrap(fn):
-    def wrapped(iface):
-      return fn()
-    return wrapped
+  if total is None:
+    try:
+      total = len(fns)
+    except TypeError: # generators don't have len
+      pass
+
+  pbar = tqdm(total=total, desc=progress, disable=(not progress))
+  results = []
+  
+  def updatefn(fn):
+    def realupdatefn(iface):
+      res = fn()
+      pbar.update(1)
+      results.append(res)
+    return realupdatefn
 
-  with ThreadedQueue(n_threads=concurrency, progress=progress) as tq:
+  with ThreadedQueue(n_threads=concurrency) as tq:
     for fn in fns:
-      tq.put(wrap(fn))
+      tq.put(updatefn(fn))
+
+  return results
 
 def schedule_green_jobs(
     fns, concurrency=DEFAULT_THREADS, 
diff --git a/cloudvolume/storage/storage.py b/cloudvolume/storage/storage.py
index 9a0d7ed..31a993a 100644
--- a/cloudvolume/storage/storage.py
+++ b/cloudvolume/storage/storage.py
@@ -7,12 +7,13 @@ import os.path
 import posixpath
 import re
 from functools import partial
+import types
 
 from tqdm import tqdm
 
 from cloudvolume import compression
 from cloudvolume.exceptions import UnsupportedProtocolError
-from cloudvolume.lib import mkdir, scatter, jsonify
+from cloudvolume.lib import mkdir, scatter, jsonify, duplicates
 from cloudvolume.threaded_queue import ThreadedQueue, DEFAULT_THREADS
 from cloudvolume.scheduler import schedule_green_jobs
 
@@ -55,6 +56,12 @@ class StorageBase(object):
   def layer_path(self):
     return self._layer_path
 
+  def progress_description(self, prefix):
+    if isinstance(self.progress, str):
+      return prefix + ' ' + self.progress
+    else:
+      return prefix if self.progress else None
+
   def get_connection(self):
     return self._interface_cls(self._path)
 
@@ -144,7 +151,8 @@ class SimpleStorage(StorageBase):
     Required:
       files: [ (filepath, content), .... ]
     """
-    for path, content in tqdm(files, disable=(not self.progress), desc="Uploading"):
+    desc = self.progress_description('Uploading')
+    for path, content in tqdm(files, disable=(not self.progress), desc=desc):
       content = compression.compress(content, method=compress, compress_level=compress_level)
       self._interface.put_file(path, content, content_type, compress, cache_control=cache_control)
     return self
@@ -178,7 +186,7 @@ class SimpleStorage(StorageBase):
     iterator = tqdm(
       zip(file_paths, starts, ends),
       disable=(not self.progress), 
-      desc="Downloading"
+      desc=self.progress_description("Downloading")
     )
 
     results = []
@@ -281,9 +289,10 @@ class GreenStorage(StorageBase):
       with self.get_connection() as conn:
         results.update(conn.files_exist(paths))
     
+    desc = self.progress_description('Existence Testing')
     schedule_green_jobs(  
       fns=( partial(exist_thunk, paths) for paths in scatter(file_paths, self.concurrency) ),
-      progress=('Existence Testing' if self.progress else None),
+      progress=(desc if self.progress else None),
       concurrency=self.concurrency,
       total=len(file_paths),
     )
@@ -333,12 +342,14 @@ class GreenStorage(StorageBase):
         "error": error,
       }
 
+    desc = self.progress_description('Downloading')
+
     return schedule_green_jobs(  
       fns=( 
         partial(getfn, path, start, end) 
         for path, start, end in zip(file_paths, starts, ends) 
       ),
-      progress=('Downloading' if self.progress else None),
+      progress=(desc if self.progress else None),
       concurrency=self.concurrency,
       total=len(file_paths),
     )
@@ -369,12 +380,18 @@ class GreenStorage(StorageBase):
           cache_control=cache_control,
         )
 
+    if not isinstance(gen, types.GeneratorType):
+      dupes = duplicates([ path for path, content in files ])
+      if dupes:
+        raise ValueError("Cannot write the same file multiple times in one pass. This causes a race condition. Files: " + ", ".join(dupes))
+
     fns = ( partial(uploadfn, path, content) for path, content in files )
 
     if block:
+      desc = desc = self.progress_description('Uploading')
       schedule_green_jobs(
         fns=fns,
-        progress=('Uploading' if self.progress else None),
+        progress=(desc if self.progress else None),
         concurrency=self.concurrency,
         total=len(files),
       )
@@ -405,9 +422,11 @@ class GreenStorage(StorageBase):
       with self.get_connection() as conn:
         conn.delete_file(path)
 
+    desc = self.progress_description('Deleting')
+
     schedule_green_jobs(
       fns=( partial(thunk_delete, path) for path in file_paths ),
-      progress=('Deleting' if self.progress else None),
+      progress=(desc if self.progress else None),
       concurrency=self.concurrency,
       total=len(file_paths),
     )
@@ -518,7 +537,12 @@ class ThreadedStorage(StorageBase, ThreadedQueue):
     def base_uploadfn(path, content, interface):
       interface.put_file(path, content, content_type, compress, cache_control=cache_control)
 
+    seen = set() 
     for path, content in files:
+      if path in seen:
+        raise ValueError("Cannot write the same file multiple times in one pass. This causes a race condition. File: " + path)
+      seen.add(path)
+
       content = compression.compress(content, method=compress, compress_level=compress_level)
       uploadfn = partial(base_uploadfn, path, content)
 
@@ -528,7 +552,7 @@ class ThreadedStorage(StorageBase, ThreadedQueue):
         uploadfn(self._interface)
 
     if block:
-      desc = 'Uploading' if self.progress else None
+      desc = self.progress_description('Uploading')
       self.wait(desc)
 
     return self
@@ -556,7 +580,7 @@ class ThreadedStorage(StorageBase, ThreadedQueue):
     else:
       exist_thunk(file_paths, self._interface)
 
-    desc = 'Existence Testing' if self.progress else None
+    desc = self.progress_description('Existence Testing')
     self.wait(desc)
 
     return results
@@ -604,7 +628,7 @@ class ThreadedStorage(StorageBase, ThreadedQueue):
       else:
         get_file_thunk(path, start, end, self._interface)
 
-    desc = 'Downloading' if self.progress else None
+    desc = self.progress_description('Downloading')
     self.wait(desc)
 
     return results
@@ -632,7 +656,7 @@ class ThreadedStorage(StorageBase, ThreadedQueue):
       else:
         thunk_delete(path, self._interface)
 
-    desc = 'Deleting' if self.progress else None
+    desc = self.progress_description('Deleting')
     self.wait(desc)
 
     return self


Test patch:
diff --git a/test/test_lru.py b/test/test_lru.py
index b5601f8..7031452 100644
--- a/test/test_lru.py
+++ b/test/test_lru.py
@@ -1,5 +1,10 @@
+import pytest
+
 from cloudvolume.lru import LRU
 
+import time
+import random
+
 def test_lru():
   lru = LRU(5)
 
@@ -34,3 +39,25 @@ def test_lru():
   assert lru.queue.tolist() == [ 
     (0,0), (4,4), (3,3), (2,2), (1,1)
   ]
+
+def test_lru_chaos():
+  lru = LRU(10)
+
+  seed = time.time()
+  print("seed", seed)
+
+  random.seed(seed)
+  for i in range(150):
+    rand = random.randint(-5000, 5000)
+    lru[rand] = random.randint(-5000, 5000)
+
+    if rand % 17 == 0:
+      keys = lru.keys()
+      key = random.choice(list(keys))
+      lru[key]
+
+
+
+
+
+