This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2082

wireproto: use maybecapturestdio() for push responses (API)
ClosedPublic

Authored by indygreg on Feb 7 2018, 7:26 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGcaca3ac2ac04: wireproto: use maybecapturestdio() for push responses (API)
Summary

The "pushres" and "pusherr" response types currently call
proto.restore() in the HTTP protocol. This completes the pairing
with proto.redirect() that occurs in the @wireprotocommand
functions. (But since the SSH protocol has a no-op redirect(),
it doesn't bother calling restore() because it would also be
a no-op.)

Having the disconnect between these paired calls is very confusing.
Knowing that you must use proto.redirect() if returning a "pushres"
or a "pusherr" is even wonkier.

We replace this confusing code with our new context manager for
[maybe] capturing output.

The "pushres" and "pusherr" types have gained an "output" argument
to their constructor and an attribute to hold captured data. The
HTTP protocol now retrieves output from these objects.

.. api::

``wireproto.pushres`` and ``wireproto.pusherr`` now explicitly
track stdio output.

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.Feb 7 2018, 7:26 PM
indygreg edited the summary of this revision. (Show Details)Feb 7 2018, 11:32 PM
indygreg updated this revision to Diff 5334.
indygreg updated this revision to Diff 5345.Feb 8 2018, 12:20 AM
durin42 accepted this revision.Feb 12 2018, 4:22 PM
This revision is now accepted and ready to land.Feb 12 2018, 4:22 PM
Closed by commit rHGcaca3ac2ac04: wireproto: use maybecapturestdio() for push responses (API) (authored by indygreg). · Explain WhyFeb 12 2018, 4:33 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 5527

hgext/largefiles/proto.py

# these will all be replaced by largefiles.uisetup # these will all be replaced by largefiles.uisetup
ssholdcallstream = None ssholdcallstream = None
httpoldcallstream = None httpoldcallstream = None
def putlfile(repo, proto, sha): def putlfile(repo, proto, sha):
'''Server command for putting a largefile into a repository's local store '''Server command for putting a largefile into a repository's local store
and into the user cache.''' and into the user cache.'''
proto.redirect() with proto.mayberedirectstdio() as output:
path = lfutil.storepath(repo, sha) path = lfutil.storepath(repo, sha)
util.makedirs(os.path.dirname(path)) util.makedirs(os.path.dirname(path))
tmpfp = util.atomictempfile(path, createmode=repo.store.createmode) tmpfp = util.atomictempfile(path, createmode=repo.store.createmode)
try: try:
proto.getfile(tmpfp) proto.getfile(tmpfp)
tmpfp._fp.seek(0) tmpfp._fp.seek(0)
if sha != lfutil.hexsha1(tmpfp._fp): if sha != lfutil.hexsha1(tmpfp._fp):
raise IOError(0, _('largefile contents do not match hash')) raise IOError(0, _('largefile contents do not match hash'))
tmpfp.close() tmpfp.close()
lfutil.linktousercache(repo, sha) lfutil.linktousercache(repo, sha)
except IOError as e: except IOError as e:
repo.ui.warn(_('largefiles: failed to put %s into store: %s\n') % repo.ui.warn(_('largefiles: failed to put %s into store: %s\n') %
(sha, e.strerror)) (sha, e.strerror))
return wireproto.pushres(1) return wireproto.pushres(1, output.getvalue() if output else '')
finally: finally:
tmpfp.discard() tmpfp.discard()
return wireproto.pushres(0) return wireproto.pushres(0, output.getvalue() if output else '')
def getlfile(repo, proto, sha): def getlfile(repo, proto, sha):
'''Server command for retrieving a largefile from the repository-local '''Server command for retrieving a largefile from the repository-local
cache or user cache.''' cache or user cache.'''
filename = lfutil.findfile(repo, sha) filename = lfutil.findfile(repo, sha)
if not filename: if not filename:
raise error.Abort(_('requested largefile %s not present in cache') raise error.Abort(_('requested largefile %s not present in cache')
% sha) % sha)

mercurial/wireproto.py

