This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3269

wireproto: implement batching on peer executor interface
ClosedPublic

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

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG2f626233859b: wireproto: implement batching on peer executor interface
Summary

This is a bit more complicated than non-batch requests because we
need to buffer sends until the last request arrives *and* we need
to support resolving futures as data arrives from the remote.

In a classical concurrent.futures executor model, the future
"starts" as soon as it is submitted. However, we have nothing to
start until the last command is submitted.

If we did nothing, calling result() would deadlock, since the future
hasn't "started." So in the case where we queue the command, we return
a special future type whose result() will trigger sendcommands().
This eliminates the deadlock potential. It also serves as a check
against callers who may be calling result() prematurely, as it will
prevent any subsequent callcommands() from working. This behavior
is slightly annoying and a bit restrictive. But it's the world
that half duplex connections forces on us.

In order to support streaming responses, we were previously using
a generator. But with a futures-based API, we're using futures
and not generators. So in order to get streaming, we need a
background thread to read data from the server.

The approach taken in this patch is to leverage the ThreadPoolExecutor
from concurrent.futures for managing a background thread. We create
an executor and future that resolves when all response data is
processed (or an error occurs). When exiting the context manager,
we wait on that background reading before returning.

I was hoping we could manually spin up a threading.Thread and this
would be simple. But I ran into a few deadlocks when implementing.
After looking at the source code to concurrent.futures, I figured
it would just be easier to use a ThreadPoolExecutor than implement
all the code needed to manually manage a thread.

To prove this works, a use of the batch API in discovery has been
updated.

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 edited the summary of this revision. (Show Details)Apr 13 2018, 3:14 PM
indygreg updated this revision to Diff 8122.
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 rHG2f626233859b: wireproto: implement batching on peer executor interface (authored by indygreg). · Explain WhyApr 13 2018, 5:45 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/setdiscovery.py (13 lines)
Mmercurial/wireprotov1peer.py (148 lines)

Diff 8166

mercurial/setdiscovery.py

