+logger = logging.getLogger('arvados.cwl-runner')
+
+pdh_size = re.compile(r'([0-9a-f]{32})\+(\d+)(\+\S+)*')
+
+class CollectionCache(object):
+ 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.num_retries = num_retries
+ self.collections = OrderedDict()
+ self.lock = threading.Lock()
+ self.total = 0
+ self.cap = cap
+ self.min_entries = min_entries
+
+ def set_cap(self, cap):
+ self.cap = cap
+
+ def cap_cache(self, required):
+ # ordered dict iterates from oldest to newest
+ for pdh, v in list(self.collections.items()):
+ available = self.cap - self.total
+ if available >= required or len(self.collections) < self.min_entries:
+ return
+ # cut it loose
+ logger.debug("Evicting collection reader %s from cache (cap %s total %s required %s)", pdh, self.cap, self.total, required)
+ del self.collections[pdh]
+ self.total -= v[1]
+
+ def get(self, locator):
+ with self.lock:
+ if locator not in self.collections:
+ m = pdh_size.match(locator)
+ if m:
+ self.cap_cache(int(m.group(2)) * 128)
+ logger.debug("Creating collection reader for %s", locator)
+ try:
+ cr = arvados.collection.CollectionReader(locator, api_client=self.api_client,
+ keep_client=self.keep_client,
+ num_retries=self.num_retries)
+ except arvados.errors.ApiError as ap:
+ raise IOError(errno.ENOENT, "Could not access collection '%s': %s" % (locator, str(ap._get_reason())))
+ sz = len(cr.manifest_text()) * 128
+ self.collections[locator] = (cr, sz)
+ self.total += sz
+ else:
+ cr, sz = self.collections[locator]
+ # bump it to the back
+ del self.collections[locator]
+ self.collections[locator] = (cr, sz)
+ return cr
+
+