[ARVADOS] updated: d52c19d19d5588c664c599aaa51c0084954d8836
git at public.curoverse.com
git at public.curoverse.com
Wed Feb 19 17:10:29 EST 2014
Summary of changes:
sdk/python/arvados/fuse.py | 2 +-
sdk/python/arvados/keep.py | 114 ++++++++++++++++++++++++++++---------------
sdk/python/bin/arv-mount | 4 ++
3 files changed, 79 insertions(+), 41 deletions(-)
via d52c19d19d5588c664c599aaa51c0084954d8836 (commit)
from 1ce079f002566b28c450959b1d7389c8c85ab6f0 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit d52c19d19d5588c664c599aaa51c0084954d8836
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Wed Feb 19 22:10:26 2014 +0000
caching wip
diff --git a/sdk/python/arvados/fuse.py b/sdk/python/arvados/fuse.py
index 5a56e85..48e9b2c 100644
--- a/sdk/python/arvados/fuse.py
+++ b/sdk/python/arvados/fuse.py
@@ -294,7 +294,7 @@ class Operations(llfuse.Operations):
def statfs(self):
st = llfuse.StatvfsData()
- st.f_bsize = 4096
+ st.f_bsize = 1024 * 1024
st.f_blocks = 0
st.f_files = 0
diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py
index fe62eb7..93abfac 100644
--- a/sdk/python/arvados/keep.py
+++ b/sdk/python/arvados/keep.py
@@ -17,6 +17,7 @@ import zlib
import fcntl
import time
import threading
+import timer
global_client_object = None
@@ -141,7 +142,7 @@ class KeepClient(object):
self._cache_lock = threading.Lock()
self._cache = []
# default 256 megabyte cache
- self._cache_max = 256 * 1024 * 1024
+ self.cache_max = 256 * 1024 * 1024
def shuffled_service_roots(self, hash):
if self.service_roots == None:
@@ -173,16 +174,72 @@ class KeepClient(object):
logging.debug(str(pseq))
return pseq
+ class CacheSlot(object):
+ def __init__(self, locator):
+ self.locator = locator
+ self.ready = threading.Event()
+ self.content = None
+
+ def get(self):
+ self.ready.wait()
+ return self.content
+
+ def set(self, value):
+ self.content = value
+ self.ready.set()
+
+ def size(self):
+ if self.content == None:
+ return 0
+ else:
+ return len(self.content)
+
+ def cap_cache(self):
+ '''Cap the cache size to self.cache_max'''
+ self._cache_lock.acquire()
+ try:
+ self._cache = filter(lambda c: not (c.ready.is_set() and c.content == None), self._cache)
+ sm = sum([slot.size() for slot in self._cache])
+ while sm > self.cache_max:
+ del self._cache[-1]
+ sm = sum([slot.size() for a in self._cache])
+ finally:
+ self._cache_lock.release()
+
+ def reserve_cache(self, locator):
+ '''Reserve a cache slot for the specified locator,
+ or return the existing slot.'''
+ self._cache_lock.acquire()
+ try:
+ # Test if the locator is already in the cache
+ for i in xrange(0, len(self._cache)):
+ if self._cache[i].locator == locator:
+ n = self._cache[i]
+ if i != 0:
+ del self._cache[i]
+ self._cache.insert(0, n)
+ return n, False
+
+ # Add a new cache slot for the locator
+ n = CacheSlot(locator)
+ self._cache.insert(0, n)
+ return n, True
+ finally:
+ self._cache_lock.release()
+
def get(self, locator):
+ logging.debug("Keep.get %s" % (locator))
+
if re.search(r',', locator):
return ''.join(self.get(x) for x in locator.split(','))
if 'KEEP_LOCAL_STORE' in os.environ:
return KeepClient.local_store_get(locator)
expect_hash = re.sub(r'\+.*', '', locator)
- c = self.check_cache(expect_hash)
- if c:
- return c
+ slot, first = self.reserve_cache(expect_hash)
+ if not first:
+ v = slot.get()
+ return v
for service_root in self.shuffled_service_roots(expect_hash):
url = service_root + expect_hash
@@ -191,7 +248,8 @@ class KeepClient(object):
'Accept': 'application/octet-stream'}
blob = self.get_url(url, headers, expect_hash)
if blob:
- self.put_cache(expect_hash, blob)
+ slot.set(blob)
+ self.cap_cache()
return blob
for location_hint in re.finditer(r'\+K@([a-z0-9]+)', locator):
@@ -199,15 +257,23 @@ class KeepClient(object):
url = 'http://keep.' + instance + '.arvadosapi.com/' + expect_hash
blob = self.get_url(url, {}, expect_hash)
if blob:
- self.put_cache(expect_hash, blob)
+ slot.set(blob)
+ self.cap_cache()
return blob
+
+ slot.set(None)
+ self.cap_cache()
+
raise arvados.errors.NotFoundError("Block not found: %s" % expect_hash)
def get_url(self, url, headers, expect_hash):
h = httplib2.Http()
try:
- resp, content = h.request(url.encode('utf-8'), 'GET',
- headers=headers)
+ logging.info("Request: GET %s" % (url))
+ with timer.Timer() as t:
+ resp, content = h.request(url.encode('utf-8'), 'GET',
+ headers=headers)
+ logging.info("Received %s bytes in %s msec (%s bytes/sec)" % (len(content), t.msecs, len(content)/t.secs))
if re.match(r'^2\d\d$', resp['status']):
m = hashlib.new('md5')
m.update(content)
@@ -248,38 +314,6 @@ class KeepClient(object):
"Write fail for %s: wanted %d but wrote %d" %
(data_hash, want_copies, have_copies))
- def put_cache(self, locator, data):
- """Put a block into the cache."""
- if self.check_cache(locator) != None:
- return
- self._cache_lock.acquire()
- try:
- # first check cache size and delete stuff from the end if necessary
- sm = sum([len(a[1]) for a in self._cache]) + len(data)
- while sm > self._cache_max:
- print sm, self._cache_max
- del self._cache[-1]
- sm = sum([len(a[1]) for a in self._cache]) + len(data)
-
- # now add the new block at the front of the list
- self._cache.insert(0, [locator, data])
- finally:
- self._cache_lock.release()
-
- def check_cache(self, locator):
- """Get a block from the cache. Also moves the block to the front of the list."""
- self._cache_lock.acquire()
- try:
- for i in xrange(0, len(self._cache)):
- if self._cache[i][0] == locator:
- n = self._cache[i]
- del self._cache[i]
- self._cache.insert(0, n)
- return n[1]
- finally:
- self._cache_lock.release()
- return None
-
@staticmethod
def sign_for_old_server(data_hash, data):
return (("-----BEGIN PGP SIGNED MESSAGE-----\n\n\n%d %s\n-----BEGIN PGP SIGNATURE-----\n\n-----END PGP SIGNATURE-----\n" % (int(time.time()), data_hash)) + data)
diff --git a/sdk/python/bin/arv-mount b/sdk/python/bin/arv-mount
index 4a8233a..aa5bd0e 100755
--- a/sdk/python/bin/arv-mount
+++ b/sdk/python/bin/arv-mount
@@ -1,8 +1,12 @@
#!/usr/bin/env python
from arvados.fuse import *
+import arvados
+
if __name__ == '__main__':
+ api = arvados.api()
+
# Handle command line parameters
parser = argparse.ArgumentParser(
description='Mount Keep data under the local filesystem.')
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list