view mercurial/sshpeer.py @ 32697:19b9fc40cc51

revlog: skeleton support for version 2 revlogs

There are a number of improvements we want to make to revlogs
that will require a new version - version 2. It is unclear what the
full set of improvements will be or when we'll be done with them.
What I do know is that the process will likely take longer than a
single release, will require input from various stakeholders to
evaluate changes, and will have many contentious debates and
bikeshedding.

It is unrealistic to develop revlog version 2 up front: there
are just too many uncertainties that we won't know until things
are implemented and experiments are run. Some changes will also
be invasive and prone to bit rot, so sitting on dozens of patches
is not practical.

This commit introduces skeleton support for version 2 revlogs in
a way that is flexible and not bound by backwards compatibility
concerns.

An experimental repo requirement for denoting revlog v2 has been
added. The requirement string has a sub-version component to it.
This will allow us to declare multiple requirements in the course
of developing revlog v2. Whenever we change the in-development
revlog v2 format, we can tweak the string, creating a new
requirement and locking out old clients. This will allow us to
make as many backwards incompatible changes and experiments to
revlog v2 as we want. In other words, we can land code and make
meaningful progress towards revlog v2 while still maintaining
extreme format flexibility up until the point we freeze the
format and remove the experimental labels.

To enable the new repo requirement, you must supply an experimental
and undocumented config option. But not just any boolean flag
will do: you need to explicitly use a value that no sane person
should ever type. This is an additional guard against enabling
revlog v2 on an installation it shouldn't be enabled on. The
specific scenario I'm trying to prevent is say a user with a
4.4 client with a frozen format enabling the option but then
downgrading to 4.3 and accidentally creating repos with an
outdated and unsupported repo format. Requiring a "challenge"
string should prevent this.

Because the format is not yet finalized and I don't want to take
any chances, revlog v2's version is currently 0xDEAD. I figure
squatting on a value we're likely never to use as an actual revlog
version to mean "internal testing only" is acceptable. And
"dead" is easily recognized as something meaningful.

There is a bunch of cleanup that is needed before work on revlog
v2 begins in earnest. I plan on doing that work once this patch
is accepted and we're comfortable with the idea of starting down
this path.

# sshpeer.py - ssh repository proxy class for mercurial
#
# Copyright 2005, 2006 Matt Mackall <[e-mail]>
#
# 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 re

from .i18n import _
from . import (
    error,
    util,
    wireproto,
)

class remotelock(object):
    def __init__(self, repo):
        self.repo = repo
    def release(self):
        self.repo.unlock()
        self.repo = None
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.repo:
            self.release()
    def __del__(self):
        if self.repo:
            self.release()

def _serverquote(s):
    if not s:
        return s
    '''quote a string for the remote shell ... which we assume is sh'''
    if re.match('[a-zA-Z0-9@%_+=:,./-]*$', s):
        return s
    return "'%s'" % s.replace("'", "'\\''")

def _forwardoutput(ui, pipe):
    """display all data currently available on pipe as remote output.

    This is non blocking."""
    s = util.readpipe(pipe)
    if s:
        for l in s.splitlines():
            ui.status(_("remote: "), l, '\n')

