( )⚙ D11212 wireprotov1peer: update all rpcs to use the new batchable scheme

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D11212

wireprotov1peer: update all rpcs to use the new batchable scheme
ClosedPublic

Authored by valentin.gatienbaron on Jul 25 2021, 4:32 PM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHGc424ff4807e6: wireprotov1peer: update all rpcs to use the new batchable scheme
rHGf4f9b0e06fb8: wireprotov1peer: update all rpcs to use the new batchable scheme
Summary

If desired, we could keep the future class and the function that
upgrades an old style rpc instead of a new style, for extensions.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

Alphare accepted this revision.Aug 24 2021, 10:09 AM
Alphare added a subscriber: Alphare.

Agreed that this is simpler, looks good to me.
I don't think the backwards compatibility concerns are too bad, but I might be wrong.

This revision is now accepted and ready to land.Aug 24 2021, 10:09 AM
baymax updated this revision to Diff 29997.Aug 24 2021, 11:14 AM

✅ refresh by Heptapod after a successful CI run (🐙 💚)
⚠ This patch is intended for stable ⚠

danchr added a subscriber: danchr.Aug 27 2021, 10:52 AM
In D11212#173796, @Alphare wrote:

I don't think the backwards compatibility concerns are too bad, but I might be wrong.

Please note that this breaks hg-git, as it currently relies on the future for wrapping a repository. We'll probably have to come up with a wrapper regardless.

See e.g. https://foss.heptapod.net/mercurial/hg-git/-/blob/branch/default/hggit/gitrepo.py#L113

Alphare added a comment.Aug 30 2021, 4:25 AM
In D11212#174426, @danchr wrote:
In D11212#173796, @Alphare wrote:

I don't think the backwards compatibility concerns are too bad, but I might be wrong.

Please note that this breaks hg-git, as it currently relies on the future for wrapping a repository. We'll probably have to come up with a wrapper regardless.
See e.g. https://foss.heptapod.net/mercurial/hg-git/-/blob/branch/default/hggit/gitrepo.py#L113

Thanks for bringing this up. Does that represent a large amount of work for you?

danchr added a comment.Sep 7 2021, 10:26 AM
In D11212#174524, @Alphare wrote:
In D11212#174426, @danchr wrote:
In D11212#173796, @Alphare wrote:

I don't think the backwards compatibility concerns are too bad, but I might be wrong.

Please note that this breaks hg-git, as it currently relies on the future for wrapping a repository. We'll probably have to come up with a wrapper regardless.
See e.g. https://foss.heptapod.net/mercurial/hg-git/-/blob/branch/default/hggit/gitrepo.py#L113

Thanks for bringing this up. Does that represent a large amount of work for you?

Well, my main challenge here is that I have absolutely no idea what's going on — I don't have much experience with the very low levels of networking in Mercurial, or the (relatively) new peer API. It seems to me that the point of that code is, essentially, to provide a minimal repository or peer that does nothing. Could something like that be moved to Mercurial core?

danchr added a comment.Sep 11 2021, 9:31 AM
In D11212#174786, @danchr wrote:
In D11212#174524, @Alphare wrote:

Thanks for bringing this up. Does that represent a large amount of work for you?

Well, my main challenge here is that I have absolutely no idea what's going on — I don't have much experience with the very low levels of networking in Mercurial, or the (relatively) new peer API. It seems to me that the point of that code is, essentially, to provide a minimal repository or peer that does nothing. Could something like that be moved to Mercurial core?

Okay, I managed to fix it:

https://foss.heptapod.net/mercurial/hg-git/-/merge_requests/130/diffs
https://foss.heptapod.net/mercurial/hg-git/-/merge_requests/131/diffs

Would one of you perhaps take a look at those changes and see if they make sense? I did it in the form of a decorator that just wraps a regular function returning a value. Something like that might be useful for core as well, and simplify a simple method like branchmap().

valentin.gatienbaron added a comment.Sep 11 2021, 11:31 AM
In D11212#174950, @danchr wrote:
In D11212#174786, @danchr wrote:
In D11212#174524, @Alphare wrote:

Thanks for bringing this up. Does that represent a large amount of work for you?

