This is an archive of the discontinued Mercurial Phabricator instance.

Differential D11211

wireprotov1peer: simplify the way batchable rpcs are defined
ClosedPublic

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

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHGcdad6560e832: wireprotov1peer: simplify the way batchable rpcs are defined
rHG1364337a4613: wireprotov1peer: simplify the way batchable rpcs are defined
Summary

The scheme with futures/generator is confusing due to the way
communication is done by side effects, especially with two different
"future" objects. Just returning a request and a function to read the
response is easier to understand.

There are tests failures with the largefiles extension due to it
aliasing one rpc to another one, which gets fixed in the next commit.

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

Alphare added a subscriber: Alphare.Jul 26 2021, 4:51 AM

Thanks for the patches, they look like a good idea, however we're in a freeze right now so we'll take a look in August.

Alphare accepted this revision.Aug 24 2021, 10:06 AM
Alphare added inline comments.
mercurial/wireprotov1peer.py
47–49

nit: s/wll/will/

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

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

Alphare added a comment.Aug 24 2021, 11:16 AM

Sorry for the noise, I should have skipped the CI refresh for the review queue, I'll take your updated patches.

Revision Contents
Changeset List

PathPackages
Mmercurial/wireprotov1peer.py (59 lines)

Diff 29998

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(f): def batchable_new_style(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),)]
# Create future for injection of encoded result: # Return it, along with a function that will receive the result
encoded_res_future = future() # from the batched request.
# Return encoded arguments and future: return encoded_args, decode
AlphareUnsubmitted
Not Done

nit: s/wll/will/

Alphare: nit: s/wll/will/
yield encoded_args, encoded_res_future
# Assuming the future to be filled with the result from the batched
# request now. Decode it:
yield decode(encoded_res_future.value)
The decorator returns a function which wraps this coroutine as a plain The decorator returns a function which wraps this coroutine as a plain
method, but adds the original method as an attribute called "batchable", method, but adds the original method as an attribute called "batchable",
which is used by remotebatch to split the call into separate encoding and which is used by remotebatch to split the call into separate encoding and
decoding phases. decoding phases.
""" """
def plain(*args, **opts): def plain(*args, **opts):
batchable = f(*args, **opts) encoded_args_or_res, decode = f(*args, **opts)
encoded_args_or_res, encoded_res_future = next(batchable) if not decode:
if not encoded_res_future:
return encoded_args_or_res # a local result in this case return encoded_args_or_res # a local result in this case
self = args[0] self = args[0]
cmd = pycompat.bytesurl(f.__name__) # ensure cmd is ascii bytestr cmd = pycompat.bytesurl(f.__name__) # ensure cmd is ascii bytestr
encoded_res_future.set(self._submitone(cmd, encoded_args_or_res)) encoded_res = self._submitone(cmd, encoded_args_or_res)
return next(batchable) 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): 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, b'value'): if util.safehasattr(self, b'value'):
raise error.RepoError(b"future is already set") raise error.RepoError(b"future is already set")
self.value = value self.value = value
states = [] states = []
for command, args, fn, f in calls: for command, args, fn, f in calls:
# Future was cancelled. Ignore it. # Future was cancelled. Ignore it.
if not f.set_running_or_notify_cancel(): if not f.set_running_or_notify_cancel():
continue continue
try: try:
batchable = fn.batchable( encoded_args_or_res, decode = fn.batchable(
fn.__self__, **pycompat.strkwargs(args) fn.__self__, **pycompat.strkwargs(args)
) )
except Exception: except Exception:
pycompat.future_set_exception_info(f, sys.exc_info()[1:]) pycompat.future_set_exception_info(f, sys.exc_info()[1:])
return return
# Encoded arguments and future holding remote result. if not decode:
try:
encoded_args_or_res, fremote = next(batchable)
except Exception:
pycompat.future_set_exception_info(f, sys.exc_info()[1:])
return
if not fremote:
f.set_result(encoded_args_or_res) f.set_result(encoded_args_or_res)
else: else:
requests.append((command, encoded_args_or_res)) requests.append((command, encoded_args_or_res))
states.append((command, f, batchable, fremote)) states.append((command, f, batchable, decode))
if not requests: if not requests:
return return
# This will emit responses in order they were executed. # This will emit responses in order they were executed.
wireresults = self._peer._submitbatch(requests) wireresults = self._peer._submitbatch(requests)
# The use of a thread pool executor here is a bit weird for something # The use of a thread pool executor here is a bit weird for something
) )
) )
self._futures = None self._futures = None
def _readbatchresponse(self, states, wireresults): def _readbatchresponse(self, states, wireresults):
# Executes in a thread to read data off the wire. # Executes in a thread to read data off the wire.
for command, f, batchable, fremote in states: for command, f, batchable, decode in states:
# Grab raw result off the wire and teach the internal future # Grab raw result off the wire and teach the internal future
# about it. # about it.
try: try:
remoteresult = next(wireresults) remoteresult = next(wireresults)
except StopIteration: except StopIteration:
# This can happen in particular because next(batchable) # This can happen in particular because next(batchable)
# in the previous iteration can call peer._abort, which # in the previous iteration can call peer._abort, which
# may close the peer. # may close the peer.
f.set_exception( f.set_exception(
error.ResponseError( error.ResponseError(
_(b'unfulfilled batch command response'), None _(b'unfulfilled batch command response'), None
) )
) )
else: else:
fremote.set(remoteresult)
# And ask the coroutine to decode that value.
try: try:
result = next(batchable) result = decode(remoteresult)
except Exception: except Exception:
pycompat.future_set_exception_info(f, sys.exc_info()[1:]) pycompat.future_set_exception_info(f, sys.exc_info()[1:])
else: else:
f.set_result(result) f.set_result(result)
@interfaceutil.implementer( @interfaceutil.implementer(
repository.ipeercommands, repository.ipeerlegacycommands repository.ipeercommands, repository.ipeerlegacycommands