exchange.py
757 lines
| 29.8 KiB
| text/x-python
|
PythonLexer
/ mercurial / exchange.py
|
r21024 | # exchange.py - utility to exchange data between repos. | ||
|
r20345 | # | ||
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> | ||||
# | ||||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
from i18n import _ | ||||
|
r20469 | from node import hex, nullid | ||
|
r21141 | import errno, urllib | ||
|
r21184 | import util, scmutil, changegroup, base85, error | ||
|
r20955 | import discovery, phases, obsolete, bookmarks, bundle2 | ||
|
r20345 | |||
|
r21064 | def readbundle(ui, fh, fname, vfs=None): | ||
|
r21065 | header = changegroup.readexactly(fh, 4) | ||
|
r21063 | |||
|
r21065 | alg = None | ||
|
r21063 | if not fname: | ||
fname = "stream" | ||||
if not header.startswith('HG') and header.startswith('\0'): | ||||
fh = changegroup.headerlessfixup(fh, header) | ||||
|
r21065 | header = "HG10" | ||
alg = 'UN' | ||||
|
r21063 | elif vfs: | ||
fname = vfs.join(fname) | ||||
|
r21065 | magic, version = header[0:2], header[2:4] | ||
|
r21063 | |||
if magic != 'HG': | ||||
raise util.Abort(_('%s: not a Mercurial bundle') % fname) | ||||
|
r21065 | if version == '10': | ||
if alg is None: | ||||
alg = changegroup.readexactly(fh, 2) | ||||
return changegroup.unbundle10(fh, alg) | ||||
|
r21144 | elif version == '2X': | ||
|
r21067 | return bundle2.unbundle20(ui, fh, header=magic + version) | ||
|
r21065 | else: | ||
|
r21063 | raise util.Abort(_('%s: unknown bundle version %s') % (fname, version)) | ||
|
r20346 | |||
class pushoperation(object): | ||||
"""A object that represent a single push operation | ||||
It purpose is to carry push related state and very common operation. | ||||
|
r21024 | A new should be created at the beginning of each push and discarded | ||
|
r20346 | afterward. | ||
""" | ||||
|
r20351 | def __init__(self, repo, remote, force=False, revs=None, newbranch=False): | ||
|
r20346 | # repo we push from | ||
self.repo = repo | ||||
|
r20347 | self.ui = repo.ui | ||
|
r20348 | # repo we push to | ||
self.remote = remote | ||||
|
r20349 | # force option provided | ||
self.force = force | ||||
|
r20350 | # revs to be pushed (None is "all") | ||
self.revs = revs | ||||
|
r20351 | # allow push of new branch | ||
self.newbranch = newbranch | ||||
|
r20436 | # did a local lock get acquired? | ||
self.locallocked = None | ||||
|
r20439 | # Integer version of the push result | ||
# - None means nothing to push | ||||
# - 0 means HTTP error | ||||
# - 1 means we pushed and remote head count is unchanged *or* | ||||
# we have outgoing changesets but refused to push | ||||
# - other values as described by addchangegroup() | ||||
self.ret = None | ||||
|
r21024 | # discover.outgoing object (contains common and outgoing data) | ||
|
r20440 | self.outgoing = None | ||
|
r20462 | # all remote heads before the push | ||
self.remoteheads = None | ||||
|
r20464 | # testable as a boolean indicating if any nodes are missing locally. | ||
self.incoming = None | ||||
|
r20467 | # set of all heads common after changeset bundle push | ||
self.commonheads = None | ||||
|
r20346 | |||
|
r20345 | def push(repo, remote, force=False, revs=None, newbranch=False): | ||
'''Push outgoing changesets (limited by revs) from a local | ||||
repository to remote. Return an integer: | ||||
- None means nothing to push | ||||
- 0 means HTTP error | ||||
- 1 means we pushed and remote head count is unchanged *or* | ||||
we have outgoing changesets but refused to push | ||||
- other values as described by addchangegroup() | ||||
''' | ||||
|
r20351 | pushop = pushoperation(repo, remote, force, revs, newbranch) | ||
|
r20348 | if pushop.remote.local(): | ||
missing = (set(pushop.repo.requirements) | ||||
- pushop.remote.local().supported) | ||||
|
r20345 | if missing: | ||
msg = _("required features are not" | ||||
" supported in the destination:" | ||||
" %s") % (', '.join(sorted(missing))) | ||||
raise util.Abort(msg) | ||||
# there are two ways to push to remote repo: | ||||
# | ||||
# addchangegroup assumes local user can lock remote | ||||
# repo (local filesystem, old ssh servers). | ||||
# | ||||
# unbundle assumes local user cannot lock remote repo (new ssh | ||||
# servers, http servers). | ||||
|
r20348 | if not pushop.remote.canpush(): | ||
|
r20345 | raise util.Abort(_("destination does not support push")) | ||
# get local lock as we might write phase data | ||||
locallock = None | ||||
try: | ||||
|
r20346 | locallock = pushop.repo.lock() | ||
|
r20436 | pushop.locallocked = True | ||
|
r20345 | except IOError, err: | ||
|
r20436 | pushop.locallocked = False | ||
|
r20345 | if err.errno != errno.EACCES: | ||
raise | ||||
# source repo cannot be locked. | ||||
# We do not abort the push, but just disable the local phase | ||||
# synchronisation. | ||||
msg = 'cannot lock source repository: %s\n' % err | ||||
|
r20347 | pushop.ui.debug(msg) | ||
|
r20345 | try: | ||
|
r20924 | pushop.repo.checkpush(pushop) | ||
|
r20345 | lock = None | ||
|
r20348 | unbundle = pushop.remote.capable('unbundle') | ||
|
r20345 | if not unbundle: | ||
|
r20348 | lock = pushop.remote.lock() | ||
|
r20345 | try: | ||
|
r20466 | _pushdiscovery(pushop) | ||
|
r20465 | if _pushcheckoutgoing(pushop): | ||
|
r21043 | pushop.repo.prepushoutgoinghooks(pushop.repo, | ||
pushop.remote, | ||||
pushop.outgoing) | ||||
|
r21148 | if (pushop.repo.ui.configbool('experimental', 'bundle2-exp', | ||
False) | ||||
and pushop.remote.capable('bundle2-exp')): | ||||
|
r21061 | _pushbundle2(pushop) | ||
else: | ||||
|
r21062 | _pushchangeset(pushop) | ||
|
r20468 | _pushcomputecommonheads(pushop) | ||
|
r20441 | _pushsyncphase(pushop) | ||
|
r20433 | _pushobsolete(pushop) | ||
|
r20345 | finally: | ||
if lock is not None: | ||||
lock.release() | ||||
finally: | ||||
if locallock is not None: | ||||
locallock.release() | ||||
|
r20431 | _pushbookmark(pushop) | ||
|
r20439 | return pushop.ret | ||
|
r20352 | |||
|
r20466 | def _pushdiscovery(pushop): | ||
# discovery | ||||
unfi = pushop.repo.unfiltered() | ||||
fci = discovery.findcommonincoming | ||||
commoninc = fci(unfi, pushop.remote, force=pushop.force) | ||||
common, inc, remoteheads = commoninc | ||||
fco = discovery.findcommonoutgoing | ||||
outgoing = fco(unfi, pushop.remote, onlyheads=pushop.revs, | ||||
commoninc=commoninc, force=pushop.force) | ||||
pushop.outgoing = outgoing | ||||
pushop.remoteheads = remoteheads | ||||
pushop.incoming = inc | ||||
|
r20465 | def _pushcheckoutgoing(pushop): | ||
outgoing = pushop.outgoing | ||||
unfi = pushop.repo.unfiltered() | ||||
if not outgoing.missing: | ||||
# nothing to push | ||||
scmutil.nochangesfound(unfi.ui, unfi, outgoing.excluded) | ||||
return False | ||||
# something to push | ||||
if not pushop.force: | ||||
# if repo.obsstore == False --> no obsolete | ||||
# then, save the iteration | ||||
if unfi.obsstore: | ||||
# this message are here for 80 char limit reason | ||||
mso = _("push includes obsolete changeset: %s!") | ||||
mst = "push includes %s changeset: %s!" | ||||
# plain versions for i18n tool to detect them | ||||
_("push includes unstable changeset: %s!") | ||||
_("push includes bumped changeset: %s!") | ||||
_("push includes divergent changeset: %s!") | ||||
# If we are to push if there is at least one | ||||
# obsolete or unstable changeset in missing, at | ||||
# least one of the missinghead will be obsolete or | ||||
# unstable. So checking heads only is ok | ||||
for node in outgoing.missingheads: | ||||
ctx = unfi[node] | ||||
if ctx.obsolete(): | ||||
raise util.Abort(mso % ctx) | ||||
elif ctx.troubled(): | ||||
raise util.Abort(_(mst) | ||||
% (ctx.troubles()[0], | ||||
ctx)) | ||||
newbm = pushop.ui.configlist('bookmarks', 'pushing') | ||||
discovery.checkheads(unfi, pushop.remote, outgoing, | ||||
pushop.remoteheads, | ||||
pushop.newbranch, | ||||
bool(pushop.incoming), | ||||
newbm) | ||||
return True | ||||
|
r21061 | def _pushbundle2(pushop): | ||
"""push data to the remote using bundle2 | ||||
The only currently supported type of data is changegroup but this will | ||||
evolve in the future.""" | ||||
# Send known head to the server for race detection. | ||||
|
r21145 | capsblob = urllib.unquote(pushop.remote.capable('bundle2-exp')) | ||
|
r21141 | caps = bundle2.decodecaps(capsblob) | ||
bundler = bundle2.bundle20(pushop.ui, caps) | ||||
|
r21142 | # create reply capability | ||
capsblob = bundle2.encodecaps(pushop.repo.bundle2caps) | ||||
|
r21146 | bundler.addpart(bundle2.bundlepart('b2x:replycaps', data=capsblob)) | ||
|
r21061 | if not pushop.force: | ||
|
r21146 | part = bundle2.bundlepart('B2X:CHECK:HEADS', | ||
data=iter(pushop.remoteheads)) | ||||
|
r21061 | bundler.addpart(part) | ||
|
r21149 | extrainfo = _pushbundle2extraparts(pushop, bundler) | ||
|
r21061 | # add the changegroup bundle | ||
cg = changegroup.getlocalbundle(pushop.repo, 'push', pushop.outgoing) | ||||
|
r21146 | cgpart = bundle2.bundlepart('B2X:CHANGEGROUP', data=cg.getchunks()) | ||
|
r21061 | bundler.addpart(cgpart) | ||
stream = util.chunkbuffer(bundler.getchunks()) | ||||
|
r21182 | try: | ||
reply = pushop.remote.unbundle(stream, ['force'], 'push') | ||||
except bundle2.UnknownPartError, exc: | ||||
raise util.Abort('missing support for %s' % exc) | ||||
|
r21061 | try: | ||
op = bundle2.processbundle(pushop.repo, reply) | ||||
|
r21180 | except bundle2.UnknownPartError, exc: | ||
|
r21061 | raise util.Abort('missing support for %s' % exc) | ||
cgreplies = op.records.getreplies(cgpart.id) | ||||
assert len(cgreplies['changegroup']) == 1 | ||||
pushop.ret = cgreplies['changegroup'][0]['return'] | ||||
|
r21149 | _pushbundle2extrareply(pushop, op, extrainfo) | ||
def _pushbundle2extraparts(pushop, bundler): | ||||
"""hook function to let extensions add parts | ||||
Return a dict to let extensions pass data to the reply processing. | ||||
""" | ||||
return {} | ||||
def _pushbundle2extrareply(pushop, op, extrainfo): | ||||
"""hook function to let extensions react to part replies | ||||
The dict from _pushbundle2extrareply is fed to this function. | ||||
""" | ||||
pass | ||||
|
r21061 | |||
|
r20463 | def _pushchangeset(pushop): | ||
"""Make the actual push of changeset bundle to remote repo""" | ||||
outgoing = pushop.outgoing | ||||
unbundle = pushop.remote.capable('unbundle') | ||||
# TODO: get bundlecaps from remote | ||||
bundlecaps = None | ||||
# create a changegroup from local | ||||
if pushop.revs is None and not (outgoing.excluded | ||||
or pushop.repo.changelog.filteredrevs): | ||||
# push everything, | ||||
# use the fast path, no race possible on push | ||||
bundler = changegroup.bundle10(pushop.repo, bundlecaps) | ||||
|
r20925 | cg = changegroup.getsubset(pushop.repo, | ||
outgoing, | ||||
bundler, | ||||
'push', | ||||
fastpath=True) | ||||
|
r20463 | else: | ||
|
r20928 | cg = changegroup.getlocalbundle(pushop.repo, 'push', outgoing, | ||
bundlecaps) | ||||
|
r20463 | |||
# apply changegroup to remote | ||||
if unbundle: | ||||
# local repo finds heads on server, finds out what | ||||
# revs it must push. once revs transferred, if server | ||||
# finds it has different heads (someone else won | ||||
# commit/push race), server aborts. | ||||
if pushop.force: | ||||
remoteheads = ['force'] | ||||
else: | ||||
remoteheads = pushop.remoteheads | ||||
# ssh: return remote's addchangegroup() | ||||
# http: return remote's addchangegroup() or 0 for error | ||||
pushop.ret = pushop.remote.unbundle(cg, remoteheads, | ||||
'push') | ||||
else: | ||||
# we return an integer indicating remote head count | ||||
# change | ||||
|
r20971 | pushop.ret = pushop.remote.addchangegroup(cg, 'push', pushop.repo.url()) | ||
|
r20463 | |||
|
r20468 | def _pushcomputecommonheads(pushop): | ||
unfi = pushop.repo.unfiltered() | ||||
if pushop.ret: | ||||
# push succeed, synchronize target of the push | ||||
cheads = pushop.outgoing.missingheads | ||||
elif pushop.revs is None: | ||||
# All out push fails. synchronize all common | ||||
cheads = pushop.outgoing.commonheads | ||||
else: | ||||
# I want cheads = heads(::missingheads and ::commonheads) | ||||
# (missingheads is revs with secret changeset filtered out) | ||||
# | ||||
# This can be expressed as: | ||||
# cheads = ( (missingheads and ::commonheads) | ||||
# + (commonheads and ::missingheads))" | ||||
# ) | ||||
# | ||||
# while trying to push we already computed the following: | ||||
# common = (::commonheads) | ||||
# missing = ((commonheads::missingheads) - commonheads) | ||||
# | ||||
# We can pick: | ||||
# * missingheads part of common (::commonheads) | ||||
common = set(pushop.outgoing.common) | ||||
nm = pushop.repo.changelog.nodemap | ||||
cheads = [node for node in pushop.revs if nm[node] in common] | ||||
# and | ||||
# * commonheads parents on missing | ||||
revset = unfi.set('%ln and parents(roots(%ln))', | ||||
pushop.outgoing.commonheads, | ||||
pushop.outgoing.missing) | ||||
cheads.extend(c.node() for c in revset) | ||||
pushop.commonheads = cheads | ||||
|
r20441 | def _pushsyncphase(pushop): | ||
|
r21024 | """synchronise phase information locally and remotely""" | ||
|
r20441 | unfi = pushop.repo.unfiltered() | ||
|
r20468 | cheads = pushop.commonheads | ||
|
r20441 | if pushop.ret: | ||
# push succeed, synchronize target of the push | ||||
cheads = pushop.outgoing.missingheads | ||||
elif pushop.revs is None: | ||||
# All out push fails. synchronize all common | ||||
cheads = pushop.outgoing.commonheads | ||||
else: | ||||
# I want cheads = heads(::missingheads and ::commonheads) | ||||
# (missingheads is revs with secret changeset filtered out) | ||||
# | ||||
# This can be expressed as: | ||||
# cheads = ( (missingheads and ::commonheads) | ||||
# + (commonheads and ::missingheads))" | ||||
# ) | ||||
# | ||||
# while trying to push we already computed the following: | ||||
# common = (::commonheads) | ||||
# missing = ((commonheads::missingheads) - commonheads) | ||||
# | ||||
# We can pick: | ||||
# * missingheads part of common (::commonheads) | ||||
common = set(pushop.outgoing.common) | ||||
nm = pushop.repo.changelog.nodemap | ||||
cheads = [node for node in pushop.revs if nm[node] in common] | ||||
# and | ||||
# * commonheads parents on missing | ||||
revset = unfi.set('%ln and parents(roots(%ln))', | ||||
pushop.outgoing.commonheads, | ||||
pushop.outgoing.missing) | ||||
cheads.extend(c.node() for c in revset) | ||||
|
r20467 | pushop.commonheads = cheads | ||
|
r20441 | # even when we don't push, exchanging phase data is useful | ||
remotephases = pushop.remote.listkeys('phases') | ||||
if (pushop.ui.configbool('ui', '_usedassubrepo', False) | ||||
and remotephases # server supports phases | ||||
and pushop.ret is None # nothing was pushed | ||||
and remotephases.get('publishing', False)): | ||||
# When: | ||||
# - this is a subrepo push | ||||
# - and remote support phase | ||||
# - and no changeset was pushed | ||||
# - and remote is publishing | ||||
# We may be in issue 3871 case! | ||||
# We drop the possible phase synchronisation done by | ||||
# courtesy to publish changesets possibly locally draft | ||||
# on the remote. | ||||
remotephases = {'publishing': 'True'} | ||||
|
r21012 | if not remotephases: # old server or public only reply from non-publishing | ||
|
r20441 | _localphasemove(pushop, cheads) | ||
# don't push any phase data as there is nothing to push | ||||
else: | ||||
ana = phases.analyzeremotephases(pushop.repo, cheads, | ||||
remotephases) | ||||
pheads, droots = ana | ||||
### Apply remote phase on local | ||||
if remotephases.get('publishing', False): | ||||
_localphasemove(pushop, cheads) | ||||
else: # publish = False | ||||
_localphasemove(pushop, pheads) | ||||
_localphasemove(pushop, cheads, phases.draft) | ||||
### Apply local phase on remote | ||||
# Get the list of all revs draft on remote by public here. | ||||
# XXX Beware that revset break if droots is not strictly | ||||
# XXX root we may want to ensure it is but it is costly | ||||
outdated = unfi.set('heads((%ln::%ln) and public())', | ||||
droots, cheads) | ||||
for newremotehead in outdated: | ||||
r = pushop.remote.pushkey('phases', | ||||
newremotehead.hex(), | ||||
str(phases.draft), | ||||
str(phases.public)) | ||||
if not r: | ||||
pushop.ui.warn(_('updating %s to public failed!\n') | ||||
% newremotehead) | ||||
|
r20438 | def _localphasemove(pushop, nodes, phase=phases.public): | ||
"""move <nodes> to <phase> in the local source repo""" | ||||
if pushop.locallocked: | ||||
phases.advanceboundary(pushop.repo, phase, nodes) | ||||
else: | ||||
# repo is not locked, do not change any phases! | ||||
# Informs the user that phases should have been moved when | ||||
# applicable. | ||||
actualmoves = [n for n in nodes if phase < pushop.repo[n].phase()] | ||||
phasestr = phases.phasenames[phase] | ||||
if actualmoves: | ||||
pushop.ui.status(_('cannot lock source repo, skipping ' | ||||
'local %s phase update\n') % phasestr) | ||||
|
r20433 | def _pushobsolete(pushop): | ||
|
r20434 | """utility function to push obsolete markers to a remote""" | ||
|
r20435 | pushop.ui.debug('try to push obsolete markers to remote\n') | ||
|
r20433 | repo = pushop.repo | ||
remote = pushop.remote | ||||
|
r20432 | if (obsolete._enabled and repo.obsstore and | ||
'obsolete' in remote.listkeys('namespaces')): | ||||
rslts = [] | ||||
remotedata = repo.listkeys('obsolete') | ||||
for key in sorted(remotedata, reverse=True): | ||||
# reverse sort to ensure we end with dump0 | ||||
data = remotedata[key] | ||||
rslts.append(remote.pushkey('obsolete', key, '', data)) | ||||
if [r for r in rslts if not r]: | ||||
msg = _('failed to push some obsolete markers!\n') | ||||
repo.ui.warn(msg) | ||||
|
r20431 | def _pushbookmark(pushop): | ||
|
r20352 | """Update bookmark position on remote""" | ||
|
r20431 | ui = pushop.ui | ||
repo = pushop.repo.unfiltered() | ||||
remote = pushop.remote | ||||
|
r20352 | ui.debug("checking for updated bookmarks\n") | ||
|
r20431 | revnums = map(repo.changelog.rev, pushop.revs or []) | ||
|
r20352 | ancestors = [a for a in repo.changelog.ancestors(revnums, inclusive=True)] | ||
(addsrc, adddst, advsrc, advdst, diverge, differ, invalid | ||||
) = bookmarks.compare(repo, repo._bookmarks, remote.listkeys('bookmarks'), | ||||
srchex=hex) | ||||
for b, scid, dcid in advsrc: | ||||
if ancestors and repo[scid].rev() not in ancestors: | ||||
continue | ||||
if remote.pushkey('bookmarks', b, dcid, scid): | ||||
ui.status(_("updating bookmark %s\n") % b) | ||||
else: | ||||
ui.warn(_('updating bookmark %s failed!\n') % b) | ||||
|
r20469 | |||
|
r20472 | class pulloperation(object): | ||
"""A object that represent a single pull operation | ||||
It purpose is to carry push related state and very common operation. | ||||
|
r21024 | A new should be created at the beginning of each pull and discarded | ||
|
r20472 | afterward. | ||
""" | ||||
|
r20475 | def __init__(self, repo, remote, heads=None, force=False): | ||
|
r20596 | # repo we pull into | ||
|
r20472 | self.repo = repo | ||
|
r20596 | # repo we pull from | ||
|
r20473 | self.remote = remote | ||
|
r20474 | # revision we try to pull (None is "all") | ||
self.heads = heads | ||||
|
r20475 | # do we force pull? | ||
self.force = force | ||||
|
r20477 | # the name the pull transaction | ||
self._trname = 'pull\n' + util.hidepassword(remote.url()) | ||||
# hold the transaction once created | ||||
self._tr = None | ||||
|
r20487 | # set of common changeset between local and remote before pull | ||
self.common = None | ||||
# set of pulled head | ||||
self.rheads = None | ||||
|
r21024 | # list of missing changeset to fetch remotely | ||
|
r20488 | self.fetch = None | ||
|
r21024 | # result of changegroup pulling (used as return code by pull) | ||
|
r20898 | self.cgresult = None | ||
|
r20901 | # list of step remaining todo (related to future bundle2 usage) | ||
self.todosteps = set(['changegroup', 'phases', 'obsmarkers']) | ||||
|
r20487 | |||
@util.propertycache | ||||
def pulledsubset(self): | ||||
"""heads of the set of changeset target by the pull""" | ||||
# compute target subset | ||||
if self.heads is None: | ||||
# We pulled every thing possible | ||||
# sync on everything common | ||||
|
r20878 | c = set(self.common) | ||
ret = list(self.common) | ||||
for n in self.rheads: | ||||
if n not in c: | ||||
ret.append(n) | ||||
return ret | ||||
|
r20487 | else: | ||
# We pulled a specific subset | ||||
# sync on this subset | ||||
return self.heads | ||||
|
r20477 | |||
def gettransaction(self): | ||||
"""get appropriate pull transaction, creating it if needed""" | ||||
if self._tr is None: | ||||
self._tr = self.repo.transaction(self._trname) | ||||
return self._tr | ||||
def closetransaction(self): | ||||
"""close transaction if created""" | ||||
if self._tr is not None: | ||||
self._tr.close() | ||||
def releasetransaction(self): | ||||
"""release transaction if created""" | ||||
if self._tr is not None: | ||||
self._tr.release() | ||||
|
r20469 | |||
def pull(repo, remote, heads=None, force=False): | ||||
|
r20476 | pullop = pulloperation(repo, remote, heads, force) | ||
|
r20473 | if pullop.remote.local(): | ||
missing = set(pullop.remote.requirements) - pullop.repo.supported | ||||
|
r20469 | if missing: | ||
msg = _("required features are not" | ||||
" supported in the destination:" | ||||
" %s") % (', '.join(sorted(missing))) | ||||
raise util.Abort(msg) | ||||
|
r20472 | lock = pullop.repo.lock() | ||
|
r20469 | try: | ||
|
r20900 | _pulldiscovery(pullop) | ||
|
r21148 | if (pullop.repo.ui.configbool('server', 'bundle2', False) | ||
and pullop.remote.capable('bundle2-exp')): | ||||
|
r20955 | _pullbundle2(pullop) | ||
|
r20901 | if 'changegroup' in pullop.todosteps: | ||
_pullchangeset(pullop) | ||||
if 'phases' in pullop.todosteps: | ||||
_pullphase(pullop) | ||||
if 'obsmarkers' in pullop.todosteps: | ||||
_pullobsolete(pullop) | ||||
|
r20477 | pullop.closetransaction() | ||
|
r20469 | finally: | ||
|
r20477 | pullop.releasetransaction() | ||
|
r20469 | lock.release() | ||
|
r20898 | return pullop.cgresult | ||
|
r20476 | |||
|
r20900 | def _pulldiscovery(pullop): | ||
"""discovery phase for the pull | ||||
Current handle changeset discovery only, will change handle all discovery | ||||
at some point.""" | ||||
tmp = discovery.findcommonincoming(pullop.repo.unfiltered(), | ||||
pullop.remote, | ||||
heads=pullop.heads, | ||||
force=pullop.force) | ||||
pullop.common, pullop.fetch, pullop.rheads = tmp | ||||
|
r20955 | def _pullbundle2(pullop): | ||
"""pull data using bundle2 | ||||
For now, the only supported data are changegroup.""" | ||||
|
r21144 | kwargs = {'bundlecaps': set(['HG2X'])} | ||
|
r21143 | capsblob = bundle2.encodecaps(pullop.repo.bundle2caps) | ||
kwargs['bundlecaps'].add('bundle2=' + urllib.quote(capsblob)) | ||||
|
r20955 | # pulling changegroup | ||
pullop.todosteps.remove('changegroup') | ||||
if not pullop.fetch: | ||||
pullop.repo.ui.status(_("no changes found\n")) | ||||
pullop.cgresult = 0 | ||||
else: | ||||
kwargs['common'] = pullop.common | ||||
kwargs['heads'] = pullop.heads or pullop.rheads | ||||
if pullop.heads is None and list(pullop.common) == [nullid]: | ||||
pullop.repo.ui.status(_("requesting all changes\n")) | ||||
|
r21159 | _pullbundle2extraprepare(pullop, kwargs) | ||
|
r20955 | if kwargs.keys() == ['format']: | ||
return # nothing to pull | ||||
bundle = pullop.remote.getbundle('pull', **kwargs) | ||||
try: | ||||
op = bundle2.processbundle(pullop.repo, bundle, pullop.gettransaction) | ||||
|
r21181 | except UnknownPartError, exc: | ||
|
r20955 | raise util.Abort('missing support for %s' % exc) | ||
assert len(op.records['changegroup']) == 1 | ||||
pullop.cgresult = op.records['changegroup'][0]['return'] | ||||
|
r21159 | def _pullbundle2extraprepare(pullop, kwargs): | ||
"""hook function so that extensions can extend the getbundle call""" | ||||
pass | ||||
|
r20489 | def _pullchangeset(pullop): | ||
"""pull changeset from unbundle into the local repo""" | ||||
# We delay the open of the transaction as late as possible so we | ||||
# don't open transaction for nothing or you break future useful | ||||
# rollback call | ||||
|
r20901 | pullop.todosteps.remove('changegroup') | ||
|
r20899 | if not pullop.fetch: | ||
pullop.repo.ui.status(_("no changes found\n")) | ||||
pullop.cgresult = 0 | ||||
return | ||||
|
r20489 | pullop.gettransaction() | ||
if pullop.heads is None and list(pullop.common) == [nullid]: | ||||
pullop.repo.ui.status(_("requesting all changes\n")) | ||||
elif pullop.heads is None and pullop.remote.capable('changegroupsubset'): | ||||
# issue1320, avoid a race if remote changed after discovery | ||||
pullop.heads = pullop.rheads | ||||
if pullop.remote.capable('getbundle'): | ||||
# TODO: get bundlecaps from remote | ||||
cg = pullop.remote.getbundle('pull', common=pullop.common, | ||||
heads=pullop.heads or pullop.rheads) | ||||
elif pullop.heads is None: | ||||
cg = pullop.remote.changegroup(pullop.fetch, 'pull') | ||||
elif not pullop.remote.capable('changegroupsubset'): | ||||
raise util.Abort(_("partial pull cannot be done because " | ||||
"other repository doesn't support " | ||||
"changegroupsubset.")) | ||||
else: | ||||
cg = pullop.remote.changegroupsubset(pullop.fetch, pullop.heads, 'pull') | ||||
|
r20933 | pullop.cgresult = changegroup.addchangegroup(pullop.repo, cg, 'pull', | ||
|
r20899 | pullop.remote.url()) | ||
|
r20489 | |||
|
r20486 | def _pullphase(pullop): | ||
# Get remote phases data from remote | ||||
|
r20901 | pullop.todosteps.remove('phases') | ||
|
r20486 | remotephases = pullop.remote.listkeys('phases') | ||
publishing = bool(remotephases.get('publishing', False)) | ||||
if remotephases and not publishing: | ||||
# remote is new and unpublishing | ||||
pheads, _dr = phases.analyzeremotephases(pullop.repo, | ||||
pullop.pulledsubset, | ||||
remotephases) | ||||
phases.advanceboundary(pullop.repo, phases.public, pheads) | ||||
phases.advanceboundary(pullop.repo, phases.draft, | ||||
pullop.pulledsubset) | ||||
else: | ||||
# Remote is old or publishing all common changesets | ||||
# should be seen as public | ||||
phases.advanceboundary(pullop.repo, phases.public, | ||||
pullop.pulledsubset) | ||||
|
r20478 | def _pullobsolete(pullop): | ||
|
r20476 | """utility function to pull obsolete markers from a remote | ||
The `gettransaction` is function that return the pull transaction, creating | ||||
one if necessary. We return the transaction to inform the calling code that | ||||
a new transaction have been created (when applicable). | ||||
Exists mostly to allow overriding for experimentation purpose""" | ||||
|
r20901 | pullop.todosteps.remove('obsmarkers') | ||
|
r20476 | tr = None | ||
if obsolete._enabled: | ||||
|
r20478 | pullop.repo.ui.debug('fetching remote obsolete markers\n') | ||
remoteobs = pullop.remote.listkeys('obsolete') | ||||
|
r20476 | if 'dump0' in remoteobs: | ||
|
r20478 | tr = pullop.gettransaction() | ||
|
r20476 | for key in sorted(remoteobs, reverse=True): | ||
if key.startswith('dump'): | ||||
data = base85.b85decode(remoteobs[key]) | ||||
|
r20478 | pullop.repo.obsstore.mergemarkers(tr, data) | ||
pullop.repo.invalidatevolatilesets() | ||||
|
r20476 | return tr | ||
|
r21157 | def getbundle(repo, source, heads=None, common=None, bundlecaps=None, | ||
**kwargs): | ||||
|
r20954 | """return a full bundle (with potentially multiple kind of parts) | ||
|
r21144 | Could be a bundle HG10 or a bundle HG2X depending on bundlecaps | ||
|
r20954 | passed. For now, the bundle can contain only changegroup, but this will | ||
changes when more part type will be available for bundle2. | ||||
This is different from changegroup.getbundle that only returns an HG10 | ||||
changegroup bundle. They may eventually get reunited in the future when we | ||||
have a clearer idea of the API we what to query different data. | ||||
The implementation is at a very early stage and will get massive rework | ||||
when the API of bundle is refined. | ||||
""" | ||||
# build bundle here. | ||||
cg = changegroup.getbundle(repo, source, heads=heads, | ||||
|
r20956 | common=common, bundlecaps=bundlecaps) | ||
|
r21144 | if bundlecaps is None or 'HG2X' not in bundlecaps: | ||
|
r20954 | return cg | ||
# very crude first implementation, | ||||
# the bundle API will change and the generation will be done lazily. | ||||
|
r21143 | b2caps = {} | ||
for bcaps in bundlecaps: | ||||
if bcaps.startswith('bundle2='): | ||||
blob = urllib.unquote(bcaps[len('bundle2='):]) | ||||
b2caps.update(bundle2.decodecaps(blob)) | ||||
bundler = bundle2.bundle20(repo.ui, b2caps) | ||||
|
r21146 | part = bundle2.bundlepart('b2x:changegroup', data=cg.getchunks()) | ||
|
r20954 | bundler.addpart(part) | ||
|
r21158 | _getbundleextrapart(bundler, repo, source, heads=None, common=None, | ||
bundlecaps=None, **kwargs) | ||||
|
r21068 | return util.chunkbuffer(bundler.getchunks()) | ||
|
r20967 | |||
|
r21158 | def _getbundleextrapart(bundler, repo, source, heads=None, common=None, | ||
bundlecaps=None, **kwargs): | ||||
"""hook function to let extensions add parts to the requested bundle""" | ||||
pass | ||||
|
r20967 | def check_heads(repo, their_heads, context): | ||
"""check if the heads of a repo have been modified | ||||
Used by peer for unbundling. | ||||
""" | ||||
heads = repo.heads() | ||||
heads_hash = util.sha1(''.join(sorted(heads))).digest() | ||||
if not (their_heads == ['force'] or their_heads == heads or | ||||
their_heads == ['hashed', heads_hash]): | ||||
# someone else committed/pushed/unbundled while we | ||||
# were transferring data | ||||
|
r21184 | raise error.PushRaced('repository changed while %s - ' | ||
'please try again' % context) | ||||
|
r20968 | |||
def unbundle(repo, cg, heads, source, url): | ||||
"""Apply a bundle to a repo. | ||||
this function makes sure the repo is locked during the application and have | ||||
|
r21024 | mechanism to check that no push race occurred between the creation of the | ||
|
r20968 | bundle and its application. | ||
If the push was raced as PushRaced exception is raised.""" | ||||
r = 0 | ||||
|
r21061 | # need a transaction when processing a bundle2 stream | ||
tr = None | ||||
|
r20968 | lock = repo.lock() | ||
try: | ||||
check_heads(repo, heads, 'uploading changes') | ||||
# push can proceed | ||||
|
r21061 | if util.safehasattr(cg, 'params'): | ||
tr = repo.transaction('unbundle') | ||||
|
r21154 | tr.hookargs['bundle2-exp'] = '1' | ||
|
r21071 | r = bundle2.processbundle(repo, cg, lambda: tr).reply | ||
|
r21155 | cl = repo.unfiltered().changelog | ||
p = cl.writepending() and repo.root or "" | ||||
repo.hook('b2x-pretransactionclose', throw=True, source=source, | ||||
url=url, pending=p, **tr.hookargs) | ||||
|
r21061 | tr.close() | ||
|
r21156 | repo.hook('b2x-transactionclose', source=source, url=url, | ||
**tr.hookargs) | ||||
|
r21061 | else: | ||
r = changegroup.addchangegroup(repo, cg, source, url) | ||||
|
r20968 | finally: | ||
|
r21061 | if tr is not None: | ||
tr.release() | ||||
|
r20968 | lock.release() | ||
return r | ||||