Well, my main challenge here is that I have absolutely no idea what's going on — I don't have much experience with the very low levels of networking in Mercurial, or the (relatively) new peer API. It seems to me that the point of that code is, essentially, to provide a minimal repository or peer that does nothing. Could something like that be moved to Mercurial core?

Okay, I managed to fix it:
https://foss.heptapod.net/mercurial/hg-git/-/merge_requests/130/diffs
https://foss.heptapod.net/mercurial/hg-git/-/merge_requests/131/diffs
Would one of you perhaps take a look at those changes and see if they make sense? I did it in the form of a decorator that just wraps a regular function returning a value. Something like that might be useful for core as well, and simplify a simple method like branchmap().

The two wrappers make sense, although it seems random to encode the query as {} in one case and None in other case (I also don't really understand why the old wrapper yields twice instead of yield+return, but I guess it doesn't really make a difference).
I can't really speak about the merge of the two classes.

danchr added a comment.Sep 11 2021, 12:12 PM
In D11212#174951, @valentin.gatienbaron wrote:

The two wrappers make sense, although it seems random to encode the query as {} in one case and None in other case

Good point! Since both work, I simplified it to just always use None.

(I also don't really understand why the old wrapper yields twice instead of yield+return, but I guess it doesn't really make a difference).

Well, yield+return is a syntax error in Python 2.7, and just yielding once raises some odd exceptions I'd rather not dig into for legacy code.

I can't really speak about the merge of the two classes.

