comparison hgext/evolve.py @ 1033:908273755215

evolve: move OBSEXC message in a dedicated function

This will help use to rework them.

1032:0fcddde7dae6

#####################################################################
### Obsolescence marker exchange experimenation                   ###
#####################################################################

@command('debugobsoleterelevant',
         [],
         'REVSET')
def debugobsoleterelevant(ui, repo, *revsets):
    """print allobsolescence marker relevant to a set of revision"""

            and remote.capable('_evoext_b2x_obsmarkers_0')):
            #
            pushop.stepsdone.add('obsmarkers')
            markers = _obsmarkersdiscovery(pushop)
            if not markers:
                repo.ui.status("OBSEXC: no marker to push\n")
                repo.ui.status("OBSEXC: DONE\n")
                return
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            repo.ui.status("OBSEXC: pushing %i markers (%i bytes)\n"
                           % (len(markers), len(obsdata.getvalue())))
            bundler.newpart('EVOLVE:B2X:OBSMARKERV1', data=obsdata)
            def callback(op):
                repo.ui.progress('OBSEXC', None)
                repo.ui.status("OBSEXC: DONE\n")
            return callback
    bundle2partsgenerators.append(_pushb2phases)
    bundle2partsgenerators.append(_pushb2obsmarker)

                               if h not in sch)
        else:
            commonheads = pushop.outgoing.missingheads
    if (obsolete._enabled and repo.obsstore and
        'obsolete' in remote.listkeys('namespaces')):
        repo.ui.status("OBSEXC: computing relevant nodes\n")
        revs = unfi.revs('::%ln', commonheads)
        common = []
        if remote.capable('_evoext_obshash_0'):
            repo.ui.status("OBSEXC: looking for common markers in %i nodes\n"
                           % len(revs))
            common = findcommonobsmarkers(pushop.ui, unfi, remote, revs)
            revs = list(unfi.revs('%ld - (::%ln)', revs, common))
        nodes = [cl.node(r) for r in revs]
        if nodes:
            repo.ui.status("OBSEXC: computing markers relevant to %i nodes\n"
                           % len(nodes))
            markers = repo.obsstore.relevantmarkers(nodes)
        else:
            repo.ui.status("OBSEXC: markers already in sync\n")
            markers = []
        return markers

@eh.wrapfunction(exchange, '_pushobsolete')
def _pushobsolete(orig, pushop):

    cl = unfi.changelog
    if (obsolete._enabled and repo.obsstore and
        'obsolete' in remote.listkeys('namespaces')):
        markers = _obsmarkersdiscovery(pushop)
        if not markers:
            repo.ui.status("OBSEXC: no marker to push\n")
        elif remote.capable('_evoext_b2x_obsmarkers_0'):
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            repo.ui.status("OBSEXC: pushing %i markers (%i bytes)\n"
                           % (len(markers), len(obsdata.getvalue())))
            bundler = bundle2.bundle20(pushop.ui, {})
            capsblob = bundle2.encodecaps(pushop.repo.bundle2caps)
            bundler.addpart(bundle2.bundlepart('b2x:replycaps', data=capsblob))
            cgpart = bundle2.bundlepart('EVOLVE:B2X:OBSMARKERV1', data=obsdata)
            bundler.addpart(cgpart)

        elif remote.capable('_evoext_pushobsmarkers_0'):
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            repo.ui.status("OBSEXC: pushing %i markers (%i bytes)\n"
                           % (len(markers), len(obsdata.getvalue())))
            remote.evoext_pushobsmarkers_0(obsdata)
            repo.ui.progress('OBSEXC', None)
        else:
            rslts = []
            remotedata = _pushkeyescape(markers).items()
            totalbytes = sum(len(d) for k,d in remotedata)
            sentbytes = 0
            repo.ui.status("OBSEXC: pushing %i markers in %i pushkey payload (%i bytes)\n"
                            % (len(markers), len(remotedata), totalbytes))
            for key, data in remotedata:
                repo.ui.progress('OBSEXC', sentbytes, item=key, unit="bytes",
                                 total=totalbytes)
                rslts.append(remote.pushkey('obsolete', key, '', data))
                sentbytes += len(data)

                                 total=totalbytes)
            repo.ui.progress('OBSEXC', None)
            if [r for r in rslts if not r]:
                msg = _('failed to push some obsolete markers!\n')
                repo.ui.warn(msg)
        repo.ui.status("OBSEXC: DONE\n")


