setdiscovery.py
336 lines
| 11.5 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 _ | ||||
from .node import ( | ||||
nullid, | ||||
nullrev, | ||||
) | ||||
from . import ( | ||||
|
r26587 | error, | ||
r32712 | util, | |||
|
r25973 | ) | ||
|
r14164 | |||
|
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) | ||||
|
r39210 | def _takequicksample(repo, headrevs, revs, size): | ||
|
r23816 | """takes a quick sample of size <size> | ||
It is meant for initial sampling and focuses on querying heads and close | ||||
ancestors of heads. | ||||
:dag: a dag object | ||||
|
r39207 | :headrevs: set of head revisions in local DAG to consider | ||
|
r39204 | :revs: set of revs to discover | ||
|
r23816 | :size: the maximum size of the sample""" | ||
|
r41146 | if len(revs) <= size: | ||
return list(revs) | ||||
|
r39205 | sample = set(repo.revs('heads(%ld)', revs)) | ||
|
r36735 | if len(sample) >= size: | ||
|
r23815 | return _limitsample(sample, size) | ||
|
r39206 | |||
|
r39210 | _updatesample(None, headrevs, sample, repo.changelog.parentrevs, | ||
quicksamplesize=size) | ||||
|
r14164 | return sample | ||
|
r39210 | def _takefullsample(repo, headrevs, revs, size): | ||
|
r41146 | if len(revs) <= size: | ||
return list(revs) | ||||
|
r39205 | sample = set(repo.revs('heads(%ld)', revs)) | ||
|
r14164 | # update from heads | ||
|
r39209 | revsheads = set(repo.revs('heads(%ld)', revs)) | ||
|
r39210 | _updatesample(revs, revsheads, sample, repo.changelog.parentrevs) | ||
|
r39214 | |||
|
r14164 | # update from roots | ||
|
r39209 | revsroots = set(repo.revs('roots(%ld)', revs)) | ||
|
r39210 | |||
|
r39214 | # _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. | ||||
# | ||||
# Because this function can be called multiple times during discovery, we | ||||
# may still perform redundant work and there is room to optimize this by | ||||
# keeping a persistent cache of children across invocations. | ||||
children = {} | ||||
|
r39210 | |||
|
r39214 | parentrevs = repo.changelog.parentrevs | ||
for rev in repo.changelog.revs(start=min(revsroots)): | ||||
# Always ensure revision has an entry so we don't need to worry about | ||||
# missing keys. | ||||
children.setdefault(rev, []) | ||||
for prev in parentrevs(rev): | ||||
if prev == nullrev: | ||||
continue | ||||
children.setdefault(prev, []).append(rev) | ||||
_updatesample(revs, revsroots, sample, children.__getitem__) | ||||
|
r14164 | assert sample | ||
|
r23810 | sample = _limitsample(sample, size) | ||
|
r41145 | if len(sample) <= size: | ||
|
r23810 | more = size - len(sample) | ||
|
r39204 | sample.update(random.sample(list(revs - sample), more)) | ||
|
r14164 | return sample | ||
|
r23083 | def _limitsample(sample, desiredlen): | ||
"""return a random subset of sample of at most desiredlen item""" | ||||
if len(sample) > desiredlen: | ||||
sample = set(random.sample(sample, desiredlen)) | ||||
return sample | ||||
|
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: | ||||
- common: own nodes I know we both know | ||||
""" | ||||
def __init__(self, repo): | ||||
self._repo = repo | ||||
self._common = repo.changelog.incrementalmissingrevs() | ||||
def addcommons(self, commons): | ||||
"""registrer nodes known as common""" | ||||
self._common.addbases(commons) | ||||
def hasinfo(self): | ||||
"""return True is we have any clue about the remote state""" | ||||
return self._common.hasbases() | ||||
|
r36732 | def findcommonheads(ui, local, remote, | ||
|
r14164 | initialsamplesize=100, | ||
fullsamplesize=200, | ||||
|
r35305 | abortwhenunrelated=True, | ||
ancestorsof=None): | ||||
|
r14206 | '''Return a tuple (common, anyincoming, remoteheads) used to identify | ||
missing nodes from or in remote. | ||||
|
r14164 | ''' | ||
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] | ||||
|
r14624 | # early exit if we know all the specified remote heads already | ||
|
r14164 | ui.debug("query 1; heads\n") | ||
roundtrips += 1 | ||||
|
r23084 | sample = _limitsample(ownheads, initialsamplesize) | ||
|
r23192 | # indices between sample and externalized version must match | ||
sample = list(sample) | ||||
|
r37649 | |||
with remote.commandexecutor() as e: | ||||
fheads = e.callcommand('heads', {}) | ||||
fknown = e.callcommand('known', { | ||||
|
r39195 | 'nodes': [clnode(r) for r in sample], | ||
|
r37649 | }) | ||
srvheadhashes, yesno = fheads.result(), fknown.result() | ||||
|
r14164 | |||
if cl.tip() == nullid: | ||||
if srvheadhashes != [nullid]: | ||||
return [nullid], True, srvheadhashes | ||||
return [nullid], False, [] | ||||
|
r14206 | # start actual discovery (we note this before the next "if" for | ||
# compatibility reasons) | ||||
|
r14164 | ui.status(_("searching for changes\n")) | ||
|
r39197 | srvheads = [] | ||
for node in srvheadhashes: | ||||
if node == nullid: | ||||
continue | ||||
try: | ||||
srvheads.append(clrev(node)) | ||||
# Catches unknown and filtered nodes. | ||||
except error.LookupError: | ||||
continue | ||||
|
r14164 | if len(srvheads) == len(srvheadhashes): | ||
|
r14833 | ui.debug("all remote heads known locally\n") | ||
|
r39195 | return srvheadhashes, False, srvheadhashes | ||
|
r14164 | |||
|
r36733 | if len(sample) == len(ownheads) and all(yesno): | ||
|
r15497 | ui.note(_("all local heads known remotely\n")) | ||
|
r39195 | ownheadhashes = [clnode(r) for r in ownheads] | ||
return ownheadhashes, True, srvheadhashes | ||||
|
r14624 | |||
|
r14164 | # full blown discovery | ||
|
r41147 | disco = partialdiscovery(local) | ||
|
r23343 | # treat remote heads (and maybe own heads) as a first implicit sample | ||
# response | ||||
|
r41147 | disco.addcommons(srvheads) | ||
|
r23343 | commoninsample = set(n for i, n in enumerate(sample) if yesno[i]) | ||
|
r41147 | disco.addcommons(commoninsample) | ||
|
r23746 | # own nodes where I don't know if remote knows them | ||
|
r41147 | undecided = set(disco._common.missingancestors(ownheads)) | ||
|
r16683 | # own nodes I know remote lacks | ||
missing = set() | ||||
|
r14624 | full = False | ||
|
r38369 | progress = ui.makeprogress(_('searching'), unit=_('queries')) | ||
|
r14624 | while undecided: | ||
|
r14164 | |||
|
r14624 | if sample: | ||
missinginsample = [n for i, n in enumerate(sample) if not yesno[i]] | ||||
|
r39202 | |||
if missing: | ||||
missing.update(local.revs('descendants(%ld) - descendants(%ld)', | ||||
missinginsample, missing)) | ||||
else: | ||||
missing.update(local.revs('descendants(%ld)', missinginsample)) | ||||
|
r14164 | |||
|
r14624 | undecided.difference_update(missing) | ||
|
r14164 | |||
if not undecided: | ||||
break | ||||
|
r41147 | if full or disco.hasinfo(): | ||
|
r23747 | if full: | ||
ui.note(_("sampling from both directions\n")) | ||||
else: | ||||
ui.debug("taking initial sample\n") | ||||
|
r23807 | samplefunc = _takefullsample | ||
|
r23130 | targetsize = fullsamplesize | ||
|
r14624 | else: | ||
# use even cheaper initial sample | ||||
ui.debug("taking quick initial sample\n") | ||||
|
r23807 | samplefunc = _takequicksample | ||
|
r23130 | targetsize = initialsamplesize | ||
|
r41146 | sample = samplefunc(local, ownheads, undecided, targetsize) | ||
|
r14164 | |||
roundtrips += 1 | ||||
|
r38369 | progress.update(roundtrips) | ||
|
r14164 | ui.debug("query %i; still undecided: %i, sample size is: %i\n" | ||
% (roundtrips, len(undecided), len(sample))) | ||||
# indices between sample and externalized version must match | ||||
sample = list(sample) | ||||
|
r37648 | |||
with remote.commandexecutor() as e: | ||||
yesno = e.callcommand('known', { | ||||
|
r39195 | 'nodes': [clnode(r) for r in sample], | ||
|
r37648 | }).result() | ||
|
r14624 | full = True | ||
|
r14164 | |||
|
r23343 | if sample: | ||
commoninsample = set(n for i, n in enumerate(sample) if yesno[i]) | ||||
|
r41147 | disco.addcommons(commoninsample) | ||
disco._common.removeancestorsfrom(undecided) | ||||
|
r23343 | |||
# heads(common) == heads(common.bases) since common represents common.bases | ||||
# and all its ancestors | ||||
|
r39212 | # The presence of nullrev will confuse heads(). So filter it out. | ||
|
r41147 | result = set(local.revs('heads(%ld)', disco._common.bases - {nullrev})) | ||
r32712 | elapsed = util.timer() - start | |||
|
r38392 | progress.complete() | ||
r32712 | ui.debug("%d total queries in %.4fs\n" % (roundtrips, elapsed)) | |||
r32768 | msg = ('found %d common and %d unknown server heads,' | |||
' %d roundtrips in %.4fs\n') | ||||
missing = set(result) - set(srvheads) | ||||
ui.log('discovery', msg, len(result), len(missing), roundtrips, | ||||
r32713 | elapsed) | |||
|
r14164 | |||
if not result and srvheadhashes != [nullid]: | ||||
if abortwhenunrelated: | ||||
|
r26587 | raise error.Abort(_("repository is unrelated")) | ||
|
r14164 | else: | ||
ui.warn(_("warning: repository is unrelated\n")) | ||||
|
r32291 | return ({nullid}, True, srvheadhashes,) | ||
|
r14164 | |||
|
r14981 | anyincoming = (srvheadhashes != [nullid]) | ||
|
r39195 | result = {clnode(r) for r in result} | ||
return result, anyincoming, srvheadhashes | ||||