setdiscovery.py
520 lines
| 18.0 KiB
| text/x-python
|
PythonLexer
/ mercurial / setdiscovery.py
|
r14164 | # setdiscovery.py - improved discovery of common nodeset for mercurial | ||
# | ||||
# Copyright 2010 Benoit Boissinot <bboissin@gmail.com> | ||||
# and Peter Arrenbrecht <peter@arrenbrecht.ch> | ||||
# | ||||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
|
r20656 | """ | ||
Algorithm works in the following way. You have two repository: local and | ||||
remote. They both contains a DAG of changelists. | ||||
The goal of the discovery protocol is to find one set of node *common*, | ||||
the set of nodes shared by local and remote. | ||||
One of the issue with the original protocol was latency, it could | ||||
potentially require lots of roundtrips to discover that the local repo was a | ||||
subset of remote (which is a very common case, you usually have few changes | ||||
compared to upstream, while upstream probably had lots of development). | ||||
The new protocol only requires one interface for the remote repo: `known()`, | ||||
which given a set of changelists tells you if they are present in the DAG. | ||||
The algorithm then works as follow: | ||||
- We will be using three sets, `common`, `missing`, `unknown`. Originally | ||||
all nodes are in `unknown`. | ||||
- Take a sample from `unknown`, call `remote.known(sample)` | ||||
- For each node that remote knows, move it and all its ancestors to `common` | ||||
- For each node that remote doesn't know, move it and all its descendants | ||||
to `missing` | ||||
- Iterate until `unknown` is empty | ||||
There are a couple optimizations, first is instead of starting with a random | ||||
sample of missing, start by sending all heads, in the case where the local | ||||
repo is a subset, you computed the answer in one round trip. | ||||
Then you can do something similar to the bisecting strategy used when | ||||
finding faulty changesets. Instead of random samples, you can try picking | ||||
nodes that will maximize the number of nodes that will be | ||||
classified with it (since all ancestors or descendants will be marked as well). | ||||
""" | ||||
|
r14164 | |||
|
r25973 | from __future__ import absolute_import | ||
|
r25113 | import collections | ||
|
r20034 | import random | ||
|
r25973 | |||
from .i18n import _ | ||||
|
r47771 | from .node import nullrev | ||
|
r25973 | from . import ( | ||
|
r26587 | error, | ||
|
r42972 | policy, | ||
r32712 | util, | |||
|
r25973 | ) | ||
|
r14164 | |||
|
r43346 | |||
|
r39210 | def _updatesample(revs, heads, sample, parentfn, quicksamplesize=0): | ||
|
r23809 | """update an existing sample to match the expected size | ||
|
r39204 | The sample is updated with revs exponentially distant from each head of the | ||
<revs> set. (H~1, H~2, H~4, H~8, etc). | ||||
|
r23809 | |||
If a target size is specified, the sampling will stop once this size is | ||||
|
r39204 | reached. Otherwise sampling will happen until roots of the <revs> set are | ||
|
r23809 | reached. | ||
|
r39204 | :revs: set of revs we want to discover (if None, assume the whole dag) | ||
|
r39206 | :heads: set of DAG head revs | ||
|
r23809 | :sample: a sample to update | ||
|
r39210 | :parentfn: a callable to resolve parents for a revision | ||
|
r23809 | :quicksamplesize: optional target size of the sample""" | ||
|
r14164 | dist = {} | ||
|
r25113 | visit = collections.deque(heads) | ||
|
r14164 | seen = set() | ||
factor = 1 | ||||
while visit: | ||||
curr = visit.popleft() | ||||
if curr in seen: | ||||
continue | ||||
d = dist.setdefault(curr, 1) | ||||
if d > factor: | ||||
factor *= 2 | ||||
if d == factor: | ||||
|
r23814 | sample.add(curr) | ||
if quicksamplesize and (len(sample) >= quicksamplesize): | ||||
return | ||||
|
r14164 | seen.add(curr) | ||
|
r39210 | |||
for p in parentfn(curr): | ||||
if p != nullrev and (not revs or p in revs): | ||||
|
r14164 | dist.setdefault(p, d + 1) | ||
visit.append(p) | ||||
|
r43346 | |||
|
r42968 | def _limitsample(sample, desiredlen, randomize=True): | ||
"""return a random subset of sample of at most desiredlen item. | ||||
If randomize is False, though, a deterministic subset is returned. | ||||
This is meant for integration tests. | ||||
""" | ||||
if len(sample) <= desiredlen: | ||||
return sample | ||||
if randomize: | ||||
return set(random.sample(sample, desiredlen)) | ||||
sample = list(sample) | ||||
sample.sort() | ||||
return set(sample[:desiredlen]) | ||||
|
r23083 | |||
|
r43346 | |||
|
r41147 | class partialdiscovery(object): | ||
"""an object representing ongoing discovery | ||||
Feed with data from the remote repository, this object keep track of the | ||||
current set of changeset in various states: | ||||
|
r41208 | - common: revs also known remotely | ||
- undecided: revs we don't have information on yet | ||||
- missing: revs missing remotely | ||||
(all tracked revisions are known locally) | ||||
|
r41147 | """ | ||
|
r42968 | def __init__(self, repo, targetheads, respectsize, randomize=True): | ||
|
r41147 | self._repo = repo | ||
|
r41203 | self._targetheads = targetheads | ||
|
r41147 | self._common = repo.changelog.incrementalmissingrevs() | ||
|
r41203 | self._undecided = None | ||
|
r41206 | self.missing = set() | ||
r42051 | self._childrenmap = None | |||
|
r42594 | self._respectsize = respectsize | ||
|
r42968 | self.randomize = randomize | ||
|
r41147 | |||
def addcommons(self, commons): | ||||
|
r42351 | """register nodes known as common""" | ||
|
r41147 | self._common.addbases(commons) | ||
|
r41374 | if self._undecided is not None: | ||
self._common.removeancestorsfrom(self._undecided) | ||||
|
r41147 | |||
|
r41206 | def addmissings(self, missings): | ||
|
r42351 | """register some nodes as missing""" | ||
|
r43347 | newmissing = self._repo.revs(b'%ld::%ld', missings, self.undecided) | ||
|
r41316 | if newmissing: | ||
self.missing.update(newmissing) | ||||
self.undecided.difference_update(newmissing) | ||||
|
r41206 | |||
|
r41207 | def addinfo(self, sample): | ||
"""consume an iterable of (rev, known) tuples""" | ||||
common = set() | ||||
missing = set() | ||||
for rev, known in sample: | ||||
if known: | ||||
common.add(rev) | ||||
else: | ||||
missing.add(rev) | ||||
if common: | ||||
self.addcommons(common) | ||||
if missing: | ||||
self.addmissings(missing) | ||||
|
r41147 | def hasinfo(self): | ||
"""return True is we have any clue about the remote state""" | ||||
return self._common.hasbases() | ||||
|
r41205 | def iscomplete(self): | ||
"""True if all the necessary data have been gathered""" | ||||
return self._undecided is not None and not self._undecided | ||||
|
r41203 | @property | ||
def undecided(self): | ||||
if self._undecided is not None: | ||||
return self._undecided | ||||
self._undecided = set(self._common.missingancestors(self._targetheads)) | ||||
return self._undecided | ||||
|
r42272 | def stats(self): | ||
return { | ||||
|
r43432 | 'undecided': len(self.undecided), | ||
|
r42272 | } | ||
|
r41148 | def commonheads(self): | ||
"""the heads of the known common set""" | ||||
# heads(common) == heads(common.bases) since common represents | ||||
# common.bases and all its ancestors | ||||
|
r41281 | return self._common.basesheads() | ||
|
r41147 | |||
r42047 | def _parentsgetter(self): | |||
getrev = self._repo.changelog.index.__getitem__ | ||||
|
r43346 | |||
r42047 | def getparents(r): | |||
r42145 | return getrev(r)[5:7] | |||
|
r43346 | |||
r42047 | return getparents | |||
r42052 | def _childrengetter(self): | |||
r42050 | ||||
r42051 | if self._childrenmap is not None: | |||
r42055 | # During discovery, the `undecided` set keep shrinking. | |||
# Therefore, the map computed for an iteration N will be | ||||
# valid for iteration N+1. Instead of computing the same | ||||
# data over and over we cached it the first time. | ||||
r42051 | return self._childrenmap.__getitem__ | |||
r42050 | # _updatesample() essentially does interaction over revisions to look | |||
# up their children. This lookup is expensive and doing it in a loop is | ||||
# quadratic. We precompute the children for all relevant revisions and | ||||
# make the lookup in _updatesample() a simple dict lookup. | ||||
r42051 | self._childrenmap = children = {} | |||
r42050 | ||||
parentrevs = self._parentsgetter() | ||||
r42052 | revs = self.undecided | |||
r42050 | ||||
for rev in sorted(revs): | ||||
# Always ensure revision has an entry so we don't need to worry | ||||
# about missing keys. | ||||
children[rev] = [] | ||||
for prev in parentrevs(rev): | ||||
if prev == nullrev: | ||||
continue | ||||
c = children.get(prev) | ||||
if c is not None: | ||||
c.append(rev) | ||||
return children.__getitem__ | ||||
|
r42045 | def takequicksample(self, headrevs, size): | ||
"""takes a quick sample of size <size> | ||||
It is meant for initial sampling and focuses on querying heads and close | ||||
ancestors of heads. | ||||
:headrevs: set of head revisions in local DAG to consider | ||||
:size: the maximum size of the sample""" | ||||
revs = self.undecided | ||||
if len(revs) <= size: | ||||
return list(revs) | ||||
|
r43347 | sample = set(self._repo.revs(b'heads(%ld)', revs)) | ||
|
r42045 | |||
if len(sample) >= size: | ||||
|
r42968 | return _limitsample(sample, size, randomize=self.randomize) | ||
|
r42045 | |||
|
r43346 | _updatesample( | ||
None, headrevs, sample, self._parentsgetter(), quicksamplesize=size | ||||
) | ||||
|
r42045 | return sample | ||
def takefullsample(self, headrevs, size): | ||||
revs = self.undecided | ||||
if len(revs) <= size: | ||||
return list(revs) | ||||
repo = self._repo | ||||
|
r43347 | sample = set(repo.revs(b'heads(%ld)', revs)) | ||
r42047 | parentrevs = self._parentsgetter() | |||
|
r42045 | |||
# update from heads | ||||
r42046 | revsheads = sample.copy() | |||
r42047 | _updatesample(revs, revsheads, sample, parentrevs) | |||
|
r42045 | |||
# update from roots | ||||
|
r43347 | revsroots = set(repo.revs(b'roots(%ld)', revs)) | ||
r42052 | childrenrevs = self._childrengetter() | |||
r42050 | _updatesample(revs, revsroots, sample, childrenrevs) | |||
|
r42045 | assert sample | ||
r42618 | ||||
if not self._respectsize: | ||||
size = max(size, min(len(revsroots), len(revsheads))) | ||||
|
r42968 | sample = _limitsample(sample, size, randomize=self.randomize) | ||
|
r42045 | if len(sample) < size: | ||
more = size - len(sample) | ||||
|
r42968 | takefrom = list(revs - sample) | ||
if self.randomize: | ||||
sample.update(random.sample(takefrom, more)) | ||||
else: | ||||
takefrom.sort() | ||||
sample.update(takefrom[:more]) | ||||
|
r42045 | return sample | ||
|
r43346 | |||
r48043 | pure_partialdiscovery = partialdiscovery | |||
|
r43346 | partialdiscovery = policy.importrust( | ||
|
r43906 | 'discovery', member='PartialDiscovery', default=partialdiscovery | ||
|
r43346 | ) | ||
|
r42972 | |||
|
r43346 | def findcommonheads( | ||
ui, | ||||
local, | ||||
remote, | ||||
abortwhenunrelated=True, | ||||
ancestorsof=None, | ||||
r46726 | audit=None, | |||
|
r43346 | ): | ||
|
r46554 | """Return a tuple (common, anyincoming, remoteheads) used to identify | ||
|
r14206 | missing nodes from or in remote. | ||
r46726 | ||||
The audit argument is an optional dictionnary that a caller can pass. it | ||||
will be updated with extra data about the discovery, this is useful for | ||||
debug. | ||||
|
r46554 | """ | ||
r47018 | ||||
samplegrowth = float(ui.config(b'devel', b'discovery.grow-sample.rate')) | ||||
r32712 | start = util.timer() | |||
|
r14164 | roundtrips = 0 | ||
cl = local.changelog | ||||
|
r39195 | clnode = cl.node | ||
|
r39197 | clrev = cl.rev | ||
|
r39195 | |||
|
r35305 | if ancestorsof is not None: | ||
|
r39201 | ownheads = [clrev(n) for n in ancestorsof] | ||
else: | ||||
ownheads = [rev for rev in cl.headrevs() if rev != nullrev] | ||||
r47034 | initial_head_exchange = ui.configbool(b'devel', b'discovery.exchange-heads') | |||
r47068 | initialsamplesize = ui.configint(b'devel', b'discovery.sample-size.initial') | |||
fullsamplesize = ui.configint(b'devel', b'discovery.sample-size') | ||||
r42335 | # We also ask remote about all the local heads. That set can be arbitrarily | |||
# large, so we used to limit it size to `initialsamplesize`. We no longer | ||||
# do as it proved counter productive. The skipped heads could lead to a | ||||
# large "undecided" set, slower to be clarified than if we asked the | ||||
# question for all heads right away. | ||||
# | ||||
# We are already fetching all server heads using the `heads` commands, | ||||
# sending a equivalent number of heads the other way should not have a | ||||
# significant impact. In addition, it is very likely that we are going to | ||||
# have to issue "known" request for an equivalent amount of revisions in | ||||
# order to decide if theses heads are common or missing. | ||||
# | ||||
# find a detailled analysis below. | ||||
# | ||||
# Case A: local and server both has few heads | ||||
# | ||||
# Ownheads is below initialsamplesize, limit would not have any effect. | ||||
# | ||||
# Case B: local has few heads and server has many | ||||
# | ||||
# Ownheads is below initialsamplesize, limit would not have any effect. | ||||
# | ||||
# Case C: local and server both has many heads | ||||
# | ||||
# We now transfert some more data, but not significantly more than is | ||||
# already transfered to carry the server heads. | ||||
# | ||||
# Case D: local has many heads, server has few | ||||
# | ||||
# D.1 local heads are mostly known remotely | ||||
# | ||||
# All the known head will have be part of a `known` request at some | ||||
# point for the discovery to finish. Sending them all earlier is | ||||
# actually helping. | ||||
# | ||||
# (This case is fairly unlikely, it requires the numerous heads to all | ||||
# be merged server side in only a few heads) | ||||
# | ||||
# D.2 local heads are mostly missing remotely | ||||
# | ||||
# To determine that the heads are missing, we'll have to issue `known` | ||||
# request for them or one of their ancestors. This amount of `known` | ||||
# request will likely be in the same order of magnitude than the amount | ||||
# of local heads. | ||||
# | ||||
# The only case where we can be more efficient using `known` request on | ||||
# ancestors are case were all the "missing" local heads are based on a | ||||
# few changeset, also "missing". This means we would have a "complex" | ||||
# graph (with many heads) attached to, but very independant to a the | ||||
# "simple" graph on the server. This is a fairly usual case and have | ||||
# not been met in the wild so far. | ||||
r47034 | if initial_head_exchange: | |||
if remote.limitedarguments: | ||||
sample = _limitsample(ownheads, initialsamplesize) | ||||
# indices between sample and externalized version must match | ||||
sample = list(sample) | ||||
else: | ||||
sample = ownheads | ||||
|
r37649 | |||
r47034 | ui.debug(b"query 1; heads\n") | |||
roundtrips += 1 | ||||
with remote.commandexecutor() as e: | ||||
fheads = e.callcommand(b'heads', {}) | ||||
fknown = e.callcommand( | ||||
b'known', | ||||
{ | ||||
b'nodes': [clnode(r) for r in sample], | ||||
}, | ||||
) | ||||
srvheadhashes, yesno = fheads.result(), fknown.result() | ||||
|
r37649 | |||
r47034 | if audit is not None: | |||
audit[b'total-roundtrips'] = 1 | ||||
r46726 | ||||
|
r47599 | if cl.tiprev() == nullrev: | ||
|
r47771 | if srvheadhashes != [cl.nullid]: | ||
return [cl.nullid], True, srvheadhashes | ||||
return [cl.nullid], False, [] | ||||
r47034 | else: | |||
# we still need the remote head for the function return | ||||
with remote.commandexecutor() as e: | ||||
fheads = e.callcommand(b'heads', {}) | ||||
srvheadhashes = fheads.result() | ||||
|
r14164 | |||
|
r14206 | # start actual discovery (we note this before the next "if" for | ||
# compatibility reasons) | ||||
|
r43347 | ui.status(_(b"searching for changes\n")) | ||
|
r14164 | |||
|
r42044 | knownsrvheads = [] # revnos of remote heads that are known locally | ||
|
r39197 | for node in srvheadhashes: | ||
|
r47771 | if node == cl.nullid: | ||
|
r39197 | continue | ||
try: | ||||
|
r42044 | knownsrvheads.append(clrev(node)) | ||
|
r39197 | # Catches unknown and filtered nodes. | ||
except error.LookupError: | ||||
continue | ||||
r47034 | if initial_head_exchange: | |||
# early exit if we know all the specified remote heads already | ||||
if len(knownsrvheads) == len(srvheadhashes): | ||||
ui.debug(b"all remote heads known locally\n") | ||||
return srvheadhashes, False, srvheadhashes | ||||
|
r14164 | |||
r47034 | if len(sample) == len(ownheads) and all(yesno): | |||
ui.note(_(b"all local changesets known remotely\n")) | ||||
ownheadhashes = [clnode(r) for r in ownheads] | ||||
return ownheadhashes, True, srvheadhashes | ||||
|
r14624 | |||
|
r14164 | # full blown discovery | ||
r47017 | # if the server has a limit to its arguments size, we can't grow the sample. | |||
r47562 | configbool = local.ui.configbool | |||
grow_sample = configbool(b'devel', b'discovery.grow-sample') | ||||
r47561 | grow_sample = grow_sample and not remote.limitedarguments | |||
r47017 | ||||
r47562 | dynamic_sample = configbool(b'devel', b'discovery.grow-sample.dynamic') | |||
hard_limit_sample = not (dynamic_sample or remote.limitedarguments) | ||||
|
r43347 | randomize = ui.configbool(b'devel', b'discovery.randomize') | ||
r48043 | if cl.index.rust_ext_compat: | |||
pd = partialdiscovery | ||||
else: | ||||
pd = pure_partialdiscovery | ||||
disco = pd(local, ownheads, hard_limit_sample, randomize=randomize) | ||||
r47034 | if initial_head_exchange: | |||
# treat remote heads (and maybe own heads) as a first implicit sample | ||||
# response | ||||
disco.addcommons(knownsrvheads) | ||||
disco.addinfo(zip(sample, yesno)) | ||||
|
r16683 | |||
r47034 | full = not initial_head_exchange | |||
|
r43347 | progress = ui.makeprogress(_(b'searching'), unit=_(b'queries')) | ||
|
r41205 | while not disco.iscomplete(): | ||
|
r14164 | |||
|
r41147 | if full or disco.hasinfo(): | ||
|
r23747 | if full: | ||
|
r43347 | ui.note(_(b"sampling from both directions\n")) | ||
|
r23747 | else: | ||
|
r43347 | ui.debug(b"taking initial sample\n") | ||
|
r42045 | samplefunc = disco.takefullsample | ||
|
r23130 | targetsize = fullsamplesize | ||
r47561 | if grow_sample: | |||
r42546 | fullsamplesize = int(fullsamplesize * samplegrowth) | |||
|
r14624 | else: | ||
# use even cheaper initial sample | ||||
|
r43347 | ui.debug(b"taking quick initial sample\n") | ||
|
r42045 | samplefunc = disco.takequicksample | ||
|
r23130 | targetsize = initialsamplesize | ||
|
r42045 | sample = samplefunc(ownheads, targetsize) | ||
|
r14164 | |||
roundtrips += 1 | ||||
|
r38369 | progress.update(roundtrips) | ||
|
r42272 | stats = disco.stats() | ||
|
r43346 | ui.debug( | ||
|
r43347 | b"query %i; still undecided: %i, sample size is: %i\n" | ||
|
r43432 | % (roundtrips, stats['undecided'], len(sample)) | ||
|
r43346 | ) | ||
|
r42272 | |||
|
r14164 | # indices between sample and externalized version must match | ||
sample = list(sample) | ||||
|
r37648 | |||
with remote.commandexecutor() as e: | ||||
|
r43346 | yesno = e.callcommand( | ||
|
r46554 | b'known', | ||
{ | ||||
b'nodes': [clnode(r) for r in sample], | ||||
}, | ||||
|
r43346 | ).result() | ||
|
r37648 | |||
|
r14624 | full = True | ||
|
r14164 | |||
|
r41207 | disco.addinfo(zip(sample, yesno)) | ||
|
r23343 | |||
|
r41148 | result = disco.commonheads() | ||
r32712 | elapsed = util.timer() - start | |||
|
r38392 | progress.complete() | ||
|
r43347 | ui.debug(b"%d total queries in %.4fs\n" % (roundtrips, elapsed)) | ||
|
r43346 | msg = ( | ||
|
r43347 | b'found %d common and %d unknown server heads,' | ||
b' %d roundtrips in %.4fs\n' | ||||
|
r43346 | ) | ||
|
r42044 | missing = set(result) - set(knownsrvheads) | ||
|
r43347 | ui.log(b'discovery', msg, len(result), len(missing), roundtrips, elapsed) | ||
|
r14164 | |||
r46726 | if audit is not None: | |||
audit[b'total-roundtrips'] = roundtrips | ||||
|
r47771 | if not result and srvheadhashes != [cl.nullid]: | ||
|
r14164 | if abortwhenunrelated: | ||
|
r43347 | raise error.Abort(_(b"repository is unrelated")) | ||
|
r14164 | else: | ||
|
r43347 | ui.warn(_(b"warning: repository is unrelated\n")) | ||
|
r43346 | return ( | ||
|
r47771 | {cl.nullid}, | ||
|
r43346 | True, | ||
srvheadhashes, | ||||
) | ||||
|
r14164 | |||
|
r47771 | anyincoming = srvheadhashes != [cl.nullid] | ||
|
r39195 | result = {clnode(r) for r in result} | ||
return result, anyincoming, srvheadhashes | ||||