feat: cache info for improved distributed performance

pychunkedgraph/graph/meta.py

@@ -1,3 +1,4 @@
+import json
 from datetime import timedelta
 from typing import Dict
 from typing import List
@@ -10,6 +11,9 @@
 
 from .utils.generic import compute_bitmasks
 from .chunks.utils import get_chunks_boundary
+from ..utils.redis import keys as r_keys
+from ..utils.redis import get_rq_queue
+from ..utils.redis import get_redis_connection
 
 
 _datasource_fields = ("EDGES", "COMPONENTS", "WATERSHED", "DATA_VERSION", "CV_MIP")
@@ -80,7 +84,20 @@ def custom_data(self):
     def ws_cv(self):
         if self._ws_cv:
             return self._ws_cv
-        self._ws_cv = CloudVolume(self._data_source.WATERSHED)
+
+        try:
+            # try reading a cached info file for distributed workers
+            # useful to avoid md5 errors on high gcs load
+            cache_key = f"{self.graph_config.ID}:cv_info_cached"
+            redis = get_redis_connection()
+            cached_info = json.loads(redis.get(cache_key))
+            self._ws_cv = CloudVolume(self._data_source.WATERSHED, info=cached_info)
+        except Exception:
+            self._ws_cv = CloudVolume(self._data_source.WATERSHED)
+            try:
+                redis.set(cache_key, json.dumps(self._ws_cv.info))
+            except Exception:
+                ...
         return self._ws_cv
 
     @property