class doublepipe(object):
    """Operate a side-channel pipe in addition of a main one

    The side-channel pipe contains server output to be forwarded to the user
    input. The double pipe will behave as the "main" pipe, but will ensure the
    content of the "side" pipe is properly processed while we wait for blocking
    call on the "main" pipe.

    If large amounts of data are read from "main", the forward will cease after
    the first bytes start to appear. This simplifies the implementation
    without affecting actual output of sshpeer too much as we rarely issue
    large read for data not yet emitted by the server.

    The main pipe is expected to be a 'bufferedinputpipe' from the util module
    that handle all the os specific bits. This class lives in this module
    because it focus on behavior specific to the ssh protocol."""

    def __init__(self, ui, main, side):
        self._ui = ui
        self._main = main
        self._side = side

    def _wait(self):
        """wait until some data are available on main or side

        return a pair of boolean (ismainready, issideready)

        (This will only wait for data if the setup is supported by `util.poll`)
        """
        if getattr(self._main, 'hasbuffer', False): # getattr for classic pipe
            return (True, True) # main has data, assume side is worth poking at.
        fds = [self._main.fileno(), self._side.fileno()]
        try:
            act = util.poll(fds)
        except NotImplementedError:
            # non supported yet case, assume all have data.
            act = fds
        return (self._main.fileno() in act, self._side.fileno() in act)

    def write(self, data):
        return self._call('write', data)

    def read(self, size):
        r = self._call('read', size)
        if size != 0 and not r:
            # We've observed a condition that indicates the
            # stdout closed unexpectedly. Check stderr one
            # more time and snag anything that's there before
            # letting anyone know the main part of the pipe
            # closed prematurely.
            _forwardoutput(self._ui, self._side)
        return r

    def readline(self):
        return self._call('readline')

    def _call(self, methname, data=None):
        """call <methname> on "main", forward output of "side" while blocking
        """
        # data can be '' or 0
        if (data is not None and not data) or self._main.closed:
            _forwardoutput(self._ui, self._side)
            return ''
        while True:
            mainready, sideready = self._wait()
            if sideready:
                _forwardoutput(self._ui, self._side)
            if mainready:
                meth = getattr(self._main, methname)
                if data is None:
                    return meth()
                else:
                    return meth(data)

    def close(self):
        return self._main.close()

    def flush(self):
        return self._main.flush()

