Merge lp:~verterok/ubuntuone-client/file_shelf-is-fast into lp:ubuntuone-client

[Guillermo Gonzalez]

[r=lucio.torre, r=dobey] minimize the times fileshelf touch the disk to do a key lookup. Also adds a new LRUCache for the Fileshelf

[Guillermo Gonzalez (verterok)]

This branch minimize the times fileshelf touch the disk to do a key lookup. Also includes a branch new LRUCache for the Filshelf this also should reduce the time we are hitting disk, as e.g: sync is issuing a lot of consecutive request to the file shelf.

[Lucio Torre (lucio.torre)]

        try:
            self.shelf['missingkey']
        except KeyError:
            self.assertEquals(self.shelf.cache_misses, 1)

This part of test_hit_miss_properties depends on the state of the cache somewhere else? dont you need to add an 'else' block that raises an error?

anyhow, big +1. i like that we now show more stats on log marks.

review: Approve

[lp:~verterok/ubuntuone-client/file_shelf-is-fast]

91. By Guillermo Gonzalez on 2009-07-17

fix lint warning

[dobey (dobey)]

review: Approve

[Guillermo Gonzalez (verterok)]

> try:
> self.shelf['missingkey']
> except KeyError:
> self.assertEquals(self.shelf.cache_misses, 1)
>
> This part of test_hit_miss_properties depends on the state of the cache
> somewhere else? dont you need to add an 'else' block that raises an error?

good point!
the cache is clean, as the shelf was initialized before starting the test, but I'll add an else clause just to make the test more complete.

>
> anyhow, big +1. i like that we now show more stats on log marks.

Thanks!

[lp:~verterok/ubuntuone-client/file_shelf-is-fast]

92. By Guillermo Gonzalez on 2009-07-17

 add else clause to cache.misses test

=== modified file 'tests/syncdaemon/test_fileshelf.py'
--- tests/syncdaemon/test_fileshelf.py	2009-07-02 20:22:51 +0000
+++ tests/syncdaemon/test_fileshelf.py	2009-07-17 20:31:26 +0000
@@ -15,57 +15,59 @@
 # You should have received a copy of the GNU General Public License along
 # with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-""" Test disk cache persistent shelf """
+""" Test file based persistent shelf """
 
 import os
 import hashlib
 import unittest
 import shutil
 
-from ubuntuone.syncdaemon.file_shelf import FileShelf
-
-TESTS_DIR = os.path.join(os.getcwd(), "tmp")
-
-class TestFileShelf(unittest.TestCase):
+from ubuntuone.syncdaemon.file_shelf import (
+    FileShelf,
+    CachedFileShelf,
+    LRUCache,
+    CacheInconsistencyError,
+)
+from contrib.testing import testcase
+
+
+class TestFileShelf(testcase.BaseTwistedTestCase):
     """ Test the FileShelf """
+    fileshelf_class = FileShelf
 
     def setUp(self):
         """ Sets up a test. """
-        try:
-            os.mkdir(TESTS_DIR)
-        except OSError:
-            # already there, remove it to clean and create again
-            shutil.rmtree(TESTS_DIR)
-            os.mkdir(TESTS_DIR)
-        self.path = os.path.join(TESTS_DIR, 'shelf')
-        self.shelf = FileShelf(self.path)
+        testcase.BaseTwistedTestCase.setUp(self)
+        self.path = self.mktemp('shelf')
+        self.shelf = self.fileshelf_class(self.path)
 
     def tearDown(self):
         '''Clean up the tests.'''
-        shutil.rmtree(TESTS_DIR)
+        shutil.rmtree(self.path)
+        testcase.BaseTwistedTestCase.tearDown(self)
 
     def test_bad_depth(self):
         """ test that the shelf reject invalid depth at creation time """
-        self.assertRaises(ValueError, FileShelf, TESTS_DIR, depth=-1)
+        self.assertRaises(ValueError, self.fileshelf_class, self.path, depth=-1)
 
     def test_bad_path(self):
         """ test that the shelf removes the previous shelve file and create a
         directory for the new file based shelf at creation time.
         """
