watchman: add the possibility to set the exact watchman binary location

This is necessary to make rolling releases of new watchman versions across
users.

Differential Revision: https://phab.mercurial-scm.org/D5954

hgext/fsmonitor/__init__.py

@@ -1,828 +1,831 b''
 # __init__.py - fsmonitor initialization and overrides
 #
 # Copyright 2013-2016 Facebook, Inc.
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 '''Faster status operations with the Watchman file monitor (EXPERIMENTAL)
 
 Integrates the file-watching program Watchman with Mercurial to produce faster
 status results.
 
 On a particular Linux system, for a real-world repository with over 400,000
 files hosted on ext4, vanilla `hg status` takes 1.3 seconds. On the same
 system, with fsmonitor it takes about 0.3 seconds.
 
 fsmonitor requires no configuration -- it will tell Watchman about your
 repository as necessary. You'll need to install Watchman from
 https://facebook.github.io/watchman/ and make sure it is in your PATH.
 
 fsmonitor is incompatible with the largefiles and eol extensions, and
 will disable itself if any of those are active.
 
 The following configuration options exist:
 
 ::
 
     [fsmonitor]
     mode = {off, on, paranoid}
 
 When `mode = off`, fsmonitor will disable itself (similar to not loading the
 extension at all). When `mode = on`, fsmonitor will be enabled (the default).
 When `mode = paranoid`, fsmonitor will query both Watchman and the filesystem,
 and ensure that the results are consistent.
 
 ::
 
     [fsmonitor]
     timeout = (float)
 
 A value, in seconds, that determines how long fsmonitor will wait for Watchman
 to return results. Defaults to `2.0`.
 
 ::
 
     [fsmonitor]
     blacklistusers = (list of userids)
 
 A list of usernames for which fsmonitor will disable itself altogether.
 
 ::
 
     [fsmonitor]
     walk_on_invalidate = (boolean)
 
 Whether or not to walk the whole repo ourselves when our cached state has been
 invalidated, for example when Watchman has been restarted or .hgignore rules
 have been changed. Walking the repo in that case can result in competing for
 I/O with Watchman. For large repos it is recommended to set this value to
 false. You may wish to set this to true if you have a very fast filesystem
 that can outpace the IPC overhead of getting the result data for the full repo
 from Watchman. Defaults to false.
 
 ::
 
     [fsmonitor]
     warn_when_unused = (boolean)
 
 Whether to print a warning during certain operations when fsmonitor would be
 beneficial to performance but isn't enabled.
 
 ::
 
     [fsmonitor]
     warn_update_file_count = (integer)
 
 If ``warn_when_unused`` is set and fsmonitor isn't enabled, a warning will
 be printed during working directory updates if this many files will be
 created.
 '''
 
 # Platforms Supported
 # ===================
 #
 # **Linux:** *Stable*. Watchman and fsmonitor are both known to work reliably,
 #   even under severe loads.
 #
 # **Mac OS X:** *Stable*. The Mercurial test suite passes with fsmonitor
 #   turned on, on case-insensitive HFS+. There has been a reasonable amount of
 #   user testing under normal loads.
 #
 # **Solaris, BSD:** *Alpha*. watchman and fsmonitor are believed to work, but
 #   very little testing has been done.
 #
 # **Windows:** *Alpha*. Not in a release version of watchman or fsmonitor yet.
 #
 # Known Issues
 # ============
 #
 # * fsmonitor will disable itself if any of the following extensions are
 #   enabled: largefiles, inotify, eol; or if the repository has subrepos.
 # * fsmonitor will produce incorrect results if nested repos that are not
 #   subrepos exist. *Workaround*: add nested repo paths to your `.hgignore`.
 #
 # The issues related to nested repos and subrepos are probably not fundamental
 # ones. Patches to fix them are welcome.
 
 from __future__ import absolute_import
 
 import codecs
 import hashlib
 import os
 import stat
 import sys
 import weakref
 
 from mercurial.i18n import _
 from mercurial.node import (
     hex,
 )
 
 from mercurial import (
     context,
     encoding,
     error,
     extensions,
     localrepo,
     merge,
     pathutil,
     pycompat,
     registrar,
     scmutil,
     util,
 )
 from mercurial import match as matchmod
 
 from . import (
     pywatchman,
     state,
     watchmanclient,
 )
 
 # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
 # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
 # be specifying the version(s) of Mercurial they are tested with, or
 # leave the attribute unspecified.
 testedwith = 'ships-with-hg-core'
 
 configtable = {}
 configitem = registrar.configitem(configtable)
 
 configitem('fsmonitor', 'mode',
     default='on',
 )
 configitem('fsmonitor', 'walk_on_invalidate',
     default=False,
 )
 configitem('fsmonitor', 'timeout',
     default='2',
 )
 configitem('fsmonitor', 'blacklistusers',
     default=list,
 )
