This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3268

wireproto: implement command executor interface for version 1 peers
ClosedPublic

Authored by indygreg on Apr 11 2018, 7:42 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGe1b32dc4646c: wireproto: implement command executor interface for version 1 peers
Summary

Now that we've defined our new interface for issuing commands,
let's implement it.

We add the interface to the base peer interface. This means all
peer types must implement it.

The only peer types that we have are the local peer in localrepo
and a shared wire peer for version 1 of the wire protocol.

The local peer implementation is pretty straightforward. We
don't do anything fancy and just return a resolved future with
the result of a method call. This is similar to what
localiterbatcher does.

The wire protocol version 1 implementation is a bit more complicated
and is a more robust implementation.

The wire executor queues commands by default. And because the new
executor interface always allows multiple commands but not all version
1 commands are @batchable, it has to check that the requested commands
are batchable if multiple commands are being requested.

The wire executor currently only supports executing a single command.
This is for simplicity reasons. Support for multiple commands will
be added in a separate commit.

To prove the new interface works, a call to the "known" command
during discovery has been updated to use the new API.

It's worth noting that both implementations require a method having
the command name to exist on the peer. There is at least one caller
in core that don't have a method calls peer._call() directly. We
may need to shore up the requirements later...

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Apr 11 2018, 7:42 PM
indygreg updated this revision to Diff 8121.Apr 13 2018, 3:14 PM
durin42 accepted this revision.Apr 13 2018, 5:34 PM
This revision is now accepted and ready to land.Apr 13 2018, 5:34 PM
Closed by commit rHGe1b32dc4646c: wireproto: implement command executor interface for version 1 peers (authored by indygreg). · Explain WhyApr 13 2018, 5:45 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 8167

mercurial/localrepo.py

