comparison mercurial/debugcommands.py @ 36557:72e487851a53

debugcommands: add debugwireproto command

We currently don't have a low-level mechanism for sending
arbitrary wire protocol commands. Having a generic and robust
mechanism for sending wire protocol commands, examining wire
data, etc would make it vastly easier to test the wire protocol
and debug server operation. This is a problem I've wanted a
solution for numerous times, especially recently as I've been
hacking on a new version of the wire protocol.

This commit establishes a `hg debugwireproto` command for sending
data to a peer.

The command invents a mini language for specifying actions to take.
This will enable a lot of flexibility for issuing commands and testing
variations for how commands are sent.

Right now, we only support low-level raw sends and receives. These
are probably the least valuable commands to intended users of this
command. But they are the most useful commands to implement to
bootstrap the feature (I've chosen to reimplement test-ssh-proto.t
using this command to prove its usefulness).

My eventual goal of `hg debugwireproto` is to allow calling wire
protocol commands with a human-friendly interface. Essentially,
people can type in a command name and arguments and
`hg debugwireproto` will figure out how to send that on the wire.
I'd love to eventually be able to save the server's raw response
to a file. This would allow us to e.g. call "getbundle" wire
protocol commands easily.

test-ssh-proto.t has been updated to use the new command in lieu
of piping directly to a server process. As part of the transition,
test behavior improved. Before, we piped all request data to the
server at once. Now, we have explicit control over the ordering of
operations. e.g. we can send one command, receive its response,
then send another command. This will allow us to more robustly
test race conditions, buffering behavior, etc.

There were some subtle changes in test behavior. For example,
previous behavior would often send trailing newlines to the server.
The new mechanism doesn't treat literal newlines specially and
requires newlines be escaped in the payload.

Because the new logging code is very low level, it is easy to
introduce race conditions in tests. For example, the number of bytes
returned by a read() may vary depending on load. This is why tests
make heavy use of "readline" for consuming data: the result of
that operation should be deterministic and not subject to race
conditions. There are still some uses of "readavailable." However,
those are only for reading from stderr. I was able to reproduce
timing issues with my system under load when using "readavailable"
globally. But if I "readline" to grab stdout, "readavailable"
appears to work deterministically for stderr. I think this is
because the server writes to stderr first. As long as the OS
delivers writes to pipes in the same order they were made, this
should work. If there are timing issues, we can introduce a
mechanism to readline from stderr.

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

36556:44dc34b8d17b

import os
import random
import socket
import ssl
import string
import sys
import tempfile
import time

from .i18n import _

    revsetlang,
    scmutil,
    setdiscovery,
    simplemerge,
    smartset,
    sslutil,
    streamclone,
    templater,
    treediscovery,
    upgrade,

    res1 = repo.debugwireargs(*vals, **args)
    res2 = repo.debugwireargs(*vals, **args)
    ui.write("%s\n" % res1)
    if res1 != res2:
        ui.warn("%s\n" % res2)

36557:72e487851a53

import os
import random
import socket
import ssl
import string
import subprocess
import sys
import tempfile
import time

from .i18n import _

    revsetlang,
    scmutil,
    setdiscovery,
    simplemerge,
    smartset,
    sshpeer,
    sslutil,
    streamclone,
    templater,
    treediscovery,
    upgrade,

    res1 = repo.debugwireargs(*vals, **args)
    res2 = repo.debugwireargs(*vals, **args)
    ui.write("%s\n" % res1)
    if res1 != res2:
        ui.warn("%s\n" % res2)

def _parsewirelangblocks(fh):
    activeaction = None
    blocklines = []

    for line in fh:
        line = line.rstrip()
        if not line:
            continue

        if line.startswith(b'#'):
            continue

        if not line.startswith(' '):
            # New block. Flush previous one.
            if activeaction:
                yield activeaction, blocklines

            activeaction = line
            blocklines = []
            continue

        # Else we start with an indent.

        if not activeaction:
            raise error.Abort(_('indented line outside of block'))

        blocklines.append(line)

    # Flush last block.
    if activeaction:
        yield activeaction, blocklines

@command('debugwireproto',
    [
        ('', 'localssh', False, _('start an SSH server for this repo')),
        ('', 'peer', '', _('construct a specific version of the peer')),
    ] + cmdutil.remoteopts,
    _('[REPO]'),
    optionalrepo=True)