+configitem('fsmonitor', 'watchman_exe',
+    default='watchman',
+)
 configitem('fsmonitor', 'verbose',
     default=True,
 )
 configitem('experimental', 'fsmonitor.transaction_notify',
     default=False,
 )
 
 # This extension is incompatible with the following blacklisted extensions
 # and will disable itself when encountering one of these:
 _blacklist = ['largefiles', 'eol']
 
 def _handleunavailable(ui, state, ex):
     """Exception handler for Watchman interaction exceptions"""
     if isinstance(ex, watchmanclient.Unavailable):
         # experimental config: fsmonitor.verbose
         if ex.warn and ui.configbool('fsmonitor', 'verbose'):
             ui.warn(str(ex) + '\n')
         if ex.invalidate:
             state.invalidate()
         # experimental config: fsmonitor.verbose
         if ui.configbool('fsmonitor', 'verbose'):
             ui.log('fsmonitor', 'Watchman unavailable: %s\n', ex.msg)
     else:
         ui.log('fsmonitor', 'Watchman exception: %s\n', ex)
 
 def _hashignore(ignore):
     """Calculate hash for ignore patterns and filenames
 
     If this information changes between Mercurial invocations, we can't
     rely on Watchman information anymore and have to re-scan the working
     copy.
 
     """
     sha1 = hashlib.sha1()
     sha1.update(repr(ignore))
     return sha1.hexdigest()
 
 _watchmanencoding = pywatchman.encoding.get_local_encoding()
 _fsencoding = sys.getfilesystemencoding() or sys.getdefaultencoding()
 _fixencoding = codecs.lookup(_watchmanencoding) != codecs.lookup(_fsencoding)
 
 def _watchmantofsencoding(path):
     """Fix path to match watchman and local filesystem encoding
 
     watchman's paths encoding can differ from filesystem encoding. For example,
     on Windows, it's always utf-8.
     """
     try:
         decoded = path.decode(_watchmanencoding)
     except UnicodeDecodeError as e:
         raise error.Abort(str(e), hint='watchman encoding error')
 
     try:
         encoded = decoded.encode(_fsencoding, 'strict')
     except UnicodeEncodeError as e:
         raise error.Abort(str(e))
 
     return encoded
 
 def overridewalk(orig, self, match, subrepos, unknown, ignored, full=True):
     '''Replacement for dirstate.walk, hooking into Watchman.
 
     Whenever full is False, ignored is False, and the Watchman client is
     available, use Watchman combined with saved state to possibly return only a
     subset of files.'''
     def bail(reason):
         self._ui.debug('fsmonitor: fallback to core status, %s\n' % reason)
         return orig(match, subrepos, unknown, ignored, full=True)
 
     if full:
         return bail('full rewalk requested')
     if ignored:
         return bail('listing ignored files')
     if not self._watchmanclient.available():
         return bail('client unavailable')
     state = self._fsmonitorstate
     clock, ignorehash, notefiles = state.get()
     if not clock:
         if state.walk_on_invalidate:
             return bail('no clock')
         # Initial NULL clock value, see
         # https://facebook.github.io/watchman/docs/clockspec.html
         clock = 'c:0:0'
         notefiles = []
 
     ignore = self._ignore
     dirignore = self._dirignore
     if unknown:
         if _hashignore(ignore) != ignorehash and clock != 'c:0:0':
             # ignore list changed -- can't rely on Watchman state any more
             if state.walk_on_invalidate:
                 return bail('ignore rules changed')
             notefiles = []
             clock = 'c:0:0'
     else:
         # always ignore
         ignore = util.always
         dirignore = util.always
 
     matchfn = match.matchfn
     matchalways = match.always()
     dmap = self._map
     if util.safehasattr(dmap, '_map'):
         # for better performance, directly access the inner dirstate map if the
         # standard dirstate implementation is in use.
         dmap = dmap._map
     nonnormalset = self._map.nonnormalset
 
     copymap = self._map.copymap
     getkind = stat.S_IFMT
     dirkind = stat.S_IFDIR
     regkind = stat.S_IFREG
     lnkkind = stat.S_IFLNK
     join = self._join
     normcase = util.normcase
     fresh_instance = False
 
     exact = skipstep3 = False
     if match.isexact():  # match.exact
         exact = True
         dirignore = util.always  # skip step 2
     elif match.prefix():  # match.match, no patterns
         skipstep3 = True
 
     if not exact and self._checkcase:
         # note that even though we could receive directory entries, we're only
         # interested in checking if a file with the same name exists. So only
         # normalize files if possible.
         normalize = self._normalizefile
         skipstep3 = False
     else:
         normalize = None
 
     # step 1: find all explicit files
     results, work, dirsnotfound = self._walkexplicit(match, subrepos)
 
     skipstep3 = skipstep3 and not (work or dirsnotfound)
     work = [d for d in work if not dirignore(d[0])]
 
     if not work and (exact or skipstep3):
         for s in subrepos:
             del results[s]
         del results['.hg']
         return results
 
     # step 2: query Watchman
     try:
         # Use the user-configured timeout for the query.
         # Add a little slack over the top of the user query to allow for
         # overheads while transferring the data
         self._watchmanclient.settimeout(state.timeout + 0.1)
         result = self._watchmanclient.command('query', {
             'fields': ['mode', 'mtime', 'size', 'exists', 'name'],
             'since': clock,
             'expression': [
                 'not', [
                     'anyof', ['dirname', '.hg'],
                     ['name', '.hg', 'wholename']
                 ]
             ],
             'sync_timeout': int(state.timeout * 1000),
             'empty_on_fresh_instance': state.walk_on_invalidate,
         })
     except Exception as ex:
         _handleunavailable(self._ui, state, ex)
         self._watchmanclient.clearconnection()
         return bail('exception during run')
     else:
         # We need to propagate the last observed clock up so that we
         # can use it for our next query
         state.setlastclock(result['clock'])
         if result['is_fresh_instance']:
             if state.walk_on_invalidate:
                 state.invalidate()
                 return bail('fresh instance')
             fresh_instance = True
             # Ignore any prior noteable files from the state info
             notefiles = []
 
     # for file paths which require normalization and we encounter a case
     # collision, we store our own foldmap
     if normalize:
         foldmap = dict((normcase(k), k) for k in results)
 
     switch_slashes = pycompat.ossep == '\\'
     # The order of the results is, strictly speaking, undefined.
     # For case changes on a case insensitive filesystem we may receive
     # two entries, one with exists=True and another with exists=False.
     # The exists=True entries in the same response should be interpreted
     # as being happens-after the exists=False entries due to the way that
     # Watchman tracks files.  We use this property to reconcile deletes
     # for name case changes.
     for entry in result['files']:
         fname = entry['name']
         if _fixencoding:
             fname = _watchmantofsencoding(fname)
         if switch_slashes:
             fname = fname.replace('\\', '/')
         if normalize:
             normed = normcase(fname)
             fname = normalize(fname, True, True)
             foldmap[normed] = fname
         fmode = entry['mode']
         fexists = entry['exists']
         kind = getkind(fmode)
 
         if '/.hg/' in fname or fname.endswith('/.hg'):
             return bail('nested-repo-detected')
 
         if not fexists:
             # if marked as deleted and we don't already have a change
             # record, mark it as deleted.  If we already have an entry
             # for fname then it was either part of walkexplicit or was
             # an earlier result that was a case change
             if fname not in results and fname in dmap and (
                     matchalways or matchfn(fname)):
                 results[fname] = None
         elif kind == dirkind:
             if fname in dmap and (matchalways or matchfn(fname)):
                 results[fname] = None
         elif kind == regkind or kind == lnkkind:
             if fname in dmap:
                 if matchalways or matchfn(fname):
                     results[fname] = entry
             elif (matchalways or matchfn(fname)) and not ignore(fname):
                 results[fname] = entry
         elif fname in dmap and (matchalways or matchfn(fname)):
             results[fname] = None
 
     # step 3: query notable files we don't already know about
     # XXX try not to iterate over the entire dmap
     if normalize:
         # any notable files that have changed case will already be handled
         # above, so just check membership in the foldmap
         notefiles = set((normalize(f, True, True) for f in notefiles
                          if normcase(f) not in foldmap))
     visit = set((f for f in notefiles if (f not in results and matchfn(f)
                                           and (f in dmap or not ignore(f)))))
 
     if not fresh_instance:
         if matchalways:
             visit.update(f for f in nonnormalset if f not in results)
             visit.update(f for f in copymap if f not in results)
         else:
             visit.update(f for f in nonnormalset
                          if f not in results and matchfn(f))
             visit.update(f for f in copymap
                          if f not in results and matchfn(f))
     else:
         if matchalways:
             visit.update(f for f, st in dmap.iteritems() if f not in results)
             visit.update(f for f in copymap if f not in results)
         else:
             visit.update(f for f, st in dmap.iteritems()
                          if f not in results and matchfn(f))
             visit.update(f for f in copymap
                          if f not in results and matchfn(f))
 
     audit = pathutil.pathauditor(self._root, cached=True).check
     auditpass = [f for f in visit if audit(f)]
     auditpass.sort()
     auditfail = visit.difference(auditpass)
     for f in auditfail:
         results[f] = None
 
     nf = iter(auditpass).next
     for st in util.statfiles([join(f) for f in auditpass]):
         f = nf()
         if st or f in dmap:
             results[f] = st
 
     for s in subrepos:
         del results[s]
     del results['.hg']
     return results
 
 def overridestatus(
         orig, self, node1='.', node2=None, match=None, ignored=False,
         clean=False, unknown=False, listsubrepos=False):
     listignored = ignored
     listclean = clean
     listunknown = unknown
 
     def _cmpsets(l1, l2):
         try:
             if 'FSMONITOR_LOG_FILE' in encoding.environ:
                 fn = encoding.environ['FSMONITOR_LOG_FILE']
                 f = open(fn, 'wb')
             else:
                 fn = 'fsmonitorfail.log'
                 f = self.vfs.open(fn, 'wb')
         except (IOError, OSError):
             self.ui.warn(_('warning: unable to write to %s\n') % fn)
             return
 
         try:
             for i, (s1, s2) in enumerate(zip(l1, l2)):
                 if set(s1) != set(s2):
                     f.write('sets at position %d are unequal\n' % i)
                     f.write('watchman returned: %s\n' % s1)
                     f.write('stat returned: %s\n' % s2)
         finally:
             f.close()
 
     if isinstance(node1, context.changectx):
         ctx1 = node1
     else:
         ctx1 = self[node1]
     if isinstance(node2, context.changectx):
         ctx2 = node2
     else:
         ctx2 = self[node2]
 
     working = ctx2.rev() is None
     parentworking = working and ctx1 == self['.']
     match = match or matchmod.always()
 
     # Maybe we can use this opportunity to update Watchman's state.
     # Mercurial uses workingcommitctx and/or memctx to represent the part of
     # the workingctx that is to be committed. So don't update the state in
     # that case.
     # HG_PENDING is set in the environment when the dirstate is being updated
     # in the middle of a transaction; we must not update our state in that
     # case, or we risk forgetting about changes in the working copy.
     updatestate = (parentworking and match.always() and
                    not isinstance(ctx2, (context.workingcommitctx,
                                          context.memctx)) and
                    'HG_PENDING' not in encoding.environ)
 
     try:
         if self._fsmonitorstate.walk_on_invalidate:
             # Use a short timeout to query the current clock.  If that
             # takes too long then we assume that the service will be slow
             # to answer our query.
             # walk_on_invalidate indicates that we prefer to walk the
             # tree ourselves because we can ignore portions that Watchman
             # cannot and we tend to be faster in the warmer buffer cache
             # cases.
             self._watchmanclient.settimeout(0.1)
         else:
             # Give Watchman more time to potentially complete its walk
             # and return the initial clock.  In this mode we assume that
             # the filesystem will be slower than parsing a potentially
             # very large Watchman result set.
             self._watchmanclient.settimeout(
                 self._fsmonitorstate.timeout + 0.1)
         startclock = self._watchmanclient.getcurrentclock()
     except Exception as ex:
         self._watchmanclient.clearconnection()
         _handleunavailable(self.ui, self._fsmonitorstate, ex)
         # boo, Watchman failed. bail
         return orig(node1, node2, match, listignored, listclean,
                     listunknown, listsubrepos)
 
     if updatestate:
         # We need info about unknown files. This may make things slower the
         # first time, but whatever.
         stateunknown = True
     else:
         stateunknown = listunknown
 
     if updatestate:
         ps = poststatus(startclock)
         self.addpostdsstatus(ps)
 
     r = orig(node1, node2, match, listignored, listclean, stateunknown,
              listsubrepos)
     modified, added, removed, deleted, unknown, ignored, clean = r
 
     if not listunknown:
         unknown = []
 
     # don't do paranoid checks if we're not going to query Watchman anyway
     full = listclean or match.traversedir is not None
     if self._fsmonitorstate.mode == 'paranoid' and not full:
         # run status again and fall back to the old walk this time
         self.dirstate._fsmonitordisable = True
 
         # shut the UI up
         quiet = self.ui.quiet
         self.ui.quiet = True
         fout, ferr = self.ui.fout, self.ui.ferr
         self.ui.fout = self.ui.ferr = open(os.devnull, 'wb')
 
         try:
             rv2 = orig(
                 node1, node2, match, listignored, listclean, listunknown,
                 listsubrepos)
         finally:
             self.dirstate._fsmonitordisable = False
             self.ui.quiet = quiet
             self.ui.fout, self.ui.ferr = fout, ferr
 
         # clean isn't tested since it's set to True above
         with self.wlock():
             _cmpsets(
                 [modified, added, removed, deleted, unknown, ignored, clean],
                 rv2)
         modified, added, removed, deleted, unknown, ignored, clean = rv2
 
     return scmutil.status(
         modified, added, removed, deleted, unknown, ignored, clean)
 
 class poststatus(object):
     def __init__(self, startclock):
         self._startclock = startclock
 
     def __call__(self, wctx, status):
         clock = wctx.repo()._fsmonitorstate.getlastclock() or self._startclock
         hashignore = _hashignore(wctx.repo().dirstate._ignore)
         notefiles = (status.modified + status.added + status.removed +
                      status.deleted + status.unknown)
         wctx.repo()._fsmonitorstate.set(clock, hashignore, notefiles)
 
 def makedirstate(repo, dirstate):
     class fsmonitordirstate(dirstate.__class__):
         def _fsmonitorinit(self, repo):
             # _fsmonitordisable is used in paranoid mode
             self._fsmonitordisable = False
             self._fsmonitorstate = repo._fsmonitorstate
             self._watchmanclient = repo._watchmanclient
             self._repo = weakref.proxy(repo)
 
         def walk(self, *args, **kwargs):
             orig = super(fsmonitordirstate, self).walk
             if self._fsmonitordisable:
                 return orig(*args, **kwargs)
             return overridewalk(orig, self, *args, **kwargs)
 
         def rebuild(self, *args, **kwargs):
             self._fsmonitorstate.invalidate()
             return super(fsmonitordirstate, self).rebuild(*args, **kwargs)
 
         def invalidate(self, *args, **kwargs):
             self._fsmonitorstate.invalidate()
             return super(fsmonitordirstate, self).invalidate(*args, **kwargs)
 
     dirstate.__class__ = fsmonitordirstate
     dirstate._fsmonitorinit(repo)
 
 def wrapdirstate(orig, self):
     ds = orig(self)
     # only override the dirstate when Watchman is available for the repo
     if util.safehasattr(self, '_fsmonitorstate'):
         makedirstate(self, ds)
     return ds
 
 def extsetup(ui):
     extensions.wrapfilecache(
         localrepo.localrepository, 'dirstate', wrapdirstate)
     if pycompat.isdarwin:
         # An assist for avoiding the dangling-symlink fsevents bug
         extensions.wrapfunction(os, 'symlink', wrapsymlink)
 
     extensions.wrapfunction(merge, 'update', wrapupdate)
 
 def wrapsymlink(orig, source, link_name):
     ''' if we create a dangling symlink, also touch the parent dir
     to encourage fsevents notifications to work more correctly '''
     try:
         return orig(source, link_name)
     finally:
         try:
             os.utime(os.path.dirname(link_name), None)
         except OSError:
             pass
 
 class state_update(object):
     ''' This context manager is responsible for dispatching the state-enter
         and state-leave signals to the watchman service. The enter and leave
         methods can be invoked manually (for scenarios where context manager
         semantics are not possible). If parameters oldnode and newnode are None,
         they will be populated based on current working copy in enter and
         leave, respectively. Similarly, if the distance is none, it will be
         calculated based on the oldnode and newnode in the leave method.'''
 
     def __init__(self, repo, name, oldnode=None, newnode=None, distance=None,
                  partial=False):
         self.repo = repo.unfiltered()
         self.name = name
         self.oldnode = oldnode
         self.newnode = newnode
         self.distance = distance
         self.partial = partial
         self._lock = None
         self.need_leave = False
 
     def __enter__(self):
         self.enter()
 
     def enter(self):
         # Make sure we have a wlock prior to sending notifications to watchman.
         # We don't want to race with other actors. In the update case,
         # merge.update is going to take the wlock almost immediately. We are
         # effectively extending the lock around several short sanity checks.
         if self.oldnode is None:
             self.oldnode = self.repo['.'].node()
 
         if self.repo.currentwlock() is None:
             if util.safehasattr(self.repo, 'wlocknostateupdate'):
                 self._lock = self.repo.wlocknostateupdate()
             else:
                 self._lock = self.repo.wlock()
         self.need_leave = self._state(
             'state-enter',
             hex(self.oldnode))
         return self
 
     def __exit__(self, type_, value, tb):
         abort = True if type_ else False
         self.exit(abort=abort)
 
     def exit(self, abort=False):
         try:
             if self.need_leave:
                 status = 'failed' if abort else 'ok'
                 if self.newnode is None:
                     self.newnode = self.repo['.'].node()
                 if self.distance is None:
                     self.distance = calcdistance(
                         self.repo, self.oldnode, self.newnode)
                 self._state(
                     'state-leave',
                     hex(self.newnode),
                     status=status)
         finally:
             self.need_leave = False
             if self._lock:
                 self._lock.release()
 
     def _state(self, cmd, commithash, status='ok'):
         if not util.safehasattr(self.repo, '_watchmanclient'):
             return False
         try:
             self.repo._watchmanclient.command(cmd, {
                 'name': self.name,
                 'metadata': {
                     # the target revision
                     'rev': commithash,
                     # approximate number of commits between current and target
                     'distance': self.distance if self.distance else 0,
                     # success/failure (only really meaningful for state-leave)
                     'status': status,
                     # whether the working copy parent is changing
                     'partial': self.partial,
             }})
             return True
         except Exception as e:
             # Swallow any errors; fire and forget
             self.repo.ui.log(
                 'watchman', 'Exception %s while running %s\n', e, cmd)
             return False
 
 # Estimate the distance between two nodes
 def calcdistance(repo, oldnode, newnode):
     anc = repo.changelog.ancestor(oldnode, newnode)
     ancrev = repo[anc].rev()
     distance = (abs(repo[oldnode].rev() - ancrev)
         + abs(repo[newnode].rev() - ancrev))
     return distance
 
 # Bracket working copy updates with calls to the watchman state-enter
 # and state-leave commands.  This allows clients to perform more intelligent
 # settling during bulk file change scenarios
 # https://facebook.github.io/watchman/docs/cmd/subscribe.html#advanced-settling
 def wrapupdate(orig, repo, node, branchmerge, force, ancestor=None,
                mergeancestor=False, labels=None, matcher=None, **kwargs):
 
     distance = 0
     partial = True
     oldnode = repo['.'].node()
     newnode = repo[node].node()
     if matcher is None or matcher.always():
         partial = False
         distance = calcdistance(repo.unfiltered(), oldnode, newnode)
 
     with state_update(repo, name="hg.update", oldnode=oldnode, newnode=newnode,
                       distance=distance, partial=partial):
         return orig(
             repo, node, branchmerge, force, ancestor, mergeancestor,
             labels, matcher, **kwargs)
 
 def repo_has_depth_one_nested_repo(repo):
     for f in repo.wvfs.listdir():
         if os.path.isdir(os.path.join(repo.root, f, '.hg')):
             msg = 'fsmonitor: sub-repository %r detected, fsmonitor disabled\n'
             repo.ui.debug(msg % f)
             return True
     return False
 
 def reposetup(ui, repo):
     # We don't work with largefiles or inotify
     exts = extensions.enabled()
     for ext in _blacklist:
         if ext in exts:
             ui.warn(_('The fsmonitor extension is incompatible with the %s '
                       'extension and has been disabled.\n') % ext)
             return
 
     if repo.local():
         # We don't work with subrepos either.
         #
         # if repo[None].substate can cause a dirstate parse, which is too
         # slow. Instead, look for a file called hgsubstate,
         if repo.wvfs.exists('.hgsubstate') or repo.wvfs.exists('.hgsub'):
             return
 
         if repo_has_depth_one_nested_repo(repo):
             return
 
         fsmonitorstate = state.state(repo)
         if fsmonitorstate.mode == 'off':
             return
 
         try:
             client = watchmanclient.client(repo)
         except Exception as ex:
             _handleunavailable(ui, fsmonitorstate, ex)
             return
 
         repo._fsmonitorstate = fsmonitorstate
         repo._watchmanclient = client
 
         dirstate, cached = localrepo.isfilecached(repo, 'dirstate')
         if cached:
             # at this point since fsmonitorstate wasn't present,
             # repo.dirstate is not a fsmonitordirstate
             makedirstate(repo, dirstate)
 
         class fsmonitorrepo(repo.__class__):
             def status(self, *args, **kwargs):
                 orig = super(fsmonitorrepo, self).status
                 return overridestatus(orig, self, *args, **kwargs)
 
             def wlocknostateupdate(self, *args, **kwargs):
                 return super(fsmonitorrepo, self).wlock(*args, **kwargs)
 
             def wlock(self, *args, **kwargs):
                 l = super(fsmonitorrepo, self).wlock(*args, **kwargs)
                 if not ui.configbool(
                     "experimental", "fsmonitor.transaction_notify"):
                     return l
                 if l.held != 1:
                     return l
                 origrelease = l.releasefn
 
                 def staterelease():
                     if origrelease:
                         origrelease()
                     if l.stateupdate:
                         l.stateupdate.exit()
                         l.stateupdate = None
 
                 try:
                     l.stateupdate = None
                     l.stateupdate = state_update(self, name="hg.transaction")
                     l.stateupdate.enter()
                     l.releasefn = staterelease
                 except Exception as e:
                     # Swallow any errors; fire and forget
                     self.ui.log(
                         'watchman', 'Exception in state update %s\n', e)
                 return l
 
         repo.__class__ = fsmonitorrepo

