14198: Update version deps for schema-salad, cwltool, ruamel.yaml, requests

[arvados.git] / sdk / cwl / arvados_cwl / fsaccess.py

diff --git a/sdk/cwl/arvados_cwl/fsaccess.py b/sdk/cwl/arvados_cwl/fsaccess.py
index 5bdd8dd085719b88d7d1173e2f03b5c24083c0cf..b22b2ffd6f4574ea6a774064d782fbfd07c7a94d 100644 (file)
--- a/sdk/cwl/arvados_cwl/fsaccess.py
+++ b/sdk/cwl/arvados_cwl/fsaccess.py
@@ -9,6 +9,7 @@ import urlparse
 import re
 import logging
 import threading
+from collections import OrderedDict
 
 import ruamel.yaml as yaml
 
@@ -21,24 +22,53 @@ import arvados.collection
 import arvados.arvfile
 import arvados.errors
 
+from googleapiclient.errors import HttpError
+
 from schema_salad.ref_resolver import DefaultFetcher
 
 logger = logging.getLogger('arvados.cwl-runner')
 
 class CollectionCache(object):
-    def __init__(self, api_client, keep_client, num_retries):
+    def __init__(self, api_client, keep_client, num_retries,
+                 cap=256*1024*1024,
+                 min_entries=2):
         self.api_client = api_client
         self.keep_client = keep_client
-        self.collections = {}
+        self.num_retries = num_retries
+        self.collections = OrderedDict()
         self.lock = threading.Lock()
+        self.total = 0
+        self.cap = cap
+        self.min_entries = min_entries
+
+    def cap_cache(self):
+        if self.total > self.cap:
+            # ordered list iterates from oldest to newest
+            for pdh, v in self.collections.items():
+                if self.total < self.cap or len(self.collections) < self.min_entries:
+                    break
+                # cut it loose
+                logger.debug("Evicting collection reader %s from cache", pdh)
+                del self.collections[pdh]
+                self.total -= v[1]
 
     def get(self, pdh):
         with self.lock:
             if pdh not in self.collections:
                 logger.debug("Creating collection reader for %s", pdh)
-                self.collections[pdh] = arvados.collection.CollectionReader(pdh, api_client=self.api_client,
-                                                                            keep_client=self.keep_client)
-            return self.collections[pdh]
+                cr = arvados.collection.CollectionReader(pdh, api_client=self.api_client,
+                                                         keep_client=self.keep_client,
+                                                         num_retries=self.num_retries)
+                sz = len(cr.manifest_text()) * 128
+                self.collections[pdh] = (cr, sz)
+                self.total += sz
+                self.cap_cache()
+            else:
+                cr, sz = self.collections[pdh]
+                # bump it to the back
+                del self.collections[pdh]
+                self.collections[pdh] = (cr, sz)
+            return cr
 
 
 class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
@@ -53,7 +83,7 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
         p = sp[0]
         if p.startswith("keep:") and arvados.util.keep_locator_pattern.match(p[5:]):
             pdh = p[5:]
-            return (self.collection_cache.get(pdh), sp[1] if len(sp) == 2 else None)
+            return (self.collection_cache.get(pdh), urlparse.unquote(sp[1]) if len(sp) == 2 else None)
         else:
             return (None, path)
 
@@ -84,7 +114,7 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
         if collection is not None and not rest:
             return [pattern]
         patternsegments = rest.split("/")
-        return self._match(collection, patternsegments, "keep:" + collection.manifest_locator())
+        return sorted(self._match(collection, patternsegments, "keep:" + collection.manifest_locator()))
 
     def open(self, fn, mode):
         collection, rest = self.get_collection(fn)
@@ -94,7 +124,13 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
             return super(CollectionFsAccess, self).open(self._abs(fn), mode)
 
     def exists(self, fn):
-        collection, rest = self.get_collection(fn)
+        try:
+            collection, rest = self.get_collection(fn)
+        except HttpError as err:
+            if err.resp.status == 404:
+                return False
+            else:
+                raise
         if collection is not None:
             if rest:
                 return collection.exists(rest)
@@ -103,6 +139,17 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
         else:
             return super(CollectionFsAccess, self).exists(fn)
 
+    def size(self, fn):  # type: (unicode) -> bool
+        collection, rest = self.get_collection(fn)
+        if collection is not None:
+            if rest:
+                arvfile = collection.find(rest)
+                if isinstance(arvfile, arvados.arvfile.ArvadosFile):
+                    return arvfile.size()
+            raise IOError(errno.EINVAL, "Not a path to a file %s" % (fn))
+        else:
+            return super(CollectionFsAccess, self).size(fn)
+
     def isfile(self, fn):  # type: (unicode) -> bool
         collection, rest = self.get_collection(fn)
         if collection is not None:
@@ -153,16 +200,13 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
             return os.path.realpath(path)
 
 class CollectionFetcher(DefaultFetcher):
-    def __init__(self, cache, session, api_client=None, fs_access=None, num_retries=4, overrides=None):
+    def __init__(self, cache, session, api_client=None, fs_access=None, num_retries=4):
         super(CollectionFetcher, self).__init__(cache, session)
         self.api_client = api_client
         self.fsaccess = fs_access
         self.num_retries = num_retries
-        self.overrides = overrides if overrides else {}
 
     def fetch_text(self, url):
-        if url in self.overrides:
-            return self.overrides[url]
         if url.startswith("keep:"):
             with self.fsaccess.open(url, "r") as f:
                 return f.read()
@@ -173,8 +217,6 @@ class CollectionFetcher(DefaultFetcher):
         return super(CollectionFetcher, self).fetch_text(url)
 
     def check_exists(self, url):
-        if url in self.overrides:
-            return True
         try:
             if url.startswith("http://arvados.org/cwl"):
                 return True
@@ -223,6 +265,12 @@ class CollectionFetcher(DefaultFetcher):
 
         return super(CollectionFetcher, self).urljoin(base_url, url)
 
+    schemes = [u"file", u"http", u"https", u"mailto", u"keep"]
+
+    def supported_schemes(self):  # type: () -> List[Text]
+        return self.schemes
+
+
 workflow_uuid_pattern = re.compile(r'[a-z0-9]{5}-7fd4e-[a-z0-9]{15}')
 pipeline_template_uuid_pattern = re.compile(r'[a-z0-9]{5}-p5p6p-[a-z0-9]{15}')