Thanks for taking a look anyway! (AFAICT the two classes were a somewhat convoluted way of handling compatibility with 4.8 and earlier; with the decorator, it's no longer needed.)

Alphare added a comment.Sep 13 2021, 5:58 AM

I'm a little late to the party, but the changes look good to me as well, thanks for fixing it in hg-git!

Revision Contents
Changeset List

Diff 29993

hgext/fastannotate/protocol.py

def peersetup(ui, peer): def peersetup(ui, peer):
class fastannotatepeer(peer.__class__): class fastannotatepeer(peer.__class__):
@wireprotov1peer.batchable @wireprotov1peer.batchable
def getannotate(self, path, lastnode=None): def getannotate(self, path, lastnode=None):
if not self.capable(b'getannotate'): if not self.capable(b'getannotate'):
ui.warn(_(b'remote peer cannot provide annotate cache\n')) ui.warn(_(b'remote peer cannot provide annotate cache\n'))
yield None, None return None, None
else: else:
args = {b'path': path, b'lastnode': lastnode or b''} args = {b'path': path, b'lastnode': lastnode or b''}
f = wireprotov1peer.future() return args, _parseresponse
yield args, f
yield _parseresponse(f.value)
peer.__class__ = fastannotatepeer peer.__class__ = fastannotatepeer
@contextlib.contextmanager @contextlib.contextmanager
def annotatepeer(repo): def annotatepeer(repo):
ui = repo.ui ui = repo.ui

hgext/infinitepush/__init__.py

return results return results
else: else:
return orig(self, namespace) return orig(self, namespace)
@wireprotov1peer.batchable @wireprotov1peer.batchable
def listkeyspatterns(self, namespace, patterns): def listkeyspatterns(self, namespace, patterns):
if not self.capable(b'pushkey'): if not self.capable(b'pushkey'):
yield {}, None return {}, None
f = wireprotov1peer.future()
self.ui.debug(b'preparing listkeys for "%s"\n' % namespace) self.ui.debug(b'preparing listkeys for "%s"\n' % namespace)
yield {
b'namespace': encoding.fromlocal(namespace), def decode(d):
b'patterns': wireprototypes.encodelist(patterns),
}, f
d = f.value
self.ui.debug( self.ui.debug(
b'received listkey for "%s": %i bytes\n' % (namespace, len(d)) b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
) )
yield pushkey.decodekeys(d) return pushkey.decodekeys(d)
return {
b'namespace': encoding.fromlocal(namespace),
b'patterns': wireprototypes.encodelist(patterns),
}, decode
def _readbundlerevs(bundlerepo): def _readbundlerevs(bundlerepo):
return list(bundlerepo.revs(b'bundle()')) return list(bundlerepo.revs(b'bundle()'))
def _includefilelogstobundle(bundlecaps, bundlerepo, bundlerevs, ui): def _includefilelogstobundle(bundlecaps, bundlerepo, bundlerevs, ui):
"""Tells remotefilelog to include all changed files to the changegroup """Tells remotefilelog to include all changed files to the changegroup

hgext/largefiles/proto.py

chunk = stream.read(1) chunk = stream.read(1)
if chunk: if chunk:
self._abort( self._abort(
error.ResponseError(_(b"unexpected response:"), chunk) error.ResponseError(_(b"unexpected response:"), chunk)
) )
@wireprotov1peer.batchable @wireprotov1peer.batchable
def statlfile(self, sha): def statlfile(self, sha):
f = wireprotov1peer.future() def decode(d):
result = {b'sha': sha}
yield result, f
try: try:
yield int(f.value) return int(d)
except (ValueError, urlerr.httperror): except (ValueError, urlerr.httperror):
# If the server returns anything but an integer followed by a # If the server returns anything but an integer followed by a
# newline, newline, it's not speaking our language; if we get # newline, newline, it's not speaking our language; if we get
# an HTTP error, we can't be sure the largefile is present; # an HTTP error, we can't be sure the largefile is present;
# either way, consider it missing. # either way, consider it missing.
yield 2 return 2
result = {b'sha': sha}
return result, decode
repo.__class__ = lfileswirerepository repo.__class__ = lfileswirerepository
# advertise the largefiles=serve capability # advertise the largefiles=serve capability
@eh.wrapfunction(wireprotov1server, b'_capabilities') @eh.wrapfunction(wireprotov1server, b'_capabilities')
def _capabilities(orig, repo, proto): def _capabilities(orig, repo, proto):
'''announce largefile server capability''' '''announce largefile server capability'''

hgext/remotefilelog/fileserverclient.py

def peersetup(ui, peer): def peersetup(ui, peer):
class remotefilepeer(peer.__class__): class remotefilepeer(peer.__class__):
@wireprotov1peer.batchable @wireprotov1peer.batchable
def x_rfl_getfile(self, file, node): def x_rfl_getfile(self, file, node):
if not self.capable(b'x_rfl_getfile'): if not self.capable(b'x_rfl_getfile'):
raise error.Abort( raise error.Abort(
b'configured remotefile server does not support getfile' b'configured remotefile server does not support getfile'
) )
f = wireprotov1peer.future()
yield {b'file': file, b'node': node}, f def decode(d):
code, data = f.value.split(b'\0', 1) code, data = d.split(b'\0', 1)
if int(code): if int(code):
raise error.LookupError(file, node, data) raise error.LookupError(file, node, data)
yield data return data
return {b'file': file, b'node': node}, decode
@wireprotov1peer.batchable @wireprotov1peer.batchable
def x_rfl_getflogheads(self, path): def x_rfl_getflogheads(self, path):
if not self.capable(b'x_rfl_getflogheads'): if not self.capable(b'x_rfl_getflogheads'):
raise error.Abort( raise error.Abort(
b'configured remotefile server does not ' b'configured remotefile server does not '
b'support getflogheads' b'support getflogheads'
) )
f = wireprotov1peer.future()
yield {b'path': path}, f def decode(d):
heads = f.value.split(b'\n') if f.value else [] return d.split(b'\n') if d else []
yield heads
return {b'path': path}, decode
def _updatecallstreamopts(self, command, opts): def _updatecallstreamopts(self, command, opts):
if command != b'getbundle': if command != b'getbundle':
return return
if ( if (
constants.NETWORK_CAP_LEGACY_SSH_GETFILES constants.NETWORK_CAP_LEGACY_SSH_GETFILES
not in self.capabilities() not in self.capabilities()
): ):

mercurial/wireprotov1peer.py

repository, repository,
util as interfaceutil, util as interfaceutil,
) )
from .utils import hashutil from .utils import hashutil
urlreq = util.urlreq urlreq = util.urlreq
def batchable_new_style(f): def batchable(f):
"""annotation for batchable methods """annotation for batchable methods
Such methods must implement a coroutine as follows: Such methods must implement a coroutine as follows:
@batchable @batchable
def sample(self, one, two=None): def sample(self, one, two=None):
# Build list of encoded arguments suitable for your wire protocol: # Build list of encoded arguments suitable for your wire protocol:
encoded_args = [('one', encode(one),), ('two', encode(two),)] encoded_args = [('one', encode(one),), ('two', encode(two),)]
encoded_res = self._submitone(cmd, encoded_args_or_res) encoded_res = self._submitone(cmd, encoded_args_or_res)
return decode(encoded_res) return decode(encoded_res)
setattr(plain, 'batchable', f) setattr(plain, 'batchable', f)
setattr(plain, '__name__', f.__name__) setattr(plain, '__name__', f.__name__)
return plain return plain
def batchable(f):
def upgraded(*args, **opts):
batchable = f(*args, **opts)
encoded_args_or_res, encoded_res_future = next(batchable)
if not encoded_res_future:
decode = None
else:
def decode(d):
encoded_res_future.set(d)
return next(batchable)
return encoded_args_or_res, decode
setattr(upgraded, '__name__', f.__name__)
return batchable_new_style(upgraded)
class future(object):
'''placeholder for a value to be set later'''
def set(self, value):
if util.safehasattr(self, b'value'):
raise error.RepoError(b"future is already set")
self.value = value
def encodebatchcmds(req): def encodebatchcmds(req):
"""Return a ``cmds`` argument value for the ``batch`` command.""" """Return a ``cmds`` argument value for the ``batch`` command."""
escapearg = wireprototypes.escapebatcharg escapearg = wireprototypes.escapebatcharg
cmds = [] cmds = []
for op, argsdict in req: for op, argsdict in req:
# Old servers didn't properly unescape argument names. So prevent # Old servers didn't properly unescape argument names. So prevent
# the sending of argument names that may not be decoded properly by # the sending of argument names that may not be decoded properly by
def clonebundles(self): def clonebundles(self):
self.requirecap(b'clonebundles', _(b'clone bundles')) self.requirecap(b'clonebundles', _(b'clone bundles'))
return self._call(b'clonebundles') return self._call(b'clonebundles')
@batchable @batchable
def lookup(self, key): def lookup(self, key):
self.requirecap(b'lookup', _(b'look up remote revision')) self.requirecap(b'lookup', _(b'look up remote revision'))
f = future()
yield {b'key': encoding.fromlocal(key)}, f def decode(d):
d = f.value
success, data = d[:-1].split(b" ", 1) success, data = d[:-1].split(b" ", 1)
if int(success): if int(success):
yield bin(data) return bin(data)
else: else:
self._abort(error.RepoError(data)) self._abort(error.RepoError(data))
return {b'key': encoding.fromlocal(key)}, decode
@batchable @batchable
def heads(self): def heads(self):
f = future() def decode(d):
yield {}, f
d = f.value
try: try:
yield wireprototypes.decodelist(d[:-1]) return wireprototypes.decodelist(d[:-1])
except ValueError: except ValueError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
return {}, decode
@batchable @batchable
def known(self, nodes): def known(self, nodes):
f = future() def decode(d):
yield {b'nodes': wireprototypes.encodelist(nodes)}, f
d = f.value
try: try:
yield [bool(int(b)) for b in pycompat.iterbytestr(d)] return [bool(int(b)) for b in pycompat.iterbytestr(d)]
except ValueError: except ValueError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
return {b'nodes': wireprototypes.encodelist(nodes)}, decode
@batchable @batchable
def branchmap(self): def branchmap(self):
f = future() def decode(d):
yield {}, f
d = f.value
try: try:
branchmap = {} branchmap = {}
for branchpart in d.splitlines(): for branchpart in d.splitlines():
branchname, branchheads = branchpart.split(b' ', 1) branchname, branchheads = branchpart.split(b' ', 1)
branchname = encoding.tolocal(urlreq.unquote(branchname)) branchname = encoding.tolocal(urlreq.unquote(branchname))
branchheads = wireprototypes.decodelist(branchheads) branchheads = wireprototypes.decodelist(branchheads)
branchmap[branchname] = branchheads branchmap[branchname] = branchheads
yield branchmap return branchmap
except TypeError: except TypeError:
self._abort(error.ResponseError(_(b"unexpected response:"), d)) self._abort(error.ResponseError(_(b"unexpected response:"), d))
return {}, decode
@batchable @batchable
def listkeys(self, namespace): def listkeys(self, namespace):
if not self.capable(b'pushkey'): if not self.capable(b'pushkey'):
yield {}, None return {}, None
f = future()
self.ui.debug(b'preparing listkeys for "%s"\n' % namespace) self.ui.debug(b'preparing listkeys for "%s"\n' % namespace)
yield {b'namespace': encoding.fromlocal(namespace)}, f
d = f.value def decode(d):
self.ui.debug( self.ui.debug(
b'received listkey for "%s": %i bytes\n' % (namespace, len(d)) b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
) )
yield pushkeymod.decodekeys(d) return pushkeymod.decodekeys(d)
return {b'namespace': encoding.fromlocal(namespace)}, decode
@batchable @batchable
def pushkey(self, namespace, key, old, new): def pushkey(self, namespace, key, old, new):
if not self.capable(b'pushkey'): if not self.capable(b'pushkey'):
yield False, None return False, None
f = future()
self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key)) self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key))
yield {
b'namespace': encoding.fromlocal(namespace), def decode(d):
b'key': encoding.fromlocal(key),
b'old': encoding.fromlocal(old),
b'new': encoding.fromlocal(new),
}, f
d = f.value
d, output = d.split(b'\n', 1) d, output = d.split(b'\n', 1)
try: try:
d = bool(int(d)) d = bool(int(d))
except ValueError: except ValueError:
raise error.ResponseError( raise error.ResponseError(
_(b'push failed (unexpected response):'), d _(b'push failed (unexpected response):'), d
) )
for l in output.splitlines(True): for l in output.splitlines(True):
self.ui.status(_(b'remote: '), l) self.ui.status(_(b'remote: '), l)
yield d return d
return {
b'namespace': encoding.fromlocal(namespace),
b'key': encoding.fromlocal(key),
b'old': encoding.fromlocal(old),
b'new': encoding.fromlocal(new),
}, decode
def stream_out(self): def stream_out(self):
return self._callstream(b'stream_out') return self._callstream(b'stream_out')
def getbundle(self, source, **kwargs): def getbundle(self, source, **kwargs):
kwargs = pycompat.byteskwargs(kwargs) kwargs = pycompat.byteskwargs(kwargs)
self.requirecap(b'getbundle', _(b'look up remote changes')) self.requirecap(b'getbundle', _(b'look up remote changes'))
opts = {} opts = {}

