This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3271

wireproto: remove iterbatch() from peer interface (API)
ClosedPublic

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

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG33a6eee08db2: wireproto: remove iterbatch() from peer interface (API)
Summary

Good riddance.

Some tests have been ported to the new API. This probably should
have been done in earlier commits. But duplicating the test coverage
would have been difficult. It was easier this way.

.. api::

The wire protocol peer's ``iterbatch()`` for bulk executing commands
has been remove.d Use ``peer.commandexecutor()`` instead.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Apr 11 2018, 7:42 PM
indygreg updated this revision to Diff 8124.Apr 13 2018, 3:14 PM
durin42 accepted this revision.Apr 13 2018, 5:35 PM
This revision is now accepted and ready to land.Apr 13 2018, 5:35 PM
Closed by commit rHG33a6eee08db2: wireproto: remove iterbatch() from peer interface (API) (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 8124

mercurial/localrepo.py

sparse, sparse,
store, store,
subrepoutil, subrepoutil,
tags as tagsmod, tags as tagsmod,
transaction, transaction,
txnutil, txnutil,
util, util,
vfs as vfsmod, vfs as vfsmod,
wireprotov1peer,
) )
from .utils import ( from .utils import (
procutil, procutil,
stringutil, stringutil,
) )
release = lockmod.release release = lockmod.release
urlerr = util.urlerr urlerr = util.urlerr
def wrapper(repo, *args, **kwargs): def wrapper(repo, *args, **kwargs):
return orig(repo.unfiltered(), *args, **kwargs) return orig(repo.unfiltered(), *args, **kwargs)
return wrapper return wrapper
moderncaps = {'lookup', 'branchmap', 'pushkey', 'known', 'getbundle', moderncaps = {'lookup', 'branchmap', 'pushkey', 'known', 'getbundle',
'unbundle'} 'unbundle'}
legacycaps = moderncaps.union({'changegroupsubset'}) legacycaps = moderncaps.union({'changegroupsubset'})
class localiterbatcher(wireprotov1peer.iterbatcher):
def __init__(self, local):
super(localiterbatcher, self).__init__()
self.local = local
def submit(self):
# submit for a local iter batcher is a noop
pass
def results(self):
for name, args, opts, resref in self.calls:
resref.set(getattr(self.local, name)(*args, **opts))
yield resref.value
@zi.implementer(repository.ipeercommandexecutor) @zi.implementer(repository.ipeercommandexecutor)
class localcommandexecutor(object): class localcommandexecutor(object):
def __init__(self, peer): def __init__(self, peer):
self._peer = peer self._peer = peer
self._sent = False self._sent = False
self._closed = False self._closed = False
def __enter__(self): def __enter__(self):
# End of _basewirecommands interface. # End of _basewirecommands interface.
# Begin of peer interface. # Begin of peer interface.
def commandexecutor(self): def commandexecutor(self):
return localcommandexecutor(self) return localcommandexecutor(self)
def iterbatch(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'''
def __init__(self, repo): def __init__(self, repo):
super(locallegacypeer, self).__init__(repo, caps=legacycaps) super(locallegacypeer, self).__init__(repo, caps=legacycaps)

mercurial/repository.py

""" """
class ipeerbase(ipeerconnection, ipeercapabilities, ipeercommands, class ipeerbase(ipeerconnection, ipeercapabilities, ipeercommands,
ipeerrequests): 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():
"""Obtain an object to be used for multiple method calls.
Various operations call several methods on peer instances. If each
method call were performed immediately and serially, this would
require round trips to remote peers and/or would slow down execution.
Some peers have the ability to "batch" method calls to avoid costly
round trips or to facilitate concurrent execution.
This method returns an object that can be used to indicate intent to
perform batched method calls.
The returned object is a proxy of this peer. It intercepts calls to
batchable methods and queues them instead of performing them
immediately. This proxy object has a ``submit`` method that will
perform all queued batchable method calls. A ``results()`` method
exposes the results of queued/batched method calls. It is a generator
of results in the order they were called.
Not all peers or wire protocol implementations may actually batch method
calls. However, they must all support this API.
"""
class ipeerbaselegacycommands(ipeerbase, ipeerlegacycommands): class ipeerbaselegacycommands(ipeerbase, ipeerlegacycommands):
"""Unified peer interface that supports legacy commands.""" """Unified peer interface that supports legacy commands."""
@zi.implementer(ipeerbase) @zi.implementer(ipeerbase)
class peer(object): class peer(object):
"""Base class for peer repositories.""" """Base class for peer repositories."""

mercurial/wireprotov1peer.py

class future(object): class future(object):
'''placeholder for a value to be set later''' '''placeholder for a value to be set later'''
def set(self, value): def set(self, value):
if util.safehasattr(self, 'value'): if util.safehasattr(self, 'value'):
raise error.RepoError("future is already set") raise error.RepoError("future is already set")
self.value = value self.value = value
class batcher(object):
'''base class for batches of commands submittable in a single request
All methods invoked on instances of this class are simply queued and
return a a future for the result. Once you call submit(), all the queued
calls are performed and the results set in their respective futures.
'''
def __init__(self):
self.calls = []
def __getattr__(self, name):
def call(*args, **opts):
resref = future()
# Please don't invent non-ascii method names, or you will
# give core hg a very sad time.
self.calls.append((name.encode('ascii'), args, opts, resref,))
return resref
return call
def submit(self):
raise NotImplementedError()
class iterbatcher(batcher):
def submit(self):
raise NotImplementedError()
def results(self):
raise NotImplementedError()
class remoteiterbatcher(iterbatcher):
def __init__(self, remote):
super(remoteiterbatcher, self).__init__()
self._remote = remote
def __getattr__(self, name):
# Validate this method is batchable, since submit() only supports
# batchable methods.
fn = getattr(self._remote, name)
if not getattr(fn, 'batchable', None):
raise error.ProgrammingError('Attempted to batch a non-batchable '
'call to %r' % name)
return super(remoteiterbatcher, self).__getattr__(name)
def submit(self):
"""Break the batch request into many patch calls and pipeline them.
This is mostly valuable over http where request sizes can be
limited, but can be used in other places as well.
"""
# 2-tuple of (command, arguments) that represents what will be
# sent over the wire.
requests = []
# 4-tuple of (command, final future, @batchable generator, remote
# future).
results = []
for command, args, opts, finalfuture in self.calls:
mtd = getattr(self._remote, command)
batchable = mtd.batchable(mtd.__self__, *args, **opts)
commandargs, fremote = next(batchable)
assert fremote
requests.append((command, commandargs))
results.append((command, finalfuture, batchable, fremote))
if requests:
self._resultiter = self._remote._submitbatch(requests)
self._results = results
def results(self):
for command, finalfuture, batchable, remotefuture in self._results:
# Get the raw result, set it in the remote future, feed it
# back into the @batchable generator so it can be decoded, and
# set the result on the final future to this value.
remoteresult = next(self._resultiter)
remotefuture.set(remoteresult)
finalfuture.set(next(batchable))
# Verify our @batchable generators only emit 2 values.
try:
next(batchable)
except StopIteration:
pass
else:
raise error.ProgrammingError('%s @batchable generator emitted '
'unexpected value count' % command)
yield finalfuture.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
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): def commandexecutor(self):
return peerexecutor(self) return peerexecutor(self)
# Begin of ipeercommands interface. # Begin of ipeercommands interface.
def iterbatch(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'))
f = future() f = future()
yield {'key': encoding.fromlocal(key)}, f yield {'key': encoding.fromlocal(key)}, f
d = f.value d = f.value
success, data = d[:-1].split(" ", 1) success, data = d[:-1].split(" ", 1)
if int(success): if int(success):

tests/test-batching.py

# test-batching.py - tests for transparent command batching # test-batching.py - tests for transparent command batching
# #
# Copyright 2011 Peter Arrenbrecht <peter@arrenbrecht.ch> # Copyright 2011 Peter Arrenbrecht <peter@arrenbrecht.ch>
# #
# 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, print_function from __future__ import absolute_import, print_function
import contextlib
from mercurial import ( from mercurial import (
error,
localrepo, localrepo,
util,
wireprotov1peer, wireprotov1peer,
) )
# equivalent of repo.repository # equivalent of repo.repository
class thing(object): class thing(object):
def hello(self): def hello(self):
return "Ready." return "Ready."
# equivalent of localrepo.localrepository # equivalent of localrepo.localrepository
class localthing(thing): class localthing(thing):
def foo(self, one, two=None): def foo(self, one, two=None):
if one: if one:
return "%s and %s" % (one, two,) return "%s and %s" % (one, two,)
return "Nope" return "Nope"
def bar(self, b, a): def bar(self, b, a):
return "%s und %s" % (b, a,) return "%s und %s" % (b, a,)
def greet(self, name=None): def greet(self, name=None):
return "Hello, %s" % name return "Hello, %s" % name
def batchiter(self):
'''Support for local batching.''' @contextlib.contextmanager
return localrepo.localiterbatcher(self) def commandexecutor(self):
e = localrepo.localcommandexecutor(self)
try:
yield e
finally:
e.close()
# usage of "thing" interface # usage of "thing" interface
def use(it): def use(it):
# Direct call to base method shared between client and server. # Direct call to base method shared between client and server.
print(it.hello()) print(it.hello())
# Direct calls to proxied methods. They cause individual roundtrips. # Direct calls to proxied methods. They cause individual roundtrips.
print(it.foo("Un", two="Deux")) print(it.foo("Un", two="Deux"))
print(it.bar("Eins", "Zwei")) print(it.bar("Eins", "Zwei"))
# Batched call to a couple of proxied methods. # Batched call to a couple of proxied methods.
batch = it.batchiter()
# The calls return futures to eventually hold results.
foo = batch.foo(one="One", two="Two")
bar = batch.bar("Eins", "Zwei")
bar2 = batch.bar(b="Uno", a="Due")
# Future shouldn't be set until we submit().
assert isinstance(foo, wireprotov1peer.future)
assert not util.safehasattr(foo, 'value')
assert not util.safehasattr(bar, 'value')
batch.submit()
# Call results() to obtain results as a generator.
results = batch.results()
# Future results shouldn't be set until we consume a value.
assert not util.safehasattr(foo, 'value')
foovalue = next(results)
assert util.safehasattr(foo, 'value')
assert foovalue == foo.value
print(foo.value)
next(results)
print(bar.value)
next(results)
print(bar2.value)
# We should be at the end of the results generator. with it.commandexecutor() as e:
try: ffoo = e.callcommand('foo', {'one': 'One', 'two': 'Two'})
next(results) fbar = e.callcommand('bar', {'b': 'Eins', 'a': 'Zwei'})
except StopIteration: fbar2 = e.callcommand('bar', {'b': 'Uno', 'a': 'Due'})
print('proper end of results generator')
else: print(ffoo.result())
print('extra emitted element!') print(fbar.result())
print(fbar2.result())
# Attempting to call a non-batchable method inside a batch fails.
batch = it.batchiter()
try:
batch.greet(name='John Smith')
except error.ProgrammingError as e:
print(e)
# Attempting to call a local method inside a batch fails.
batch = it.batchiter()
try:
batch.hello()
except error.ProgrammingError as e:
print(e)
# local usage # local usage
mylocal = localthing() mylocal = localthing()
print() print()
print("== Local") print("== Local")
use(mylocal) use(mylocal)
# demo remoting; mimicks what wireproto and HTTP/SSH do # demo remoting; mimicks what wireproto and HTTP/SSH do
for name, args in cmds: for name, args in cmds:
args = ','.join(n + '=' + escapearg(v) for n, v in args) args = ','.join(n + '=' + escapearg(v) for n, v in args)
req.append(name + ':' + args) req.append(name + ':' + args)
req = ';'.join(req) req = ';'.join(req)
res = self._submitone('batch', [('cmds', req,)]) res = self._submitone('batch', [('cmds', req,)])
for r in res.split(';'): for r in res.split(';'):
yield r yield r
def batchiter(self): @contextlib.contextmanager
return wireprotov1peer.remoteiterbatcher(self) def commandexecutor(self):
e = wireprotov1peer.peerexecutor(self)
try:
yield e
finally:
e.close()
@wireprotov1peer.batchable @wireprotov1peer.batchable
def foo(self, one, two=None): def foo(self, one, two=None):
encargs = [('one', mangle(one),), ('two', mangle(two),)] encargs = [('one', mangle(one),), ('two', mangle(two),)]
encresref = wireprotov1peer.future() encresref = wireprotov1peer.future()
yield encargs, encresref yield encargs, encresref
yield unmangle(encresref.value) yield unmangle(encresref.value)

tests/test-batching.py.out

== Local == Local
Ready. Ready.
Un and Deux Un and Deux
Eins und Zwei Eins und Zwei
One and Two One and Two
Eins und Zwei Eins und Zwei
Uno und Due Uno und Due
proper end of results generator
== Remote == Remote
Ready. Ready.
REQ: foo?one=Vo&two=Efvy REQ: foo?one=Vo&two=Efvy
-> Vo!boe!Efvy -> Vo!boe!Efvy
Un and Deux Un and Deux
REQ: bar?b=Fjot&a=[xfj REQ: bar?b=Fjot&a=[xfj
-> Fjot!voe![xfj -> Fjot!voe![xfj
Eins und Zwei Eins und Zwei
REQ: batch?cmds=foo:one=Pof,two=Uxp;bar:b=Fjot,a=[xfj;bar:b=Vop,a=Evf REQ: batch?cmds=foo:one=Pof,two=Uxp;bar:b=Fjot,a=[xfj;bar:b=Vop,a=Evf
-> Pof!boe!Uxp;Fjot!voe![xfj;Vop!voe!Evf -> Pof!boe!Uxp;Fjot!voe![xfj;Vop!voe!Evf
One and Two One and Two
Eins und Zwei Eins und Zwei
Uno und Due Uno und Due
proper end of results generator
Attempted to batch a non-batchable call to 'greet'
Attempted to batch a non-batchable call to 'hello'

tests/test-wireproto.py

return mangle(repo.greet(unmangle(name))) return mangle(repo.greet(unmangle(name)))
wireproto.commands[b'greet'] = (greet, b'name',) wireproto.commands[b'greet'] = (greet, b'name',)
srv = serverrepo() srv = serverrepo()
clt = clientpeer(srv, uimod.ui()) clt = clientpeer(srv, uimod.ui())
print(clt.greet(b"Foobar")) print(clt.greet(b"Foobar"))
b = clt.iterbatch()
list(map(b.greet, (b'Fo, =;:<o', b'Bar'))) with clt.commandexecutor() as e:
b.submit() fgreet1 = e.callcommand(b'greet', {b'name': b'Fo, =;:<o'})
print([r for r in b.results()]) fgreet2 = e.callcommand(b'greet', {b'name': b'Bar'})
print([f.result() for f in (fgreet1, fgreet2)])