mercurial/wireproto.py
980	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?

Diff	ID	Description	Created	Lint	Unit
Base		Base
Diff 1	6148		Feb 26 2018, 8:38 PM	★	★
Diff 2	6177		Feb 27 2018, 5:25 PM	★	★
Diff 3	6178		Feb 27 2018, 5:26 PM	★	★
Diff 4	7816		Apr 6 2018, 6:43 PM	★	★
Diff 5	7817	rHG2d965bfeb8f6326eed0b86e5070366a67f074387	Feb 26 2018, 8:37 PM	★	★

Diff 6178

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')			@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:			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
				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 because it can be big			# write bundle data to temporary file because 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-')
				repo.ui.debug('redirecting incoming bundle to %s\n' %
				tempname)
	fp = os.fdopen(fd, pycompat.sysstr('wb+'))			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 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			# The return looks streamable, we are in the bundle2 case
	# and should return a stream.			# and should return a stream.
	return streamres_legacy(gen=r.getchunks())			return streamres_legacy(gen=r.getchunks())
	return pushres(r, output.getvalue() if output else '')			return pushres(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 util.stderr directly.			# The old code we moved used util.stderr directly.

mercurial/wireprotoserver.py

	args.update(urlreq.parseqs(			args.update(urlreq.parseqs(
	self._req.read(postlen), keep_blank_values=True))			self._req.read(postlen), keep_blank_values=True))
	return args			return args

	argvalue = decodevaluefromheaders(self._req, r'X-HgArg')			argvalue = decodevaluefromheaders(self._req, r'X-HgArg')
	args.update(urlreq.parseqs(argvalue, keep_blank_values=True))			args.update(urlreq.parseqs(argvalue, keep_blank_values=True))
	return args			return args

	def forwardpayload(self, fp):			def getpayload(self):
	if r'HTTP_CONTENT_LENGTH' in self._req.env:			if r'HTTP_CONTENT_LENGTH' in self._req.env:
	length = int(self._req.env[r'HTTP_CONTENT_LENGTH'])			length = int(self._req.env[r'HTTP_CONTENT_LENGTH'])
	else:			else:
	length = int(self._req.env[r'CONTENT_LENGTH'])			length = int(self._req.env[r'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.env.get(r'HTTP_X_HGARGS_POST', 0))			length -= int(self._req.env.get(r'HTTP_X_HGARGS_POST', 0))
	for s in util.filechunkiter(self._req, limit=length):			return util.filechunkiter(self._req, 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()

	val = self._fin.read(int(l))			val = self._fin.read(int(l))
	star[arg] = val			star[arg] = val
	data['*'] = star			data['*'] = star
	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 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


	@abc.abstractmethod			@abc.abstractmethod
	def getargs(self, args):			def getargs(self, args):
	"""return the value for arguments in <args>			"""return the value for arguments in <args>

	returns a list of values (same order as <args>)"""			returns a list of values (same order as <args>)"""

	@abc.abstractmethod			@abc.abstractmethod
	def forwardpayload(self, fp):			def getpayload(self):
	"""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.
	"""			"""

	@abc.abstractmethod			@abc.abstractmethod
	def mayberedirectstdio(self):			def mayberedirectstdio(self):
	"""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

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
	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
	% serve errors			% serve errors

	#endif			#endif

	$ cd ..			$ cd ..

				$ killdaemons.py

			Path	Packages
M			hgext/largefiles/proto.py (3 lines)
M			mercurial/configitems.py (3 lines)
M			mercurial/help/config.txt (5 lines)
M			mercurial/wireproto.py (36 lines)
M			mercurial/wireprotoserver.py (9 lines)
M			mercurial/wireprototypes.py (7 lines)
M			tests/test-push-http.t (60 lines)

This is an archive of the discontinued Mercurial Phabricator instance.

wireproto: allow direct stream processing for unbundle
ClosedPublic

Details

Diff Detail

Event Timeline

Revision Contents
Changeset List

Diff 6178

hgext/largefiles/proto.py

mercurial/configitems.py

mercurial/help/config.txt

mercurial/wireproto.py

mercurial/wireprotoserver.py

mercurial/wireprototypes.py

tests/test-push-http.t

This is an archive of the discontinued Mercurial Phabricator instance.

wireproto: allow direct stream processing for unbundleClosedPublic

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 6178

hgext/largefiles/proto.py

mercurial/configitems.py

mercurial/help/config.txt

mercurial/wireproto.py

mercurial/wireprotoserver.py

mercurial/wireprototypes.py

tests/test-push-http.t

wireproto: allow direct stream processing for unbundle
ClosedPublic

Revision Contents
Changeset List