-        path = os.path.join(TESTS_DIR, 'shelf_file')
+        path = os.path.join(self.path, 'shelf_file')
         open(path, 'w').close()
-        FileShelf(path)
+        self.fileshelf_class(path)
         self.assertTrue(os.path.isdir(path))
 
     def test_different_depth_sizes(self):
         """ test the basic operations (delitem, getitem, setitem) with
         depths between 0 and len(hashlib.sha1().hexdigest())
         """
-        base_path = os.path.join(TESTS_DIR, 'shelf_depth-')
+        base_path = os.path.join(self.path, 'shelf_depth-')
         sha1 = hashlib.sha1()
         for idx in xrange(0, len(sha1.hexdigest())):
             path = base_path + str(idx)
-            shelf = FileShelf(path, depth=idx)
+            shelf = self.fileshelf_class(path, depth=idx)
             key = sha1.hexdigest()
             # test __setitem__
             shelf[key] = 'foo'
@@ -85,8 +87,8 @@
 
     def test_contains(self):
         """ test that it behaves with the 'in' """
-        path = os.path.join(TESTS_DIR, 'shelf_depth')
-        shelf = FileShelf(path)
+        path = os.path.join(self.path, 'shelf_depth')
+        shelf = self.fileshelf_class(path)
         shelf["foo"] = "bar"
         self.assertTrue("foo" in shelf)
         self.assertFalse("baz" in shelf)
@@ -95,8 +97,8 @@
 
     def test_pop(self):
         """ test that it behaves with the .pop() """
-        path = os.path.join(TESTS_DIR, 'shelf_depth')
-        shelf = FileShelf(path)
+        path = os.path.join(self.path, 'shelf_depth')
+        shelf = self.fileshelf_class(path)
         shelf["foo"] = "bar"
         self.assertEqual("bar", shelf.pop("foo"))
         self.assertFalse("foo" in shelf)
@@ -106,8 +108,8 @@
 
     def test_get(self):
         """ test that it behaves with the .get(key, default) """
-        path = os.path.join(TESTS_DIR, 'shelf_get')
-        shelf = FileShelf(path)
+        path = os.path.join(self.path, 'shelf_get')
+        shelf = self.fileshelf_class(path)
         shelf["foo"] = "bar"
         self.assertEquals('bar', shelf.get('foo'))
         self.assertEquals('bar', shelf.get('foo', None))
@@ -116,8 +118,8 @@
 
     def test_items(self):
         """ test that it behaves with the .items() """
-        path = os.path.join(TESTS_DIR, 'shelf_get')
-        shelf = FileShelf(path)
+        path = os.path.join(self.path, 'shelf_get')
+        shelf = self.fileshelf_class(path)
         shelf["foo"] = "bar"
         # k, v are temp variables, pylint: disable-msg=W0631
         self.assertEquals([('foo', 'bar')],
@@ -127,6 +129,131 @@
                         [(k, v) for k, v in shelf.items()])
 
 
