This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2470

wireproto: allow direct stream processing for unbundle
ClosedPublic

Authored by joerg.sonnenberger on Feb 26 2018, 8:38 PM.

Details

Reviewers
lothiraldan
indygreg
Group Reviewers
hg-reviewers
Commits
rHG2d965bfeb8f6: wireproto: allow direct stream processing for unbundle
Summary

Introduce a new option server.streamunbundle which starts a transaction
immediately to apply a bundle instead of writing it to a temporary file
first. This side steps the need for a large tmp directory at the cost of
preventing concurrent pushes. This is a reasonable trade-off for many
setups as concurrent pushes for the main branch at least are disallowed
anyway. The option defaults to off to preserve existing behavior.

Change the wireproto interface to provide a generator for reading the
payload and make callers responsible for consuming all data.

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

lothiraldan added a subscriber: lothiraldan.Feb 27 2018, 5:56 AM
lothiraldan added inline comments.
mercurial/wireproto.py
1090

It's not perfectly clear why do we need to do this loop? Is it to consume the end of the payload before ending? Would it be possible to add a comment here?

joerg.sonnenberger updated this revision to Diff 6177.Feb 27 2018, 5:25 PM
joerg.sonnenberger updated this revision to Diff 6178.
lothiraldan accepted this revision.Feb 28 2018, 4:01 AM
indygreg accepted this revision.Apr 6 2018, 7:07 PM
indygreg added a subscriber: indygreg.

LGTM. Thanks for the feature!

This revision is now accepted and ready to land.Apr 6 2018, 7:07 PM
Closed by commit rHG2d965bfeb8f6: wireproto: allow direct stream processing for unbundle (authored by joerg.sonnenberger). · Explain WhyApr 6 2018, 7:08 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7817

hgext/largefiles/proto.py

'''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.'''
with proto.mayberedirectstdio() as output: 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.forwardpayload(tmpfp) for p in proto.getpayload():
tmpfp.write(p)
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))

mercurial/configitems.py

