diff hgext/inotify/server.py @ 6239:39cfcef4f463

Add inotify extension

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/hgext/inotify/server.py	Wed Mar 12 15:30:11 2008 -0700
@@ -0,0 +1,717 @@
+# server.py - inotify status server
+#
+# Copyright 2006, 2007, 2008 Bryan O'Sullivan <[e-mail]>
+# Copyright 2007, 2008 Brendan Cully <[e-mail]>
+#
+# This software may be used and distributed according to the terms
+# of the GNU General Public License, incorporated herein by reference.
+
+from mercurial.i18n import gettext as _
+from mercurial import osutil, ui, util
+import common
+import errno, os, select, socket, stat, struct, sys, time
+
+try:
+    import hgext.inotify.linux as inotify
+    from hgext.inotify.linux import watcher
+except ImportError:
+    print >> sys.stderr, '*** native support is required for this extension'
+    raise
+
+class AlreadyStartedException(Exception): pass
+
+def join(a, b):
+    if a:
+        if a[-1] == '/':
+            return a + b
+        return a + '/' + b
+    return b
+
+walk_ignored_errors = (errno.ENOENT, errno.ENAMETOOLONG)
+
+def walkrepodirs(repo):
+    '''Iterate over all subdirectories of this repo.
+    Exclude the .hg directory, any nested repos, and ignored dirs.'''
+    rootslash = repo.root + os.sep
+    def walkit(dirname, top):
+        hginside = False
+        try:
+            for name, kind in osutil.listdir(rootslash + dirname):
+                if kind == stat.S_IFDIR:
+                    if name == '.hg':
+                        hginside = True
+                        if not top: break
+                    else:
+                        d = join(dirname, name)
+                        if repo.dirstate._ignore(d):
+                            continue
+                        for subdir, hginsub in walkit(d, False):
+                            if not hginsub:
+                                yield subdir, False
+        except OSError, err:
+            if err.errno not in walk_ignored_errors:
+                raise
+        yield rootslash + dirname, hginside
+    for dirname, hginside in walkit('', True):
+        yield dirname
+
+def walk(repo, root):
+    '''Like os.walk, but only yields regular files.'''
+
+    # This function is critical to performance during startup.
+
+    reporoot = root == ''
+    rootslash = repo.root + os.sep
+
+    def walkit(root, reporoot):
+        files, dirs = [], []
+        hginside = False
+
+        try:
+            fullpath = rootslash + root
+            for name, kind in osutil.listdir(fullpath):
+                if kind == stat.S_IFDIR:
+                    if name == '.hg':
+                        hginside = True
+                        if reporoot:
+                            continue
+                        else:
+                            break
+                    dirs.append(name)
+                elif kind in (stat.S_IFREG, stat.S_IFLNK):
+                    path = join(root, name)
+                    files.append((name, kind))
+
+            yield hginside, fullpath, dirs, files
+
+            for subdir in dirs:
+                path = join(root, subdir)
+                if repo.dirstate._ignore(path):
+                    continue
+                for result in walkit(path, False):
+                    if not result[0]:
+                        yield result
+        except OSError, err:
+            if err.errno not in walk_ignored_errors:
+                raise
+    for result in walkit(root, reporoot):
+        yield result[1:]
+
+def _explain_watch_limit(ui, repo, count):
+    path = '/proc/sys/fs/inotify/max_user_watches'
+    try:
+        limit = int(file(path).read())
+    except IOError, err:
+        if err.errno != errno.ENOENT:
+            raise
+        raise util.Abort(_('this system does not seem to '
+                           'support inotify'))
+    ui.warn(_('*** the current per-user limit on the number '
+              'of inotify watches is %s\n') % limit)
+    ui.warn(_('*** this limit is too low to watch every '
+              'directory in this repository\n'))
+    ui.warn(_('*** counting directories: '))
+    ndirs = len(list(walkrepodirs(repo)))
+    ui.warn(_('found %d\n') % ndirs)
+    newlimit = min(limit, 1024)
+    while newlimit < ((limit + ndirs) * 1.1):
+        newlimit *= 2
+    ui.warn(_('*** to raise the limit from %d to %d (run as root):\n') %
+            (limit, newlimit))
+    ui.warn(_('***  echo %d > %s\n') % (newlimit, path))
+    raise util.Abort(_('cannot watch %s until inotify watch limit is raised')
+                     % repo.root)
+
+class Watcher(object):
+    poll_events = select.POLLIN
+    statuskeys = 'almr!?'
+
+    def __init__(self, ui, repo, master):
+        self.ui = ui
+        self.repo = repo
+        self.wprefix = self.repo.wjoin('')
+        self.timeout = None
+        self.master = master
+        self.mask = (
+            inotify.IN_ATTRIB |
+            inotify.IN_CREATE |
+            inotify.IN_DELETE |
+            inotify.IN_DELETE_SELF |
+            inotify.IN_MODIFY |
+            inotify.IN_MOVED_FROM |
+            inotify.IN_MOVED_TO |
+            inotify.IN_MOVE_SELF |
+            inotify.IN_ONLYDIR |
+            inotify.IN_UNMOUNT |
+            0)
+        try:
+            self.watcher = watcher.Watcher()
+        except OSError, err:
+            raise util.Abort(_('inotify service not available: %s') %
+                             err.strerror)
+        self.threshold = watcher.Threshold(self.watcher)
+        self.registered = True
+        self.fileno = self.watcher.fileno
+
+        self.repo.dirstate.__class__.inotifyserver = True
+
+        self.tree = {}
+        self.statcache = {}
+        self.statustrees = dict([(s, {}) for s in self.statuskeys])
+
+        self.watches = 0
+        self.last_event = None
+
+        self.eventq = {}
+        self.deferred = 0
+
+        self.ds_info = self.dirstate_info()
+        self.scan()
+
+    def event_time(self):
+        last = self.last_event
+        now = time.time()
+        self.last_event = now
+
+        if last is None:
+            return 'start'
+        delta = now - last
+        if delta < 5:
+            return '+%.3f' % delta
+        if delta < 50:
+            return '+%.2f' % delta
+        return '+%.1f' % delta
+
+    def dirstate_info(self):
+        try:
+            st = os.lstat(self.repo.join('dirstate'))
+            return st.st_mtime, st.st_ino
+        except OSError, err:
+            if err.errno != errno.ENOENT:
+                raise
+            return 0, 0
+
+    def add_watch(self, path, mask):
+        if not path:
+            return
+        if self.watcher.path(path) is None:
+            if self.ui.debugflag:
+                self.ui.note(_('watching %r\n') % path[len(self.wprefix):])
+            try:
+                self.watcher.add(path, mask)
+                self.watches += 1
+            except OSError, err:
+                if err.errno in (errno.ENOENT, errno.ENOTDIR):
+                    return
+                if err.errno != errno.ENOSPC:
+                    raise
+                _explain_watch_limit(self.ui, self.repo, self.watches)
+
+    def setup(self):
+        self.ui.note(_('watching directories under %r\n') % self.repo.root)
+        self.add_watch(self.repo.path, inotify.IN_DELETE)
+        self.check_dirstate()
+
+    def wpath(self, evt):
+        path = evt.fullpath
+        if path == self.repo.root:
+            return ''
+        if path.startswith(self.wprefix):
+            return path[len(self.wprefix):]
+        raise 'wtf? ' + path
+
+    def dir(self, tree, path):
+        if path:
+            for name in path.split('/'):
+                tree.setdefault(name, {})
+                tree = tree[name]
+        return tree
+
+    def lookup(self, path, tree):
+        if path:
+            try:
+                for name in path.split('/'):
+                    tree = tree[name]
+            except KeyError:
+                return 'x'
+            except TypeError:
+                return 'd'
+        return tree
+
+    def split(self, path):
+        c = path.rfind('/')
+        if c == -1:
+            return '', path
+        return path[:c], path[c+1:]
+
+    def filestatus(self, fn, st):
+        try:
+            type_, mode, size, time = self.repo.dirstate._map[fn][:4]
+        except KeyError:
+            type_ = '?'
+        if type_ == 'n':
+            if not st:
+                return '!'
+            st_mode, st_size, st_mtime = st
+            if size and (size != st_size or (mode ^ st_mode) & 0100):
+                return 'm'
+            if time != int(st_mtime):
+                return 'l'
+            return 'n'
+        if type_ in 'ma' and not st:
+            return '!'
+        if type_ == '?' and self.repo.dirstate._ignore(fn):
+            return 'i'
+        return type_
+
+    def updatestatus(self, wfn, st=None, status=None, oldstatus=None):
+        if st:
+            status = self.filestatus(wfn, st)
+        else:
+            self.statcache.pop(wfn, None)
+        root, fn = self.split(wfn)
+        d = self.dir(self.tree, root)
+        if oldstatus is None:
+            oldstatus = d.get(fn)
+        isdir = False
+        if oldstatus:
+            try:
+                if not status:
+                    if oldstatus in 'almn':
+                        status = '!'
+                    elif oldstatus == 'r':
+                        status = 'r'
+            except TypeError:
+                # oldstatus may be a dict left behind by a deleted
+                # directory
+                isdir = True
+            else:
+                if oldstatus in self.statuskeys and oldstatus != status:
+                    del self.dir(self.statustrees[oldstatus], root)[fn]
+        if self.ui.debugflag and oldstatus != status:
+            if isdir:
+                self.ui.note('status: %r dir(%d) -> %s\n' %
+                             (wfn, len(oldstatus), status))
+            else:
+                self.ui.note('status: %r %s -> %s\n' %
+                             (wfn, oldstatus, status))
+        if not isdir:
+            if status and status != 'i':
+                d[fn] = status
+                if status in self.statuskeys:
+                    dd = self.dir(self.statustrees[status], root)
+                    if oldstatus != status or fn not in dd:
+                        dd[fn] = status
+            else:
+                d.pop(fn, None)
+
+    def check_deleted(self, key):
+        # Files that had been deleted but were present in the dirstate
+        # may have vanished from the dirstate; we must clean them up.
+        nuke = []
+        for wfn, ignore in self.walk(key, self.statustrees[key]):
+            if wfn not in self.repo.dirstate:
+                nuke.append(wfn)
+        for wfn in nuke:
+            root, fn = self.split(wfn)
+            del self.dir(self.statustrees[key], root)[fn]
+            del self.dir(self.tree, root)[fn]
+        
+    def scan(self, topdir=''):
+        self.handle_timeout()
+        ds = self.repo.dirstate._map.copy()
+        self.add_watch(join(self.repo.root, topdir), self.mask)
+        for root, dirs, entries in walk(self.repo, topdir):
+            for d in dirs:
+                self.add_watch(join(root, d), self.mask)
+            wroot = root[len(self.wprefix):]
+            d = self.dir(self.tree, wroot)
+            for fn, kind in entries:
+                wfn = join(wroot, fn)
+                self.updatestatus(wfn, self.getstat(wfn))
+                ds.pop(wfn, None)
+        wtopdir = topdir
+        if wtopdir and wtopdir[-1] != '/':
+            wtopdir += '/'
+        for wfn, state in ds.iteritems():
+            if not wfn.startswith(wtopdir):
+                continue
+            status = state[0]
+            st = self.getstat(wfn)
+            if status == 'r' and not st:
+                self.updatestatus(wfn, st, status=status)
+            else:
+                self.updatestatus(wfn, st, oldstatus=status)
+        self.check_deleted('!')
+        self.check_deleted('r')
+
+    def check_dirstate(self):
+        ds_info = self.dirstate_info()
+        if ds_info == self.ds_info:
+            return
+        self.ds_info = ds_info
+        if not self.ui.debugflag:
+            self.last_event = None
+        self.ui.note(_('%s dirstate reload\n') % self.event_time())
+        self.repo.dirstate.invalidate()
+        self.scan()
+        self.ui.note(_('%s end dirstate reload\n') % self.event_time())
+
+    def walk(self, states, tree, prefix=''):
+        # This is the "inner loop" when talking to the client.
+        
+        for name, val in tree.iteritems():
+            path = join(prefix, name)
+            try:
+                if val in states:
+                    yield path, val
+            except TypeError:
+                for p in self.walk(states, val, path):
+                    yield p
+
+    def update_hgignore(self):
+        # An update of the ignore file can potentially change the
+        # states of all unknown and ignored files.
+
+        # XXX If the user has other ignore files outside the repo, or
+        # changes their list of ignore files at run time, we'll
+        # potentially never see changes to them.  We could get the
+        # client to report to us what ignore data they're using.
+        # But it's easier to do nothing than to open that can of
+        # worms.
+
+        if self.repo.dirstate.ignorefunc is not None:
+            self.repo.dirstate.ignorefunc = None
+            self.ui.note('rescanning due to .hgignore change\n')
+            self.scan()
+        
+    def getstat(self, wpath):
+        try:
+            return self.statcache[wpath]
+        except KeyError:
+            try:
+                return self.stat(wpath)
+            except OSError, err:
+                if err.errno != errno.ENOENT:
+                    raise
+        
+    def stat(self, wpath):
+        try:
+            st = os.lstat(join(self.wprefix, wpath))
+            ret = st.st_mode, st.st_size, st.st_mtime
+            self.statcache[wpath] = ret
+            return ret
+        except OSError, err:
+            self.statcache.pop(wpath, None)
+            raise
+            
+    def created(self, wpath):
+        if wpath == '.hgignore':
+            self.update_hgignore()
+        try:
+            st = self.stat(wpath)
+            if stat.S_ISREG(st[0]):
+                self.updatestatus(wpath, st)
+        except OSError, err:
+            pass
+
+    def modified(self, wpath):
+        if wpath == '.hgignore':
+            self.update_hgignore()
+        try:
+            st = self.stat(wpath)
+            if stat.S_ISREG(st[0]):
+                if self.repo.dirstate[wpath] in 'lmn':
+                    self.updatestatus(wpath, st)
+        except OSError:
+            pass
+
+    def deleted(self, wpath):
+        if wpath == '.hgignore':
+            self.update_hgignore()
+        elif wpath.startswith('.hg/'):
+            if wpath == '.hg/wlock':
+                self.check_dirstate()
+            return
+
+        self.updatestatus(wpath, None)
+        
+    def schedule_work(self, wpath, evt):
+        self.eventq.setdefault(wpath, [])
+        prev = self.eventq[wpath]
+        try:
+            if prev and evt == 'm' and prev[-1] in 'cm':
+                return
+            self.eventq[wpath].append(evt)
+        finally:
+            self.deferred += 1
+            self.timeout = 250
+
+    def deferred_event(self, wpath, evt):
+        if evt == 'c':
+            self.created(wpath)
+        elif evt == 'm':
+            self.modified(wpath)
+        elif evt == 'd':
+            self.deleted(wpath)
+            
+    def process_create(self, wpath, evt):
+        if self.ui.debugflag:
+            self.ui.note(_('%s event: created %s\n') %
+                         (self.event_time(), wpath))
+
+        if evt.mask & inotify.IN_ISDIR:
+            self.scan(wpath)
+        else:
+            self.schedule_work(wpath, 'c')
+
+    def process_delete(self, wpath, evt):
+        if self.ui.debugflag:
+            self.ui.note(('%s event: deleted %s\n') %
+                         (self.event_time(), wpath))
+
+        if evt.mask & inotify.IN_ISDIR:
+            self.scan(wpath)
+        else:
+            self.schedule_work(wpath, 'd')
+
+    def process_modify(self, wpath, evt):
+        if self.ui.debugflag:
+            self.ui.note(_('%s event: modified %s\n') %
+                         (self.event_time(), wpath))
+
+        if not (evt.mask & inotify.IN_ISDIR):
+            self.schedule_work(wpath, 'm')
+
+    def process_unmount(self, evt):
+        self.ui.warn(_('filesystem containing %s was unmounted\n') %
+                     evt.fullpath)
+        sys.exit(0)
+
+    def handle_event(self, fd, event):
+        if self.ui.debugflag:
+            self.ui.note('%s readable: %d bytes\n' %
+                         (self.event_time(), self.threshold.readable()))
+        if not self.threshold():
+            if self.registered:
+                if self.ui.debugflag:
+                    self.ui.note('%s below threshold - unhooking\n' %
+                                 (self.event_time()))
+                self.master.poll.unregister(fd)
+                self.registered = False
+                self.timeout = 250
+        else:
+            self.read_events()
+
+    def read_events(self, bufsize=None):
+        events = self.watcher.read(bufsize)
+        if self.ui.debugflag:
+            self.ui.note('%s reading %d events\n' %
+                         (self.event_time(), len(events)))
+        for evt in events:
+            wpath = self.wpath(evt)
+            if evt.mask & inotify.IN_UNMOUNT:
+                self.process_unmount(wpath, evt)
+            elif evt.mask & (inotify.IN_MODIFY | inotify.IN_ATTRIB):
+                self.process_modify(wpath, evt)
+            elif evt.mask & (inotify.IN_DELETE | inotify.IN_DELETE_SELF |
+                             inotify.IN_MOVED_FROM):
+                self.process_delete(wpath, evt)
+            elif evt.mask & (inotify.IN_CREATE | inotify.IN_MOVED_TO):
+                self.process_create(wpath, evt)
+
+    def handle_timeout(self):
+        if not self.registered:
+            if self.ui.debugflag:
+                self.ui.note('%s hooking back up with %d bytes readable\n' %
+                             (self.event_time(), self.threshold.readable()))
+            self.read_events(0)
+            self.master.poll.register(self, select.POLLIN)
+            self.registered = True
+
+        if self.eventq:
+            if self.ui.debugflag:
+                self.ui.note('%s processing %d deferred events as %d\n' %
+                             (self.event_time(), self.deferred,
+                              len(self.eventq)))
+            eventq = self.eventq.items()
+            eventq.sort()
+            for wpath, evts in eventq:
+                for evt in evts:
+                    self.deferred_event(wpath, evt)
+            self.eventq.clear()
+            self.deferred = 0
+        self.timeout = None
+
+    def shutdown(self):
+        self.watcher.close()
+
+class Server(object):
+    poll_events = select.POLLIN
+
+    def __init__(self, ui, repo, watcher, timeout):
+        self.ui = ui
+        self.repo = repo
+        self.watcher = watcher
+        self.timeout = timeout
+        self.sock = socket.socket(socket.AF_UNIX)
+        self.sockpath = self.repo.join('inotify.sock')
+        try:
+            self.sock.bind(self.sockpath)
+        except socket.error, err:
+            if err[0] == errno.EADDRINUSE:
+                raise AlreadyStartedException(_('could not start server: %s') \
+                                              % err[1])
+            raise
+        self.sock.listen(5)
+        self.fileno = self.sock.fileno
+
+    def handle_timeout(self):
+        pass
+
+    def handle_event(self, fd, event):
+        sock, addr = self.sock.accept()
+
+        cs = common.recvcs(sock)
+        version = ord(cs.read(1))
+
+        sock.sendall(chr(common.version))
+
+        if version != common.version:
+            self.ui.warn(_('received query from incompatible client '
+                           'version %d\n') % version)
+            return
+
+        names = cs.read().split('\0')
+        
+        states = names.pop()
+
+        self.ui.note(_('answering query for %r\n') % states)
+
+        if self.watcher.timeout:
+            # We got a query while a rescan is pending.  Make sure we
+            # rescan before responding, or we could give back a wrong
+            # answer.
+            self.watcher.handle_timeout()
+
+        if not names:
+            def genresult(states, tree):
+                for fn, state in self.watcher.walk(states, tree):
+                    yield fn
+        else:
+            def genresult(states, tree):
+                for fn in names:
+                    l = self.watcher.lookup(fn, tree)
+                    try:
+                        if l in states:
+                            yield fn
+                    except TypeError:
+                        for f, s in self.watcher.walk(states, l, fn):
+                            yield f
+
+        results = ['\0'.join(r) for r in [
+            genresult('l', self.watcher.statustrees['l']),
+            genresult('m', self.watcher.statustrees['m']),
+            genresult('a', self.watcher.statustrees['a']),
+            genresult('r', self.watcher.statustrees['r']),
+            genresult('!', self.watcher.statustrees['!']),
+            '?' in states and genresult('?', self.watcher.statustrees['?']) or [],
+            [],
+            'c' in states and genresult('n', self.watcher.tree) or [],
+            ]]
+
+        try:
+            try:
+                sock.sendall(struct.pack(common.resphdrfmt,
+                                         *map(len, results)))
+                sock.sendall(''.join(results))
+            finally:
+                sock.shutdown(socket.SHUT_WR)
+        except socket.error, err:
+            if err[0] != errno.EPIPE:
+                raise
+
+    def shutdown(self):
+        self.sock.close()
+        try:
+            os.unlink(self.sockpath)
+        except OSError, err:
+            if err.errno != errno.ENOENT:
+                raise
+
+class Master(object):
+    def __init__(self, ui, repo, timeout=None):
+        self.ui = ui
+        self.repo = repo
+        self.poll = select.poll()
+        self.watcher = Watcher(ui, repo, self)
+        self.server = Server(ui, repo, self.watcher, timeout)
+        self.table = {}
+        for obj in (self.watcher, self.server):
+            fd = obj.fileno()
+            self.table[fd] = obj
+            self.poll.register(fd, obj.poll_events)
+
+    def register(self, fd, mask):
+        self.poll.register(fd, mask)
+
+    def shutdown(self):
+        for obj in self.table.itervalues():
+            obj.shutdown()
+
+    def run(self):
+        self.watcher.setup()
+        self.ui.note(_('finished setup\n'))
+        if os.getenv('TIME_STARTUP'):
+            sys.exit(0)
+        while True:
+            timeout = None
+            timeobj = None
+            for obj in self.table.itervalues():
+                if obj.timeout is not None and (timeout is None or obj.timeout < timeout):
+                    timeout, timeobj = obj.timeout, obj
+            try:
+                if self.ui.debugflag:
+                    if timeout is None:
+                        self.ui.note('polling: no timeout\n')
+                    else:
+                        self.ui.note('polling: %sms timeout\n' % timeout)
+                events = self.poll.poll(timeout)
+            except select.error, err:
+                if err[0] == errno.EINTR:
+                    continue
+                raise
+            if events:
+                for fd, event in events:
+                    self.table[fd].handle_event(fd, event)
+            elif timeobj:
+                timeobj.handle_timeout()
+
+def start(ui, repo):
+    m = Master(ui, repo)
+    sys.stdout.flush()
+    sys.stderr.flush()
+
+    pid = os.fork()
+    if pid:
+        return pid
+
+    os.setsid()
+
+    fd = os.open('/dev/null', os.O_RDONLY)
+    os.dup2(fd, 0)
+    if fd > 0:
+        os.close(fd)
+
+    fd = os.open(ui.config('inotify', 'log', '/dev/null'),
+                 os.O_RDWR | os.O_CREAT | os.O_TRUNC)
+    os.dup2(fd, 1)
+    os.dup2(fd, 2)
+    if fd > 2:
+        os.close(fd)
+
+    try:
+        m.run()
+    finally:
+        m.shutdown()
+        os._exit(0)