This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4482

exchangev2: fetch changeset revisions
ClosedPublic

Authored by indygreg on Sep 5 2018, 12:20 PM.

Download Raw Diff

Details

Reviewers

durin42

Group Reviewers

hg-reviewers

Commits

rHGb9e453d683a1: exchangev2: fetch changeset revisions

Summary

All Mercurial repository data is derived from changesets:
you can't do anything unless you have changesets. Therefore,
it makes sense for changesets to be the first piece of data
that we transfer as part of pull.

To do this, we call our new "changesetdata" command, requesting
parents and revision data. This gives us all the data that a
changegroup delta group would give us. We simply normalize
this data into what addgroup() expects and call that API on
the changelog to bulk insert revisions into the changelog.

Code in this commit is heavily borrowed from
changegroup.cg1unpacker.apply().

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.Sep 5 2018, 12:20 PM

Herald added a reviewer: hg-reviewers. · View Herald TranscriptSep 5 2018, 12:20 PM

Herald added a subscriber: mercurial-devel. · View Herald Transcript

indygreg added a child revision: D4483: wireprotov2: add phases to "changesetdata" command.Sep 5 2018, 12:20 PM

durin42 accepted this revision.Sep 12 2018, 11:16 AM

This revision is now accepted and ready to land.Sep 12 2018, 11:16 AM

Closed by commit rHGb9e453d683a1: exchangev2: fetch changeset revisions (authored by indygreg). · Explain WhySep 14 2018, 11:15 PM

This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

			Path	Packages
M			mercurial/exchangev2.py (83 lines)
M			tests/test-wireproto-exchangev2.t (161 lines)

Status	Author	Revision
Closed	indygreg	D4492 wireprotov2: let clients drive delta behavior
Closed	indygreg	D4491 exchangev2: fetch file revisions
Closed	indygreg	D4490 wireprotov2: define and implement "filedata" command
Closed	indygreg	D4489 exchangev2: fetch manifest revisions
Closed	indygreg	D4488 wireprotov2: define and implement "manifestdata" command
Closed	indygreg	D4487 wireprotov2: add TODOs around extending changesetdata fields
Closed	indygreg	D4486 exchangev2: fetch and apply bookmarks
Closed	indygreg	D4485 wireprotov2: add bookmarks to "changesetdata" command
Closed	indygreg	D4484 exchangev2: fetch and apply phases data
Closed	indygreg	D4483 wireprotov2: add phases to "changesetdata" command
Closed	indygreg	D4482 exchangev2: fetch changeset revisions
Closed	indygreg	D4481 wireprotov2: define and implement "changesetdata" command
Closed	indygreg	D4480 exchangev2: start to implement pull with wire protocol v2
Closed	indygreg	D4436 httppeer: expose capabilities for each command
Closed	indygreg	D4474 wireprotov2peer: stream decoded responses
Closed	indygreg	D4473 wireprotoframing: buffer emitted data to reduce frame count
Closed	indygreg	D4472 wireprotov2: implement commands as a generator of objects
Closed	indygreg	D4471 state: use our CBOR module
Closed	indygreg	D4470 wireprotov2peer: use our CBOR decoder
Closed	indygreg	D4469 debugcommands: use our CBOR decoder
Closed	indygreg	D4468 wireprotov2server: use our CBOR encoder
Closed	indygreg	D4467 wireprotoserver: use our CBOR encoder
Closed	indygreg	D4466 wireprotoframing: use our CBOR module
Closed	indygreg	D4465 httppeer: use our CBOR decoder
Closed	indygreg	D4443 internals: extract frame-based protocol docs to own document
Closed	indygreg	D4442 internals: extract wire protocol version 2 commands to standalone doc
Closed	indygreg	D4441 wireprotov2peer: properly format errors
Closed	indygreg	D4440 wireprotov2peer: report exceptions in frame handling against request future
Closed	indygreg	D4439 wireprotov2peer: split responsedata handling into separate function
Closed	indygreg	D4438 httppeer: add TODO about delayed handling of ^C
Closed	indygreg	D4437 httppeer: log commands for version 2 peer
Closed	indygreg	D4435 wireprotov2peer: add TODO about streaming responses
Closed	indygreg	D4434 cborutil: add a buffering decoder
Closed	indygreg	D4433 cborutil: remove readindefinitebytestringtoiter()
Closed	indygreg	D4414 cborutil: implement sans I/O decoder
Closed	indygreg	D4413 tests: remove use of string in CBOR test
Closed	indygreg	D4412 internals: document CBOR utilization

Diff 11060

