apenwarr-redo/state.py

import sys, os, errno, glob, stat, sqlite3
import vars
from helpers import unlink, err, debug2, debug3, mkdirp, close_on_exec

SCHEMA_VER=1
TIMEOUT=60

_db = None
def db():
    global _db
    if _db:
        return _db
    dbdir = '%s/.redo' % vars.BASE
    dbfile = '%s/db.sqlite3' % dbdir
    mkdirp(dbdir)
    must_create = not os.path.exists(dbfile)
    if not must_create:
        _db = sqlite3.connect(dbfile, timeout=TIMEOUT)
        try:
            row = _db.cursor().execute("select version from Schema").fetchone()
        except sqlite3.OperationalError:
            row = None
        ver = row and row[0] or None
        if ver != SCHEMA_VER:
            err("state database: discarding v%s (wanted v%s)\n"
                % (ver, SCHEMA_VER))
            must_create = True
            _db = None
    if must_create:
        unlink(dbfile)
        _db = sqlite3.connect(dbfile, timeout=TIMEOUT)
        _db.execute("create table Schema "
                    "    (version int)")
        _db.execute("create table Runid "
                    "    (id integer primary key autoincrement)")
        _db.execute("create table Files "
                    "    (name not null primary key, "
                    "     is_generated int, "
                    "     checked_runid int, "
                    "     changed_runid int, "
                    "     stamp, "
                    "     csum)")
        _db.execute("create table Deps "
                    "    (target int, "
                    "     source int, "
                    "     mode not null, "
                    "     primary key (target,source))")
        _db.execute("insert into Schema (version) values (?)", [SCHEMA_VER])
        _db.execute("insert into Runid default values")  # eat the '0' runid
        _db.commit()

    if not vars.RUNID:
        _db.execute("insert into Runid default values")
        _db.commit()
        vars.RUNID = _db.execute("select last_insert_rowid()").fetchone()[0]
        os.environ['REDO_RUNID'] = str(vars.RUNID)
    
    _db.execute("pragma journal_mode = PERSIST")
    _db.execute("pragma synchronous = off")
    return _db
    

def init():
    # FIXME: just wiping out all the locks is kind of cheating.  But we
    # only do this from the toplevel redo process, so unless the user
    # deliberately starts more than one redo on the same repository, it's
    # sort of ok.
    db()
    for f in glob.glob('%s/.redo/lock*' % vars.BASE):
        os.unlink(f)


_insane = None
def is_sane():
    global _insane
    if not _insane:
        _insane = not os.path.exists('%s/.redo' % vars.BASE)
        if _insane:
            err('.redo directory disappeared; cannot continue.\n')
    return not _insane


_cwd = None
def relpath(t, base):
    global _cwd
    if not _cwd:
        _cwd = os.getcwd()
    t = os.path.normpath(os.path.join(_cwd, t))
    tparts = t.split('/')
    bparts = base.split('/')
    for tp,bp in zip(tparts,bparts):
        if tp != bp:
            break
        tparts.pop(0)
        bparts.pop(0)
    while bparts:
        tparts.insert(0, '..')
        bparts.pop(0)
    return '/'.join(tparts)


def xx_sname(typ, t):
    # FIXME: t.replace(...) is non-reversible and non-unique here!
    tnew = relpath(t, vars.BASE)
    v = vars.BASE + ('/.redo/%s^%s' % (typ, tnew.replace('/', '^')))
    if vars.DEBUG >= 3:
        debug3('sname: (%r) %r -> %r\n' % (os.getcwd(), t, tnew))
    return v