tests/test-batching.py

b'one', b'one',
mangle(one), mangle(one),
), ),
( (
b'two', b'two',
mangle(two), mangle(two),
), ),
] ]
encoded_res_future = wireprotov1peer.future() return encoded_args, unmangle
yield encoded_args, encoded_res_future
yield unmangle(encoded_res_future.value)
@wireprotov1peer.batchable @wireprotov1peer.batchable
def bar(self, b, a): def bar(self, b, a):
encresref = wireprotov1peer.future() return [
yield [
( (
b'b', b'b',
mangle(b), mangle(b),
), ),
( (
b'a', b'a',
mangle(a), mangle(a),
), ),
], encresref ], unmangle
yield unmangle(encresref.value)
# greet is coded directly. It therefore does not support batching. If it # greet is coded directly. It therefore does not support batching. If it
# does appear in a batch, the batch is split around greet, and the call to # does appear in a batch, the batch is split around greet, and the call to
# greet is done in its own roundtrip. # greet is done in its own roundtrip.
def greet(self, name=None): def greet(self, name=None):
return unmangle( return unmangle(
self._submitone( self._submitone(
b'greet', b'greet',

tests/test-wireproto.py

else: else:
raise error.Abort('dummy client does not support response type') raise error.Abort('dummy client does not support response type')
def _callstream(self, cmd, **args): def _callstream(self, cmd, **args):
return stringio(self._call(cmd, **args)) return stringio(self._call(cmd, **args))
@wireprotov1peer.batchable @wireprotov1peer.batchable
def greet(self, name): def greet(self, name):
f = wireprotov1peer.future() return {b'name': mangle(name)}, unmangle
yield {b'name': mangle(name)}, f
yield unmangle(f.value)
class serverrepo(object): class serverrepo(object):
def __init__(self, ui): def __init__(self, ui):
self.ui = ui self.ui = ui
def greet(self, name): def greet(self, name):
return b"Hello, " + name return b"Hello, " + name