# early exit if we know all the specified remote heads already # early exit if we know all the specified remote heads already
ui.debug("query 1; heads\n") ui.debug("query 1; heads\n")
roundtrips += 1 roundtrips += 1
ownheads = dag.heads() ownheads = dag.heads()
sample = _limitsample(ownheads, initialsamplesize) sample = _limitsample(ownheads, initialsamplesize)
# indices between sample and externalized version must match # indices between sample and externalized version must match
sample = list(sample) sample = list(sample)
batch = remote.iterbatch()
batch.heads() with remote.commandexecutor() as e:
batch.known(dag.externalizeall(sample)) fheads = e.callcommand('heads', {})
batch.submit() fknown = e.callcommand('known', {
srvheadhashes, yesno = batch.results() 'nodes': dag.externalizeall(sample),
})
srvheadhashes, yesno = fheads.result(), fknown.result()
if cl.tip() == nullid: if cl.tip() == nullid:
if srvheadhashes != [nullid]: if srvheadhashes != [nullid]:
return [nullid], True, srvheadhashes return [nullid], True, srvheadhashes
return [nullid], False, [] return [nullid], False, []
# start actual discovery (we note this before the next "if" for # start actual discovery (we note this before the next "if" for
# compatibility reasons) # compatibility reasons)

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 import sys
import weakref
from .i18n import _ from .i18n import _
from .node import ( from .node import (
bin, bin,
) )
from .thirdparty.zope import ( from .thirdparty.zope import (
interface as zi, interface as zi,
) )
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)
class unsentfuture(pycompat.futures.Future):
"""A Future variation to represent an unsent command.
Because we buffer commands and don't submit them immediately, calling
``result()`` on an unsent future could deadlock. Futures for buffered
commands are represented by this type, which wraps ``result()`` to
call ``sendcommands()``.
"""
def result(self, timeout=None):
if self.done():
return pycompat.futures.Future.result(self, timeout)
self._peerexecutor.sendcommands()
# This looks like it will infinitely recurse. However,
# sendcommands() should modify __class__. This call serves as a check
# on that.
return self.result(timeout)
@zi.implementer(repository.ipeercommandexecutor) @zi.implementer(repository.ipeercommandexecutor)
class peerexecutor(object): class peerexecutor(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
self._calls = [] self._calls = []
self._futures = weakref.WeakSet()
self._responseexecutor = None
self._responsef = None
def __enter__(self): def __enter__(self):
return self return self
def __exit__(self, exctype, excvalee, exctb): def __exit__(self, exctype, excvalee, exctb):
self.close() self.close()
def callcommand(self, command, args): def callcommand(self, command, args):
if not fn: if not fn:
raise error.ProgrammingError( raise error.ProgrammingError(
'cannot call command %s: method of same name not available ' 'cannot call command %s: method of same name not available '
'on peer' % command) 'on peer' % command)
# Commands are either batchable or they aren't. If a command # Commands are either batchable or they aren't. If a command
# isn't batchable, we send it immediately because the executor # isn't batchable, we send it immediately because the executor
# can no longer accept new commands after a non-batchable command. # can no longer accept new commands after a non-batchable command.
# If a command is batchable, we queue it for later. # If a command is batchable, we queue it for later. But we have
# to account for the case of a non-batchable command arriving after
# a batchable one and refuse to service it.
def addcall():
f = pycompat.futures.Future()
self._futures.add(f)
self._calls.append((command, args, fn, f))
return f
if getattr(fn, 'batchable', False): if getattr(fn, 'batchable', False):
pass f = addcall()
# But since we don't issue it immediately, we wrap its result()
# to trigger sending so we avoid deadlocks.
f.__class__ = unsentfuture
f._peerexecutor = self
else: else:
if self._calls: if self._calls:
raise error.ProgrammingError( raise error.ProgrammingError(
'%s is not batchable and cannot be called on a command ' '%s is not batchable and cannot be called on a command '
'executor along with other commands' % command) 'executor along with other commands' % command)
# We don't support batching yet. So resolve it immediately. f = addcall()
f = pycompat.futures.Future()
self._calls.append((command, args, fn, f)) # Non-batchable commands can never coexist with another command
# in this executor. So send the command immediately.
self.sendcommands() self.sendcommands()
return f return f
def sendcommands(self): def sendcommands(self):
if self._sent: if self._sent:
return return
if not self._calls: if not self._calls:
return return
self._sent = True self._sent = True
# Unhack any future types so caller seens a clean type and to break
# cycle between us and futures.
for f in self._futures:
if isinstance(f, unsentfuture):
f.__class__ = pycompat.futures.Future
f._peerexecutor = None
calls = self._calls calls = self._calls
# Mainly to destroy references to futures. # Mainly to destroy references to futures.
self._calls = None self._calls = None
# Simple case of a single command. We call it synchronously.
if len(calls) == 1: if len(calls) == 1:
command, args, fn, f = calls[0] command, args, fn, f = calls[0]
# 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():
return return
try: try:
result = fn(**pycompat.strkwargs(args)) result = fn(**pycompat.strkwargs(args))
except Exception: except Exception:
f.set_exception_info(*sys.exc_info()[1:]) f.set_exception_info(*sys.exc_info()[1:])
else: else:
f.set_result(result) f.set_result(result)
return return
raise error.ProgrammingError('support for multiple commands not ' # Batch commands are a bit harder. First, we have to deal with the
'yet implemented') # @batchable coroutine. That's a bit annoying. Furthermore, we also
# need to preserve streaming. i.e. it should be possible for the
# futures to resolve as data is coming in off the wire without having
# to wait for the final byte of the final response. We do this by
# spinning up a thread to read the responses.
requests = []
states = []
for command, args, fn, f in calls:
# Future was cancelled. Ignore it.
if not f.set_running_or_notify_cancel():
continue
try:
batchable = fn.batchable(fn.__self__,
**pycompat.strkwargs(args))
except Exception:
f.set_exception_info(*sys.exc_info()[1:])
return
# Encoded arguments and future holding remote result.
try:
encodedargs, fremote = next(batchable)
except Exception:
f.set_exception_info(*sys.exc_info()[1:])
return
requests.append((command, encodedargs))
states.append((command, f, batchable, fremote))
if not requests:
return
# This will emit responses in order they were executed.
wireresults = self._peer._submitbatch(requests)
# The use of a thread pool executor here is a bit weird for something
# that only spins up a single thread. However, thread management is
# hard and it is easy to encounter race conditions, deadlocks, etc.
# concurrent.futures already solves these problems and its thread pool
# executor has minimal overhead. So we use it.
self._responseexecutor = pycompat.futures.ThreadPoolExecutor(1)
self._responsef = self._responseexecutor.submit(self._readbatchresponse,
states, wireresults)
def close(self): def close(self):
self.sendcommands() self.sendcommands()
if self._closed:
return
self._closed = True self._closed = True
if not self._responsef:
return
# We need to wait on our in-flight response and then shut down the
# executor once we have a result.
try:
self._responsef.result()
finally:
self._responseexecutor.shutdown(wait=True)
self._responsef = None
self._responseexecutor = None
# If any of our futures are still in progress, mark them as
# errored. Otherwise a result() could wait indefinitely.
for f in self._futures:
if not f.done():
f.set_exception(error.ResponseError(
_('unfulfilled batch command response')))
self._futures = None
def _readbatchresponse(self, states, wireresults):
# Executes in a thread to read data off the wire.
for command, f, batchable, fremote in states:
# Grab raw result off the wire and teach the internal future
# about it.
remoteresult = next(wireresults)
fremote.set(remoteresult)
# And ask the coroutine to decode that value.
try:
result = next(batchable)
except Exception:
f.set_exception_info(*sys.exc_info()[1:])
else:
f.set_result(result)
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.
""" """