import re
import logging
import threading
+from collections import OrderedDict
import ruamel.yaml as yaml
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.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)
+ 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):
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)
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()
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