# localrepo.py - read/write repository class for mercurial # localrepo.py - read/write repository class for mercurial
# #
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> # Copyright 2005-2007 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import errno import errno
import hashlib import hashlib
import os import os
import random import random
import sys
import time import time
import weakref import weakref
from .i18n import _ from .i18n import _
from .node import ( from .node import (
hex, hex,
nullid, nullid,
short, short,
# submit for a local iter batcher is a noop # submit for a local iter batcher is a noop
pass pass
def results(self): def results(self):
for name, args, opts, resref in self.calls: for name, args, opts, resref in self.calls:
resref.set(getattr(self.local, name)(*args, **opts)) resref.set(getattr(self.local, name)(*args, **opts))
yield resref.value yield resref.value
@zi.implementer(repository.ipeercommandexecutor)
class localcommandexecutor(object):
def __init__(self, peer):
self._peer = peer
self._sent = False
self._closed = False
def __enter__(self):
return self
def __exit__(self, exctype, excvalue, exctb):
self.close()
def callcommand(self, command, args):
if self._sent:
raise error.ProgrammingError('callcommand() cannot be used after '
'sendcommands()')
if self._closed:
raise error.ProgrammingError('callcommand() cannot be used after '
'close()')
# We don't need to support anything fancy. Just call the named
# method on the peer and return a resolved future.
fn = getattr(self._peer, pycompat.sysstr(command))
f = pycompat.futures.Future()
try:
result = fn(**args)
except Exception:
f.set_exception_info(*sys.exc_info()[1:])
else:
f.set_result(result)
return f
def sendcommands(self):
self._sent = True
def close(self):
self._closed = True
class localpeer(repository.peer): class localpeer(repository.peer):
'''peer for a local repo; reflects only the most recent API''' '''peer for a local repo; reflects only the most recent API'''
def __init__(self, repo, caps=None): def __init__(self, repo, caps=None):
super(localpeer, self).__init__() super(localpeer, self).__init__()
if caps is None: if caps is None:
caps = moderncaps.copy() caps = moderncaps.copy()
except error.PushRaced as exc: except error.PushRaced as exc:
raise error.ResponseError(_('push failed:'), raise error.ResponseError(_('push failed:'),
stringutil.forcebytestr(exc)) stringutil.forcebytestr(exc))
# End of _basewirecommands interface. # End of _basewirecommands interface.
# Begin of peer interface. # Begin of peer interface.
def commandexecutor(self):
return localcommandexecutor(self)
def iterbatch(self): def iterbatch(self):
return localiterbatcher(self) return localiterbatcher(self)
# End of peer interface. # End of peer interface.
class locallegacypeer(repository.legacypeer, localpeer): class locallegacypeer(repository.legacypeer, localpeer):
'''peer extension which implements legacy methods too; used for tests with '''peer extension which implements legacy methods too; used for tests with
restricted capabilities''' restricted capabilities'''

mercurial/repository.py

a single HTTP request. While some peers may support multiple command a single HTTP request. While some peers may support multiple command
sends over the wire per executor, consumers need to code to the least sends over the wire per executor, consumers need to code to the least
capable peer. So it should be assumed that command executors buffer capable peer. So it should be assumed that command executors buffer
called commands until they are told to send them and that each called commands until they are told to send them and that each
command executor could result in a new connection or wire-level request command executor could result in a new connection or wire-level request
being issued. being issued.
""" """
class ipeerbase(ipeerconnection, ipeercapabilities, ipeercommands): class ipeerbase(ipeerconnection, ipeercapabilities, ipeercommands,
ipeerrequests):
"""Unified interface for peer repositories. """Unified interface for peer repositories.
All peer instances must conform to this interface. All peer instances must conform to this interface.
""" """
def iterbatch(): def iterbatch():
"""Obtain an object to be used for multiple method calls. """Obtain an object to be used for multiple method calls.
Various operations call several methods on peer instances. If each Various operations call several methods on peer instances. If each

mercurial/setdiscovery.py

sample = samplefunc(dag, undecided, targetsize) sample = samplefunc(dag, undecided, targetsize)
roundtrips += 1 roundtrips += 1
ui.progress(_('searching'), roundtrips, unit=_('queries')) ui.progress(_('searching'), roundtrips, unit=_('queries'))
ui.debug("query %i; still undecided: %i, sample size is: %i\n" ui.debug("query %i; still undecided: %i, sample size is: %i\n"
% (roundtrips, len(undecided), len(sample))) % (roundtrips, len(undecided), len(sample)))
# indices between sample and externalized version must match # indices between sample and externalized version must match
sample = list(sample) sample = list(sample)
yesno = remote.known(dag.externalizeall(sample))
with remote.commandexecutor() as e:
yesno = e.callcommand('known', {
'nodes': dag.externalizeall(sample),
}).result()
full = True full = True
if sample: if sample:
commoninsample = set(n for i, n in enumerate(sample) if yesno[i]) commoninsample = set(n for i, n in enumerate(sample) if yesno[i])
common.addbases(commoninsample) common.addbases(commoninsample)
common.removeancestorsfrom(undecided) common.removeancestorsfrom(undecided)
# heads(common) == heads(common.bases) since common represents common.bases # heads(common) == heads(common.bases) since common represents common.bases

mercurial/wireprotov1peer.py

# wireprotov1peer.py - Client-side functionality for wire protocol version 1. # wireprotov1peer.py - Client-side functionality for wire protocol version 1.
# #
# Copyright 2005-2010 Matt Mackall <mpm@selenic.com> # Copyright 2005-2010 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import hashlib import hashlib
import sys
from .i18n import _ from .i18n import _
from .node import ( from .node import (
bin, bin,
) )
from .thirdparty.zope import (
interface as zi,
)
from . import ( from . import (
bundle2, bundle2,
changegroup as changegroupmod, changegroup as changegroupmod,
encoding, encoding,
error, error,
pushkey as pushkeymod, pushkey as pushkeymod,
pycompat, pycompat,
repository, repository,
assert all(escapearg(k) == k for k in argsdict) assert all(escapearg(k) == k for k in argsdict)
args = ','.join('%s=%s' % (escapearg(k), escapearg(v)) args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
for k, v in argsdict.iteritems()) for k, v in argsdict.iteritems())
cmds.append('%s %s' % (op, args)) cmds.append('%s %s' % (op, args))
return ';'.join(cmds) return ';'.join(cmds)
@zi.implementer(repository.ipeercommandexecutor)
class peerexecutor(object):
def __init__(self, peer):
self._peer = peer
self._sent = False
self._closed = False
self._calls = []
def __enter__(self):
return self
def __exit__(self, exctype, excvalee, exctb):
self.close()
def callcommand(self, command, args):
if self._sent:
raise error.ProgrammingError('callcommand() cannot be used '
'after commands are sent')
if self._closed:
raise error.ProgrammingError('callcommand() cannot be used '
'after close()')
# Commands are dispatched through methods on the peer.
fn = getattr(self._peer, pycompat.sysstr(command), None)
if not fn:
raise error.ProgrammingError(
'cannot call command %s: method of same name not available '
'on peer' % command)
# Commands are either batchable or they aren't. If a command
# isn't batchable, we send it immediately because the executor
# can no longer accept new commands after a non-batchable command.
# If a command is batchable, we queue it for later.
if getattr(fn, 'batchable', False):
pass
else:
if self._calls:
raise error.ProgrammingError(
'%s is not batchable and cannot be called on a command '
'executor along with other commands' % command)
# We don't support batching yet. So resolve it immediately.
f = pycompat.futures.Future()
self._calls.append((command, args, fn, f))
self.sendcommands()
return f
def sendcommands(self):
if self._sent:
return
if not self._calls:
return
self._sent = True
calls = self._calls
# Mainly to destroy references to futures.
self._calls = None
if len(calls) == 1:
command, args, fn, f = calls[0]
# Future was cancelled. Ignore it.
if not f.set_running_or_notify_cancel():
return
try:
result = fn(**pycompat.strkwargs(args))
except Exception:
f.set_exception_info(*sys.exc_info()[1:])
else:
f.set_result(result)
return
raise error.ProgrammingError('support for multiple commands not '
'yet implemented')
def close(self):
self.sendcommands()
self._closed = True
class wirepeer(repository.legacypeer): class wirepeer(repository.legacypeer):
"""Client-side interface for communicating with a peer repository. """Client-side interface for communicating with a peer repository.
Methods commonly call wire protocol commands of the same name. Methods commonly call wire protocol commands of the same name.
See also httppeer.py and sshpeer.py for protocol-specific See also httppeer.py and sshpeer.py for protocol-specific
implementations of this interface. implementations of this interface.
""" """
def commandexecutor(self):
return peerexecutor(self)
# Begin of ipeercommands interface. # Begin of ipeercommands interface.
def iterbatch(self): def iterbatch(self):
return remoteiterbatcher(self) return remoteiterbatcher(self)
@batchable @batchable
def lookup(self, key): def lookup(self, key):
self.requirecap('lookup', _('look up remote revision')) self.requirecap('lookup', _('look up remote revision'))

tests/test-check-interfaces.py

repository, repository,
sshpeer, sshpeer,
statichttprepo, statichttprepo,
ui as uimod, ui as uimod,
unionrepo, unionrepo,
vfs as vfsmod, vfs as vfsmod,
wireprotoserver, wireprotoserver,
wireprototypes, wireprototypes,
wireprotov1peer,
wireprotov2server, wireprotov2server,
) )
rootdir = os.path.normpath(os.path.join(os.path.dirname(__file__), '..')) rootdir = os.path.normpath(os.path.join(os.path.dirname(__file__), '..'))
def checkzobject(o, allowextra=False): def checkzobject(o, allowextra=False):
"""Verify an object with a zope interface.""" """Verify an object with a zope interface."""
ifaces = zi.providedBy(o) ifaces = zi.providedBy(o)
ziverify.verifyClass(repository.ipeercapabilities, ziverify.verifyClass(repository.ipeercapabilities,
httppeer.httpv2peer) httppeer.httpv2peer)
checkzobject(httppeer.httpv2peer(None, '', None, None, None, None)) checkzobject(httppeer.httpv2peer(None, '', None, None, None, None))
ziverify.verifyClass(repository.ipeerbase, ziverify.verifyClass(repository.ipeerbase,
localrepo.localpeer) localrepo.localpeer)
checkzobject(localrepo.localpeer(dummyrepo())) checkzobject(localrepo.localpeer(dummyrepo()))
ziverify.verifyClass(repository.ipeercommandexecutor,
localrepo.localcommandexecutor)
checkzobject(localrepo.localcommandexecutor(None))
ziverify.verifyClass(repository.ipeercommandexecutor,
wireprotov1peer.peerexecutor)
checkzobject(wireprotov1peer.peerexecutor(None))
ziverify.verifyClass(repository.ipeerbaselegacycommands, ziverify.verifyClass(repository.ipeerbaselegacycommands,
sshpeer.sshv1peer) sshpeer.sshv1peer)
checkzobject(sshpeer.sshv1peer(ui, 'ssh://localhost/foo', None, dummypipe(), checkzobject(sshpeer.sshv1peer(ui, 'ssh://localhost/foo', None, dummypipe(),
dummypipe(), None, None)) dummypipe(), None, None))
ziverify.verifyClass(repository.ipeerbaselegacycommands, ziverify.verifyClass(repository.ipeerbaselegacycommands,
sshpeer.sshv2peer) sshpeer.sshv2peer)
checkzobject(sshpeer.sshv2peer(ui, 'ssh://localhost/foo', None, dummypipe(), checkzobject(sshpeer.sshv2peer(ui, 'ssh://localhost/foo', None, dummypipe(),