class File(object):
    __slots__ = ['id', 'name', 'is_generated',
                 'checked_runid', 'changed_runid',
                 'stamp', 'csum']
    
    def __init__(self, id=None, name=None):
        q = ('select rowid, name, is_generated, checked_runid, changed_runid, '
             '    stamp, csum '
             '  from Files ')
        if id != None:
            q += 'where rowid=?'
            l = [id]
        elif name != None:
            name = relpath(name, vars.BASE)
            q += 'where name=?'
            l = [name]
        else:
            raise Exception('name or id must be set')
        d = db()
        row = d.execute(q, l).fetchone()
        if not row:
            if not name:
                raise Exception('File with id=%r not found and '
                                'name not given' % id)
            d.execute('insert into Files (name) values (?)', [name])
            d.commit()
            row = d.execute(q, l).fetchone()
            assert(row)
        (self.id, self.name, self.is_generated,
         self.checked_runid, self.changed_runid,
         self.stamp, self.csum) = row

    def save(self):
        if not os.path.exists('%s/.redo' % vars.BASE):
            # this might happen if 'make clean' removes the .redo dir
            return
        d = db()
        d.execute('update Files set '
                  '    is_generated=?, checked_runid=?, changed_runid=?, '
                  '    stamp=?, csum=? '
                  '    where rowid=?',
                  [self.is_generated, self.checked_runid, self.changed_runid,
                   self.stamp, self.csum,
                   self.id])
        d.commit()

    def set_checked(self):
        self.checked_runid = vars.RUNID
        
    def set_changed(self):
        debug2('BUILT: %r (%r)\n' % (self.name, self.stamp))
        self.changed_runid = vars.RUNID

    def set_static(self):
        self.update_stamp()

    def update_stamp(self):
        newstamp = self.read_stamp()
        if newstamp != self.stamp:
            debug2("STAMP: %s: %r -> %r\n" % (self.name, self.stamp, newstamp))
            self.stamp = newstamp
            self.set_changed()

    def is_changed(self):
        return self.changed_runid and self.changed_runid >= vars.RUNID

    def is_checked(self):
        return (self.checked_runid and self.checked_runid >= vars.RUNID
                and not (self.changed_runid 
                         and self.changed_runid >= self.checked_runid))

    def deps(self):
        q = "select mode, source from Deps where target=?"
        for mode,source_id in db().execute(q, [self.id]).fetchall():
            assert(mode in ('c', 'm'))
            name = File(id=source_id).name
            yield mode,name

    def zap_deps(self):
        debug2('zap-deps: %r\n' % self.name)
        db().execute('delete from Deps where target=?', [self.id])
        db().commit()

    def add_dep(self, mode, dep):
        src = File(name=dep)
        reldep = relpath(dep, vars.BASE)
        debug2('add-dep: %r < %s %r\n' % (self.name, mode, reldep))
        assert(src.name == reldep)
        d = db()
        d.execute("delete from Deps where target=? and source=?",
                  [self.id, src.id])
        d.execute("insert into Deps "
                  "    (target, mode, source) values (?,?,?)",
                  [self.id, mode, src.id])
        d.commit()

    def read_stamp(self):
        try:
            st = os.stat(os.path.join(vars.BASE, self.name))
        except OSError:
            return '0'  # does not exist
        if stat.S_ISDIR(st.st_mode):
            return 'dir'  # the timestamp of a directory is meaningless
        else:
            # a "unique identifier" stamp for a regular file
            return str((st.st_ctime, st.st_mtime, st.st_size, st.st_ino))
        


class Lock:
    def __init__(self, t):
        self.owned = False
        self.rfd = self.wfd = None
        self.lockname = xx_sname('lock', t)

    def __del__(self):
        if self.owned:
            self.unlock()

    def trylock(self):
        try:
            os.mkfifo(self.lockname, 0600)
            self.owned = True
            self.rfd = os.open(self.lockname, os.O_RDONLY|os.O_NONBLOCK)
            self.wfd = os.open(self.lockname, os.O_WRONLY)
            close_on_exec(self.rfd, True)
            close_on_exec(self.wfd, True)
        except OSError, e:
            if e.errno == errno.EEXIST:
                pass
            else:
                raise

    def waitlock(self):
        while not self.owned:
            self.wait()
            self.trylock()
        assert(self.owned)
            
    def unlock(self):
        if not self.owned:
            raise Exception("can't unlock %r - we don't own it" 
                            % self.lockname)
        unlink(self.lockname)
        # ping any connected readers
        os.close(self.rfd)
        os.close(self.wfd)
        self.rfd = self.wfd = None
        self.owned = False

    def wait(self):
        if self.owned:
            raise Exception("can't wait on %r - we own it" % self.lockname)
        try:
            # open() will finish only when a writer exists and does close()
            fd = os.open(self.lockname, os.O_RDONLY)
            try:
                os.read(fd, 1)
            finally:
                os.close(fd)
        except OSError, e:
            if e.errno == errno.ENOENT:
                pass  # it's not even unlocked or was unlocked earlier
            else:
                raise