hgext/fsmonitor/pywatchman/__init__.py

@@ -1,1031 +1,1035 b''
 # Copyright 2014-present Facebook, Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are met:
 #
 #  * Redistributions of source code must retain the above copyright notice,
 #    this list of conditions and the following disclaimer.
 #
 #  * Redistributions in binary form must reproduce the above copyright notice,
 #    this list of conditions and the following disclaimer in the documentation
 #    and/or other materials provided with the distribution.
 #
 #  * Neither the name Facebook nor the names of its contributors may be used to
 #    endorse or promote products derived from this software without specific
 #    prior written permission.
 #
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 # DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 # SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 # no unicode literals
 
 import inspect
 import math
 import os
 import socket
 import subprocess
 import time
 
 # Sometimes it's really hard to get Python extensions to compile,
 # so fall back to a pure Python implementation.
 try:
     from . import bser
     # Demandimport causes modules to be loaded lazily. Force the load now
     # so that we can fall back on pybser if bser doesn't exist
     bser.pdu_info
 except ImportError:
     from . import pybser as bser
 
 from mercurial.utils import (
     procutil,
 )
 
 from mercurial import (
     pycompat,
 )
 
 from . import (
     capabilities,
     compat,
     encoding,
     load,
 )
 
 
 if os.name == 'nt':
     import ctypes
     import ctypes.wintypes
 
     wintypes = ctypes.wintypes
     GENERIC_READ = 0x80000000
     GENERIC_WRITE = 0x40000000
     FILE_FLAG_OVERLAPPED = 0x40000000
     OPEN_EXISTING = 3
     INVALID_HANDLE_VALUE = -1
     FORMAT_MESSAGE_FROM_SYSTEM = 0x00001000
     FORMAT_MESSAGE_ALLOCATE_BUFFER = 0x00000100
     FORMAT_MESSAGE_IGNORE_INSERTS = 0x00000200
     WAIT_FAILED = 0xFFFFFFFF
     WAIT_TIMEOUT = 0x00000102
     WAIT_OBJECT_0 = 0x00000000
     WAIT_IO_COMPLETION = 0x000000C0
     INFINITE = 0xFFFFFFFF
 
     # Overlapped I/O operation is in progress. (997)
     ERROR_IO_PENDING = 0x000003E5
 
     # The pointer size follows the architecture
     # We use WPARAM since this type is already conditionally defined
     ULONG_PTR = ctypes.wintypes.WPARAM
 
     class OVERLAPPED(ctypes.Structure):
         _fields_ = [
             ("Internal", ULONG_PTR), ("InternalHigh", ULONG_PTR),
             ("Offset", wintypes.DWORD), ("OffsetHigh", wintypes.DWORD),
             ("hEvent", wintypes.HANDLE)
         ]
 
         def __init__(self):
             self.Internal = 0
             self.InternalHigh = 0
             self.Offset = 0
             self.OffsetHigh = 0
             self.hEvent = 0
 
     LPDWORD = ctypes.POINTER(wintypes.DWORD)
 
     CreateFile = ctypes.windll.kernel32.CreateFileA
     CreateFile.argtypes = [wintypes.LPSTR, wintypes.DWORD, wintypes.DWORD,
                            wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD,
                            wintypes.HANDLE]
     CreateFile.restype = wintypes.HANDLE
 
     CloseHandle = ctypes.windll.kernel32.CloseHandle
     CloseHandle.argtypes = [wintypes.HANDLE]
     CloseHandle.restype = wintypes.BOOL
 
     ReadFile = ctypes.windll.kernel32.ReadFile
     ReadFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD,
                          LPDWORD, ctypes.POINTER(OVERLAPPED)]
     ReadFile.restype = wintypes.BOOL
 
     WriteFile = ctypes.windll.kernel32.WriteFile
     WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD,
                           LPDWORD, ctypes.POINTER(OVERLAPPED)]
     WriteFile.restype = wintypes.BOOL
 
     GetLastError = ctypes.windll.kernel32.GetLastError
     GetLastError.argtypes = []
     GetLastError.restype = wintypes.DWORD
 
     SetLastError = ctypes.windll.kernel32.SetLastError
     SetLastError.argtypes = [wintypes.DWORD]
     SetLastError.restype = None
 
     FormatMessage = ctypes.windll.kernel32.FormatMessageA
     FormatMessage.argtypes = [wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD,
                               wintypes.DWORD, ctypes.POINTER(wintypes.LPSTR),
                               wintypes.DWORD, wintypes.LPVOID]
     FormatMessage.restype = wintypes.DWORD
 
     LocalFree = ctypes.windll.kernel32.LocalFree
 
     GetOverlappedResult = ctypes.windll.kernel32.GetOverlappedResult
     GetOverlappedResult.argtypes = [wintypes.HANDLE,
                                     ctypes.POINTER(OVERLAPPED), LPDWORD,
                                     wintypes.BOOL]
     GetOverlappedResult.restype = wintypes.BOOL
 
     GetOverlappedResultEx = getattr(ctypes.windll.kernel32,
                                     'GetOverlappedResultEx', None)
     if GetOverlappedResultEx is not None:
         GetOverlappedResultEx.argtypes = [wintypes.HANDLE,
                                           ctypes.POINTER(OVERLAPPED), LPDWORD,
                                           wintypes.DWORD, wintypes.BOOL]
         GetOverlappedResultEx.restype = wintypes.BOOL
 
     WaitForSingleObjectEx = ctypes.windll.kernel32.WaitForSingleObjectEx
     WaitForSingleObjectEx.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.BOOL]
     WaitForSingleObjectEx.restype = wintypes.DWORD
 
     CreateEvent = ctypes.windll.kernel32.CreateEventA
     CreateEvent.argtypes = [LPDWORD, wintypes.BOOL, wintypes.BOOL,
                             wintypes.LPSTR]
     CreateEvent.restype = wintypes.HANDLE
 
     # Windows Vista is the minimum supported client for CancelIoEx.
     CancelIoEx = ctypes.windll.kernel32.CancelIoEx
     CancelIoEx.argtypes = [wintypes.HANDLE, ctypes.POINTER(OVERLAPPED)]
     CancelIoEx.restype = wintypes.BOOL
 
 # 2 bytes marker, 1 byte int size, 8 bytes int64 value
 sniff_len = 13
 
 # This is a helper for debugging the client.
 _debugging = False
 if _debugging:
 
     def log(fmt, *args):
         print('[%s] %s' %
               (time.strftime("%a, %d %b %Y %H:%M:%S", time.gmtime()),
                fmt % args[:]))
 else:
 
     def log(fmt, *args):
         pass
 
 
 def _win32_strerror(err):
     """ expand a win32 error code into a human readable message """
 
     # FormatMessage will allocate memory and assign it here
     buf = ctypes.c_char_p()
     FormatMessage(
         FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_ALLOCATE_BUFFER
         | FORMAT_MESSAGE_IGNORE_INSERTS, None, err, 0, buf, 0, None)
     try:
         return buf.value
     finally:
         LocalFree(buf)
 
 
 class WatchmanError(Exception):
     def __init__(self, msg=None, cmd=None):
         self.msg = msg
         self.cmd = cmd
 
     def setCommand(self, cmd):
         self.cmd = cmd
 
     def __str__(self):
         if self.cmd:
             return '%s, while executing %s' % (self.msg, self.cmd)
         return self.msg
 
 
 class WatchmanEnvironmentError(WatchmanError):
     def __init__(self, msg, errno, errmsg, cmd=None):
         super(WatchmanEnvironmentError, self).__init__(
             '{0}: errno={1} errmsg={2}'.format(msg, errno, errmsg),
             cmd)
 
 
 class SocketConnectError(WatchmanError):
     def __init__(self, sockpath, exc):
         super(SocketConnectError, self).__init__(
             'unable to connect to %s: %s' % (sockpath, exc))
         self.sockpath = sockpath
         self.exc = exc
 
 
 class SocketTimeout(WatchmanError):
     """A specialized exception raised for socket timeouts during communication to/from watchman.
        This makes it easier to implement non-blocking loops as callers can easily distinguish
        between a routine timeout and an actual error condition.
 
        Note that catching WatchmanError will also catch this as it is a super-class, so backwards
        compatibility in exception handling is preserved.
     """
 
 
 class CommandError(WatchmanError):
     """error returned by watchman
 
     self.msg is the message returned by watchman.
     """
     def __init__(self, msg, cmd=None):
         super(CommandError, self).__init__(
             'watchman command error: %s' % (msg, ),
             cmd,
         )
 
 
 class Transport(object):
     """ communication transport to the watchman server """
     buf = None
 
     def close(self):
         """ tear it down """
         raise NotImplementedError()
 
     def readBytes(self, size):
         """ read size bytes """
         raise NotImplementedError()
 
     def write(self, buf):
         """ write some data """
         raise NotImplementedError()
 
     def setTimeout(self, value):
         pass
 
     def readLine(self):
         """ read a line
         Maintains its own buffer, callers of the transport should not mix
         calls to readBytes and readLine.
         """
         if self.buf is None:
             self.buf = []
 
         # Buffer may already have a line if we've received unilateral
         # response(s) from the server
         if len(self.buf) == 1 and b"\n" in self.buf[0]:
             (line, b) = self.buf[0].split(b"\n", 1)
             self.buf = [b]
             return line
 
         while True:
             b = self.readBytes(4096)
             if b"\n" in b:
                 result = b''.join(self.buf)
                 (line, b) = b.split(b"\n", 1)
                 self.buf = [b]
                 return result + line
             self.buf.append(b)
 
 
 class Codec(object):
     """ communication encoding for the watchman server """
     transport = None
 
     def __init__(self, transport):
         self.transport = transport
 
     def receive(self):
         raise NotImplementedError()
 
     def send(self, *args):
         raise NotImplementedError()
 
     def setTimeout(self, value):
         self.transport.setTimeout(value)
 
 
 class UnixSocketTransport(Transport):
     """ local unix domain socket transport """
     sock = None
 