def __init__(self, gen=None): def __init__(self, gen=None):
self.gen = gen self.gen = gen
class pushres(object): class pushres(object):
"""wireproto reply: success with simple integer return """wireproto reply: success with simple integer return
The call was successful and returned an integer contained in `self.res`. The call was successful and returned an integer contained in `self.res`.
""" """
def __init__(self, res): def __init__(self, res, output):
self.res = res self.res = res
self.output = output
class pusherr(object): class pusherr(object):
"""wireproto reply: failure """wireproto reply: failure
The call failed. The `self.res` attribute contains the error message. The call failed. The `self.res` attribute contains the error message.
""" """
def __init__(self, res): def __init__(self, res, output):
self.res = res self.res = res
self.output = output
class ooberror(object): class ooberror(object):
"""wireproto reply: failure of a batch of operation """wireproto reply: failure of a batch of operation
Something failed during a batch call. The error message is stored in Something failed during a batch call. The error message is stored in
`self.message`. `self.message`.
""" """
def __init__(self, message): def __init__(self, message):
it is serving. Client checks to see if it understands the format. it is serving. Client checks to see if it understands the format.
''' '''
return streamres_legacy(streamclone.generatev1wireproto(repo)) return streamres_legacy(streamclone.generatev1wireproto(repo))
@wireprotocommand('unbundle', 'heads') @wireprotocommand('unbundle', 'heads')
def unbundle(repo, proto, heads): def unbundle(repo, proto, heads):
their_heads = decodelist(heads) their_heads = decodelist(heads)
with proto.mayberedirectstdio() as output:
try: try:
proto.redirect()
exchange.check_heads(repo, their_heads, 'preparing changes') exchange.check_heads(repo, their_heads, 'preparing changes')
# write bundle data to temporary file because it can be big # write bundle data to temporary file because it can be big
fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-') fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-')
fp = os.fdopen(fd, pycompat.sysstr('wb+')) fp = os.fdopen(fd, pycompat.sysstr('wb+'))
r = 0 r = 0
try: try:
proto.getfile(fp) proto.getfile(fp)
fp.seek(0) fp.seek(0)
gen = exchange.readbundle(repo.ui, fp, None) gen = exchange.readbundle(repo.ui, fp, None)
if (isinstance(gen, changegroupmod.cg1unpacker) if (isinstance(gen, changegroupmod.cg1unpacker)
and not bundle1allowed(repo, 'push')): and not bundle1allowed(repo, 'push')):
if proto.name == 'http': if proto.name == 'http':
# need to special case http because stderr do not get to # need to special case http because stderr do not get to
# the http client on failed push so we need to abuse some # the http client on failed push so we need to abuse
# other error type to make sure the message get to the # some other error type to make sure the message get to
# user. # the user.
return ooberror(bundle2required) return ooberror(bundle2required)
raise error.Abort(bundle2requiredmain, raise error.Abort(bundle2requiredmain,
hint=bundle2requiredhint) hint=bundle2requiredhint)
r = exchange.unbundle(repo, gen, their_heads, 'serve', r = exchange.unbundle(repo, gen, their_heads, 'serve',
proto._client()) proto._client())
if util.safehasattr(r, 'addpart'): if util.safehasattr(r, 'addpart'):
# The return looks streamable, we are in the bundle2 case and # The return looks streamable, we are in the bundle2 case
# should return a stream. # and should return a stream.
return streamres_legacy(gen=r.getchunks()) return streamres_legacy(gen=r.getchunks())
return pushres(r) return pushres(r, output.getvalue() if output else '')
finally: finally:
fp.close() fp.close()
os.unlink(tempname) os.unlink(tempname)
except (error.BundleValueError, error.Abort, error.PushRaced) as exc: except (error.BundleValueError, error.Abort, error.PushRaced) as exc:
# handle non-bundle2 case first # handle non-bundle2 case first
if not getattr(exc, 'duringunbundle2', False): if not getattr(exc, 'duringunbundle2', False):
try: try:
raise raise
except error.Abort: except error.Abort:
# The old code we moved used util.stderr directly. # The old code we moved used util.stderr directly.
# We did not change it to minimise code change. # We did not change it to minimise code change.
# This need to be moved to something proper. # This need to be moved to something proper.
# Feel free to do it. # Feel free to do it.
util.stderr.write("abort: %s\n" % exc) util.stderr.write("abort: %s\n" % exc)
if exc.hint is not None: if exc.hint is not None:
util.stderr.write("(%s)\n" % exc.hint) util.stderr.write("(%s)\n" % exc.hint)
return pushres(0) return pushres(0, output.getvalue() if output else '')
except error.PushRaced: except error.PushRaced:
return pusherr(str(exc)) return pusherr(str(exc),
output.getvalue() if output else '')
bundler = bundle2.bundle20(repo.ui) bundler = bundle2.bundle20(repo.ui)
for out in getattr(exc, '_bundle2salvagedoutput', ()): for out in getattr(exc, '_bundle2salvagedoutput', ()):
bundler.addpart(out) bundler.addpart(out)
try: try:
try: try:
raise raise
except error.PushkeyFailed as exc: except error.PushkeyFailed as exc:
# check client caps # check client caps
remotecaps = getattr(exc, '_replycaps', None) remotecaps = getattr(exc, '_replycaps', None)
if (remotecaps is not None if (remotecaps is not None
and 'pushkey' not in remotecaps.get('error', ())): and 'pushkey' not in remotecaps.get('error', ())):
# no support remote side, fallback to Abort handler. # no support remote side, fallback to Abort handler.
raise raise
part = bundler.newpart('error:pushkey') part = bundler.newpart('error:pushkey')
part.addparam('in-reply-to', exc.partid) part.addparam('in-reply-to', exc.partid)
if exc.namespace is not None: if exc.namespace is not None:
part.addparam('namespace', exc.namespace, mandatory=False) part.addparam('namespace', exc.namespace,
mandatory=False)
if exc.key is not None: if exc.key is not None:
part.addparam('key', exc.key, mandatory=False) part.addparam('key', exc.key, mandatory=False)
if exc.new is not None: if exc.new is not None:
part.addparam('new', exc.new, mandatory=False) part.addparam('new', exc.new, mandatory=False)
if exc.old is not None: if exc.old is not None:
part.addparam('old', exc.old, mandatory=False) part.addparam('old', exc.old, mandatory=False)
if exc.ret is not None: if exc.ret is not None:
part.addparam('ret', exc.ret, mandatory=False) part.addparam('ret', exc.ret, mandatory=False)
except error.BundleValueError as exc: except error.BundleValueError as exc:
errpart = bundler.newpart('error:unsupportedcontent') errpart = bundler.newpart('error:unsupportedcontent')
if exc.parttype is not None: if exc.parttype is not None:
errpart.addparam('parttype', exc.parttype) errpart.addparam('parttype', exc.parttype)
if exc.params: if exc.params:
errpart.addparam('params', '\0'.join(exc.params)) errpart.addparam('params', '\0'.join(exc.params))
except error.Abort as exc: except error.Abort as exc:
manargs = [('message', str(exc))] manargs = [('message', str(exc))]
advargs = [] advargs = []
if exc.hint is not None: if exc.hint is not None:
advargs.append(('hint', exc.hint)) advargs.append(('hint', exc.hint))
bundler.addpart(bundle2.bundlepart('error:abort', bundler.addpart(bundle2.bundlepart('error:abort',
manargs, advargs)) manargs, advargs))
except error.PushRaced as exc: except error.PushRaced as exc:
bundler.newpart('error:pushraced', [('message', str(exc))]) bundler.newpart('error:pushraced', [('message', str(exc))])
return streamres_legacy(gen=bundler.getchunks()) return streamres_legacy(gen=bundler.getchunks())