+class CachedFileShelfTests(TestFileShelf):
+    """TestFileShelf tests but using CachedFileShelf"""
+    fileshelf_class = CachedFileShelf
+
+    def test_hit_miss_properties(self):
+        """test the cache hits/misses properties"""
+        # yes, the statement has some effect, pylint: disable-msg=W0104
+        try:
+            self.shelf['missingkey']
+        except KeyError:
+            self.assertEquals(self.shelf.cache_misses, 1)
+        self.shelf['realkey'] = 'realvalue'
+        self.shelf['realkey']
+        self.shelf['realkey']
+        self.assertEquals(self.shelf.cache_hits, 1)
+
+
+class LRUCacheTests(unittest.TestCase):
+    '''Test the LRUCache class'''
+    # I'm a unittest of a class, I need to access protected members
+    # pylint: disable-msg=W0212
+
+    def test_setitem(self):
+        """test __delitem__ method"""
+        cache = LRUCache(100, 4)
+        # set some data in the cache
+        values = [('key'+str(i), i) for i in range(100)]
+        for i, j in values:
+            cache[i] = j
+        self.assertEquals(len(cache._queue), len(values))
+        self.assertEquals(len(cache._cache), len(values))
+
+    def test_getitem(self):
+        """test __delitem__ method"""
+        cache = LRUCache(100, 4)
+        # set some data in the cache
+        values = [('key'+str(i), i) for i in range(100)]
+        for i, j in values:
+            cache[i] = j
+        self.assertEquals(len(cache._queue), len(values))
+        self.assertEquals(len(cache._cache), len(values))
+        # compare all the items with the values
+        for i, j in values:
+            self.assertEquals(cache[i], j)
+
+    def test_delitem(self):
+        """test __delitem__ method"""
+        cache = LRUCache(100, 4)
+        values = [('key'+str(i), i) for i in range(100)]
+        for i, j in values:
+            cache[i] = j
+        self.assertEquals(len(cache._queue), len(values))
+        self.assertEquals(len(cache._cache), len(values))
+        for item in cache._cache.copy():
+            del cache[item]
+        self.assertEquals(len(cache._queue), 0)
+        self.assertEquals(len(cache._cache), 0)
+
+    def test_update(self):
+        """test the update method, chacking the refcount and queue."""
+        cache = LRUCache(100, 4)
+        cache.update('key1')
+        self.assertEquals(len(cache._queue), 1)
+        self.assertEquals(len(cache._refcount), 1)
+        self.assertEquals(cache._refcount['key1'], 1)
+        cache.update('key1')
+        self.assertEquals(len(cache._queue), 2)
+        self.assertEquals(len(cache._refcount), 1)
+        self.assertEquals(cache._refcount['key1'], 2)
+
+    def test_purge(self):
+        """Test the queue compact and cache purge"""
+        cache = LRUCache(100, 4)
+        values = [('key'+str(i), j) for i in range(50) for j in range(8)]
+        for i, j in values:
+            cache[i] = j
+        # we hit the limit
+        self.assertEquals(len(cache._queue), 400)
+        self.assertEquals(len(cache._cache), 50)
+        # insert key1 item to compact the queue
+        cache[values[0][0]] = values[0][1]
+        self.assertEquals(len(cache._queue), 50)
+        self.assertEquals(len(cache._cache), 50)
+
+        # now with a key not present in the cache
+        cache = LRUCache(100, 4)
+        for i, j in values:
+            cache[i] = j
+        # we hit the limit
+        self.assertEquals(len(cache._queue), 400)
+        self.assertEquals(len(cache._cache), 50)
+        # insert key1 item to compact the queue
+        cache['non-present-key'] = 'some value'
+        self.assertEquals(len(cache._queue), 51)
+        self.assertEquals(len(cache._cache), 51)
+
+    def test_statistics(self):
+        """Tests if the cache correclty keeps track of misses and hits."""
+        cache = LRUCache(100, 4)
+        # set some data in the cache
+        values = [('key'+str(i), i) for i in range(10)]
+        for i, j in values:
+            cache[i] = j
+        self.assertEquals(len(cache._queue), len(values))
+        self.assertEquals(len(cache._cache), len(values))
+        # compare 4 items with the values
+        for i, j in values[:4]:
+            self.assertEquals(cache[i], j)
+        # check the hits value
+        self.assertEquals(cache.hits, 4)
+        # try to get items not present in the cache
+        for i, j in values[5:10]:
+            self.assertRaises(KeyError, cache.__getitem__, i*10)
+        self.assertEquals(cache.misses, 5)
+
+    def test_inconsistency(self):
+        """Test that the consistency checking works as expected"""
+        cache = LRUCache(2, 1)
+        cache['foo'] = 'bar'
+        # add it again to the _queue to force a inconsistency
+        cache._queue.append('foo')
+        self.assertRaises(CacheInconsistencyError,
+                          cache.__setitem__, 'bar', 'foo')
+
+
 def test_suite():
     # pylint: disable-msg=C0111
     return unittest.TestLoader().loadTestsFromName(__name__)