class sshpeer(wireproto.wirepeer):
    def __init__(self, ui, path, create=False):
        self._url = path
        self.ui = ui
        self.pipeo = self.pipei = self.pipee = None

        u = util.url(path, parsequery=False, parsefragment=False)
        if u.scheme != 'ssh' or not u.host or u.path is None:
            self._abort(error.RepoError(_("couldn't parse location %s") % path))

        self.user = u.user
        if u.passwd is not None:
            self._abort(error.RepoError(_("password in URL not supported")))
        self.host = u.host
        self.port = u.port
        self.path = u.path or "."

        sshcmd = self.ui.config("ui", "ssh", "ssh")
        remotecmd = self.ui.config("ui", "remotecmd", "hg")

        args = util.sshargs(sshcmd,
                            _serverquote(self.host),
                            _serverquote(self.user),
                            _serverquote(self.port))

        if create:
            cmd = '%s %s %s' % (sshcmd, args,
                util.shellquote("%s init %s" %
                    (_serverquote(remotecmd), _serverquote(self.path))))
            ui.debug('running %s\n' % cmd)
            res = ui.system(cmd, blockedtag='sshpeer')
            if res != 0:
                self._abort(error.RepoError(_("could not create remote repo")))

        self._validaterepo(sshcmd, args, remotecmd)

    def url(self):
        return self._url

    def _validaterepo(self, sshcmd, args, remotecmd):
        # cleanup up previous run
        self.cleanup()

        cmd = '%s %s %s' % (sshcmd, args,
            util.shellquote("%s -R %s serve --stdio" %
                (_serverquote(remotecmd), _serverquote(self.path))))
        self.ui.debug('running %s\n' % cmd)
        cmd = util.quotecommand(cmd)

        # while self.subprocess isn't used, having it allows the subprocess to
        # to clean up correctly later
        #
        # no buffer allow the use of 'select'
        # feel free to remove buffering and select usage when we ultimately
        # move to threading.
        sub = util.popen4(cmd, bufsize=0)
        self.pipeo, self.pipei, self.pipee, self.subprocess = sub

        self.pipei = util.bufferedinputpipe(self.pipei)
        self.pipei = doublepipe(self.ui, self.pipei, self.pipee)
        self.pipeo = doublepipe(self.ui, self.pipeo, self.pipee)

        # skip any noise generated by remote shell
        self._callstream("hello")
        r = self._callstream("between", pairs=("%s-%s" % ("0"*40, "0"*40)))
        lines = ["", "dummy"]
        max_noise = 500
        while lines[-1] and max_noise:
            l = r.readline()
            self.readerr()
            if lines[-1] == "1\n" and l == "\n":
                break
            if l:
                self.ui.debug("remote: ", l)
            lines.append(l)
            max_noise -= 1
        else:
            self._abort(error.RepoError(_('no suitable response from '
                                          'remote hg')))

        self._caps = set()
        for l in reversed(lines):
            if l.startswith("capabilities:"):
                self._caps.update(l[:-1].split(":")[1].split())
                break

    def _capabilities(self):
        return self._caps

    def readerr(self):
        _forwardoutput(self.ui, self.pipee)

    def _abort(self, exception):
        self.cleanup()
        raise exception

    def cleanup(self):
        if self.pipeo is None:
            return
        self.pipeo.close()
        self.pipei.close()
        try:
            # read the error descriptor until EOF
            for l in self.pipee:
                self.ui.status(_("remote: "), l)
        except (IOError, ValueError):
            pass
        self.pipee.close()

    __del__ = cleanup

    def _submitbatch(self, req):
        rsp = self._callstream("batch", cmds=wireproto.encodebatchcmds(req))
        available = self._getamount()
        # TODO this response parsing is probably suboptimal for large
        # batches with large responses.
        toread = min(available, 1024)
        work = rsp.read(toread)
        available -= toread
        chunk = work
        while chunk:
            while ';' in work:
                one, work = work.split(';', 1)
                yield wireproto.unescapearg(one)
            toread = min(available, 1024)
            chunk = rsp.read(toread)
            available -= toread
            work += chunk
        yield wireproto.unescapearg(work)

    def _callstream(self, cmd, **args):
        self.ui.debug("sending %s command\n" % cmd)
        self.pipeo.write("%s\n" % cmd)
        _func, names = wireproto.commands[cmd]
        keys = names.split()
        wireargs = {}
        for k in keys:
            if k == '*':
                wireargs['*'] = args
                break
            else:
                wireargs[k] = args[k]
                del args[k]
        for k, v in sorted(wireargs.iteritems()):
            self.pipeo.write("%s %d\n" % (k, len(v)))
            if isinstance(v, dict):
                for dk, dv in v.iteritems():
                    self.pipeo.write("%s %d\n" % (dk, len(dv)))
                    self.pipeo.write(dv)
            else:
                self.pipeo.write(v)
        self.pipeo.flush()

        return self.pipei

    def _callcompressable(self, cmd, **args):
        return self._callstream(cmd, **args)

    def _call(self, cmd, **args):
        self._callstream(cmd, **args)
        return self._recv()

    def _callpush(self, cmd, fp, **args):
        r = self._call(cmd, **args)
        if r:
            return '', r
        for d in iter(lambda: fp.read(4096), ''):
            self._send(d)
        self._send("", flush=True)
        r = self._recv()
        if r:
            return '', r
        return self._recv(), ''

    def _calltwowaystream(self, cmd, fp, **args):
        r = self._call(cmd, **args)
        if r:
            # XXX needs to be made better
            raise error.Abort(_('unexpected remote reply: %s') % r)
        for d in iter(lambda: fp.read(4096), ''):
            self._send(d)
        self._send("", flush=True)
        return self.pipei

    def _getamount(self):
        l = self.pipei.readline()
        if l == '\n':
            self.readerr()
            msg = _('check previous remote output')
            self._abort(error.OutOfBandError(hint=msg))
        self.readerr()
        try:
            return int(l)
        except ValueError:
            self._abort(error.ResponseError(_("unexpected response:"), l))

    def _recv(self):
        return self.pipei.read(self._getamount())

    def _send(self, data, flush=False):
        self.pipeo.write("%d\n" % len(data))
        if data:
            self.pipeo.write(data)
        if flush:
            self.pipeo.flush()
        self.readerr()

    def lock(self):
        self._call("lock")
        return remotelock(self)

    def unlock(self):
        self._call("unlock")

    def addchangegroup(self, cg, source, url, lock=None):
        '''Send a changegroup to the remote server.  Return an integer
        similar to unbundle(). DEPRECATED, since it requires locking the
        remote.'''
        d = self._call("addchangegroup")
        if d:
            self._abort(error.RepoError(_("push refused: %s") % d))
        for d in iter(lambda: cg.read(4096), ''):
            self.pipeo.write(d)
            self.readerr()

        self.pipeo.flush()

        self.readerr()
        r = self._recv()
        if not r:
            return 1
        try:
            return int(r)
        except ValueError:
            self._abort(error.ResponseError(_("unexpected response:"), r))

instance = sshpeer