mercurial/wireprotoserver.py

gen = engine.compressstream(gen, engineopts) gen = engine.compressstream(gen, engineopts)
if mediatype == HGTYPE2: if mediatype == HGTYPE2:
gen = genversion2(gen, engine, engineopts) gen = genversion2(gen, engine, engineopts)
req.respond(HTTP_OK, mediatype) req.respond(HTTP_OK, mediatype)
return gen return gen
elif isinstance(rsp, wireproto.pushres): elif isinstance(rsp, wireproto.pushres):
val = proto.restore() rsp = '%d\n%s' % (rsp.res, rsp.output)
rsp = '%d\n%s' % (rsp.res, val)
req.respond(HTTP_OK, HGTYPE, body=rsp) req.respond(HTTP_OK, HGTYPE, body=rsp)
return [] return []
elif isinstance(rsp, wireproto.pusherr): elif isinstance(rsp, wireproto.pusherr):
# This is the httplib workaround documented in _handlehttperror(). # This is the httplib workaround documented in _handlehttperror().
req.drain() req.drain()
proto.restore()
rsp = '0\n%s\n' % rsp.res rsp = '0\n%s\n' % rsp.res
req.respond(HTTP_OK, HGTYPE, body=rsp) req.respond(HTTP_OK, HGTYPE, body=rsp)
return [] return []
elif isinstance(rsp, wireproto.ooberror): elif isinstance(rsp, wireproto.ooberror):
rsp = rsp.message rsp = rsp.message
req.respond(HTTP_OK, HGERRTYPE, body=rsp) req.respond(HTTP_OK, HGERRTYPE, body=rsp)
return [] return []
raise error.ProgrammingError('hgweb.protocol internal failure', rsp) raise error.ProgrammingError('hgweb.protocol internal failure', rsp)