def debugwireproto(ui, repo, **opts):
    """send wire protocol commands to a server

    This command can be used to issue wire protocol commands to remote
    peers and to debug the raw data being exchanged.

    ``--localssh`` will start an SSH server against the current repository
    and connect to that. By default, the connection will perform a handshake
    and establish an appropriate peer instance.

    ``--peer`` can be used to bypass the handshake protocol and construct a
    peer instance using the specified class type. Valid values are ``raw``,
    ``ssh1``, and ``ssh2``. ``raw`` instances only allow sending raw data
    payloads and don't support higher-level command actions.

    Commands are issued via a mini language which is specified via stdin.
    The language consists of individual actions to perform. An action is
    defined by a block. A block is defined as a line with no leading
    space followed by 0 or more lines with leading space. Blocks are
    effectively a high-level command with additional metadata.

    Lines beginning with ``#`` are ignored.

    The following sections denote available actions.

    raw
    ---

    Send raw data to the server.

    The block payload contains the raw data to send as one atomic send
    operation. The data may not actually be delivered in a single system
    call: it depends on the abilities of the transport being used.

    Each line in the block is de-indented and concatenated. Then, that
    value is evaluated as a Python b'' literal. This allows the use of
    backslash escaping, etc.

    raw+
    ----

    Behaves like ``raw`` except flushes output afterwards.

    close
    -----

    Close the connection to the server.

    flush
    -----

    Flush data written to the server.

    readavailable
    -------------

    Read all available data from the server.

    If the connection to the server encompasses multiple pipes, we poll both
    pipes and read available data.

    readline
    --------

    Read a line of output from the server. If there are multiple output
    pipes, reads only the main pipe.
    """
    opts = pycompat.byteskwargs(opts)

    if opts['localssh'] and not repo:
        raise error.Abort(_('--localssh requires a repository'))

    if opts['peer'] and opts['peer'] not in ('raw', 'ssh1', 'ssh2'):
        raise error.Abort(_('invalid value for --peer'),
                          hint=_('valid values are "raw", "ssh1", and "ssh2"'))

    if ui.interactive():
        ui.write(_('(waiting for commands on stdin)\n'))

    blocks = list(_parsewirelangblocks(ui.fin))

    proc = None

    if opts['localssh']:
        # We start the SSH server in its own process so there is process
        # separation. This prevents a whole class of potential bugs around
        # shared state from interfering with server operation.
        args = util.hgcmd() + [
            '-R', repo.root,
            'debugserve', '--sshstdio',
        ]
        proc = subprocess.Popen(args, stdin=subprocess.PIPE,
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                bufsize=0)

        stdin = proc.stdin
        stdout = proc.stdout
        stderr = proc.stderr

        # We turn the pipes into observers so we can log I/O.
        if ui.verbose or opts['peer'] == 'raw':
            stdin = util.makeloggingfileobject(ui, proc.stdin, b'i',
                                               logdata=True)
            stdout = util.makeloggingfileobject(ui, proc.stdout, b'o',
                                                logdata=True)
            stderr = util.makeloggingfileobject(ui, proc.stderr, b'e',
                                                logdata=True)

        # --localssh also implies the peer connection settings.

        url = 'ssh://localserver'

        if opts['peer'] == 'ssh1':
            ui.write(_('creating ssh peer for wire protocol version 1\n'))
            peer = sshpeer.sshv1peer(ui, url, proc, stdin, stdout, stderr,
                                     None)
        elif opts['peer'] == 'ssh2':
            ui.write(_('creating ssh peer for wire protocol version 2\n'))
            peer = sshpeer.sshv2peer(ui, url, proc, stdin, stdout, stderr,
                                     None)
        elif opts['peer'] == 'raw':
            ui.write(_('using raw connection to peer\n'))
            peer = None
        else:
            ui.write(_('creating ssh peer from handshake results\n'))
            peer = sshpeer.makepeer(ui, url, proc, stdin, stdout, stderr)

    else:
        raise error.Abort(_('only --localssh is currently supported'))

    # Now perform actions based on the parsed wire language instructions.
    for action, lines in blocks:
        if action in ('raw', 'raw+'):
            # Concatenate the data together.
            data = ''.join(l.lstrip() for l in lines)
            data = util.unescapestr(data)
            stdin.write(data)

            if action == 'raw+':
                stdin.flush()
        elif action == 'flush':
            stdin.flush()
        elif action == 'close':
            peer.close()
        elif action == 'readavailable':
            fds = util.poll([stdout.fileno(), stderr.fileno()])

            if stdout.fileno() in fds:
                util.readpipe(stdout)
            if stderr.fileno() in fds:
                util.readpipe(stderr)
        elif action == 'readline':
            stdout.readline()
        else:
            raise error.Abort(_('unknown action: %s') % action)

    if peer:
        peer.close()

    if proc:
        proc.kill()