default=list, default=list,
) )
coreconfigitem('server', 'concurrent-push-mode', coreconfigitem('server', 'concurrent-push-mode',
default='strict', default='strict',
) )
coreconfigitem('server', 'disablefullbundle', coreconfigitem('server', 'disablefullbundle',
default=False, default=False,
) )
coreconfigitem('server', 'streamunbundle',
default=False,
)
coreconfigitem('server', 'maxhttpheaderlen', coreconfigitem('server', 'maxhttpheaderlen',
default=1024, default=1024,
) )
coreconfigitem('server', 'preferuncompressed', coreconfigitem('server', 'preferuncompressed',
default=False, default=False,
) )
coreconfigitem('server', 'uncompressed', coreconfigitem('server', 'uncompressed',
default=True, default=True,

mercurial/help/config.txt

protocol. (default: False) protocol. (default: False)
``disablefullbundle`` ``disablefullbundle``
When set, servers will refuse attempts to do pull-based clones. When set, servers will refuse attempts to do pull-based clones.
If this option is set, ``preferuncompressed`` and/or clone bundles If this option is set, ``preferuncompressed`` and/or clone bundles
are highly recommended. Partial clones will still be allowed. are highly recommended. Partial clones will still be allowed.
(default: False) (default: False)
``streamunbundle``
When set, servers will apply data sent from the client directly,
otherwise it will be written to a temporary file first. This option
effectively prevents concurrent pushes.
``concurrent-push-mode`` ``concurrent-push-mode``
Level of allowed race condition between two pushing clients. Level of allowed race condition between two pushing clients.
- 'strict': push is abort if another client touched the repository - 'strict': push is abort if another client touched the repository
while the push was preparing. (default) while the push was preparing. (default)
- 'check-related': push is only aborted if it affects head that got also - 'check-related': push is only aborted if it affects head that got also
affected while the push was preparing. affected while the push was preparing.

mercurial/wireproto.py

@wireprotocommand('unbundle', 'heads', permission='push') @wireprotocommand('unbundle', 'heads', permission='push')
def unbundle(repo, proto, heads): def unbundle(repo, proto, heads):
their_heads = decodelist(heads) their_heads = decodelist(heads)
with proto.mayberedirectstdio() as output: with proto.mayberedirectstdio() as output:
try: try:
exchange.check_heads(repo, their_heads, 'preparing changes') exchange.check_heads(repo, their_heads, 'preparing changes')
cleanup = lambda: None
# write bundle data to temporary file because it can be big try:
payload = proto.getpayload()
if repo.ui.configbool('server', 'streamunbundle'):
def cleanup():
# Ensure that the full payload is consumed, so
lothiraldanUnsubmitted
Not Done

It's not perfectly clear why do we need to do this loop? Is it to consume the end of the payload before ending? Would it be possible to add a comment here?

lothiraldan: It's not perfectly clear why do we need to do this loop? Is it to consume the end of the…
# that the connection doesn't contain trailing garbage.
for p in payload:
pass
fp = util.chunkbuffer(payload)
else:
# write bundle data to temporary file as it can be big
fp, tempname = None, None
def cleanup():
if fp:
fp.close()
if tempname:
os.unlink(tempname)
fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-') fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-')
fp = os.fdopen(fd, r'wb+') repo.ui.debug('redirecting incoming bundle to %s\n' %
tempname)
fp = os.fdopen(fd, pycompat.sysstr('wb+'))
r = 0 r = 0
try: for p in payload:
proto.forwardpayload(fp) fp.write(p)
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-v1': if proto.name == 'http-v1':
# 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 # the http client on failed push so we need to abuse
# some other error type to make sure the message get to # some other error type to make sure the message get to
# the user. # the user.
return wireprototypes.ooberror(bundle2required) return wireprototypes.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 # The return looks streamable, we are in the bundle2 case
# and should return a stream. # and should return a stream.
return wireprototypes.streamreslegacy(gen=r.getchunks()) return wireprototypes.streamreslegacy(gen=r.getchunks())
return wireprototypes.pushres( return wireprototypes.pushres(
r, output.getvalue() if output else '') r, output.getvalue() if output else '')
finally: finally:
fp.close() cleanup()
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 procutil.stderr directly. # The old code we moved used procutil.stderr directly.

mercurial/wireprotoserver.py

return args return args
def getprotocaps(self): def getprotocaps(self):
if self._protocaps is None: if self._protocaps is None:
value = decodevaluefromheaders(self._req, r'X-HgProto') value = decodevaluefromheaders(self._req, r'X-HgProto')
self._protocaps = set(value.split(' ')) self._protocaps = set(value.split(' '))
return self._protocaps return self._protocaps
def forwardpayload(self, fp): def getpayload(self):
# Existing clients *always* send Content-Length. # Existing clients *always* send Content-Length.
length = int(self._req.headers[b'Content-Length']) length = int(self._req.headers[b'Content-Length'])
# If httppostargs is used, we need to read Content-Length # If httppostargs is used, we need to read Content-Length
# minus the amount that was consumed by args. # minus the amount that was consumed by args.
length -= int(self._req.headers.get(b'X-HgArgs-Post', 0)) length -= int(self._req.headers.get(b'X-HgArgs-Post', 0))
for s in util.filechunkiter(self._req.bodyfh, limit=length): return util.filechunkiter(self._req.bodyfh, limit=length)
fp.write(s)
@contextlib.contextmanager @contextlib.contextmanager
def mayberedirectstdio(self): def mayberedirectstdio(self):
oldout = self._ui.fout oldout = self._ui.fout
olderr = self._ui.ferr olderr = self._ui.ferr
out = util.stringio() out = util.stringio()
data[k] = self._args[k] data[k] = self._args[k]
return [data[k] for k in args.split()] return [data[k] for k in args.split()]
def getprotocaps(self): def getprotocaps(self):
# Protocol capabilities are currently not implemented for HTTP V2. # Protocol capabilities are currently not implemented for HTTP V2.
return set() return set()
def forwardpayload(self, fp): def getpayload(self):
raise NotImplementedError raise NotImplementedError
@contextlib.contextmanager @contextlib.contextmanager
def mayberedirectstdio(self): def mayberedirectstdio(self):
raise NotImplementedError raise NotImplementedError
def client(self): def client(self):
raise NotImplementedError raise NotImplementedError
else: else:
val = self._fin.read(int(l)) val = self._fin.read(int(l))
data[arg] = val data[arg] = val
return [data[k] for k in keys] return [data[k] for k in keys]
def getprotocaps(self): def getprotocaps(self):
return self._protocaps return self._protocaps
def forwardpayload(self, fpout): def getpayload(self):
# We initially send an empty response. This tells the client it is # We initially send an empty response. This tells the client it is
# OK to start sending data. If a client sees any other response, it # OK to start sending data. If a client sees any other response, it
# interprets it as an error. # interprets it as an error.
_sshv1respondbytes(self._fout, b'') _sshv1respondbytes(self._fout, b'')
# The file is in the form: # The file is in the form:
# #
# <chunk size>\n<chunk> # <chunk size>\n<chunk>
# ... # ...
# 0\n # 0\n
count = int(self._fin.readline()) count = int(self._fin.readline())
while count: while count:
fpout.write(self._fin.read(count)) yield self._fin.read(count)
count = int(self._fin.readline()) count = int(self._fin.readline())
@contextlib.contextmanager @contextlib.contextmanager
def mayberedirectstdio(self): def mayberedirectstdio(self):
yield None yield None
def client(self): def client(self):
client = encoding.environ.get('SSH_CLIENT', '').split(' ', 1)[0] client = encoding.environ.get('SSH_CLIENT', '').split(' ', 1)[0]

mercurial/wireprototypes.py

returns a list of values (same order as <args>)""" returns a list of values (same order as <args>)"""
def getprotocaps(): def getprotocaps():
"""Returns the list of protocol-level capabilities of client """Returns the list of protocol-level capabilities of client
Returns a list of capabilities as declared by the client for Returns a list of capabilities as declared by the client for
the current request (or connection for stateful protocol handlers).""" the current request (or connection for stateful protocol handlers)."""
def forwardpayload(fp): def getpayload():
"""Read the raw payload and forward to a file. """Provide a generator for the raw payload.
The payload is read in full before the function returns. The caller is responsible for ensuring that the full payload is
processed.
""" """
def mayberedirectstdio(): def mayberedirectstdio():
"""Context manager to possibly redirect stdio. """Context manager to possibly redirect stdio.
The context manager yields a file-object like object that receives The context manager yields a file-object like object that receives
stdout and stderr output when the context manager is active. Or it stdout and stderr output when the context manager is active. Or it
yields ``None`` if no I/O redirection occurs. yields ``None`` if no I/O redirection occurs.

tests/test-push-http.t

$ cd .. $ cd ..
$ hg clone test test2 $ hg clone test test2
updating to branch default updating to branch default
1 files updated, 0 files merged, 0 files removed, 0 files unresolved 1 files updated, 0 files merged, 0 files removed, 0 files unresolved
$ cd test2 $ cd test2
$ echo a >> a $ echo a >> a
$ hg ci -mb $ hg ci -mb
$ req() { $ req() {
> hg serve -p $HGPORT -d --pid-file=hg.pid -E errors.log > hg $1 serve -p $HGPORT -d --pid-file=hg.pid -E errors.log
> cat hg.pid >> $DAEMON_PIDS > cat hg.pid >> $DAEMON_PIDS
> hg --cwd ../test2 push http://localhost:$HGPORT/ > hg --cwd ../test2 push http://localhost:$HGPORT/
> exitstatus=$? > exitstatus=$?
> killdaemons.py > killdaemons.py
> echo % serve errors > echo % serve errors
> cat errors.log > cat errors.log
> return $exitstatus > return $exitstatus
> } > }
[255] [255]
expect success expect success
$ cat > $TESTTMP/hook.sh <<'EOF' $ cat > $TESTTMP/hook.sh <<'EOF'
> echo "phase-move: $HG_NODE: $HG_OLDPHASE -> $HG_PHASE" > echo "phase-move: $HG_NODE: $HG_OLDPHASE -> $HG_PHASE"
> EOF > EOF
#if bundle1
$ cat >> .hg/hgrc <<EOF
> allow_push = *
> [hooks]
> changegroup = sh -c "printenv.py changegroup 0"
> pushkey = sh -c "printenv.py pushkey 0"
> txnclose-phase.test = sh $TESTTMP/hook.sh
> EOF
$ req "--debug --config extensions.blackbox="
listening at http://localhost:$HGPORT/ (bound to $LOCALIP:$HGPORT)
pushing to http://localhost:$HGPORT/
searching for changes
remote: redirecting incoming bundle to */hg-unbundle-* (glob)
remote: adding changesets
remote: add changeset ba677d0156c1
remote: adding manifests
remote: adding file changes
remote: adding a revisions
remote: added 1 changesets with 1 changes to 1 files
remote: updating the branch cache
remote: running hook txnclose-phase.test: sh $TESTTMP/hook.sh
remote: phase-move: cb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b: draft -> public
remote: running hook txnclose-phase.test: sh $TESTTMP/hook.sh
remote: phase-move: ba677d0156c1196c1a699fa53f390dcfc3ce3872: -> public
remote: running hook changegroup: sh -c "printenv.py changegroup 0"
remote: changegroup hook: HG_HOOKNAME=changegroup HG_HOOKTYPE=changegroup HG_NODE=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_NODE_LAST=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_SOURCE=serve HG_TXNID=TXN:$ID$ HG_URL=remote:http:$LOCALIP: (glob)
% serve errors
$ hg rollback
repository tip rolled back to revision 0 (undo serve)
$ req "--debug --config server.streamunbundle=True --config extensions.blackbox="
listening at http://localhost:$HGPORT/ (bound to $LOCALIP:$HGPORT)
pushing to http://localhost:$HGPORT/
searching for changes
remote: adding changesets
remote: add changeset ba677d0156c1
remote: adding manifests
remote: adding file changes
remote: adding a revisions
remote: added 1 changesets with 1 changes to 1 files
remote: updating the branch cache
remote: running hook txnclose-phase.test: sh $TESTTMP/hook.sh
remote: phase-move: cb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b: draft -> public
remote: running hook txnclose-phase.test: sh $TESTTMP/hook.sh
remote: phase-move: ba677d0156c1196c1a699fa53f390dcfc3ce3872: -> public
remote: running hook changegroup: sh -c "printenv.py changegroup 0"
remote: changegroup hook: HG_HOOKNAME=changegroup HG_HOOKTYPE=changegroup HG_NODE=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_NODE_LAST=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_SOURCE=serve HG_TXNID=TXN:$ID$ HG_URL=remote:http:$LOCALIP: (glob)
% serve errors
$ hg rollback
repository tip rolled back to revision 0 (undo serve)
#endif
#if bundle2
$ cat >> .hg/hgrc <<EOF $ cat >> .hg/hgrc <<EOF
> allow_push = * > allow_push = *
> [hooks] > [hooks]
> changegroup = sh -c "printenv.py changegroup 0" > changegroup = sh -c "printenv.py changegroup 0"
> pushkey = sh -c "printenv.py pushkey 0" > pushkey = sh -c "printenv.py pushkey 0"
> txnclose-phase.test = sh $TESTTMP/hook.sh > txnclose-phase.test = sh $TESTTMP/hook.sh
> EOF > EOF
$ req $ req
pushing to http://localhost:$HGPORT/ pushing to http://localhost:$HGPORT/
searching for changes searching for changes
remote: adding changesets remote: adding changesets
remote: adding manifests remote: adding manifests
remote: adding file changes remote: adding file changes
remote: added 1 changesets with 1 changes to 1 files remote: added 1 changesets with 1 changes to 1 files
remote: phase-move: cb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b: draft -> public remote: phase-move: cb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b: draft -> public
remote: phase-move: ba677d0156c1196c1a699fa53f390dcfc3ce3872: -> public remote: phase-move: ba677d0156c1196c1a699fa53f390dcfc3ce3872: -> public
remote: changegroup hook: HG_HOOKNAME=changegroup HG_HOOKTYPE=changegroup HG_NODE=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_NODE_LAST=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_SOURCE=serve HG_TXNID=TXN:$ID$ HG_URL=remote:http:$LOCALIP: (glob) (bundle1 !) remote: changegroup hook: HG_BUNDLE2=1 HG_HOOKNAME=changegroup HG_HOOKTYPE=changegroup HG_NODE=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_NODE_LAST=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_SOURCE=serve HG_TXNID=TXN:$ID$ HG_URL=remote:http:$LOCALIP: (glob)
remote: changegroup hook: HG_BUNDLE2=1 HG_HOOKNAME=changegroup HG_HOOKTYPE=changegroup HG_NODE=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_NODE_LAST=ba677d0156c1196c1a699fa53f390dcfc3ce3872 HG_SOURCE=serve HG_TXNID=TXN:$ID$ HG_URL=remote:http:$LOCALIP: (glob) (bundle2 !)
% serve errors % serve errors
$ hg rollback $ hg rollback
repository tip rolled back to revision 0 (undo serve) repository tip rolled back to revision 0 (undo serve)
#endif
expect success, server lacks the httpheader capability expect success, server lacks the httpheader capability
$ CAP=httpheader $ CAP=httpheader
$ . "$TESTDIR/notcapable" $ . "$TESTDIR/notcapable"
$ req $ req
pushing to http://localhost:$HGPORT/ pushing to http://localhost:$HGPORT/
searching for changes searching for changes