a simple key/value store

example.py

wiki = KVStore("/tmp/wiki.kvstore")
with wiki.write_transaction:
    wiki["home1"] = "This is the home page 1"
    wiki["home2"] = "This is the home page 2"

with wiki.read_transaction:
    for name in wiki:
        print name, "=", wiki[name]

kvstore.mdown

Wie kann ein einfacher Key/Value-Store (KVS) in Python aussehen?

Hier ist ein Beispiel:

wiki = KVStore("/tmp/wiki.kvstore")
with wiki.write_transaction:
    wiki["home1"] = "This is the home page 1"
    wiki["home2"] = "This is the home page 2"

with wiki.read_transaction:
    for name in wiki:
        print name, "=", wiki[name]

Jeder Zugriff auf den KVS muss innerhalb einer mit with gebildeten Transaktion stattfinden. Ich könnte mir vorstellen, alternativ auch wiki.in_transaction(func) zu unterstützen, aber das wäre nur eine andere Art, das selbe zu erreichen. Aus Gründen der Effizienz unterscheide ich zwei Arten von Transaktionen: Lesende und Schreibende. Letztere brauchen einen exklusiven Zugriff auf den Store, was ich dadurch sicherstelle, dass ich mir einen SHA-1 digest über den Inhalt des Stores am Anfang der Transaktion merke und sicherstelle, dass am Ende der Transaktion der SHA-1 noch immer der selbe ist. Andernfalls ist die Transaktion gescheitert. Eine lesende Transaktion kann mit anderen Transaktionen parallel ausgeführt werden.

Ansonsten verhält sich der Store wie ein dict. Neben [] und []= gibt es noch get und setdefault mit der üblichen Semantik. Natürlich geht auch del und in. Und wie man sieht sind die Schlüssel aufzählbar.

Bei meiner ersten Implementierung will ich parallele Dateizugriffe ausklammern und hoffen, dass das Betriebssystem Dateien zum Schreiben exklusiv öffnet (was Unix nicht macht). Ich weiß, dass meine Implementierung nicht threadsafe ist, aber ich möchte wenigstens sicherstellen, dass verschiedene Prozesse auf einem KVS arbeiten können.

Hier ist read_transaction:

@property
def read_transaction(self):
    return self.begin_transaction("read")

def begin_transaction(mode):
    self._check_not_in_transaction()
    if os.path.isfile(self.name):
        with open(self.name, "rb" if mode == "read" else "r+b") as f:
            sha1 = f.read(20)
            if self.sha1 != sha1:
                self.sha1 = sha1
                self.data = pickle.load(f)
    else:
        self.sha1 = ''
        self.data = {}
    self.transaction = mode
    return self

Zunächst prüfe ich, dass ich nicht schon in einer Transaktion bin. Schachteln ist nicht erlaubt. Gibt es eine Datei, öffne ich sie (nicht-exklusiv) zum Lesen und versuche, einen SHA-1 digest als Prüfsumme über die Daten und die Daten selbst zu lesen und mittels pickle.load daraus wieder Python-Objekte zu machen. Ich habe eine Optimierung eingebaut: Stimmt der eingelesene SHA-1 digest mit dem eigenen (zuvor eingelesenen) überein, kann ich mir das Lesen der Daten und das Umwandeln in Python-Objekte sparen. Ich habe noch einen aktuellen Stand. Gibt es keine Datei (oder, das ist eine Unterlassung meinerseits, zeigt self.name fälschlicherweise auf irgendwas anderes) nehme ich an, der KVS ist leer. In beiden Fällen bin ich danach in einer Transaktion.

Hier ist, wie ich eine Transaktion beginne, in der ich auch ändern darf:

@property
def write_transaction(self):
    return self.begin_transaction("write")

Der feine Unterschied ist, dass ich die Datei zum (exklusiven) Lesen und Schreiben öffne, selbst wenn ich nicht schreibe. Meine Hoffnung hier ist, dass ich auf diese Weise verhindere, dass ich die Datei öffne, während ein andere Prozess sie gerade schreibt und so einen inkonsistenten Zustand verhindere.

Am Ende der Transaktion muss ich die Daten wieder sichern. Hier ist die Implementierung:

def __exit__(self, exc, value, tb):
    if self.transaction == "write":
        if exc:
            self.sha1 = None
            self.data = None
        else:
            new_data = pickle.dumps(self.data)
            new_sha1 = hashlib.sha1(new_data).digest()
            if not os.path.isfile(self.name):
                open(self.name, "a").close()
            with open(self.name, "r+b") as f:
                sha1 = f.read(20)
                if sha1 and sha1 != self.sha1:
                    self.sha1 = None
                    self.data = None
                    raise KVStore.Error("concurrency error")
                f.truncate()
                f.write(new_sha1)
                f.write(new_data)
            self.sha1 = new_sha1
    self.transaction = None

