Synnefo » Pithos

from sqlite3 import connect

from os import makedirs

from os.path import realpath, dirname, exists, isdir

from pithos.lib.hashfiler import (Blocker, Mapper, Noder, AccessController,

                                  inf, SERIAL, PARENT, PATH, SIZE,

                                  POPULATION, POPSIZE, SOURCE, MTIME, CLUSTER)

def packversion(serial, mtime):

    """Return a version string combining serial and mtime"""

    return "%x:%.8f" % (serial, mtime)

def unpackversion(strver):

    """Return a serial, mtime tuple from a combined version string"""

    s, t = strver.split(':')

    return int(s, 16), float(t)

class Filer(Noder, AccessController, Mapper, Blocker):

    version = None

    def __init__(self, **params):

        filepath = realpath(params['filepath'])

        basepath = dirname(filepath)

        if not exists(basepath):

            makedirs(basepath)

        elif not isdir(basepath):

            raise ValueError("%s is not a directory" % (basepath,))

        conn = connect(filepath, check_same_thread=0)

        cur = conn.cursor()

        self.filepath = filepath

        self.conn = conn

        self.cur = cur

        self.execute = cur.execute

        self.executemany = cur.executemany

        self.fetchone = cur.fetchone

        self.fetchall = cur.fetchall

        self.serial = 0

        self.limit_serial = 0

        self.serial_step = 256

        self.filer_init()

        params['connection'] = conn

        params['cursor'] = cur

        Blocker.__init__(self, **params)

        params['namelen'] = self.hashlen

        Noder.__init__(self, **params)

        AccessController.__init__(self, **params)

        Mapper.__init__(self, **params)

    def filer_init(self):

        execute = self.execute

        execute(""" create table if not exists conf

                          ( param text,

                            value text,

                            primary key(param) )""")

        execute(""" select value from conf where param = 'serial' """)

        r = self.fetchone()

        if r is None:

            execute(""" insert into conf values('serial', 100) """)

    def alloc_serial(self, nr=1):

        """Allocate global serial identifiers.

           Pre-allocate a buffer of identifiers locally,

           to reduce load to the database.

        """

        serial = self.serial + nr

        if serial >= self.limit_serial:

            params = self.params

            db_limit_serial = int(self.get_parameter('serial'), 16)

            limit_serial = db_limit_serial + self.serial_step

            self.set_parameter('serial', hex(limit_serial))

            self.limit_serial = limit_serial

            serial = db_limit_serial

        self.serial = serial

        return serial

    def quit(self):

        return

    def begin(self):

        """Commits the current transaction (if any)

           and starts a new one.

        """

        self.cur.execute("begin deferred")

    def commit(self):

        """Commit the current transaction (if any)

           but do not start a new one.

        """

        self.conn.commit()

    def rollback(self):

        """Abort the current transaction and undo the changes it made."""

        self.conn.rollback()

    def __enter__(self):

        self.begin()

        return self

    def __exit__(self, exc, val, trace):

        if not exc:

            self.commit()

        else:

            self.rollback()

    def get_parameter(self, name):

        c = self.cur.execute("select value from conf where param = ?", (name,))

        r = c.fetchone()

        if r is None:

            return r

        return r[0]

    def set_parameter(self, name, value):

        q = "insert or replace into conf values (?, ?)"

        self.cur.execute(q, (name, value))

    def list_parameters(self):

        c = self.cur.execute("select * from conf")

        return c.fetchall()

    def file_create(self, parent, path, size, cluster=0):

        """Create a new file on path in parent's namespace with the given size.

           Return the (serial, timestamp) of the created file.

        """

        return self.node_create(parent, path, size, None, cluster=cluster)

    def file_list(self, parent, prefix='',

                  start='', delimiter=None,

                  after=0.0, before=inf,

                  filterq=None, versions=0,

                  cluster=0, limit=10000):

        return self.node_list(parent, prefix=prefix,

                              start=start, delimiter=delimiter,

                              after=after, before=before,

                              filterq=filterq, versions=versions,

                              cluster=cluster, limit=limit)

    def file_get_properties(self, serial):

        """Alias for node_get_properties()"""

        return self.node_get_properties(serial)

    def file_set_attributes(self, serial, items):

        """Alias for node_attr_set()."""

        return self.node_attr_set(serial, items)

    def file_increment(self, serial, size=None):

        """Alias for node_increment()."""

        return self.node_increment(serial, size=None)

    def file_list_attributes(self, parent):

        """List file attributes. Alias for node_attr_list_keys()."""

        return self.node_attr_list_keys(parent)

    def file_get_attributes(self, serial, keys=()):

        """Alias for node_attr_get()."""

        return self.node_attr_get(serial, keys=keys)

    def file_del_attributes(self, serial):

        """Alias for node_attr_del()."""

        return self.node_attr_del(serial)

    def file_remove(self, serial, recursive=0):

        """Remove the file given.

           If the file's namespace is not empty,

           remove nothing and return false.

           Return true if the file was removed.

        """

        r = self.node_remove(serial, recursive=recursive)

        if not r:

            return 0

        r = self.map_remv(serial)

        return 1

    def file_purge_cluster(self, parent, cluster=0):

        """Purge a whole cluster from the namespace of a node.

           Return the serials of the purged nodes.

        """

        purged = self.node_delete(parent, cluster=0)

        purged = [props[0] for props in purged]

        map_remv = self.map_remv

        for serial in purged:

            # XXX: victim caches?

            map_remv(serial)

        return purged

    def file_purge_path(self, parent, path, after=0, before=inf, cluster=0):

        """Purge all versions in (after, before)

           for the given parent and path and cluster.

           Return the list of serials deleted.

        """

        serials = self.node_purge(parent, path, after, cluster=cluster)

        map_remv = self.map_remv

        for serial in serials:

            map_remv(serial)

        return serials

    def file_lookup(self, parent, path, before=inf,

                    create=0, size=0, cluster=0):

        """Lookup the latest node in parent and path.

           If the object is not found and create is true,

           it is created with the given size.

           Also, accept a before time limit and a cluster.

        """

        props = self.node_lookup(parent, path, before=inf, cluster=cluster)

        if props or not create:

            return props

        source = None

        population = 0

        popsize = 0

        r = self.node_create(parent, path, size, source, cluster=cluster)

        serial, mtime = r

        return (serial, parent, path, size,

                population, popsize, source, mtime)

    def file_copy(self, source, parent=None, path=None):

        """Create a file as a copy (data and attributes) of another.

           If parent or path is not specified, it remains the same.

        """

        serial, mtime = self.node_copy(source, parent, path)

        self.node_attr_copy(source, serial)

        return serial, mtime

    def file_read(self, serial, nr, blkoff=0):

        """Read a number of blocks at block offset from a file."""

        hashes = self.map_retr(serial, blkoff=blkoff, nr=nr)

        blocks = self.block_retr(hashes)

        return blocks

    def file_write(self, serial, blocks, blkoff=0, fork=0):

        """Write blocks at block offset and create a new version.

           If fork is true, the old version is preserved.

        """

        blocksize = self.blocksize

        datasize = (len(blocks) -1) * blocksize + len(blocks[-1])

        props = self.node_get_properties(serial)

        oldsize = props[3]

        newsize = blkoff * blocksize + datasize

        if fork:

            serial, mtime = self.file_clone(serial, props[1], props[2])

        hashes, missing = self.block_stor(blocks)

        self.map_stor(serial, hashes, blkoff=blkoff)

        node_increment = self.node_increment

        if newsize > oldsize:

            mtime = node_increment(serial, newsize)

        elif not fork:

            mtime = node_increment(serial, oldsize)

        return serial, mtime, newsize, hashes, missing

    def file_move(self, serial, parent, path=None):

        """Move a file, possibly to another parent and path."""

        if path is None:

            props = self.node_get_properties(serial)

            path = props[PATH]

        return self.node_move(serial, parent, path)

    def file_recluster(self, serial, cluster):

        """Send file to the given cluster."""

        self.node_set_properties(serial, (('cluster', cluster),))

if __name__ == "__main__":

    hf = Filer( filepath=".hf-test/nodes.sqlite3",

                mappath=".hf-test/maps",

                blockpath=".hf-test/blocks",

                blocksize=64*1024,

                hashtype='sha256')

    #hf.quit()