-    def __init__(self, sockpath, timeout):
+    def __init__(self, sockpath, timeout, watchman_exe):
         self.sockpath = sockpath
         self.timeout = timeout
 
         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
         try:
             sock.settimeout(self.timeout)
             sock.connect(self.sockpath)
             self.sock = sock
         except socket.error as e:
             sock.close()
             raise SocketConnectError(self.sockpath, e)
 
     def close(self):
         self.sock.close()
         self.sock = None
 
     def setTimeout(self, value):
         self.timeout = value
         self.sock.settimeout(self.timeout)
 
     def readBytes(self, size):
         try:
             buf = [self.sock.recv(size)]
             if not buf[0]:
                 raise WatchmanError('empty watchman response')
             return buf[0]
         except socket.timeout:
             raise SocketTimeout('timed out waiting for response')
 
     def write(self, data):
         try:
             self.sock.sendall(data)
         except socket.timeout:
             raise SocketTimeout('timed out sending query command')
 
 
 def _get_overlapped_result_ex_impl(pipe, olap, nbytes, millis, alertable):
     """ Windows 7 and earlier does not support GetOverlappedResultEx. The
     alternative is to use GetOverlappedResult and wait for read or write
     operation to complete. This is done be using CreateEvent and
     WaitForSingleObjectEx. CreateEvent, WaitForSingleObjectEx
     and GetOverlappedResult are all part of Windows API since WindowsXP.
     This is the exact same implementation that can be found in the watchman
     source code (see get_overlapped_result_ex_impl in stream_win.c). This
     way, maintenance should be simplified.
     """
     log('Preparing to wait for maximum %dms', millis )
     if millis != 0:
         waitReturnCode = WaitForSingleObjectEx(olap.hEvent, millis, alertable)
         if waitReturnCode == WAIT_OBJECT_0:
             # Event is signaled, overlapped IO operation result should be available.
             pass
         elif waitReturnCode == WAIT_IO_COMPLETION:
             # WaitForSingleObjectEx returnes because the system added an I/O completion
             # routine or an asynchronous procedure call (APC) to the thread queue.
             SetLastError(WAIT_IO_COMPLETION)
             pass
         elif waitReturnCode == WAIT_TIMEOUT:
             # We reached the maximum allowed wait time, the IO operation failed
             # to complete in timely fashion.
             SetLastError(WAIT_TIMEOUT)
             return False
         elif waitReturnCode == WAIT_FAILED:
             # something went wrong calling WaitForSingleObjectEx
             err = GetLastError()
             log('WaitForSingleObjectEx failed: %s', _win32_strerror(err))
             return False
         else:
             # unexpected situation deserving investigation.
             err = GetLastError()
             log('Unexpected error: %s', _win32_strerror(err))
             return False
 
     return GetOverlappedResult(pipe, olap, nbytes, False)
 
 
 class WindowsNamedPipeTransport(Transport):
     """ connect to a named pipe """
 
