This is an archive of the discontinued Mercurial Phabricator instance.

Differential D320

peer: remove non iterating batcher (API)
ClosedPublic

Authored by indygreg on Aug 10 2017, 2:47 AM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGb47fe9733d76: peer: remove non iterating batcher (API)
Summary

The last use of this API was removed in b6e71f8af5b8 in 2016. While
not formally deprecated, as of the last commit the code is no longer
explicitly tested. I think the new API has existed long enough for
people to transition to it.

I also have plans to more formalize the peer API and removing batch()
makes that work easier.

I'm not convinced the current client-side API around batching is
great. But it's the best we have at the moment.

.. api:: remove peer.batch()

Replace with peer.iterbatch().

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Revision Contents
Changeset List

PathPackages
Mmercurial/peer.py (13 lines)
Mmercurial/wireproto.py (38 lines)
Mtests/test-wireproto.py (6 lines)

Diff 735

mercurial/peer.py

class iterbatcher(batcher): class iterbatcher(batcher):
def submit(self): def submit(self):
raise NotImplementedError() raise NotImplementedError()
def results(self): def results(self):
raise NotImplementedError() raise NotImplementedError()
class localbatch(batcher):
'''performs the queued calls directly'''
def __init__(self, local):
batcher.__init__(self)
self.local = local
def submit(self):
for name, args, opts, resref in self.calls:
resref.set(getattr(self.local, name)(*args, **opts))
class localiterbatcher(iterbatcher): class localiterbatcher(iterbatcher):
def __init__(self, local): def __init__(self, local):
super(iterbatcher, self).__init__() super(iterbatcher, self).__init__()
self.local = local self.local = local
def submit(self): def submit(self):
# submit for a local iter batcher is a noop # submit for a local iter batcher is a noop
pass pass
return encargsorres # a local result in this case return encargsorres # a local result in this case
self = args[0] self = args[0]
encresref.set(self._submitone(f.func_name, encargsorres)) encresref.set(self._submitone(f.func_name, encargsorres))
return next(batchable) return next(batchable)
setattr(plain, 'batchable', f) setattr(plain, 'batchable', f)
return plain return plain
class peerrepository(object): class peerrepository(object):
def batch(self):
return localbatch(self)
def iterbatch(self): def iterbatch(self):
"""Batch requests but allow iterating over the results. """Batch requests but allow iterating over the results.
This is to allow interleaving responses with things like This is to allow interleaving responses with things like
progress updates for clients. progress updates for clients.
""" """
return localiterbatcher(self) return localiterbatcher(self)

mercurial/wireproto.py

# wireproto.py - generic wire protocol support functions # wireproto.py - generic wire protocol support functions
# #
# 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 itertools
import os import os
import tempfile import tempfile
from .i18n import _ from .i18n import _
from .node import ( from .node import (
bin, bin,
hex, hex,
nullid, nullid,
# #
# left commented here on purpose # left commented here on purpose
# #
#def restore(self): #def restore(self):
# """reinstall previous stdout and stderr and return intercepted stdout # """reinstall previous stdout and stderr and return intercepted stdout
# """ # """
# raise NotImplementedError() # raise NotImplementedError()
class remotebatch(peer.batcher):
'''batches the queued calls; uses as few roundtrips as possible'''
def __init__(self, remote):
'''remote must support _submitbatch(encbatch) and
_submitone(op, encargs)'''
peer.batcher.__init__(self)
self.remote = remote
def submit(self):
req, rsp = [], []
for name, args, opts, resref in self.calls:
mtd = getattr(self.remote, name)
batchablefn = getattr(mtd, 'batchable', None)
if batchablefn is not None:
batchable = batchablefn(mtd.im_self, *args, **opts)
encargsorres, encresref = next(batchable)
assert encresref
req.append((name, encargsorres,))
rsp.append((batchable, encresref, resref,))
else:
if req:
self._submitreq(req, rsp)
req, rsp = [], []
resref.set(mtd(*args, **opts))
if req:
self._submitreq(req, rsp)
def _submitreq(self, req, rsp):
encresults = self.remote._submitbatch(req)
for encres, r in zip(encresults, rsp):
batchable, encresref, resref = r
encresref.set(encres)
resref.set(next(batchable))
class remoteiterbatcher(peer.iterbatcher): class remoteiterbatcher(peer.iterbatcher):
def __init__(self, remote): def __init__(self, remote):
super(remoteiterbatcher, self).__init__() super(remoteiterbatcher, self).__init__()
self._remote = remote self._remote = remote
def __getattr__(self, name): def __getattr__(self, name):
# Validate this method is batchable, since submit() only supports # Validate this method is batchable, since submit() only supports
# batchable methods. # batchable methods.
class wirepeer(peer.peerrepository): class wirepeer(peer.peerrepository):
"""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 batch(self):
if self.capable('batch'):
return remotebatch(self)
else:
return peer.localbatch(self)
def _submitbatch(self, req): def _submitbatch(self, req):
"""run batch request <req> on the server """run batch request <req> on the server
Returns an iterator of the raw responses from the server. Returns an iterator of the raw responses from the server.
""" """
rsp = self._callstream("batch", cmds=encodebatchcmds(req)) rsp = self._callstream("batch", cmds=encodebatchcmds(req))
chunk = rsp.read(1024) chunk = rsp.read(1024)
work = [chunk] work = [chunk]

tests/test-wireproto.py

return mangle(repo.greet(unmangle(name))) return mangle(repo.greet(unmangle(name)))
wireproto.commands['greet'] = (greet, 'name',) wireproto.commands['greet'] = (greet, 'name',)
srv = serverrepo() srv = serverrepo()
clt = clientpeer(srv) clt = clientpeer(srv)
print(clt.greet("Foobar")) print(clt.greet("Foobar"))
b = clt.batch() b = clt.iterbatch()
fs = [b.greet(s) for s in ["Fo, =;:<o", "Bar"]] map(b.greet, ('Fo, =;:<o', 'Bar'))
b.submit() b.submit()
print([f.value for f in fs]) print([r for r in b.results()])