mercurial/exchangev2.py

	# exchangev2.py - repository exchange for wire protocol version 2			# exchangev2.py - repository exchange for wire protocol version 2
	#			#
	# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com>			# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.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 weakref

				from .i18n import _
	from .node import (			from .node import (
	nullid,			nullid,
				short,
	)			)
	from . import (			from . import (
				mdiff,
				pycompat,
	setdiscovery,			setdiscovery,
	)			)

	def pull(pullop):			def pull(pullop):
	"""Pull using wire protocol version 2."""			"""Pull using wire protocol version 2."""
	repo = pullop.repo			repo = pullop.repo
	remote = pullop.remote			remote = pullop.remote
				tr = pullop.trmanager.transaction()

	# Figure out what needs to be fetched.			# Figure out what needs to be fetched.
	common, fetch, remoteheads = _pullchangesetdiscovery(			common, fetch, remoteheads = _pullchangesetdiscovery(
	repo, remote, pullop.heads, abortwhenunrelated=pullop.force)			repo, remote, pullop.heads, abortwhenunrelated=pullop.force)

				pullheads = pullop.heads or remoteheads
				_fetchchangesets(repo, tr, remote, common, fetch, pullheads)

	def _pullchangesetdiscovery(repo, remote, heads, abortwhenunrelated=True):			def _pullchangesetdiscovery(repo, remote, heads, abortwhenunrelated=True):
	"""Determine which changesets need to be pulled."""			"""Determine which changesets need to be pulled."""

	if heads:			if heads:
	knownnode = repo.changelog.hasnode			knownnode = repo.changelog.hasnode
	if all(knownnode(head) for head in heads):			if all(knownnode(head) for head in heads):
	return heads, False, heads			return heads, False, heads

	common \|= {head for head in remoteheads if head in nodemap}			common \|= {head for head in remoteheads if head in nodemap}

	if set(remoteheads).issubset(common):			if set(remoteheads).issubset(common):
	fetch = []			fetch = []

	common.discard(nullid)			common.discard(nullid)

	return common, fetch, remoteheads			return common, fetch, remoteheads

				def _fetchchangesets(repo, tr, remote, common, fetch, remoteheads):
				if not fetch:
				return

				# TODO consider adding a step here where we obtain the DAG shape first
				# (or ask the server to slice changesets into chunks for us) so that
				# we can perform multiple fetches in batches. This will facilitate
				# resuming interrupted clones, higher server-side cache hit rates due
				# to smaller segments, etc.
				with remote.commandexecutor() as e:
				objs = e.callcommand(b'changesetdata', {
				b'noderange': [sorted(common), sorted(remoteheads)],
				b'fields': {b'parents', b'revision'},
				}).result()

				# The context manager waits on all response data when exiting. So
				# we need to remain in the context manager in order to stream data.
				return _processchangesetdata(repo, tr, objs)

				def _processchangesetdata(repo, tr, objs):
				repo.hook('prechangegroup', throw=True,
				**pycompat.strkwargs(tr.hookargs))

				urepo = repo.unfiltered()
				cl = urepo.changelog

				cl.delayupdate(tr)

				# The first emitted object is a header describing the data that
				# follows.
				meta = next(objs)

				progress = repo.ui.makeprogress(_('changesets'),
				unit=_('chunks'),
				total=meta.get(b'totalitems'))

				def linkrev(node):
				repo.ui.debug('add changeset %s\n' % short(node))
				# Linkrev for changelog is always self.
				return len(cl)

				def onchangeset(cl, node):
				progress.increment()

				# addgroup() expects a 7-tuple describing revisions. This normalizes
				# the wire data to that format.
				def iterrevisions():
				for cset in objs:
				assert b'revisionsize' in cset
				data = next(objs)

				yield (
				cset[b'node'],
				cset[b'parents'][0],
				cset[b'parents'][1],
				# Linknode is always itself for changesets.
				cset[b'node'],
				# We always send full revisions. So delta base is not set.
				nullid,
				mdiff.trivialdiffheader(len(data)) + data,
				# Flags not yet supported.
				0,
				)

				added = cl.addgroup(iterrevisions(), linkrev, weakref.proxy(tr),
				addrevisioncb=onchangeset)

				progress.complete()

				return {
				'added': added,
				}