-    def __init__(self, sockpath, timeout):
+    def __init__(self, sockpath, timeout, watchman_exe):
         self.sockpath = sockpath
         self.timeout = int(math.ceil(timeout * 1000))
         self._iobuf = None
 
         self.pipe = CreateFile(sockpath, GENERIC_READ | GENERIC_WRITE, 0, None,
                                OPEN_EXISTING, FILE_FLAG_OVERLAPPED, None)
 
         if self.pipe == INVALID_HANDLE_VALUE:
             self.pipe = None
             self._raise_win_err('failed to open pipe %s' % sockpath,
                                 GetLastError())
 
         # event for the overlapped I/O operations
         self._waitable = CreateEvent(None, True, False, None)
         if self._waitable is None:
             self._raise_win_err('CreateEvent failed', GetLastError())
 
         self._get_overlapped_result_ex = GetOverlappedResultEx
         if (os.getenv('WATCHMAN_WIN7_COMPAT') == '1' or
             self._get_overlapped_result_ex is None):
             self._get_overlapped_result_ex = _get_overlapped_result_ex_impl
 
     def _raise_win_err(self, msg, err):
         raise IOError('%s win32 error code: %d %s' %
                       (msg, err, _win32_strerror(err)))
 
     def close(self):
         if self.pipe:
             log('Closing pipe')
             CloseHandle(self.pipe)
         self.pipe = None
 
         if self._waitable is not None:
             # We release the handle for the event
             CloseHandle(self._waitable)
         self._waitable = None
 
     def setTimeout(self, value):
         # convert to milliseconds
         self.timeout = int(value * 1000)
 
     def readBytes(self, size):
         """ A read can block for an unbounded amount of time, even if the
             kernel reports that the pipe handle is signalled, so we need to
             always perform our reads asynchronously
         """
 
         # try to satisfy the read from any buffered data
         if self._iobuf:
             if size >= len(self._iobuf):
                 res = self._iobuf
                 self.buf = None
                 return res
             res = self._iobuf[:size]
             self._iobuf = self._iobuf[size:]
             return res
 
         # We need to initiate a read
         buf = ctypes.create_string_buffer(size)
         olap = OVERLAPPED()
         olap.hEvent = self._waitable
 
         log('made read buff of size %d', size)
 
         # ReadFile docs warn against sending in the nread parameter for async
         # operations, so we always collect it via GetOverlappedResultEx
         immediate = ReadFile(self.pipe, buf, size, None, olap)
 
         if not immediate:
             err = GetLastError()
             if err != ERROR_IO_PENDING:
                 self._raise_win_err('failed to read %d bytes' % size,
                                     GetLastError())
 
         nread = wintypes.DWORD()
         if not self._get_overlapped_result_ex(self.pipe, olap, nread,
                                               0 if immediate else self.timeout,
                                               True):
             err = GetLastError()
             CancelIoEx(self.pipe, olap)
 
             if err == WAIT_TIMEOUT:
                 log('GetOverlappedResultEx timedout')
                 raise SocketTimeout('timed out after waiting %dms for read' %
                                     self.timeout)
 
             log('GetOverlappedResultEx reports error %d', err)
             self._raise_win_err('error while waiting for read', err)
 
         nread = nread.value
         if nread == 0:
             # Docs say that named pipes return 0 byte when the other end did
             # a zero byte write.  Since we don't ever do that, the only
             # other way this shows up is if the client has gotten in a weird
             # state, so let's bail out
             CancelIoEx(self.pipe, olap)
             raise IOError('Async read yielded 0 bytes; unpossible!')
 
         # Holds precisely the bytes that we read from the prior request
         buf = buf[:nread]
 
         returned_size = min(nread, size)
         if returned_size == nread:
             return buf
 
         # keep any left-overs around for a later read to consume
         self._iobuf = buf[returned_size:]
         return buf[:returned_size]
 
     def write(self, data):
         olap = OVERLAPPED()
         olap.hEvent = self._waitable
 
         immediate = WriteFile(self.pipe, ctypes.c_char_p(data), len(data),
                               None, olap)
 
         if not immediate:
             err = GetLastError()
             if err != ERROR_IO_PENDING:
                 self._raise_win_err('failed to write %d bytes' % len(data),
                                     GetLastError())
 
         # Obtain results, waiting if needed
         nwrote = wintypes.DWORD()
         if self._get_overlapped_result_ex(self.pipe, olap, nwrote,
                                           0 if immediate else self.timeout,
                                           True):
             log('made write of %d bytes', nwrote.value)
             return nwrote.value
 
         err = GetLastError()
 
         # It's potentially unsafe to allow the write to continue after
         # we unwind, so let's make a best effort to avoid that happening
         CancelIoEx(self.pipe, olap)
 
         if err == WAIT_TIMEOUT:
             raise SocketTimeout('timed out after waiting %dms for write' %
                                 self.timeout)
         self._raise_win_err('error while waiting for write of %d bytes' %
                             len(data), err)
 
 
 class CLIProcessTransport(Transport):
     """ open a pipe to the cli to talk to the service
     This intended to be used only in the test harness!
 
     The CLI is an oddball because we only support JSON input
     and cannot send multiple commands through the same instance,
     so we spawn a new process for each command.
 
     We disable server spawning for this implementation, again, because
     it is intended to be used only in our test harness.  You really
     should not need to use the CLI transport for anything real.
 
     While the CLI can output in BSER, our Transport interface doesn't
     support telling this instance that it should do so.  That effectively
     limits this implementation to JSON input and output only at this time.
 
     It is the responsibility of the caller to set the send and
     receive codecs appropriately.
     """
     proc = None
     closed = True
 
