Merge lp:~mwhudson/loggerhead/concurrency-thoughts into lp:loggerhead

[Michael Hudson-Doyle]

[Michael Hudson-Doyle (mwhudson)]

Hi.

Currently all accesses to the filechanges.sql cache db are serialized. I'd like to experiment on launchpad with having one database for all branches, so this serialization won't really do.

This branch switches to a more optimistic approach to concurrency -- if, by the time you're trying to add the data to the database, someone else has already added it, oh well. The extremely paranoid locking was left over from the days when we used bsddb, sqlite should be much safer!

I also delete some of the locking infrastructure that is no longer used.

[Martin Albisetti (beuno)]

Looks good

review: Approve

=== modified file 'loggerhead/changecache.py'
--- loggerhead/changecache.py	2009-03-18 20:57:19 +0000
+++ loggerhead/changecache.py	2009-03-19 21:01:11 +0000
@@ -17,9 +17,10 @@
 #
 
 """
-a cache for chewed-up "change" data structures, which are basically just a
-different way of storing a revision.  the cache improves lookup times 10x
-over bazaar's xml revision structure, though, so currently still worth doing.
+a cache for chewed-up 'file change' data structures, which are basically just
+a different way of storing a revision delta.  the cache improves lookup times
+10x over bazaar's xml revision structure, though, so currently still worth
+doing.
 
 once a revision is committed in bazaar, it never changes, so once we have
 cached a change, it's good forever.
@@ -27,36 +28,40 @@
 
 import cPickle
 import os
-
-from loggerhead import util
-from loggerhead.lockfile import LockFile
-
-with_lock = util.with_lock('_lock', 'ChangeCache')
+import tempfile
 
 try:
     from sqlite3 import dbapi2
 except ImportError:
     from pysqlite2 import dbapi2
 
+# We take an optimistic approach to concurrency here: we might do work twice
+# in the case of races, but not crash or corrupt data.
 
 class FakeShelf(object):
 
     def __init__(self, filename):
         create_table = not os.path.exists(filename)
+        if create_table:
+            # To avoid races around creating the database, we create the db in
+            # a temporary file and rename it into the ultimate location.
+            fd, path = tempfile.mkstemp(dir=os.path.dirname(filename))
+            self._create_table(path)
+            os.rename(path, filename)
         self.connection = dbapi2.connect(filename)
         self.cursor = self.connection.cursor()
-        if create_table:
-            self._create_table()
 
-    def _create_table(self):
-        self.cursor.execute(
+    def _create_table(self, filename):
+        con = dbapi2.connect(filename)
+        cur = con.cursor()
+        cur.execute(
             "create table RevisionData "
             "(revid binary primary key, data binary)")
-        self.connection.commit()
+        con.commit()
+        con.close()
 
     def _serialize(self, obj):
-        r = dbapi2.Binary(cPickle.dumps(obj, protocol=2))
-        return r
+        return dbapi2.Binary(cPickle.dumps(obj, protocol=2))
 
     def _unserialize(self, data):
         return cPickle.loads(str(data))
@@ -71,10 +76,15 @@
             return self._unserialize(filechange[0])
 
     def add(self, revid, object):
-        self.cursor.execute(
-            "insert into revisiondata (revid, data) values (?, ?)",
-            (revid, self._serialize(object)))
-        self.connection.commit()
+        try:
+            self.cursor.execute(
+                "insert into revisiondata (revid, data) values (?, ?)",
+                (revid, self._serialize(object)))
+            self.connection.commit()
+        except dbapi2.IntegrityError:
+            # If another thread or process attempted to set the same key, we
+            # assume it set it to the same value and carry on with our day.
+            pass
 
 
 class FileChangeCache(object):
@@ -87,11 +97,6 @@
 
         self._changes_filename = os.path.join(cache_path, 'filechanges.sql')
 
-        # use a lockfile since the cache folder could be shared across
-        # different processes.
-        self._lock = LockFile(os.path.join(cache_path, 'filechange-lock'))
-
-    @with_lock
     def get_file_changes(self, entry):
         cache = FakeShelf(self._changes_filename)
         changes = cache.get(entry.revid)

=== removed file 'loggerhead/lockfile.py'
--- loggerhead/lockfile.py	2008-10-24 02:26:05 +0000
+++ loggerhead/lockfile.py	1970-01-01 00:00:00 +0000
@@ -1,84 +0,0 @@
-#
-# Copyright (C) 2006  Robey Pointer <[e-mail]>
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
-#
-
-import os
-import threading
-import time
-
-from loggerhead import util
-
-
-with_lock = util.with_lock('_tlock', 'LockFile')
-
-MAX_STALE_TIME = 5 * 60
-
-
-class LockFile (object):
-    """
-    simple lockfile implementation that mimics the API of threading.Lock, so
-    it can be used interchangably.  it's actually a reentrant lock, so the
-    lock may be acquired multiple times by the same thread, as long as it's
-    released an equal number of times.  unlike threading.Lock, this lock can
-    be used across processes.
-
-    this uses os.open(O_CREAT|O_EXCL), which apparently works even on windows,
-    but will not work over NFS, if anyone still uses that.  so don't put the
-    cache folder on an NFS server...
-    """
-
-    def __init__(self, filename):
-        self._filename = filename
-        # thread lock to maintain internal consistency
-        self._tlock = threading.Lock()
-        self._count = 0
-        if os.path.exists(filename):
-            # remove stale locks left over from a previous run
-            if time.time() - os.stat(filename).st_mtime > MAX_STALE_TIME:
-                os.remove(filename)
-
-    @with_lock
-    def _try_acquire(self):
-        if self._count > 0:
-            self._count += 1
-            return True
-        try:
-            fd = os.open(self._filename,
-                         os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0600)
-            os.close(fd)
-            self._count += 1
-            return True
-        except OSError:
-            return False
-
-    def acquire(self):
-        # try over and over, sleeping on exponential backoff with
-        # an upper limit of about 5 seconds
-        pause = 0.1
-        #max_pause = 5.0
-        max_pause = 0.1
-        while True:
-            if self._try_acquire():
-                return
-            time.sleep(pause)
-            pause = min(pause * 2.0, max_pause)
-
-    @with_lock
-    def release(self):
-        self._count -= 1
-        if self._count == 0:
-            os.remove(self._filename)

=== modified file 'loggerhead/util.py'
--- loggerhead/util.py	2009-03-06 09:57:54 +0000
+++ loggerhead/util.py	2009-03-19 20:31:49 +0000
@@ -449,25 +449,6 @@
     return new_decorator
 
 
-# common threading-lock decorator
-
-
-def with_lock(lockname, debug_name=None):
-    if debug_name is None:
-        debug_name = lockname
-
-    @decorator
-    def _decorator(unbound):
-
-        def locked(self, *args, **kw):
-            getattr(self, lockname).acquire()
-            try:
-                return unbound(self, *args, **kw)
-            finally:
-                getattr(self, lockname).release()
-        return locked
-    return _decorator
-
 
 @decorator
 def lsprof(f):