tests/test-wireproto-exchangev2.t

	'nodes': []			'nodes': []
	}			}
	received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)			received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
	received frame(size=43; request=1; stream=2; streamflags=; type=command-response; flags=continuation)			received frame(size=43; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
	received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)			received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
	received frame(size=11; request=3; stream=2; streamflags=; type=command-response; flags=continuation)			received frame(size=11; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
	received frame(size=1; request=3; stream=2; streamflags=; type=command-response; flags=continuation)			received frame(size=1; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
	received frame(size=0; request=3; stream=2; streamflags=; type=command-response; flags=eos)			received frame(size=0; request=3; stream=2; streamflags=; type=command-response; flags=eos)
				sending 1 commands
				sending command changesetdata: {
				'fields': set([
				'parents',
				'revision'
				]),
				'noderange': [
				[],
				[
				'\xca\xa2\xa4eE\x1d\xd1\xfa\xcd\xa0\xf5\xb1#\x12\xc3UXA\x88\xa1',
				'\xcd%4vk\xec\xe18\xc7\xc1\xaf\xdch%0/\x0fb\xd8\x1f'
				]
				]
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=809; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				add changeset 3390ef850073
				add changeset 4432d83626e8
				add changeset cd2534766bec
				add changeset e96ae20f4188
				add changeset caa2a465451d
				updating the branch cache
				new changesets 3390ef850073:caa2a465451d

				All changesets should have been transferred

				$ hg -R client-simple debugindex -c
				rev linkrev nodeid p1 p2
				0 0 3390ef850073 000000000000 000000000000
				1 1 4432d83626e8 3390ef850073 000000000000
				2 2 cd2534766bec 4432d83626e8 000000000000
				3 3 e96ae20f4188 3390ef850073 000000000000
				4 4 caa2a465451d e96ae20f4188 000000000000

				$ hg -R client-simple log -G -T '{rev} {node} {phase}\n'
				o 4 caa2a465451dd1facda0f5b12312c355584188a1 public
				\|
				o 3 e96ae20f4188487b9ae4ef3941c27c81143146e5 public
				\|
				\| o 2 cd2534766bece138c7c1afdc6825302f0f62d81f public
				\| \|
				\| o 1 4432d83626e8a98655f062ec1f2a43b07f7fbbb0 public
				\|/
				o 0 3390ef850073fbc2f0dfff2244342c8e9229013a public


				Cloning only a specific revision works

				$ hg --debug clone -U -r 4432d83626e8 http://localhost:$HGPORT client-singlehead
				using http://localhost:$HGPORT/
				sending capabilities command
				sending 1 commands
				sending command lookup: {
				'key': '4432d83626e8'
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=21; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				query 1; heads
				sending 2 commands
				sending command heads: {}
				sending command known: {
				'nodes': []
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=43; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				received frame(size=11; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=1; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=3; stream=2; streamflags=; type=command-response; flags=eos)
				sending 1 commands
				sending command changesetdata: {
				'fields': set([
				'parents',
				'revision'
				]),
				'noderange': [
				[],
				[
				'D2\xd86&\xe8\xa9\x86U\xf0b\xec\x1f*C\xb0\x7f\x7f\xbb\xb0'
				]
				]
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=327; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				add changeset 3390ef850073
				add changeset 4432d83626e8
				updating the branch cache
				new changesets 3390ef850073:4432d83626e8

				$ cd client-singlehead

				$ hg log -G -T '{rev} {node} {phase}\n'
				o 1 4432d83626e8a98655f062ec1f2a43b07f7fbbb0 public
				\|
				o 0 3390ef850073fbc2f0dfff2244342c8e9229013a public


				Incremental pull works

				$ hg --debug pull
				pulling from http://localhost:$HGPORT/
				using http://localhost:$HGPORT/
				sending capabilities command
				query 1; heads
				sending 2 commands
				sending command heads: {}
				sending command known: {
				'nodes': [
				'D2\xd86&\xe8\xa9\x86U\xf0b\xec\x1f*C\xb0\x7f\x7f\xbb\xb0'
				]
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=43; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				received frame(size=11; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=2; request=3; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=3; stream=2; streamflags=; type=command-response; flags=eos)
				searching for changes
				all local heads known remotely
				sending 1 commands
				sending command changesetdata: {
				'fields': set([
				'parents',
				'revision'
				]),
				'noderange': [
				[
				'D2\xd86&\xe8\xa9\x86U\xf0b\xec\x1f*C\xb0\x7f\x7f\xbb\xb0'
				],
				[
				'\xca\xa2\xa4eE\x1d\xd1\xfa\xcd\xa0\xf5\xb1#\x12\xc3UXA\x88\xa1',
				'\xcd%4vk\xec\xe18\xc7\xc1\xaf\xdch%0/\x0fb\xd8\x1f'
				]
				]
				}
				received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
				received frame(size=495; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
				received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
				add changeset cd2534766bec
				add changeset e96ae20f4188
				add changeset caa2a465451d
				updating the branch cache
				new changesets cd2534766bec:caa2a465451d
				(run 'hg update' to get a working copy)

				$ hg log -G -T '{rev} {node} {phase}\n'
				o 4 caa2a465451dd1facda0f5b12312c355584188a1 public
				\|
				o 3 e96ae20f4188487b9ae4ef3941c27c81143146e5 public
				\|
				\| o 2 cd2534766bece138c7c1afdc6825302f0f62d81f public
				\| \|
				\| o 1 4432d83626e8a98655f062ec1f2a43b07f7fbbb0 public
				\|/
				o 0 3390ef850073fbc2f0dfff2244342c8e9229013a public


				$ cd ..

Diff	ID	Description	Created	Lint	Unit
Base		Base
Diff 1	10795		Sep 5 2018, 12:20 PM	★	★
Diff 2	11060	rHGb9e453d683a1799f818f62254d73c1aa3cd09fc1	Sep 12 2018, 1:01 PM	★	★