##// END OF EJS Templates
changegroup: refactor outgoing logic into a function...
Durham Goode -
r21260:aa3e5660 default
parent child Browse files
Show More
@@ -1,741 +1,752 b''
1 1 # changegroup.py - Mercurial changegroup manipulation functions
2 2 #
3 3 # Copyright 2006 Matt Mackall <mpm@selenic.com>
4 4 #
5 5 # This software may be used and distributed according to the terms of the
6 6 # GNU General Public License version 2 or any later version.
7 7
8 8 import weakref
9 9 from i18n import _
10 10 from node import nullrev, nullid, hex, short
11 11 import mdiff, util, dagutil
12 12 import struct, os, bz2, zlib, tempfile
13 13 import discovery, error, phases, branchmap
14 14
15 15 _BUNDLE10_DELTA_HEADER = "20s20s20s20s"
16 16
17 17 def readexactly(stream, n):
18 18 '''read n bytes from stream.read and abort if less was available'''
19 19 s = stream.read(n)
20 20 if len(s) < n:
21 21 raise util.Abort(_("stream ended unexpectedly"
22 22 " (got %d bytes, expected %d)")
23 23 % (len(s), n))
24 24 return s
25 25
26 26 def getchunk(stream):
27 27 """return the next chunk from stream as a string"""
28 28 d = readexactly(stream, 4)
29 29 l = struct.unpack(">l", d)[0]
30 30 if l <= 4:
31 31 if l:
32 32 raise util.Abort(_("invalid chunk length %d") % l)
33 33 return ""
34 34 return readexactly(stream, l - 4)
35 35
36 36 def chunkheader(length):
37 37 """return a changegroup chunk header (string)"""
38 38 return struct.pack(">l", length + 4)
39 39
40 40 def closechunk():
41 41 """return a changegroup chunk header (string) for a zero-length chunk"""
42 42 return struct.pack(">l", 0)
43 43
44 44 class nocompress(object):
45 45 def compress(self, x):
46 46 return x
47 47 def flush(self):
48 48 return ""
49 49
50 50 bundletypes = {
51 51 "": ("", nocompress), # only when using unbundle on ssh and old http servers
52 52 # since the unification ssh accepts a header but there
53 53 # is no capability signaling it.
54 54 "HG10UN": ("HG10UN", nocompress),
55 55 "HG10BZ": ("HG10", lambda: bz2.BZ2Compressor()),
56 56 "HG10GZ": ("HG10GZ", lambda: zlib.compressobj()),
57 57 }
58 58
59 59 # hgweb uses this list to communicate its preferred type
60 60 bundlepriority = ['HG10GZ', 'HG10BZ', 'HG10UN']
61 61
62 62 def writebundle(cg, filename, bundletype, vfs=None):
63 63 """Write a bundle file and return its filename.
64 64
65 65 Existing files will not be overwritten.
66 66 If no filename is specified, a temporary file is created.
67 67 bz2 compression can be turned off.
68 68 The bundle file will be deleted in case of errors.
69 69 """
70 70
71 71 fh = None
72 72 cleanup = None
73 73 try:
74 74 if filename:
75 75 if vfs:
76 76 fh = vfs.open(filename, "wb")
77 77 else:
78 78 fh = open(filename, "wb")
79 79 else:
80 80 fd, filename = tempfile.mkstemp(prefix="hg-bundle-", suffix=".hg")
81 81 fh = os.fdopen(fd, "wb")
82 82 cleanup = filename
83 83
84 84 header, compressor = bundletypes[bundletype]
85 85 fh.write(header)
86 86 z = compressor()
87 87
88 88 # parse the changegroup data, otherwise we will block
89 89 # in case of sshrepo because we don't know the end of the stream
90 90
91 91 # an empty chunkgroup is the end of the changegroup
92 92 # a changegroup has at least 2 chunkgroups (changelog and manifest).
93 93 # after that, an empty chunkgroup is the end of the changegroup
94 94 for chunk in cg.getchunks():
95 95 fh.write(z.compress(chunk))
96 96 fh.write(z.flush())
97 97 cleanup = None
98 98 return filename
99 99 finally:
100 100 if fh is not None:
101 101 fh.close()
102 102 if cleanup is not None:
103 103 if filename and vfs:
104 104 vfs.unlink(cleanup)
105 105 else:
106 106 os.unlink(cleanup)
107 107
108 108 def decompressor(fh, alg):
109 109 if alg == 'UN':
110 110 return fh
111 111 elif alg == 'GZ':
112 112 def generator(f):
113 113 zd = zlib.decompressobj()
114 114 for chunk in util.filechunkiter(f):
115 115 yield zd.decompress(chunk)
116 116 elif alg == 'BZ':
117 117 def generator(f):
118 118 zd = bz2.BZ2Decompressor()
119 119 zd.decompress("BZ")
120 120 for chunk in util.filechunkiter(f, 4096):
121 121 yield zd.decompress(chunk)
122 122 else:
123 123 raise util.Abort("unknown bundle compression '%s'" % alg)
124 124 return util.chunkbuffer(generator(fh))
125 125
126 126 class unbundle10(object):
127 127 deltaheader = _BUNDLE10_DELTA_HEADER
128 128 deltaheadersize = struct.calcsize(deltaheader)
129 129 def __init__(self, fh, alg):
130 130 self._stream = decompressor(fh, alg)
131 131 self._type = alg
132 132 self.callback = None
133 133 def compressed(self):
134 134 return self._type != 'UN'
135 135 def read(self, l):
136 136 return self._stream.read(l)
137 137 def seek(self, pos):
138 138 return self._stream.seek(pos)
139 139 def tell(self):
140 140 return self._stream.tell()
141 141 def close(self):
142 142 return self._stream.close()
143 143
144 144 def chunklength(self):
145 145 d = readexactly(self._stream, 4)
146 146 l = struct.unpack(">l", d)[0]
147 147 if l <= 4:
148 148 if l:
149 149 raise util.Abort(_("invalid chunk length %d") % l)
150 150 return 0
151 151 if self.callback:
152 152 self.callback()
153 153 return l - 4
154 154
155 155 def changelogheader(self):
156 156 """v10 does not have a changelog header chunk"""
157 157 return {}
158 158
159 159 def manifestheader(self):
160 160 """v10 does not have a manifest header chunk"""
161 161 return {}
162 162
163 163 def filelogheader(self):
164 164 """return the header of the filelogs chunk, v10 only has the filename"""
165 165 l = self.chunklength()
166 166 if not l:
167 167 return {}
168 168 fname = readexactly(self._stream, l)
169 169 return {'filename': fname}
170 170
171 171 def _deltaheader(self, headertuple, prevnode):
172 172 node, p1, p2, cs = headertuple
173 173 if prevnode is None:
174 174 deltabase = p1
175 175 else:
176 176 deltabase = prevnode
177 177 return node, p1, p2, deltabase, cs
178 178
179 179 def deltachunk(self, prevnode):
180 180 l = self.chunklength()
181 181 if not l:
182 182 return {}
183 183 headerdata = readexactly(self._stream, self.deltaheadersize)
184 184 header = struct.unpack(self.deltaheader, headerdata)
185 185 delta = readexactly(self._stream, l - self.deltaheadersize)
186 186 node, p1, p2, deltabase, cs = self._deltaheader(header, prevnode)
187 187 return {'node': node, 'p1': p1, 'p2': p2, 'cs': cs,
188 188 'deltabase': deltabase, 'delta': delta}
189 189
190 190 def getchunks(self):
191 191 """returns all the chunks contains in the bundle
192 192
193 193 Used when you need to forward the binary stream to a file or another
194 194 network API. To do so, it parse the changegroup data, otherwise it will
195 195 block in case of sshrepo because it don't know the end of the stream.
196 196 """
197 197 # an empty chunkgroup is the end of the changegroup
198 198 # a changegroup has at least 2 chunkgroups (changelog and manifest).
199 199 # after that, an empty chunkgroup is the end of the changegroup
200 200 empty = False
201 201 count = 0
202 202 while not empty or count <= 2:
203 203 empty = True
204 204 count += 1
205 205 while True:
206 206 chunk = getchunk(self)
207 207 if not chunk:
208 208 break
209 209 empty = False
210 210 yield chunkheader(len(chunk))
211 211 pos = 0
212 212 while pos < len(chunk):
213 213 next = pos + 2**20
214 214 yield chunk[pos:next]
215 215 pos = next
216 216 yield closechunk()
217 217
218 218 class headerlessfixup(object):
219 219 def __init__(self, fh, h):
220 220 self._h = h
221 221 self._fh = fh
222 222 def read(self, n):
223 223 if self._h:
224 224 d, self._h = self._h[:n], self._h[n:]
225 225 if len(d) < n:
226 226 d += readexactly(self._fh, n - len(d))
227 227 return d
228 228 return readexactly(self._fh, n)
229 229
230 230 class bundle10(object):
231 231 deltaheader = _BUNDLE10_DELTA_HEADER
232 232 def __init__(self, repo, bundlecaps=None):
233 233 """Given a source repo, construct a bundler.
234 234
235 235 bundlecaps is optional and can be used to specify the set of
236 236 capabilities which can be used to build the bundle.
237 237 """
238 238 # Set of capabilities we can use to build the bundle.
239 239 if bundlecaps is None:
240 240 bundlecaps = set()
241 241 self._bundlecaps = bundlecaps
242 242 self._changelog = repo.changelog
243 243 self._manifest = repo.manifest
244 244 reorder = repo.ui.config('bundle', 'reorder', 'auto')
245 245 if reorder == 'auto':
246 246 reorder = None
247 247 else:
248 248 reorder = util.parsebool(reorder)
249 249 self._repo = repo
250 250 self._reorder = reorder
251 251 self._progress = repo.ui.progress
252 252 def close(self):
253 253 return closechunk()
254 254
255 255 def fileheader(self, fname):
256 256 return chunkheader(len(fname)) + fname
257 257
258 258 def group(self, nodelist, revlog, lookup, units=None, reorder=None):
259 259 """Calculate a delta group, yielding a sequence of changegroup chunks
260 260 (strings).
261 261
262 262 Given a list of changeset revs, return a set of deltas and
263 263 metadata corresponding to nodes. The first delta is
264 264 first parent(nodelist[0]) -> nodelist[0], the receiver is
265 265 guaranteed to have this parent as it has all history before
266 266 these changesets. In the case firstparent is nullrev the
267 267 changegroup starts with a full revision.
268 268
269 269 If units is not None, progress detail will be generated, units specifies
270 270 the type of revlog that is touched (changelog, manifest, etc.).
271 271 """
272 272 # if we don't have any revisions touched by these changesets, bail
273 273 if len(nodelist) == 0:
274 274 yield self.close()
275 275 return
276 276
277 277 # for generaldelta revlogs, we linearize the revs; this will both be
278 278 # much quicker and generate a much smaller bundle
279 279 if (revlog._generaldelta and reorder is not False) or reorder:
280 280 dag = dagutil.revlogdag(revlog)
281 281 revs = set(revlog.rev(n) for n in nodelist)
282 282 revs = dag.linearize(revs)
283 283 else:
284 284 revs = sorted([revlog.rev(n) for n in nodelist])
285 285
286 286 # add the parent of the first rev
287 287 p = revlog.parentrevs(revs[0])[0]
288 288 revs.insert(0, p)
289 289
290 290 # build deltas
291 291 total = len(revs) - 1
292 292 msgbundling = _('bundling')
293 293 for r in xrange(len(revs) - 1):
294 294 if units is not None:
295 295 self._progress(msgbundling, r + 1, unit=units, total=total)
296 296 prev, curr = revs[r], revs[r + 1]
297 297 linknode = lookup(revlog.node(curr))
298 298 for c in self.revchunk(revlog, curr, prev, linknode):
299 299 yield c
300 300
301 301 yield self.close()
302 302
303 303 # filter any nodes that claim to be part of the known set
304 304 def prune(self, revlog, missing, commonrevs, source):
305 305 rr, rl = revlog.rev, revlog.linkrev
306 306 return [n for n in missing if rl(rr(n)) not in commonrevs]
307 307
308 308 def generate(self, commonrevs, clnodes, fastpathlinkrev, source):
309 309 '''yield a sequence of changegroup chunks (strings)'''
310 310 repo = self._repo
311 311 cl = self._changelog
312 312 mf = self._manifest
313 313 reorder = self._reorder
314 314 progress = self._progress
315 315
316 316 # for progress output
317 317 msgbundling = _('bundling')
318 318
319 319 mfs = {} # needed manifests
320 320 fnodes = {} # needed file nodes
321 321 changedfiles = set()
322 322
323 323 # Callback for the changelog, used to collect changed files and manifest
324 324 # nodes.
325 325 # Returns the linkrev node (identity in the changelog case).
326 326 def lookupcl(x):
327 327 c = cl.read(x)
328 328 changedfiles.update(c[3])
329 329 # record the first changeset introducing this manifest version
330 330 mfs.setdefault(c[0], x)
331 331 return x
332 332
333 333 # Callback for the manifest, used to collect linkrevs for filelog
334 334 # revisions.
335 335 # Returns the linkrev node (collected in lookupcl).
336 336 def lookupmf(x):
337 337 clnode = mfs[x]
338 338 if not fastpathlinkrev:
339 339 mdata = mf.readfast(x)
340 340 for f, n in mdata.iteritems():
341 341 if f in changedfiles:
342 342 # record the first changeset introducing this filelog
343 343 # version
344 344 fnodes[f].setdefault(n, clnode)
345 345 return clnode
346 346
347 347 for chunk in self.group(clnodes, cl, lookupcl, units=_('changesets'),
348 348 reorder=reorder):
349 349 yield chunk
350 350 progress(msgbundling, None)
351 351
352 352 for f in changedfiles:
353 353 fnodes[f] = {}
354 354 mfnodes = self.prune(mf, mfs, commonrevs, source)
355 355 for chunk in self.group(mfnodes, mf, lookupmf, units=_('manifests'),
356 356 reorder=reorder):
357 357 yield chunk
358 358 progress(msgbundling, None)
359 359
360 360 mfs.clear()
361 361 needed = set(cl.rev(x) for x in clnodes)
362 362
363 363 def linknodes(filerevlog, fname):
364 364 if fastpathlinkrev:
365 365 llr = filerevlog.linkrev
366 366 def genfilenodes():
367 367 for r in filerevlog:
368 368 linkrev = llr(r)
369 369 if linkrev in needed:
370 370 yield filerevlog.node(r), cl.node(linkrev)
371 371 fnodes[fname] = dict(genfilenodes())
372 372 return fnodes.get(fname, {})
373 373
374 374 for chunk in self.generatefiles(changedfiles, linknodes, commonrevs,
375 375 source):
376 376 yield chunk
377 377
378 378 yield self.close()
379 379 progress(msgbundling, None)
380 380
381 381 if clnodes:
382 382 repo.hook('outgoing', node=hex(clnodes[0]), source=source)
383 383
384 384 def generatefiles(self, changedfiles, linknodes, commonrevs, source):
385 385 repo = self._repo
386 386 progress = self._progress
387 387 reorder = self._reorder
388 388 msgbundling = _('bundling')
389 389
390 390 total = len(changedfiles)
391 391 # for progress output
392 392 msgfiles = _('files')
393 393 for i, fname in enumerate(sorted(changedfiles)):
394 394 filerevlog = repo.file(fname)
395 395 if not filerevlog:
396 396 raise util.Abort(_("empty or missing revlog for %s") % fname)
397 397
398 398 linkrevnodes = linknodes(filerevlog, fname)
399 399 # Lookup for filenodes, we collected the linkrev nodes above in the
400 400 # fastpath case and with lookupmf in the slowpath case.
401 401 def lookupfilelog(x):
402 402 return linkrevnodes[x]
403 403
404 404 filenodes = self.prune(filerevlog, linkrevnodes, commonrevs, source)
405 405 if filenodes:
406 406 progress(msgbundling, i + 1, item=fname, unit=msgfiles,
407 407 total=total)
408 408 yield self.fileheader(fname)
409 409 for chunk in self.group(filenodes, filerevlog, lookupfilelog,
410 410 reorder=reorder):
411 411 yield chunk
412 412
413 413 def revchunk(self, revlog, rev, prev, linknode):
414 414 node = revlog.node(rev)
415 415 p1, p2 = revlog.parentrevs(rev)
416 416 base = prev
417 417
418 418 prefix = ''
419 419 if base == nullrev:
420 420 delta = revlog.revision(node)
421 421 prefix = mdiff.trivialdiffheader(len(delta))
422 422 else:
423 423 delta = revlog.revdiff(base, rev)
424 424 p1n, p2n = revlog.parents(node)
425 425 basenode = revlog.node(base)
426 426 meta = self.builddeltaheader(node, p1n, p2n, basenode, linknode)
427 427 meta += prefix
428 428 l = len(meta) + len(delta)
429 429 yield chunkheader(l)
430 430 yield meta
431 431 yield delta
432 432 def builddeltaheader(self, node, p1n, p2n, basenode, linknode):
433 433 # do nothing with basenode, it is implicitly the previous one in HG10
434 434 return struct.pack(self.deltaheader, node, p1n, p2n, linknode)
435 435
436 436 def _changegroupinfo(repo, nodes, source):
437 437 if repo.ui.verbose or source == 'bundle':
438 438 repo.ui.status(_("%d changesets found\n") % len(nodes))
439 439 if repo.ui.debugflag:
440 440 repo.ui.debug("list of changesets:\n")
441 441 for node in nodes:
442 442 repo.ui.debug("%s\n" % hex(node))
443 443
444 444 def getsubset(repo, outgoing, bundler, source, fastpath=False):
445 445 repo = repo.unfiltered()
446 446 commonrevs = outgoing.common
447 447 csets = outgoing.missing
448 448 heads = outgoing.missingheads
449 449 # We go through the fast path if we get told to, or if all (unfiltered
450 450 # heads have been requested (since we then know there all linkrevs will
451 451 # be pulled by the client).
452 452 heads.sort()
453 453 fastpathlinkrev = fastpath or (
454 454 repo.filtername is None and heads == sorted(repo.heads()))
455 455
456 456 repo.hook('preoutgoing', throw=True, source=source)
457 457 _changegroupinfo(repo, csets, source)
458 458 gengroup = bundler.generate(commonrevs, csets, fastpathlinkrev, source)
459 459 return unbundle10(util.chunkbuffer(gengroup), 'UN')
460 460
461 461 def changegroupsubset(repo, roots, heads, source):
462 462 """Compute a changegroup consisting of all the nodes that are
463 463 descendants of any of the roots and ancestors of any of the heads.
464 464 Return a chunkbuffer object whose read() method will return
465 465 successive changegroup chunks.
466 466
467 467 It is fairly complex as determining which filenodes and which
468 468 manifest nodes need to be included for the changeset to be complete
469 469 is non-trivial.
470 470
471 471 Another wrinkle is doing the reverse, figuring out which changeset in
472 472 the changegroup a particular filenode or manifestnode belongs to.
473 473 """
474 474 cl = repo.changelog
475 475 if not roots:
476 476 roots = [nullid]
477 477 # TODO: remove call to nodesbetween.
478 478 csets, roots, heads = cl.nodesbetween(roots, heads)
479 479 discbases = []
480 480 for n in roots:
481 481 discbases.extend([p for p in cl.parents(n) if p != nullid])
482 482 outgoing = discovery.outgoing(cl, discbases, heads)
483 483 bundler = bundle10(repo)
484 484 return getsubset(repo, outgoing, bundler, source)
485 485
486 486 def getlocalbundle(repo, source, outgoing, bundlecaps=None):
487 487 """Like getbundle, but taking a discovery.outgoing as an argument.
488 488
489 489 This is only implemented for local repos and reuses potentially
490 490 precomputed sets in outgoing."""
491 491 if not outgoing.missing:
492 492 return None
493 493 bundler = bundle10(repo, bundlecaps)
494 494 return getsubset(repo, outgoing, bundler, source)
495 495
496 def _computeoutgoing(repo, heads, common):
497 """Computes which revs are outgoing given a set of common
498 and a set of heads.
499
500 This is a separate function so extensions can have access to
501 the logic.
502
503 Returns a discovery.outgoing object.
504 """
505 cl = repo.changelog
506 if common:
507 hasnode = cl.hasnode
508 common = [n for n in common if hasnode(n)]
509 else:
510 common = [nullid]
511 if not heads:
512 heads = cl.heads()
513 return discovery.outgoing(cl, common, heads)
514
496 515 def getbundle(repo, source, heads=None, common=None, bundlecaps=None):
497 516 """Like changegroupsubset, but returns the set difference between the
498 517 ancestors of heads and the ancestors common.
499 518
500 519 If heads is None, use the local heads. If common is None, use [nullid].
501 520
502 521 The nodes in common might not all be known locally due to the way the
503 522 current discovery protocol works.
504 523 """
505 cl = repo.changelog
506 if common:
507 hasnode = cl.hasnode
508 common = [n for n in common if hasnode(n)]
509 else:
510 common = [nullid]
511 if not heads:
512 heads = cl.heads()
513 outgoing = discovery.outgoing(cl, common, heads)
524 outgoing = _computeoutgoing(repo, heads, common)
514 525 return getlocalbundle(repo, source, outgoing, bundlecaps=bundlecaps)
515 526
516 527 def changegroup(repo, basenodes, source):
517 528 # to avoid a race we use changegroupsubset() (issue1320)
518 529 return changegroupsubset(repo, basenodes, repo.heads(), source)
519 530
520 531 def addchangegroupfiles(repo, source, revmap, trp, pr, needfiles):
521 532 revisions = 0
522 533 files = 0
523 534 while True:
524 535 chunkdata = source.filelogheader()
525 536 if not chunkdata:
526 537 break
527 538 f = chunkdata["filename"]
528 539 repo.ui.debug("adding %s revisions\n" % f)
529 540 pr()
530 541 fl = repo.file(f)
531 542 o = len(fl)
532 543 if not fl.addgroup(source, revmap, trp):
533 544 raise util.Abort(_("received file revlog group is empty"))
534 545 revisions += len(fl) - o
535 546 files += 1
536 547 if f in needfiles:
537 548 needs = needfiles[f]
538 549 for new in xrange(o, len(fl)):
539 550 n = fl.node(new)
540 551 if n in needs:
541 552 needs.remove(n)
542 553 else:
543 554 raise util.Abort(
544 555 _("received spurious file revlog entry"))
545 556 if not needs:
546 557 del needfiles[f]
547 558 repo.ui.progress(_('files'), None)
548 559
549 560 for f, needs in needfiles.iteritems():
550 561 fl = repo.file(f)
551 562 for n in needs:
552 563 try:
553 564 fl.rev(n)
554 565 except error.LookupError:
555 566 raise util.Abort(
556 567 _('missing file data for %s:%s - run hg verify') %
557 568 (f, hex(n)))
558 569
559 570 return revisions, files
560 571
561 572 def addchangegroup(repo, source, srctype, url, emptyok=False):
562 573 """Add the changegroup returned by source.read() to this repo.
563 574 srctype is a string like 'push', 'pull', or 'unbundle'. url is
564 575 the URL of the repo where this changegroup is coming from.
565 576
566 577 Return an integer summarizing the change to this repo:
567 578 - nothing changed or no source: 0
568 579 - more heads than before: 1+added heads (2..n)
569 580 - fewer heads than before: -1-removed heads (-2..-n)
570 581 - number of heads stays the same: 1
571 582 """
572 583 repo = repo.unfiltered()
573 584 def csmap(x):
574 585 repo.ui.debug("add changeset %s\n" % short(x))
575 586 return len(cl)
576 587
577 588 def revmap(x):
578 589 return cl.rev(x)
579 590
580 591 if not source:
581 592 return 0
582 593
583 594 repo.hook('prechangegroup', throw=True, source=srctype, url=url)
584 595
585 596 changesets = files = revisions = 0
586 597 efiles = set()
587 598
588 599 # write changelog data to temp files so concurrent readers will not see
589 600 # inconsistent view
590 601 cl = repo.changelog
591 602 cl.delayupdate()
592 603 oldheads = cl.heads()
593 604
594 605 tr = repo.transaction("\n".join([srctype, util.hidepassword(url)]))
595 606 try:
596 607 trp = weakref.proxy(tr)
597 608 # pull off the changeset group
598 609 repo.ui.status(_("adding changesets\n"))
599 610 clstart = len(cl)
600 611 class prog(object):
601 612 step = _('changesets')
602 613 count = 1
603 614 ui = repo.ui
604 615 total = None
605 616 def __call__(repo):
606 617 repo.ui.progress(repo.step, repo.count, unit=_('chunks'),
607 618 total=repo.total)
608 619 repo.count += 1
609 620 pr = prog()
610 621 source.callback = pr
611 622
612 623 source.changelogheader()
613 624 srccontent = cl.addgroup(source, csmap, trp)
614 625 if not (srccontent or emptyok):
615 626 raise util.Abort(_("received changelog group is empty"))
616 627 clend = len(cl)
617 628 changesets = clend - clstart
618 629 for c in xrange(clstart, clend):
619 630 efiles.update(repo[c].files())
620 631 efiles = len(efiles)
621 632 repo.ui.progress(_('changesets'), None)
622 633
623 634 # pull off the manifest group
624 635 repo.ui.status(_("adding manifests\n"))
625 636 pr.step = _('manifests')
626 637 pr.count = 1
627 638 pr.total = changesets # manifests <= changesets
628 639 # no need to check for empty manifest group here:
629 640 # if the result of the merge of 1 and 2 is the same in 3 and 4,
630 641 # no new manifest will be created and the manifest group will
631 642 # be empty during the pull
632 643 source.manifestheader()
633 644 repo.manifest.addgroup(source, revmap, trp)
634 645 repo.ui.progress(_('manifests'), None)
635 646
636 647 needfiles = {}
637 648 if repo.ui.configbool('server', 'validate', default=False):
638 649 # validate incoming csets have their manifests
639 650 for cset in xrange(clstart, clend):
640 651 mfest = repo.changelog.read(repo.changelog.node(cset))[0]
641 652 mfest = repo.manifest.readdelta(mfest)
642 653 # store file nodes we must see
643 654 for f, n in mfest.iteritems():
644 655 needfiles.setdefault(f, set()).add(n)
645 656
646 657 # process the files
647 658 repo.ui.status(_("adding file changes\n"))
648 659 pr.step = _('files')
649 660 pr.count = 1
650 661 pr.total = efiles
651 662 source.callback = None
652 663
653 664 newrevs, newfiles = addchangegroupfiles(repo, source, revmap, trp, pr,
654 665 needfiles)
655 666 revisions += newrevs
656 667 files += newfiles
657 668
658 669 dh = 0
659 670 if oldheads:
660 671 heads = cl.heads()
661 672 dh = len(heads) - len(oldheads)
662 673 for h in heads:
663 674 if h not in oldheads and repo[h].closesbranch():
664 675 dh -= 1
665 676 htext = ""
666 677 if dh:
667 678 htext = _(" (%+d heads)") % dh
668 679
669 680 repo.ui.status(_("added %d changesets"
670 681 " with %d changes to %d files%s\n")
671 682 % (changesets, revisions, files, htext))
672 683 repo.invalidatevolatilesets()
673 684
674 685 if changesets > 0:
675 686 p = lambda: cl.writepending() and repo.root or ""
676 687 if 'node' not in tr.hookargs:
677 688 tr.hookargs['node'] = hex(cl.node(clstart))
678 689 repo.hook('pretxnchangegroup', throw=True, source=srctype,
679 690 url=url, pending=p, **tr.hookargs)
680 691
681 692 added = [cl.node(r) for r in xrange(clstart, clend)]
682 693 publishing = repo.ui.configbool('phases', 'publish', True)
683 694 if srctype in ('push', 'serve'):
684 695 # Old servers can not push the boundary themselves.
685 696 # New servers won't push the boundary if changeset already
686 697 # exists locally as secret
687 698 #
688 699 # We should not use added here but the list of all change in
689 700 # the bundle
690 701 if publishing:
691 702 phases.advanceboundary(repo, phases.public, srccontent)
692 703 else:
693 704 phases.advanceboundary(repo, phases.draft, srccontent)
694 705 phases.retractboundary(repo, phases.draft, added)
695 706 elif srctype != 'strip':
696 707 # publishing only alter behavior during push
697 708 #
698 709 # strip should not touch boundary at all
699 710 phases.retractboundary(repo, phases.draft, added)
700 711
701 712 # make changelog see real files again
702 713 cl.finalize(trp)
703 714
704 715 tr.close()
705 716
706 717 if changesets > 0:
707 718 if srctype != 'strip':
708 719 # During strip, branchcache is invalid but coming call to
709 720 # `destroyed` will repair it.
710 721 # In other case we can safely update cache on disk.
711 722 branchmap.updatecache(repo.filtered('served'))
712 723 def runhooks():
713 724 # These hooks run when the lock releases, not when the
714 725 # transaction closes. So it's possible for the changelog
715 726 # to have changed since we last saw it.
716 727 if clstart >= len(repo):
717 728 return
718 729
719 730 # forcefully update the on-disk branch cache
720 731 repo.ui.debug("updating the branch cache\n")
721 732 repo.hook("changegroup", source=srctype, url=url,
722 733 **tr.hookargs)
723 734
724 735 for n in added:
725 736 repo.hook("incoming", node=hex(n), source=srctype,
726 737 url=url)
727 738
728 739 newheads = [h for h in repo.heads() if h not in oldheads]
729 740 repo.ui.log("incoming",
730 741 "%s incoming changes - new heads: %s\n",
731 742 len(added),
732 743 ', '.join([hex(c[:6]) for c in newheads]))
733 744 repo._afterlock(runhooks)
734 745
735 746 finally:
736 747 tr.release()
737 748 # never return 0 here:
738 749 if dh < 0:
739 750 return dh - 1
740 751 else:
741 752 return dh + 1
General Comments 0
You need to be logged in to leave comments. Login now