@eh.addattr(wireproto.wirepeer, 'evoext_pushobsmarkers_0')
def client_pushobsmarkers(self, obsfile):
    """wireprotocol peer method"""

    remote = pullop.remote
    unfi = repo.unfiltered()
    revs = unfi.revs('::%ln', pullop.pulledsubset)
    common = [nullid]
    if remote.capable('_evoext_obshash_0'):
        repo.ui.status("OBSEXC: looking for common markers in %i nodes\n"
                       % len(revs))
        common = findcommonobsmarkers(repo.ui, repo, remote, revs)
    return {'heads': pullop.pulledsubset, 'common': common}

@eh.uisetup
def addgetbundleargs(self):

        return orig(pullop)
    if 'obsolete' not in pullop.remote.listkeys('namespaces'):
        return None # remote opted out of obsolescence marker exchange
    tr = None
    ui = pullop.repo.ui
    ui.status("OBSEXC: pull obsolescence markers\n")
    boundaries = _buildpullobsmerkersboundaries(pullop)
    new = 0

    if b2xpull:
        kwargs = {'bundlecaps': set(['HG2X'])}

        bytes = new = 0
        for entry in op.records['evo_obsmarkers']:
            bytes += entry.get('bytes', 0)
            new += entry.get('new', 0)
        if 5 < bytes:
            ui.status("OBSEXC: merging obsolescence markers (%i bytes)\n"
                      % bytes)
            ui.status("OBSEXC: %i markers added\n" % new)
            tr = op.gettransaction()
        else:
            ui.status("OBSEXC: no unknown remote markers\n")
        ui.status("OBSEXC: DONE\n")
    elif wirepull:
        obsdata = pullop.remote.evoext_pullobsmarkers_0(**boundaries)
        obsdata = obsdata.read()
        if len(obsdata) > 5:
            ui.status("OBSEXC: merging obsolescence markers (%i bytes)\n"
                           % len(obsdata))
            tr = pullop.gettransaction()
            old = len(pullop.repo.obsstore._all)
            pullop.repo.obsstore.mergemarkers(tr, obsdata)
            new = len(pullop.repo.obsstore._all) - old
            ui.status("OBSEXC: %i markers added\n" % new)
        else:
            ui.status("OBSEXC: no unknown remote markers\n")
        ui.status("OBSEXC: DONE\n")
    if new:
        pullop.repo.invalidatevolatilesets()
    return tr

def _getobsmarkersstream(repo, heads=None, common=None):

1033:908273755215

#####################################################################
### Obsolescence marker exchange experimenation                   ###
#####################################################################

def obsexcmsg(ui, message):
    message = 'OBSEXC: ' + message
    ui.status(message)


@command('debugobsoleterelevant',
         [],
         'REVSET')
def debugobsoleterelevant(ui, repo, *revsets):
    """print allobsolescence marker relevant to a set of revision"""

            and remote.capable('_evoext_b2x_obsmarkers_0')):
            #
            pushop.stepsdone.add('obsmarkers')
            markers = _obsmarkersdiscovery(pushop)
            if not markers:
                obsexcmsg(repo.ui, "no marker to push\n")
                obsexcmsg(repo.ui, "DONE\n")
                return
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            obsexcmsg(repo.ui, "pushing %i markers (%i bytes)\n"
                               % (len(markers), len(obsdata.getvalue())))
            bundler.newpart('EVOLVE:B2X:OBSMARKERV1', data=obsdata)
            def callback(op):
                repo.ui.progress('OBSEXC', None)
                obsexcmsg(repo.ui, "DONE\n")
            return callback
    bundle2partsgenerators.append(_pushb2phases)
    bundle2partsgenerators.append(_pushb2obsmarker)

                               if h not in sch)
        else:
            commonheads = pushop.outgoing.missingheads
    if (obsolete._enabled and repo.obsstore and
        'obsolete' in remote.listkeys('namespaces')):
        obsexcmsg(repo.ui, "computing relevant nodes\n")
        revs = unfi.revs('::%ln', commonheads)
        common = []
        if remote.capable('_evoext_obshash_0'):
            obsexcmsg(repo.ui, "looking for common markers in %i nodes\n"
                               % len(revs))
            common = findcommonobsmarkers(pushop.ui, unfi, remote, revs)
            revs = list(unfi.revs('%ld - (::%ln)', revs, common))
        nodes = [cl.node(r) for r in revs]
        if nodes:
            obsexcmsg(repo.ui, "computing markers relevant to %i nodes\n"
                               % len(nodes))
            markers = repo.obsstore.relevantmarkers(nodes)
        else:
            obsexcmsg(repo.ui, "markers already in sync\n")
            markers = []
        return markers