-    def __init__(self, sockpath, timeout):
+    def __init__(self, sockpath, timeout, watchman_exe):
         self.sockpath = sockpath
         self.timeout = timeout
+        self.watchman_exe = watchman_exe
 
     def close(self):
         if self.proc:
             if self.proc.pid is not None:
                 self.proc.kill()
             self.proc.stdin.close()
             self.proc.stdout.close()
             self.proc = None
 
     def _connect(self):
         if self.proc:
             return self.proc
         args = [
-            'watchman',
+            self.watchman_exe,
             '--sockname={0}'.format(self.sockpath),
             '--logfile=/BOGUS',
             '--statefile=/BOGUS',
             '--no-spawn',
             '--no-local',
             '--no-pretty',
             '-j',
         ]
         self.proc = subprocess.Popen(pycompat.rapply(procutil.tonativestr,
                                                      args),
                                      stdin=subprocess.PIPE,
                                      stdout=subprocess.PIPE)
         return self.proc
 
     def readBytes(self, size):
         self._connect()
         res = self.proc.stdout.read(size)
         if res == '':
             raise WatchmanError('EOF on CLI process transport')
         return res
 
     def write(self, data):
         if self.closed:
             self.close()
             self.closed = False
         self._connect()
         res = self.proc.stdin.write(data)
         self.proc.stdin.close()
         self.closed = True
         return res
 
 
 class BserCodec(Codec):
     """ use the BSER encoding.  This is the default, preferred codec """
 
     def _loads(self, response):
         return bser.loads(response) # Defaults to BSER v1
 
     def receive(self):
         buf = [self.transport.readBytes(sniff_len)]
         if not buf[0]:
             raise WatchmanError('empty watchman response')
 
         _1, _2, elen = bser.pdu_info(buf[0])
 
         rlen = len(buf[0])
         while elen > rlen:
             buf.append(self.transport.readBytes(elen - rlen))
             rlen += len(buf[-1])
 
         response = b''.join(buf)
         try:
             res = self._loads(response)
             return res
         except ValueError as e:
             raise WatchmanError('watchman response decode error: %s' % e)
 
     def send(self, *args):
         cmd = bser.dumps(*args) # Defaults to BSER v1
         self.transport.write(cmd)
 
 
 class ImmutableBserCodec(BserCodec):
     """ use the BSER encoding, decoding values using the newer
         immutable object support """
 
     def _loads(self, response):
         return bser.loads(response, False) # Defaults to BSER v1
 
 
 class Bser2WithFallbackCodec(BserCodec):
     """ use BSER v2 encoding """
 
     def __init__(self, transport):
         super(Bser2WithFallbackCodec, self).__init__(transport)
         # Once the server advertises support for bser-v2 we should switch this
         # to 'required' on Python 3.
         self.send(["version", {"optional": ["bser-v2"]}])
 
         capabilities = self.receive()
 
         if 'error' in capabilities:
           raise Exception('Unsupported BSER version')
 
         if capabilities['capabilities']['bser-v2']:
             self.bser_version = 2
             self.bser_capabilities = 0
         else:
             self.bser_version = 1
             self.bser_capabilities = 0
 
     def _loads(self, response):
         return bser.loads(response)
 
     def receive(self):
         buf = [self.transport.readBytes(sniff_len)]
         if not buf[0]:
             raise WatchmanError('empty watchman response')
 
         recv_bser_version, recv_bser_capabilities, elen = bser.pdu_info(buf[0])
 
         if hasattr(self, 'bser_version'):
           # Readjust BSER version and capabilities if necessary
           self.bser_version = max(self.bser_version, recv_bser_version)
           self.capabilities = self.bser_capabilities & recv_bser_capabilities
 
         rlen = len(buf[0])
         while elen > rlen:
             buf.append(self.transport.readBytes(elen - rlen))
             rlen += len(buf[-1])
 
         response = b''.join(buf)
         try:
             res = self._loads(response)
             return res
         except ValueError as e:
             raise WatchmanError('watchman response decode error: %s' % e)
 
     def send(self, *args):
         if hasattr(self, 'bser_version'):
             cmd = bser.dumps(*args, version=self.bser_version,
                 capabilities=self.bser_capabilities)
         else:
             cmd = bser.dumps(*args)
         self.transport.write(cmd)
 
 
 class JsonCodec(Codec):
     """ Use json codec.  This is here primarily for testing purposes """
     json = None
 
     def __init__(self, transport):
         super(JsonCodec, self).__init__(transport)
         # optional dep on json, only if JsonCodec is used
         import json
         self.json = json
 
     def receive(self):
         line = self.transport.readLine()
         try:
             # In Python 3, json.loads is a transformation from Unicode string to
             # objects possibly containing Unicode strings. We typically expect
             # the JSON blob to be ASCII-only with non-ASCII characters escaped,
             # but it's possible we might get non-ASCII bytes that are valid
             # UTF-8.
             if compat.PYTHON3:
                 line = line.decode('utf-8')
             return self.json.loads(line)
         except Exception as e:
             print(e, line)
             raise
 
     def send(self, *args):
         cmd = self.json.dumps(*args)
         # In Python 3, json.dumps is a transformation from objects possibly
         # containing Unicode strings to Unicode string. Even with (the default)
         # ensure_ascii=True, dumps returns a Unicode string.
         if compat.PYTHON3:
             cmd = cmd.encode('ascii')
         self.transport.write(cmd + b"\n")
 
 
 class client(object):
     """ Handles the communication with the watchman service """
     sockpath = None
     transport = None
     sendCodec = None
     recvCodec = None
     sendConn = None
     recvConn = None
     subs = {}  # Keyed by subscription name
     sub_by_root = {}  # Keyed by root, then by subscription name
     logs = []  # When log level is raised
     unilateral = ['log', 'subscription']
     tport = None
     useImmutableBser = None