Für "read"-Transaktionen muss ich nichts tun, außer die Transaktion wieder zurückzusetzen. Bei "write"-Transaktionen ist die Sache komplizierter: Wenn exc einen Wert hat, erreiche ich das Ende des with-Blocks aufgrund einer Exception. In diesem Fall verwerfe ich alle Daten. Dadurch werden sie am Anfang einer neuen Transaktion neu aus der Datei gelesen und alles ist wieder gut. Habe ich das Ende des Blocks auf normale Weise erreicht, muss ich die Python-Objekte zurück in die Datei als Daten schreiben. Zuvor berechne ich jedoch den neuen SHA-1 digest. Das hält den Teil des Programms, der exklusiven Zugriff benötigt, kleiner. Gibt es die Datei noch nicht, versuche ich sie anzulegen. Leider gibt es keinen Mode für open, der dies direkt erlaubt. Mit r+ kann ich nur existierende Dateien zum Schreiben öffnen und mit w+ könnte ich zwar neue Dateien anlegen, existierende würden aber überschrieben und ich kann den SHA-1 digest nicht mehr auslesen. a kann ich allerdings auch nicht weiter verwenden, da dies nur Anhängen und nicht überschreiben erlaubt. Bevor ich die Daten schreibe, prüfe ich, ob der SHA-1 digest aus der Datei immer noch der ist, den ich an Anfang der Transaktion ausgelesen habe. Ist das nicht der Fall, war eine andere parallele Transaktion schneller und ich breche ab. Andernfalls kann ich die neuen Daten (beginnend vom Dateianfang) schreiben. Ich mache dieses beides mit nur einer geöffneten Datei aufgrund der Hoffnung, dass so andere Prozesse nicht meine Datei beschreiben können. Am Schluss aktualisiere ich noch den neuen SHA-1 digest. Dies hilft der nächsten Transaktion, das Einlesen der Daten möglicherweise zu überspringen.

Leider hilft hoffen nicht.

Ich muss die low-level-Operationen aus os benutzen, wenn ich Dateizugriffe sperren will und dann immer noch hoffen, dass das Betriebssystem das überhaupt unterstützt.

So kann ich eine Datei zum gemeinsamen Lesen öffnen:

fd = os.open("/tmp/x", os.O_RDONLY + os.O_SHLOCK)

Ein Prozess, der die selbe Datei für exklusiven Zugriff zu öffnen, bleibt so lange hängen, bis ich meine Datei wieder schließe. Erst dann erhält er seinen File-Descriptor fd. Füge ich noch os.O_CREAT hinzu, wird die Datei angelegt, sollte sie noch nicht existieren. Zum Schreiben brauche ich einen exklusiven Lock, damit nicht zwei Prozesse gleichzeitig die selbe Datei schreiben. Das geht mit os.O_EXLOCK statt os.O_SHLOCK.

So sieht der Anfang einer Transaktion aus:

try:
    fd = os.open(self.name, os.O_RDONLY + os.O_SHLOCK)
    try:
        sha1 = os.read(fd, 20)
        if sha1 and len(sha1) != 20:
            raise IOError("file truncated")
        if self.sha1 != sha1:
            self.sha1 = sha1
            self.data = pickle.loads(os.read(fd, os.fstat(fd).st_size))
    except OSError:
        raise IOError
    finally:
        os.close(fd)
except OSError:
    self.sha1 = ''
    self.data = {}

Ich öffne die Datei (oder das, woraus self.name zeigt; ist es Verzeichnis, gelingt es auch, aber der erste Aufruf von os.read wird scheitern) und lese den SHA-1 digest. Gibt es die Datei nicht, nehme ich an, dass mein KVS leer ist. Kann ich nicht genug Bytes lesen, ist dies ein Fehler. Stimmt der SHA-1 nicht mit dem vorhandenen überein, muss ich die anderen Datei lesen. Statt in einer Schleife Blöcke zu lesen (was möglicherweise effizienter ist) bestimme ich die Dateigröße (wobei ich natürlich 20 Bytes zu viel habe) und lese alles in einem Schwung (hoffentlich) und lasse das von pickle.loads in Python-Objekte verwandeln. Geht das Lesen irgendwie schief, ist das ein Fehler. Ansonsten kann ich die Datei schließen. Sollte dies schief gehen, melde ich fälschlicherweise, dass die Datei leer ist. Low-level-Fehlerbehandlung ist extrem lästig.

So sieht das Ende einer Transaktion aus:

