##// END OF EJS Templates
discovery: always use batching now that all peers support batching...
Augie Fackler -
r25914:4d77e896 default
parent child Browse files
Show More
@@ -1,251 +1,241 b''
1 # setdiscovery.py - improved discovery of common nodeset for mercurial
1 # setdiscovery.py - improved discovery of common nodeset for mercurial
2 #
2 #
3 # Copyright 2010 Benoit Boissinot <bboissin@gmail.com>
3 # Copyright 2010 Benoit Boissinot <bboissin@gmail.com>
4 # and Peter Arrenbrecht <peter@arrenbrecht.ch>
4 # and Peter Arrenbrecht <peter@arrenbrecht.ch>
5 #
5 #
6 # This software may be used and distributed according to the terms of the
6 # This software may be used and distributed according to the terms of the
7 # GNU General Public License version 2 or any later version.
7 # GNU General Public License version 2 or any later version.
8 """
8 """
9 Algorithm works in the following way. You have two repository: local and
9 Algorithm works in the following way. You have two repository: local and
10 remote. They both contains a DAG of changelists.
10 remote. They both contains a DAG of changelists.
11
11
12 The goal of the discovery protocol is to find one set of node *common*,
12 The goal of the discovery protocol is to find one set of node *common*,
13 the set of nodes shared by local and remote.
13 the set of nodes shared by local and remote.
14
14
15 One of the issue with the original protocol was latency, it could
15 One of the issue with the original protocol was latency, it could
16 potentially require lots of roundtrips to discover that the local repo was a
16 potentially require lots of roundtrips to discover that the local repo was a
17 subset of remote (which is a very common case, you usually have few changes
17 subset of remote (which is a very common case, you usually have few changes
18 compared to upstream, while upstream probably had lots of development).
18 compared to upstream, while upstream probably had lots of development).
19
19
20 The new protocol only requires one interface for the remote repo: `known()`,
20 The new protocol only requires one interface for the remote repo: `known()`,
21 which given a set of changelists tells you if they are present in the DAG.
21 which given a set of changelists tells you if they are present in the DAG.
22
22
23 The algorithm then works as follow:
23 The algorithm then works as follow:
24
24
25 - We will be using three sets, `common`, `missing`, `unknown`. Originally
25 - We will be using three sets, `common`, `missing`, `unknown`. Originally
26 all nodes are in `unknown`.
26 all nodes are in `unknown`.
27 - Take a sample from `unknown`, call `remote.known(sample)`
27 - Take a sample from `unknown`, call `remote.known(sample)`
28 - For each node that remote knows, move it and all its ancestors to `common`
28 - For each node that remote knows, move it and all its ancestors to `common`
29 - For each node that remote doesn't know, move it and all its descendants
29 - For each node that remote doesn't know, move it and all its descendants
30 to `missing`
30 to `missing`
31 - Iterate until `unknown` is empty
31 - Iterate until `unknown` is empty
32
32
33 There are a couple optimizations, first is instead of starting with a random
33 There are a couple optimizations, first is instead of starting with a random
34 sample of missing, start by sending all heads, in the case where the local
34 sample of missing, start by sending all heads, in the case where the local
35 repo is a subset, you computed the answer in one round trip.
35 repo is a subset, you computed the answer in one round trip.
36
36
37 Then you can do something similar to the bisecting strategy used when
37 Then you can do something similar to the bisecting strategy used when
38 finding faulty changesets. Instead of random samples, you can try picking
38 finding faulty changesets. Instead of random samples, you can try picking
39 nodes that will maximize the number of nodes that will be
39 nodes that will maximize the number of nodes that will be
40 classified with it (since all ancestors or descendants will be marked as well).
40 classified with it (since all ancestors or descendants will be marked as well).
41 """
41 """
42
42
43 import collections
43 import collections
44 from node import nullid, nullrev
44 from node import nullid, nullrev
45 from i18n import _
45 from i18n import _
46 import random
46 import random
47 import util, dagutil
47 import util, dagutil
48
48
49 def _updatesample(dag, nodes, sample, quicksamplesize=0):
49 def _updatesample(dag, nodes, sample, quicksamplesize=0):
50 """update an existing sample to match the expected size
50 """update an existing sample to match the expected size
51
51
52 The sample is updated with nodes exponentially distant from each head of the
52 The sample is updated with nodes exponentially distant from each head of the
53 <nodes> set. (H~1, H~2, H~4, H~8, etc).
53 <nodes> set. (H~1, H~2, H~4, H~8, etc).
54
54
55 If a target size is specified, the sampling will stop once this size is
55 If a target size is specified, the sampling will stop once this size is
56 reached. Otherwise sampling will happen until roots of the <nodes> set are
56 reached. Otherwise sampling will happen until roots of the <nodes> set are
57 reached.
57 reached.
58
58
59 :dag: a dag object from dagutil
59 :dag: a dag object from dagutil
60 :nodes: set of nodes we want to discover (if None, assume the whole dag)
60 :nodes: set of nodes we want to discover (if None, assume the whole dag)
61 :sample: a sample to update
61 :sample: a sample to update
62 :quicksamplesize: optional target size of the sample"""
62 :quicksamplesize: optional target size of the sample"""
63 # if nodes is empty we scan the entire graph
63 # if nodes is empty we scan the entire graph
64 if nodes:
64 if nodes:
65 heads = dag.headsetofconnecteds(nodes)
65 heads = dag.headsetofconnecteds(nodes)
66 else:
66 else:
67 heads = dag.heads()
67 heads = dag.heads()
68 dist = {}
68 dist = {}
69 visit = collections.deque(heads)
69 visit = collections.deque(heads)
70 seen = set()
70 seen = set()
71 factor = 1
71 factor = 1
72 while visit:
72 while visit:
73 curr = visit.popleft()
73 curr = visit.popleft()
74 if curr in seen:
74 if curr in seen:
75 continue
75 continue
76 d = dist.setdefault(curr, 1)
76 d = dist.setdefault(curr, 1)
77 if d > factor:
77 if d > factor:
78 factor *= 2
78 factor *= 2
79 if d == factor:
79 if d == factor:
80 sample.add(curr)
80 sample.add(curr)
81 if quicksamplesize and (len(sample) >= quicksamplesize):
81 if quicksamplesize and (len(sample) >= quicksamplesize):
82 return
82 return
83 seen.add(curr)
83 seen.add(curr)
84 for p in dag.parents(curr):
84 for p in dag.parents(curr):
85 if not nodes or p in nodes:
85 if not nodes or p in nodes:
86 dist.setdefault(p, d + 1)
86 dist.setdefault(p, d + 1)
87 visit.append(p)
87 visit.append(p)
88
88
89 def _takequicksample(dag, nodes, size):
89 def _takequicksample(dag, nodes, size):
90 """takes a quick sample of size <size>
90 """takes a quick sample of size <size>
91
91
92 It is meant for initial sampling and focuses on querying heads and close
92 It is meant for initial sampling and focuses on querying heads and close
93 ancestors of heads.
93 ancestors of heads.
94
94
95 :dag: a dag object
95 :dag: a dag object
96 :nodes: set of nodes to discover
96 :nodes: set of nodes to discover
97 :size: the maximum size of the sample"""
97 :size: the maximum size of the sample"""
98 sample = dag.headsetofconnecteds(nodes)
98 sample = dag.headsetofconnecteds(nodes)
99 if size <= len(sample):
99 if size <= len(sample):
100 return _limitsample(sample, size)
100 return _limitsample(sample, size)
101 _updatesample(dag, None, sample, quicksamplesize=size)
101 _updatesample(dag, None, sample, quicksamplesize=size)
102 return sample
102 return sample
103
103
104 def _takefullsample(dag, nodes, size):
104 def _takefullsample(dag, nodes, size):
105 sample = dag.headsetofconnecteds(nodes)
105 sample = dag.headsetofconnecteds(nodes)
106 # update from heads
106 # update from heads
107 _updatesample(dag, nodes, sample)
107 _updatesample(dag, nodes, sample)
108 # update from roots
108 # update from roots
109 _updatesample(dag.inverse(), nodes, sample)
109 _updatesample(dag.inverse(), nodes, sample)
110 assert sample
110 assert sample
111 sample = _limitsample(sample, size)
111 sample = _limitsample(sample, size)
112 if len(sample) < size:
112 if len(sample) < size:
113 more = size - len(sample)
113 more = size - len(sample)
114 sample.update(random.sample(list(nodes - sample), more))
114 sample.update(random.sample(list(nodes - sample), more))
115 return sample
115 return sample
116
116
117 def _limitsample(sample, desiredlen):
117 def _limitsample(sample, desiredlen):
118 """return a random subset of sample of at most desiredlen item"""
118 """return a random subset of sample of at most desiredlen item"""
119 if len(sample) > desiredlen:
119 if len(sample) > desiredlen:
120 sample = set(random.sample(sample, desiredlen))
120 sample = set(random.sample(sample, desiredlen))
121 return sample
121 return sample
122
122
123 def findcommonheads(ui, local, remote,
123 def findcommonheads(ui, local, remote,
124 initialsamplesize=100,
124 initialsamplesize=100,
125 fullsamplesize=200,
125 fullsamplesize=200,
126 abortwhenunrelated=True):
126 abortwhenunrelated=True):
127 '''Return a tuple (common, anyincoming, remoteheads) used to identify
127 '''Return a tuple (common, anyincoming, remoteheads) used to identify
128 missing nodes from or in remote.
128 missing nodes from or in remote.
129 '''
129 '''
130 roundtrips = 0
130 roundtrips = 0
131 cl = local.changelog
131 cl = local.changelog
132 dag = dagutil.revlogdag(cl)
132 dag = dagutil.revlogdag(cl)
133
133
134 # early exit if we know all the specified remote heads already
134 # early exit if we know all the specified remote heads already
135 ui.debug("query 1; heads\n")
135 ui.debug("query 1; heads\n")
136 roundtrips += 1
136 roundtrips += 1
137 ownheads = dag.heads()
137 ownheads = dag.heads()
138 sample = _limitsample(ownheads, initialsamplesize)
138 sample = _limitsample(ownheads, initialsamplesize)
139 # indices between sample and externalized version must match
139 # indices between sample and externalized version must match
140 sample = list(sample)
140 sample = list(sample)
141 if remote.local():
142 # stopgap until we have a proper localpeer that supports batch()
143 srvheadhashes = remote.heads()
144 yesno = remote.known(dag.externalizeall(sample))
145 elif remote.capable('batch'):
146 batch = remote.batch()
141 batch = remote.batch()
147 srvheadhashesref = batch.heads()
142 srvheadhashesref = batch.heads()
148 yesnoref = batch.known(dag.externalizeall(sample))
143 yesnoref = batch.known(dag.externalizeall(sample))
149 batch.submit()
144 batch.submit()
150 srvheadhashes = srvheadhashesref.value
145 srvheadhashes = srvheadhashesref.value
151 yesno = yesnoref.value
146 yesno = yesnoref.value
152 else:
153 # compatibility with pre-batch, but post-known remotes during 1.9
154 # development
155 srvheadhashes = remote.heads()
156 sample = []
157
147
158 if cl.tip() == nullid:
148 if cl.tip() == nullid:
159 if srvheadhashes != [nullid]:
149 if srvheadhashes != [nullid]:
160 return [nullid], True, srvheadhashes
150 return [nullid], True, srvheadhashes
161 return [nullid], False, []
151 return [nullid], False, []
162
152
163 # start actual discovery (we note this before the next "if" for
153 # start actual discovery (we note this before the next "if" for
164 # compatibility reasons)
154 # compatibility reasons)
165 ui.status(_("searching for changes\n"))
155 ui.status(_("searching for changes\n"))
166
156
167 srvheads = dag.internalizeall(srvheadhashes, filterunknown=True)
157 srvheads = dag.internalizeall(srvheadhashes, filterunknown=True)
168 if len(srvheads) == len(srvheadhashes):
158 if len(srvheads) == len(srvheadhashes):
169 ui.debug("all remote heads known locally\n")
159 ui.debug("all remote heads known locally\n")
170 return (srvheadhashes, False, srvheadhashes,)
160 return (srvheadhashes, False, srvheadhashes,)
171
161
172 if sample and len(ownheads) <= initialsamplesize and all(yesno):
162 if sample and len(ownheads) <= initialsamplesize and all(yesno):
173 ui.note(_("all local heads known remotely\n"))
163 ui.note(_("all local heads known remotely\n"))
174 ownheadhashes = dag.externalizeall(ownheads)
164 ownheadhashes = dag.externalizeall(ownheads)
175 return (ownheadhashes, True, srvheadhashes,)
165 return (ownheadhashes, True, srvheadhashes,)
176
166
177 # full blown discovery
167 # full blown discovery
178
168
179 # own nodes I know we both know
169 # own nodes I know we both know
180 # treat remote heads (and maybe own heads) as a first implicit sample
170 # treat remote heads (and maybe own heads) as a first implicit sample
181 # response
171 # response
182 common = cl.incrementalmissingrevs(srvheads)
172 common = cl.incrementalmissingrevs(srvheads)
183 commoninsample = set(n for i, n in enumerate(sample) if yesno[i])
173 commoninsample = set(n for i, n in enumerate(sample) if yesno[i])
184 common.addbases(commoninsample)
174 common.addbases(commoninsample)
185 # own nodes where I don't know if remote knows them
175 # own nodes where I don't know if remote knows them
186 undecided = set(common.missingancestors(ownheads))
176 undecided = set(common.missingancestors(ownheads))
187 # own nodes I know remote lacks
177 # own nodes I know remote lacks
188 missing = set()
178 missing = set()
189
179
190 full = False
180 full = False
191 while undecided:
181 while undecided:
192
182
193 if sample:
183 if sample:
194 missinginsample = [n for i, n in enumerate(sample) if not yesno[i]]
184 missinginsample = [n for i, n in enumerate(sample) if not yesno[i]]
195 missing.update(dag.descendantset(missinginsample, missing))
185 missing.update(dag.descendantset(missinginsample, missing))
196
186
197 undecided.difference_update(missing)
187 undecided.difference_update(missing)
198
188
199 if not undecided:
189 if not undecided:
200 break
190 break
201
191
202 if full or common.hasbases():
192 if full or common.hasbases():
203 if full:
193 if full:
204 ui.note(_("sampling from both directions\n"))
194 ui.note(_("sampling from both directions\n"))
205 else:
195 else:
206 ui.debug("taking initial sample\n")
196 ui.debug("taking initial sample\n")
207 samplefunc = _takefullsample
197 samplefunc = _takefullsample
208 targetsize = fullsamplesize
198 targetsize = fullsamplesize
209 else:
199 else:
210 # use even cheaper initial sample
200 # use even cheaper initial sample
211 ui.debug("taking quick initial sample\n")
201 ui.debug("taking quick initial sample\n")
212 samplefunc = _takequicksample
202 samplefunc = _takequicksample
213 targetsize = initialsamplesize
203 targetsize = initialsamplesize
214 if len(undecided) < targetsize:
204 if len(undecided) < targetsize:
215 sample = list(undecided)
205 sample = list(undecided)
216 else:
206 else:
217 sample = samplefunc(dag, undecided, targetsize)
207 sample = samplefunc(dag, undecided, targetsize)
218 sample = _limitsample(sample, targetsize)
208 sample = _limitsample(sample, targetsize)
219
209
220 roundtrips += 1
210 roundtrips += 1
221 ui.progress(_('searching'), roundtrips, unit=_('queries'))
211 ui.progress(_('searching'), roundtrips, unit=_('queries'))
222 ui.debug("query %i; still undecided: %i, sample size is: %i\n"
212 ui.debug("query %i; still undecided: %i, sample size is: %i\n"
223 % (roundtrips, len(undecided), len(sample)))
213 % (roundtrips, len(undecided), len(sample)))
224 # indices between sample and externalized version must match
214 # indices between sample and externalized version must match
225 sample = list(sample)
215 sample = list(sample)
226 yesno = remote.known(dag.externalizeall(sample))
216 yesno = remote.known(dag.externalizeall(sample))
227 full = True
217 full = True
228
218
229 if sample:
219 if sample:
230 commoninsample = set(n for i, n in enumerate(sample) if yesno[i])
220 commoninsample = set(n for i, n in enumerate(sample) if yesno[i])
231 common.addbases(commoninsample)
221 common.addbases(commoninsample)
232 common.removeancestorsfrom(undecided)
222 common.removeancestorsfrom(undecided)
233
223
234 # heads(common) == heads(common.bases) since common represents common.bases
224 # heads(common) == heads(common.bases) since common represents common.bases
235 # and all its ancestors
225 # and all its ancestors
236 result = dag.headsetofconnecteds(common.bases)
226 result = dag.headsetofconnecteds(common.bases)
237 # common.bases can include nullrev, but our contract requires us to not
227 # common.bases can include nullrev, but our contract requires us to not
238 # return any heads in that case, so discard that
228 # return any heads in that case, so discard that
239 result.discard(nullrev)
229 result.discard(nullrev)
240 ui.progress(_('searching'), None)
230 ui.progress(_('searching'), None)
241 ui.debug("%d total queries\n" % roundtrips)
231 ui.debug("%d total queries\n" % roundtrips)
242
232
243 if not result and srvheadhashes != [nullid]:
233 if not result and srvheadhashes != [nullid]:
244 if abortwhenunrelated:
234 if abortwhenunrelated:
245 raise util.Abort(_("repository is unrelated"))
235 raise util.Abort(_("repository is unrelated"))
246 else:
236 else:
247 ui.warn(_("warning: repository is unrelated\n"))
237 ui.warn(_("warning: repository is unrelated\n"))
248 return (set([nullid]), True, srvheadhashes,)
238 return (set([nullid]), True, srvheadhashes,)
249
239
250 anyincoming = (srvheadhashes != [nullid])
240 anyincoming = (srvheadhashes != [nullid])
251 return dag.externalizeall(result), anyincoming, srvheadhashes
241 return dag.externalizeall(result), anyincoming, srvheadhashes
General Comments 0
You need to be logged in to leave comments. Login now