+    watchman_exe = None
 
     def __init__(self,
                  sockpath=None,
                  timeout=1.0,
                  transport=None,
                  sendEncoding=None,
                  recvEncoding=None,
-                 useImmutableBser=False):
+                 useImmutableBser=False,
+                 watchman_exe=None):
         self.sockpath = sockpath
         self.timeout = timeout
         self.useImmutableBser = useImmutableBser
+        self.watchman_exe = watchman_exe
 
         if inspect.isclass(transport) and issubclass(transport, Transport):
             self.transport = transport
         else:
             transport = transport or os.getenv('WATCHMAN_TRANSPORT') or 'local'
             if transport == 'local' and os.name == 'nt':
                 self.transport = WindowsNamedPipeTransport
             elif transport == 'local':
                 self.transport = UnixSocketTransport
             elif transport == 'cli':
                 self.transport = CLIProcessTransport
                 if sendEncoding is None:
                     sendEncoding = 'json'
                 if recvEncoding is None:
                     recvEncoding = sendEncoding
             else:
                 raise WatchmanError('invalid transport %s' % transport)
 
         sendEncoding = str(sendEncoding or os.getenv('WATCHMAN_ENCODING') or
                            'bser')
         recvEncoding = str(recvEncoding or os.getenv('WATCHMAN_ENCODING') or
                            'bser')
 
         self.recvCodec = self._parseEncoding(recvEncoding)
         self.sendCodec = self._parseEncoding(sendEncoding)
 
     def _parseEncoding(self, enc):
         if enc == 'bser':
             if self.useImmutableBser:
                 return ImmutableBserCodec
             return BserCodec
         elif enc == 'experimental-bser-v2':
           return Bser2WithFallbackCodec
         elif enc == 'json':
             return JsonCodec
         else:
             raise WatchmanError('invalid encoding %s' % enc)
 
     def _hasprop(self, result, name):
         if self.useImmutableBser:
             return hasattr(result, name)
         return name in result
 
     def _resolvesockname(self):
         # if invoked via a trigger, watchman will set this env var; we
         # should use it unless explicitly set otherwise
         path = os.getenv('WATCHMAN_SOCK')
         if path:
             return path
 
-        cmd = ['watchman', '--output-encoding=bser', 'get-sockname']
+        cmd = [self.watchman_exe, '--output-encoding=bser', 'get-sockname']
         try:
             args = dict(stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         close_fds=os.name != 'nt')
 
             if os.name == 'nt':
                 # if invoked via an application with graphical user interface,
                 # this call will cause a brief command window pop-up.
                 # Using the flag STARTF_USESHOWWINDOW to avoid this behavior.
                 startupinfo = subprocess.STARTUPINFO()
                 startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
                 args['startupinfo'] = startupinfo
 
             p = subprocess.Popen(pycompat.rapply(procutil.tonativestr, cmd),
                                  **args)
 
         except OSError as e:
             raise WatchmanError('"watchman" executable not in PATH (%s)' % e)
 
         stdout, stderr = p.communicate()
         exitcode = p.poll()
 
         if exitcode:
             raise WatchmanError("watchman exited with code %d" % exitcode)
 
         result = bser.loads(stdout)
         if b'error' in result:
             raise WatchmanError('get-sockname error: %s' % result['error'])
 
         return result[b'sockname']
 
     def _connect(self):
         """ establish transport connection """
 
         if self.recvConn:
             return
 
         if self.sockpath is None:
             self.sockpath = self._resolvesockname()
 