@eh.wrapfunction(exchange, '_pushobsolete')
def _pushobsolete(orig, pushop):

    cl = unfi.changelog
    if (obsolete._enabled and repo.obsstore and
        'obsolete' in remote.listkeys('namespaces')):
        markers = _obsmarkersdiscovery(pushop)
        if not markers:
            obsexcmsg(repo.ui, "no marker to push\n")
        elif remote.capable('_evoext_b2x_obsmarkers_0'):
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            obsexcmsg(repo.ui, "pushing %i markers (%i bytes)\n"
                               % (len(markers), len(obsdata.getvalue())))
            bundler = bundle2.bundle20(pushop.ui, {})
            capsblob = bundle2.encodecaps(pushop.repo.bundle2caps)
            bundler.addpart(bundle2.bundlepart('b2x:replycaps', data=capsblob))
            cgpart = bundle2.bundlepart('EVOLVE:B2X:OBSMARKERV1', data=obsdata)
            bundler.addpart(cgpart)

        elif remote.capable('_evoext_pushobsmarkers_0'):
            obsdata = pushobsmarkerStringIO()
            _encodemarkersstream(obsdata, markers)
            obsdata.seek(0)
            obsdata.ui = repo.ui
            obsexcmsg(repo.ui, "pushing %i markers (%i bytes)\n"
                               % (len(markers), len(obsdata.getvalue())))
            remote.evoext_pushobsmarkers_0(obsdata)
            repo.ui.progress('OBSEXC', None)
        else:
            rslts = []
            remotedata = _pushkeyescape(markers).items()
            totalbytes = sum(len(d) for k,d in remotedata)
            sentbytes = 0
            obsexcmsg(repo.ui, "pushing %i markers in %i pushkey payload (%i bytes)\n"
                                % (len(markers), len(remotedata), totalbytes))
            for key, data in remotedata:
                repo.ui.progress('OBSEXC', sentbytes, item=key, unit="bytes",
                                 total=totalbytes)
                rslts.append(remote.pushkey('obsolete', key, '', data))
                sentbytes += len(data)

                                 total=totalbytes)
            repo.ui.progress('OBSEXC', None)
            if [r for r in rslts if not r]:
                msg = _('failed to push some obsolete markers!\n')
                repo.ui.warn(msg)
        obsexcmsg(repo.ui, "DONE\n")


@eh.addattr(wireproto.wirepeer, 'evoext_pushobsmarkers_0')
def client_pushobsmarkers(self, obsfile):
    """wireprotocol peer method"""

    remote = pullop.remote
    unfi = repo.unfiltered()
    revs = unfi.revs('::%ln', pullop.pulledsubset)
    common = [nullid]
    if remote.capable('_evoext_obshash_0'):
        obsexcmsg(repo.ui, "looking for common markers in %i nodes\n"
                           % len(revs))
        common = findcommonobsmarkers(repo.ui, repo, remote, revs)
    return {'heads': pullop.pulledsubset, 'common': common}

@eh.uisetup
def addgetbundleargs(self):

        return orig(pullop)
    if 'obsolete' not in pullop.remote.listkeys('namespaces'):
        return None # remote opted out of obsolescence marker exchange
    tr = None
    ui = pullop.repo.ui
    obsexcmsg(ui, "pull obsolescence markers\n")
    boundaries = _buildpullobsmerkersboundaries(pullop)
    new = 0

    if b2xpull:
        kwargs = {'bundlecaps': set(['HG2X'])}

        bytes = new = 0
        for entry in op.records['evo_obsmarkers']:
            bytes += entry.get('bytes', 0)
            new += entry.get('new', 0)
        if 5 < bytes:
            obsexcmsg(ui, "merging obsolescence markers (%i bytes)\n"
                      % bytes)
            obsexcmsg(ui, "%i markers added\n" % new)
            tr = op.gettransaction()
        else:
            obsexcmsg(ui, "no unknown remote markers\n")
        obsexcmsg(ui, "DONE\n")
    elif wirepull:
        obsdata = pullop.remote.evoext_pullobsmarkers_0(**boundaries)
        obsdata = obsdata.read()
        if len(obsdata) > 5:
            obsexcmsg(ui, "merging obsolescence markers (%i bytes)\n"
                           % len(obsdata))
            tr = pullop.gettransaction()
            old = len(pullop.repo.obsstore._all)
            pullop.repo.obsstore.mergemarkers(tr, obsdata)
            new = len(pullop.repo.obsstore._all) - old
            obsexcmsg(ui, "%i markers added\n" % new)
        else:
            obsexcmsg(ui, "no unknown remote markers\n")
        obsexcmsg(ui, "DONE\n")
    if new:
        pullop.repo.invalidatevolatilesets()
    return tr

def _getobsmarkersstream(repo, heads=None, common=None):