=== modified file 'ubuntuone/syncdaemon/file_shelf.py'
--- ubuntuone/syncdaemon/file_shelf.py	2009-07-15 07:55:08 +0000
+++ ubuntuone/syncdaemon/file_shelf.py	2009-07-17 20:37:43 +0000
@@ -22,6 +22,10 @@
 from UserDict import DictMixin
 import cPickle
 import os
+import stat
+import errno
+from collections import deque
+
 
 class FileShelf(object, DictMixin):
     """ File based shelf.
@@ -46,10 +50,19 @@
         """ check if the path isn't a file and in case it don't exists,
         creates it
         """
-        if os.path.exists(path) and os.path.isfile(path):
-            os.unlink(path)
-        if not os.path.exists(path):
-            os.makedirs(path)
+        try:
+            stat_result = os.stat(path)
+            # is a regular file?
+            if stat.S_ISREG(stat_result.st_mode):
+                os.unlink(path)
+                os.makedirs(path)
+            # else, the dir is already there
+        except OSError, e:
+            if e.errno == errno.ENOENT:
+                # the file or dir don't exist
+                os.makedirs(path)
+            else:
+                raise
 
     def key_file(self, key):
         """ get the real key used by the storage from a key """
@@ -62,9 +75,7 @@
             raise ValueError("The key (%r) needs to be longer!" % key)
 
         letters = [key[i] for i in xrange(0, self._depth)]
-        path = os.path.join(self._path, *letters)
-        self._check_and_create_dirs(path)
-        return os.path.join(path, key)
+        return os.path.join(os.path.join(self._path, *letters), key)
 
     def has_key(self, key):
         """ True if the the shelf has the key """
@@ -105,7 +116,9 @@
 
     def __setitem__(self, key, value):
         """ setitem backed by the file storage """
-        with open(self.key_file(key), "wb") as fh:
+        path = self.key_file(key)
+        self._check_and_create_dirs(os.path.dirname(path))
+        with open(path, "wb") as fh:
             cPickle.dump(value, fh, protocol=2)
             fh.flush()
             os.fsync(fh.fileno())
@@ -124,3 +137,143 @@
             counter += 1
         return counter
 
+
+class CachedFileShelf(FileShelf):
+    """A extension of FileShelf that uses a cache of 1500 items"""
+
+    def __init__(self, *args, **kwargs):
+        """Create the instance"""
+        super(CachedFileShelf, self).__init__(*args, **kwargs)
+        # XXX: the size of the cache and the compact threshold needs to be
+        # tweaked once we get more statistics from real usage
+        self._cache = LRUCache(1500, 4)
+
+    @property
+    def cache_misses(self):
+        """proterty to access the internal cache misses"""
+        return self._cache.misses
+
+    @property
+    def cache_hits(self):
+        """proterty to access the internal cache hits"""
+        return self._cache.hits
+
+    def __getitem__(self, key):
+        """ getitem backed by the file storage """
+        try:
+            return self._cache[key]
+        except KeyError:
+            value = super(CachedFileShelf, self).__getitem__(key)
+            # add it to the cache
+            self._cache[key] = value
+            return value
+
+    def __setitem__(self, key, value):
+        """ setitem backed by the file storage """
+        super(CachedFileShelf, self).__setitem__(key, value)
+        if key in self._cache:
+            self._cache[key] = value
+
+    def __delitem__(self, key):
+        """ delitem backed by the file storage """
+        super(CachedFileShelf, self).__delitem__(key)
+        if key in self._cache:
+            del self._cache[key]
+
+
+class LRUCache(object):
+    """A least-recently-used|updated cache with maximum size.
+
+    The object(s) added to the cache must be hashable.
+    Cache performance statistics stored in self.hits and self.misses.
+
+    Based on recipe #252524 by Raymond Hettinger
+    """
+    def __init__(self, maxsize, compact_threshold=4):
+        """Create the instance.
+        @param maxsize:
+        @param compact_threshold:
+        """
+        self._maxsize = maxsize
+        self._compact_threshold = compact_threshold
+        self._cache = {}      # mapping of args to results
+        self._queue = deque() # order that keys have been accessed
+        self._refcount = {}   # number of times each key is in the access queue
+        self.hits = 0
+        self.misses = 0
+
+    def __getitem__(self, key):
+        """return the item from the cache or raise KeyError."""
+        try:
+            result = self._cache[key]
+            self.hits += 1
+        except KeyError:
+            # get the value and increase misses
+            self.misses += 1
+            raise
+        else:
+            self.update(key)
+            self.purge()
+        return result
+
+    def __setitem__(self, key, value):
+        """set the key, value in the cache"""
+        self._cache[key] = value
+        self.update(key)
+        self.purge()
+
+    def __delitem__(self, key):
+        """removes the key, value from the cache"""
+        del self._cache[key]
+        self._refcount[key] -= 1
+        self._queue.remove(key)
+        self.purge()
+
+    def __contains__(self, key):
+        """returns True if key is in the cache"""
+        return key in self._cache
+
+    def update(self, key):
+        """Update the least recently used|updated and refcount"""
+        self._queue.append(key)
+        self._refcount[key] = self._refcount.get(key, 0) + 1
+
+    def purge(self):
+        """Purge least recently accessed cache contents and periodically
+        compact the queue by duplicate keys.
+        """
+        while len(self._cache) > self._maxsize:
+            k = self._queue.popleft()
+            self._refcount[k] -= 1
+            if not self._refcount[k]:
+                if k in self._cache:
+                    del self._cache[k]
+                if k in self._refcount:
+                    del self._refcount[k]
+
+        # Periodically compact the queue by duplicate keys
+        queue_len = len(self._queue)
+        if queue_len > self._maxsize * self._compact_threshold:
+            for _ in xrange(queue_len):
+                k = self._queue.popleft()
+                if self._refcount[k] == 1:
+                    self._queue.append(k)
+                else:
+                    self._refcount[k] -= 1
+            if not (len(self._queue) == len(self._cache) \
+                    == len(self._refcount) \
+                    == sum(self._refcount.itervalues())):
+                # create a custom exception for this error
+                raise CacheInconsistencyError(len(self._queue),
+                                              len(self._cache),
+                                              len(self._refcount),
+                                              sum(self._refcount.itervalues()))
+
+
+class CacheInconsistencyError(Exception):
+    """Exception representing a inconsistency in the cache"""
+
+    def __str__(self):
+        return "Inconsistency in the cache: queue: %d cache: %d refcount: %d" \
+                " sum(refcount.values): %d" % self.args
+

=== modified file 'ubuntuone/syncdaemon/filesystem_manager.py'
--- ubuntuone/syncdaemon/filesystem_manager.py	2009-07-14 13:37:22 +0000
+++ ubuntuone/syncdaemon/filesystem_manager.py	2009-07-16 22:52:12 +0000
@@ -187,7 +187,7 @@
             raise TypeError("data_dir should be a string instead of %s" % \
                             type(data_dir))
         fsmdir = os.path.join(data_dir, 'fsm')
-        self.fs = file_shelf.FileShelf(fsmdir)
+        self.fs = file_shelf.CachedFileShelf(fsmdir)
         self.shares = {}
         self.vm = vm
 

=== modified file 'ubuntuone/syncdaemon/main.py'
--- ubuntuone/syncdaemon/main.py	2009-07-14 05:49:49 +0000
+++ ubuntuone/syncdaemon/main.py	2009-07-17 20:31:26 +0000
@@ -94,10 +94,12 @@
     def log_mark(self):
         """ log a "mark" that includes the current AQ state and queue size"""
         self.logger.info("%s %s (state: %s; queues: metadata: %d; content: %d;"
-                         " hash: %d) %s" % ('-'*4, 'MARK', self.state.name,
+                         " hash: %d, fsm-cache: hit=%d miss=%d) %s" % ('-'*4, 'MARK', self.state.name,
                                          len(self.action_q.meta_queue),
                                          len(self.action_q.content_queue),
-                                         len(self.hash_q), '-'*4))
+                                         len(self.hash_q),
+                                         self.fs.fs.cache_hits,
+                                         self.fs.fs.cache_misses, '-'*4))
 
     def wait_for_nirvana(self, last_event_interval=0.5):
         """Get a deferred that will fire when there are no more