-        self.tport = self.transport(self.sockpath, self.timeout)
+        self.tport = self.transport(self.sockpath, self.timeout, self.watchman_exe)
         self.sendConn = self.sendCodec(self.tport)
         self.recvConn = self.recvCodec(self.tport)
 
     def __del__(self):
         self.close()
 
     def close(self):
         if self.tport:
             self.tport.close()
             self.tport = None
             self.recvConn = None
             self.sendConn = None
 
     def receive(self):
         """ receive the next PDU from the watchman service
 
         If the client has activated subscriptions or logs then
         this PDU may be a unilateral PDU sent by the service to
         inform the client of a log event or subscription change.
 
         It may also simply be the response portion of a request
         initiated by query.
 
         There are clients in production that subscribe and call
         this in a loop to retrieve all subscription responses,
         so care should be taken when making changes here.
         """
 
         self._connect()
         result = self.recvConn.receive()
         if self._hasprop(result, 'error'):
             error = result['error']
             if compat.PYTHON3 and isinstance(self.recvConn, BserCodec):
                 error = result['error'].decode('utf-8', 'surrogateescape')
             raise CommandError(error)
 
         if self._hasprop(result, 'log'):
             log = result['log']
             if compat.PYTHON3 and isinstance(self.recvConn, BserCodec):
                 log = log.decode('utf-8', 'surrogateescape')
             self.logs.append(log)
 
         if self._hasprop(result, 'subscription'):
             sub = result['subscription']
             if not (sub in self.subs):
                 self.subs[sub] = []
             self.subs[sub].append(result)
 
             # also accumulate in {root,sub} keyed store
             root = os.path.normcase(result['root'])
             if not root in self.sub_by_root:
                 self.sub_by_root[root] = {}
             if not sub in self.sub_by_root[root]:
                 self.sub_by_root[root][sub] = []
             self.sub_by_root[root][sub].append(result)
 
         return result
 
     def isUnilateralResponse(self, res):
         if 'unilateral' in res and res['unilateral']:
             return True
         # Fall back to checking for known unilateral responses
         for k in self.unilateral:
             if k in res:
                 return True
         return False
 
     def getLog(self, remove=True):
         """ Retrieve buffered log data
 
         If remove is true the data will be removed from the buffer.
         Otherwise it will be left in the buffer
         """
         res = self.logs
         if remove:
             self.logs = []
         return res
 
     def getSubscription(self, name, remove=True, root=None):
         """ Retrieve the data associated with a named subscription
 
         If remove is True (the default), the subscription data is removed
         from the buffer.  Otherwise the data is returned but left in
         the buffer.
 
         Returns None if there is no data associated with `name`
 
         If root is not None, then only return the subscription
         data that matches both root and name.  When used in this way,
         remove processing impacts both the unscoped and scoped stores
         for the subscription data.
         """
         if compat.PYTHON3 and issubclass(self.recvCodec, BserCodec):
             # People may pass in Unicode strings here -- but currently BSER only
             # returns bytestrings. Deal with that.
             if isinstance(root, str):
                 root = encoding.encode_local(root)
             if isinstance(name, str):
                 name = name.encode('utf-8')
 
         if root is not None:
             if not root in self.sub_by_root:
                 return None
             if not name in self.sub_by_root[root]:
                 return None
             sub = self.sub_by_root[root][name]
             if remove:
                 del self.sub_by_root[root][name]
                 # don't let this grow unbounded
                 if name in self.subs:
                     del self.subs[name]
             return sub
 
         if not (name in self.subs):
             return None
         sub = self.subs[name]
         if remove:
             del self.subs[name]
         return sub
 
     def query(self, *args):
         """ Send a query to the watchman service and return the response
 
         This call will block until the response is returned.
         If any unilateral responses are sent by the service in between
         the request-response they will be buffered up in the client object
         and NOT returned via this method.
         """
 
         log('calling client.query')
         self._connect()
         try:
             self.sendConn.send(args)
 
             res = self.receive()
             while self.isUnilateralResponse(res):
                 res = self.receive()
 
             return res
         except EnvironmentError as ee:
             # When we can depend on Python 3, we can use PEP 3134
             # exception chaining here.
             raise WatchmanEnvironmentError(
                 'I/O error communicating with watchman daemon',
                 ee.errno,
                 ee.strerror,
                 args)
         except WatchmanError as ex:
             ex.setCommand(args)
             raise
 
     def capabilityCheck(self, optional=None, required=None):
         """ Perform a server capability check """
         res = self.query('version', {
             'optional': optional or [],
             'required': required or []
         })
 
         if not self._hasprop(res, 'capabilities'):
             # Server doesn't support capabilities, so we need to
             # synthesize the results based on the version
             capabilities.synthesize(res, optional)
             if 'error' in res:
                 raise CommandError(res['error'])
 
         return res
 
     def setTimeout(self, value):
         self.recvConn.setTimeout(value)
         self.sendConn.setTimeout(value)

hgext/fsmonitor/watchmanclient.py

@@ -1,109 +1,111 b''
 # watchmanclient.py - Watchman client for the fsmonitor extension
 #
 # Copyright 2013-2016 Facebook, Inc.
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import getpass
 
 from mercurial import util
 
 from . import pywatchman
 
 class Unavailable(Exception):
     def __init__(self, msg, warn=True, invalidate=False):
         self.msg = msg
         self.warn = warn
         if self.msg == 'timed out waiting for response':
             self.warn = False
         self.invalidate = invalidate
 
     def __str__(self):
         if self.warn:
             return 'warning: Watchman unavailable: %s' % self.msg
         else:
             return 'Watchman unavailable: %s' % self.msg
 
 class WatchmanNoRoot(Unavailable):
     def __init__(self, root, msg):
         self.root = root
         super(WatchmanNoRoot, self).__init__(msg)
 
 class client(object):
     def __init__(self, repo, timeout=1.0):
         err = None
         if not self._user:
             err = "couldn't get user"
             warn = True
         if self._user in repo.ui.configlist('fsmonitor', 'blacklistusers'):
             err = 'user %s in blacklist' % self._user
             warn = False
 
         if err:
             raise Unavailable(err, warn)
 
         self._timeout = timeout
         self._watchmanclient = None
         self._root = repo.root
         self._ui = repo.ui
         self._firsttime = True
 
     def settimeout(self, timeout):
         self._timeout = timeout
         if self._watchmanclient is not None:
             self._watchmanclient.setTimeout(timeout)
 
     def getcurrentclock(self):
         result = self.command('clock')
         if not util.safehasattr(result, 'clock'):
             raise Unavailable('clock result is missing clock value',
                               invalidate=True)
         return result.clock
 
     def clearconnection(self):
         self._watchmanclient = None
 
     def available(self):
         return self._watchmanclient is not None or self._firsttime
 
     @util.propertycache
     def _user(self):
         try:
             return getpass.getuser()
         except KeyError:
             # couldn't figure out our user
             return None
 
     def _command(self, *args):
         watchmanargs = (args[0], self._root) + args[1:]
         try:
             if self._watchmanclient is None:
                 self._firsttime = False
+                watchman_exe = self._ui.configpath('fsmonitor', 'watchman_exe')
                 self._watchmanclient = pywatchman.client(
                     timeout=self._timeout,
-                    useImmutableBser=True)
+                    useImmutableBser=True,
+                    watchman_exe=watchman_exe)
             return self._watchmanclient.query(*watchmanargs)
         except pywatchman.CommandError as ex:
             if 'unable to resolve root' in ex.msg:
                 raise WatchmanNoRoot(self._root, ex.msg)
             raise Unavailable(ex.msg)
         except pywatchman.WatchmanError as ex:
             raise Unavailable(str(ex))
 
     def command(self, *args):
         try:
             try:
                 return self._command(*args)
             except WatchmanNoRoot:
                 # this 'watch' command can also raise a WatchmanNoRoot if
                 # watchman refuses to accept this root
                 self._command('watch')
                 return self._command(*args)
         except Unavailable:
             # this is in an outer scope to catch Unavailable form any of the
             # above _command calls
             self._watchmanclient = None
             raise