new_data = pickle.dumps(self.data)
new_sha1 = hashlib.sha1(new_data).digest()
try:
    fd = os.open(self.name, os.O_RDWR + os.O_CREAT + os.O_EXLOCK)
    try:
        sha1 = os.read(fd, 20)
        if sha1:
            if len(sha1) != 20:
                raise IOError("file truncated")
            if self.sha1 != sha1:
                self.sha1 = None
                self.data = None
                raise KVStore.Error("concurrency error")
            os.lseek(fd, 0, 0)
        os.write(fd, new_sha1)
        os.write(fd, new_data)
        self.sha1 = new_sha1
    except OSError:
        raise IOError # at this point, the store is corrupted!
    finally:
        os.close(fd)
except OSError:
    raise IOError("cannot write %s" % self.name)

Wieder öffne ich die Datei, diesmal exklusiv. Ich lese und vergleiche den SHA-1 digest und werfe einen Fehler, wenn sich die Datei in der Zwischenzeit verändert hat. Dann schreibe ich die neuen Daten. Tritt dabei ein Fehler auf, habe ich meinen Store kaputt gemacht. Ansonsten schließe ich die Datei (wobei auch ein Fehler auftreten könnte, der ebenfalls den Store kaputt zurück lässt).

So kann das nicht bleiben. Ich muss die alte Datei zwar exklusiv zum Lesen öffnen und den SHA-1 vergleichen und so die Datei blockieren, dann aber die neuen Daten in eine neue Datei speichern und erst wenn dies geklappt hat, die alte Datei löschen und die neue umbenennen.

Ich meine mich zu erinnern, dass ich unter Unix geöffnete Dateien löschen kann. Unter Windows geht das definitiv nicht. Keine gute Lösung. Ich könnte die Originaldatei exklusiv zum Lesen und Schreiben öffnen und statt umzubenennen, dann kopieren. Aber das kann wieder schief gehen. Muss ich zuvor noch die alte Datei zur Sicherheit an eine andere Stelle kopieren?

Sollte ich statt des Dateisystems z.B. SQLite als Backend benutzen?

kvstore.py

import hashlib, pickle, os

class KVStore(object):
    class Error(Exception): pass
    
    def __init__(self, name):
        self.name = name
        self.sha1 = ''
        self.transaction = None

    @property
    def read_transaction(self):
        return self.in_transaction("read")

    @property
    def write_transaction(self):
        return self.in_transaction("write")
    
    def in_transaction(self, mode):
        self._check_not_in_transaction()
        try:
            fd = os.open(self.name, os.O_RDONLY + os.O_SHLOCK)
        except OSError:
            raise KVStore.Error("cannot open %s" % self.name)
        try:
            sha1 = os.read(fd, 20)
            if len(sha1) != 20:
                raise KVStore.Error("file corrupt")
            if self.sha1 != sha1:
                self.sha1 = sha1
                self.data = pickle.loads(os.read(fd, os.fstat(fd).st_size))
        except OSError:
            raise KVStore.Error("cannot read %s" % self.name)
        finally:
            os.close(fd)
        self.transaction = mode
        return self
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc, value, tb):
        if exc:
            self.sha1 = None
            self.data = None
        elif self.transaction == "write":
            new_data = pickle.dumps(self.data)
            new_sha1 = hashlib.sha1(new_data).digest()
            try:
                fd = os.open(self.name, os.O_RDWR + os.O_CREAT + os.O_EXLOCK)
            except OSError:
                raise KVStore.Error("cannot access %s" % self.name)
            try:
                sha1 = os.read(fd, 20)
                if sha1 and sha1 != self.sha1:
                    raise KVStore.Error("concurrency error")
                os.lseek(fd, 0, 0)
                os.write(fd, new_sha1)
                os.write(fd, new_data)
            except OSError:
                raise KVStore.Error("store corrupt, sorry")
            finally:
                os.close(fd)
            self.sha1 = new_sha1
        self.transaction = None
    
    def __iter__(self):
        self._check_transaction("read")
        return self.data.iterkeys()

    def __getitem__(self, name):
        self._check_transaction("read")
        return self.data[name]
    
    def get(self, name, default=None):
        self._check_transaction("read")
        return self.data.get(name, default)
    
    def __contains__(self, name):
        self._check_transaction("read")
        return name in self.data
    
    def setdefault(self, name, default):
        self._check_transaction("write")
        return self.setdefault(name, default)
    
    def __setitem__(self, name, value):
        self._check_transaction("write")
        self.data[name] = value
    
    def __delitem__(self, name):
        self._check_transaction("write")
        del self.data[name]
    
    def _check_not_in_transaction(self):
        if self.transaction:
            raise KVStore.Error("nested transaction")
    
    def _check_transaction(self, mode):
        if self.transaction != mode:
            raise KVStore.Error("not in transaction")