obsdiscovery.py 30.5 KB
Newer Older
1
2
3
4
5
6
7
# Code dedicated to the discovery of obsolescence marker "over the wire"
#
# Copyright 2017 Pierre-Yves David <pierre-yves.david@ens-lyon.org>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

8
9
10
11
12
13
14
15
# Status: Experiment in progress // open question
#
#   The final discovery algorithm and protocol will go into core when we'll be
#   happy with it.
#
#   Some of the code in this module is for compatiblity with older version
#   of evolve and will be eventually dropped.

16
17
from __future__ import absolute_import

18
import hashlib
19
import heapq
20
import os
21
import sqlite3
22
import struct
23
import weakref
24

25
from mercurial.i18n import _
26
from mercurial import (
27
    encoding,
28
29
    error,
    exchange,
30
    extensions,
31
    localrepo,
32
    node as nodemod,
33
    obsolete,
34
    scmutil,
35
    store,
36
    util,
37
)
38

39
from mercurial.utils.stringutil import forcebytestr
40
41

from . import (
42
    compat,
43
    exthelper,
44
    obscache,
45
    utility,
46
    stablerange,
47
    stablerangecache,
48
49
)

50
from mercurial import wireprotov1server
51
52
from mercurial.wireprotov1peer import wirepeer
from mercurial.wireprototypes import encodelist, decodelist
53

54
55
_pack = struct.pack
_unpack = struct.unpack
56
_calcsize = struct.calcsize
57

58
59
60
eh = exthelper.exthelper()
obsexcmsg = utility.obsexcmsg

61
# Config
62
63
64
65
66
eh.configitem(b'experimental', b'evolution.obsdiscovery', True)
eh.configitem(b'experimental', b'obshashrange', True)
eh.configitem(b'experimental', b'obshashrange.warm-cache', b'auto')
eh.configitem(b'experimental', b'obshashrange.max-revs', None)
eh.configitem(b'experimental', b'obshashrange.lru-size', 2000)
67

68
69
70
##################################
###  Code performing discovery ###
##################################
71

72
73
74
75
def findmissingrange(ui, local, remote, probeset,
                     initialsamplesize=100,
                     fullsamplesize=200):
    missing = set()
76
    starttime = util.timer()
77

78
    heads = local.revs(b'heads(%ld)', probeset)
79
    local.stablerange.warmup(local)
80

81
    rangelength = local.stablerange.rangelength
82
    subranges = local.stablerange.subranges
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
    # size of slice ?
    heappop = heapq.heappop
    heappush = heapq.heappush
    heapify = heapq.heapify

    tested = set()

    sample = []
    samplesize = initialsamplesize

    def addentry(entry):
        if entry in tested:
            return False
        sample.append(entry)
        tested.add(entry)
        return True

    for h in heads:
101
        entry = (h, 0)
102
103
        addentry(entry)

104
    local.obsstore.rangeobshashcache.update(local)
105
    querycount = 0
106
107
    compat.progress(ui, _(b"comparing obsmarker with other"), querycount,
                    unit=_(b"queries"))
108
109
110
111
112
113
114
115
116
117
118
    overflow = []
    while sample or overflow:
        if overflow:
            sample.extend(overflow)
            overflow = []

        if samplesize < len(sample):
            # too much sample already
            overflow = sample[samplesize:]
            sample = sample[:samplesize]
        elif len(sample) < samplesize:
119
            ui.debug(b"query %i; add more sample (target %i, current %i)\n"
120
                     % (querycount, samplesize, len(sample)))
121
122
123
124
125
            # we need more sample !
            needed = samplesize - len(sample)
            sliceme = []
            heapify(sliceme)
            for entry in sample:
126
127
                if 1 < rangelength(local, entry):
                    heappush(sliceme, (-rangelength(local, entry), entry))
128
129
130

            while sliceme and 0 < needed:
                _key, target = heappop(sliceme)
131
                for new in subranges(local, target):
132
                    # XXX we could record hierarchy to optimise drop
133
134
135
                    if addentry(new):
                        if 1 < len(new):
                            heappush(sliceme, (-rangelength(local, new), new))
136
137
138
139
140
141
142
143
                        needed -= 1
                        if needed <= 0:
                            break

        # no longer the first interation
        samplesize = fullsamplesize

        nbsample = len(sample)
144
        maxsize = max([rangelength(local, r) for r in sample])
145
        ui.debug(b"query %i; sample size is %i, largest range %i\n"
146
                 % (querycount, nbsample, maxsize))
147
148
149
        nbreplies = 0
        replies = list(_queryrange(ui, local, remote, sample))
        sample = []
150
        n = local.changelog.node
151
152
        for entry, remotehash in replies:
            nbreplies += 1
153
            if remotehash == _obshashrange(local, entry):
154
                continue
155
            elif 1 == rangelength(local, entry):
156
                missing.add(n(entry[0]))
157
            else:
158
                for new in subranges(local, entry):
159
160
161
                    addentry(new)
        assert nbsample == nbreplies
        querycount += 1
162
163
164
        compat.progress(ui, _(b"comparing obsmarker with other"), querycount,
                        unit=_(b"queries"))
    compat.progress(ui, _(b"comparing obsmarker with other"), None)
165
    local.obsstore.rangeobshashcache.save(local)
166
    duration = util.timer() - starttime
167
168
    logmsg = (b'obsdiscovery, %d/%d mismatch'
              b' - %d obshashrange queries in %.4f seconds\n')
169
    logmsg %= (len(missing), len(probeset), querycount, duration)
170
    ui.log(b'evoext-obsdiscovery', logmsg)
171
    ui.debug(logmsg)
172
    return sorted(missing)
173
174

def _queryrange(ui, repo, remote, allentries):
175
    #  question are asked with node
176
    n = repo.changelog.node
177
    noderanges = [(n(entry[0]), entry[1]) for entry in allentries]
178
    replies = remote.evoext_obshashrange_v1(noderanges)
179
180
181
182
    result = []
    for idx, entry in enumerate(allentries):
        result.append((entry, replies[idx]))
    return result
183

184
185
186
##############################
### Range Hash computation ###
##############################
187

188
@eh.command(
189
    b'debugobshashrange',
190
    [
191
192
        (b'', b'rev', [], b'display obshash for all (rev, 0) range in REVS'),
        (b'', b'subranges', False, b'display all subranges'),
193
    ],
194
    _(b''))
195
def debugobshashrange(ui, repo, **opts):
196
197
    """display the ::REVS set topologically sorted in a stable way
    """
198
    s = nodemod.short
199
    revs = scmutil.revrange(repo, opts['rev'])
200
    # prewarm depth cache
201
202
    if revs:
        repo.stablerange.warmup(repo, max(revs))
203
    cl = repo.changelog
204
    rangelength = repo.stablerange.rangelength
205
    depthrev = repo.stablerange.depthrev
206
    if opts['subranges']:
207
        ranges = stablerange.subrangesclosure(repo, repo.stablerange, revs)
208
209
    else:
        ranges = [(r, 0) for r in revs]
210
211
212
    headers = (b'rev', b'node', b'index', b'size', b'depth', b'obshash')
    linetemplate = b'%12d %12s %12d %12d %12d %12s\n'
    headertemplate = linetemplate.replace(b'd', b's')
213
    ui.status(headertemplate % headers)
214
    repo.obsstore.rangeobshashcache.update(repo)
215
    for r in ranges:
216
        d = (r[0],
217
             s(cl.node(r[0])),
218
             r[1],
219
             rangelength(repo, r),
220
             depthrev(repo, r[0]),
221
             s(_obshashrange(repo, r)))
222
        ui.status(linetemplate % d)
223
    repo.obsstore.rangeobshashcache.save(repo)
224

225
226
def _obshashrange(repo, rangeid):
    """return the obsolete hash associated to a range"""
227
    cache = repo.obsstore.rangeobshashcache
228
    cl = repo.changelog
229
230
231
232
    obshash = cache.get(rangeid)
    if obshash is not None:
        return obshash
    pieces = []
233
    nullid = nodemod.nullid
234
    if repo.stablerange.rangelength(repo, rangeid) == 1:
235
236
        rangenode = cl.node(rangeid[0])
        tmarkers = repo.obsstore.relevantmarkers([rangenode])
237
238
239
240
241
242
        pieces = []
        for m in tmarkers:
            mbin = obsolete._fm1encodeonemarker(m)
            pieces.append(mbin)
        pieces.sort()
    else:
243
        for subrange in repo.stablerange.subranges(repo, rangeid):
244
245
246
247
248
249
250
251
            obshash = _obshashrange(repo, subrange)
            if obshash != nullid:
                pieces.append(obshash)

    sha = hashlib.sha1()
    # note: if there is only one subrange with actual data, we'll just
    # reuse the same hash.
    if not pieces:
252
        obshash = nodemod.nullid
253
254
255
256
    elif len(pieces) != 1 or obshash is None:
        sha = hashlib.sha1()
        for p in pieces:
            sha.update(p)
257
258
        obshash = sha.digest()
    cache[rangeid] = obshash
259
    return obshash
260

261
262
263
### sqlite caching

_sqliteschema = [
264
    r"""CREATE TABLE obshashrange(rev     INTEGER NOT NULL,
265
266
267
                                 idx     INTEGER NOT NULL,
                                 obshash BLOB    NOT NULL,
                                 PRIMARY KEY(rev, idx));""",
268
269
    r"CREATE INDEX range_index ON obshashrange(rev, idx);",
    r"""CREATE TABLE meta(schemaversion INTEGER NOT NULL,
270
271
                         tiprev        INTEGER NOT NULL,
                         tipnode       BLOB    NOT NULL,
272
                         nbobsmarker   INTEGER NOT NULL,
273
274
                         obssize       BLOB    NOT NULL,
                         obskey        BLOB    NOT NULL
275
276
                        );""",
]
277
278
279
_queryexist = r"SELECT name FROM sqlite_master WHERE type='table' AND name='meta';"
_clearmeta = r"""DELETE FROM meta;"""
_newmeta = r"""INSERT INTO meta (schemaversion, tiprev, tipnode, nbobsmarker, obssize, obskey)
280
            VALUES (?,?,?,?,?,?);"""
281
282
283
284
_updateobshash = r"INSERT INTO obshashrange(rev, idx, obshash) VALUES (?,?,?);"
_querymeta = r"SELECT schemaversion, tiprev, tipnode, nbobsmarker, obssize, obskey FROM meta;"
_queryobshash = r"SELECT obshash FROM obshashrange WHERE (rev = ? AND idx = ?);"
_query_max_stored = r"SELECT MAX(rev) FROM obshashrange"
285

286
287
_reset = r"DELETE FROM obshashrange;"
_delete = r"DELETE FROM obshashrange WHERE (rev = ? AND idx = ?);"
288

289
290
291
292
293
294
295
def _affectedby(repo, markers):
    """return all nodes whose relevant set is affected by this changeset

    This is a reversed version of obsstore.relevantmarkers
    """
    affected_nodes = set()
    known_markers = set(markers)
296
297
    nodes_to_proceed = set()
    markers_to_proceed = set(known_markers)
298

299
300
    successors = repo.obsstore.successors
    predecessors = repo.obsstore.predecessors
301

302
303
304
    while nodes_to_proceed or markers_to_proceed:
        while markers_to_proceed:
            marker = markers_to_proceed.pop()
305
            # check successors and parent
306
307
            if marker[1]:
                relevant = (marker[1], )
308
            else: # prune case
309
310
311
                relevant = ((marker[0], ), marker[5])
            for relnodes in relevant:
                if relnodes is None:
312
                    continue
313
314
                for node in relnodes:
                    if node not in affected_nodes:
315
                        nodes_to_proceed.add(node)
316
                    affected_nodes.add(node)
317
318
319
        # markers_to_proceed is now empty:
        if nodes_to_proceed:
            node = nodes_to_proceed.pop()
320
            markers = set()
321
322
            markers.update(successors.get(node, ()))
            markers.update(predecessors.get(node, ()))
323
            markers -= known_markers
324
            markers_to_proceed.update(markers)
325
326
327
328
            known_markers.update(markers)

    return affected_nodes

329
330
331
# if there is that many new obsmarkers, reset without analysing them
RESET_ABOVE = 10000

332
class _obshashcache(obscache.dualsourcecache):
333

334
    _schemaversion = 3
335

336
337
    _cachename = b'evo-ext-obshashrange' # used for error message
    _filename = b'evoext_obshashrange_v2.sqlite'
338

339
340
    def __init__(self, repo):
        super(_obshashcache, self).__init__()
341
        self._vfs = repo.vfs
342
        self._path = repo.cachevfs.join(self._filename)
343
344
345
346
347
        self._new = set()
        self._valid = True
        self._repo = weakref.ref(repo.unfiltered())
        # cache status
        self._ondiskcachekey = None
348
        self._data = {}
349
        self._createmode = store._calcmode(self._vfs)
350

351
    def clear(self, reset=False):
352
        super(_obshashcache, self).clear(reset=reset)
353
        self._data.clear()
354
        self._new.clear()
355
356
        if reset:
            self._valid = False
357
        if r'_con' in vars(self):
358
            del self._con
359
360

    def get(self, rangeid):
361
        # revision should be covered by the tiprev
362
363
364
        #
        # XXX there are issue with cache warming, we hack around it for now
        if not getattr(self, '_updating', False):
365
            if self._cachekey[0] < rangeid[0]:
366
                msg = (b'using unwarmed obshashrangecache (%s %s)'
367
368
                       % (rangeid[0], self._cachekey[0]))
                raise error.ProgrammingError(msg)
369

370
        value = self._data.get(rangeid)
371
372
        if value is None and self._con is not None:
            nrange = (rangeid[0], rangeid[1])
373
374
375
376
377
            try:
                obshash = self._con.execute(_queryobshash, nrange).fetchone()
                if obshash is not None:
                    value = obshash[0]
                self._data[rangeid] = value
378
            except (sqlite3.DatabaseError, sqlite3.OperationalError):
379
380
                # something is wrong with the sqlite db
                # Since this is a cache, we ignore it.
381
                if r'_con' in vars(self):
382
383
                    del self._con
                self._new.clear()
384
385
386
387
        return value

    def __setitem__(self, rangeid, obshash):
        self._new.add(rangeid)
388
        self._data[rangeid] = obshash
389

390
391
392
393
394
395
396
397
398
399
400
401
    def _updatefrom(self, repo, revs, obsmarkers):
        """override this method to update your cache data incrementally

        revs:      list of new revision in the changelog
        obsmarker: list of new obsmarkers in the obsstore
        """
        # XXX for now, we'll not actually update the cache, but we'll be
        # smarter at invalidating it.
        #
        # 1) new revisions does not get their entry updated (not update)
        # 2) if we detect markers affecting non-new revision we reset the cache

402
403
        self._updating = True

404
405
        con = self._con
        if con is not None:
406
407
408
409
            reset = False
            affected = []
            if RESET_ABOVE < len(obsmarkers):
                # lots of new obsmarkers, probably smarter to reset the cache
410
411
                repo.ui.log(b'evoext-cache', b'obshashcache reset - '
                            b'many new markers (%d)\n'
412
413
414
415
416
                            % len(obsmarkers))
                reset = True
            elif obsmarkers:
                max_stored = con.execute(_query_max_stored).fetchall()[0][0]
                affected_nodes = _affectedby(repo, obsmarkers)
417

418
419
                getrev = compat.getgetrev(repo.changelog)
                affected = [getrev(n) for n in affected_nodes]
420
421
                affected = [r for r in affected
                            if r is not None and r <= max_stored]
422

423
            if RESET_ABOVE < len(affected):
424
425
                repo.ui.log(b'evoext-cache', b'obshashcache reset - '
                            b'new markers affect many changeset (%d)\n'
426
                            % len(affected))
427
428
429
430
                reset = True

            if affected or reset:
                if not reset:
431
432
                    repo.ui.log(b'evoext-cache', b'obshashcache clean - '
                                b'new markers affect %d changeset and cached ranges\n'
433
                                % len(affected))
434
435
436
                if con is not None:
                    # always reset for now, the code detecting affect is buggy
                    # so we need to reset more broadly than we would like.
437
438
                    try:
                        if repo.stablerange._con is None:
439
440
                            repo.ui.log(b'evoext-cache', b'obshashcache reset - '
                                        b'underlying stablerange cache unavailable\n')
441
442
                            reset = True
                        if reset:
443
444
445
446
447
448
449
                            con.execute(_reset)
                            self._data.clear()
                        else:
                            ranges = repo.stablerange.contains(repo, affected)
                            con.executemany(_delete, ranges)
                            for r in ranges:
                                self._data.pop(r, None)
450
                    except (sqlite3.DatabaseError, sqlite3.OperationalError) as exc:
451
                        repo.ui.log(b'evoext-cache',
452
                                    b'error while updating obshashrange cache: %s\n'
453
                                    % forcebytestr(exc))
454
455
                        del self._updating
                        return
456

457
458
459
460
461
462
                # rewarm key revisions
                #
                # (The current invalidation is too wide, but rewarming every
                # single revision is quite costly)
                newrevs = []
                stop = self._cachekey[0] # tiprev
463
                for h in repo.filtered(b'immutable').changelog.headrevs():
464
465
466
467
                    if h <= stop and h in affected:
                        newrevs.append(h)
                newrevs.extend(revs)
                revs = newrevs
468

469
        repo.depthcache.update(repo)
470
471
        total = len(revs)

472
        def progress(pos, rev=None):
473
474
            revstr = b'' if rev is None else (b'rev %d' % rev)
            compat.progress(repo.ui, b'updating obshashrange cache',
475
                            pos, revstr, unit=_(b'changesets'), total=total)
476
        # warm the cache for the new revs
477
        progress(0)
478
        for idx, r in enumerate(revs):
479
            _obshashrange(repo, (r, 0))
480
            progress(idx, r)
481
        progress(None)
482
483
484

        del self._updating

485
486
487
488
489
490
491
492
493
    @property
    def _fullcachekey(self):
        return (self._schemaversion, ) + self._cachekey

    def load(self, repo):
        if self._con is None:
            self._cachekey = self.emptykey
            self._ondiskcachekey = self.emptykey
        assert self._cachekey is not None
494

495
496
    def _db(self):
        try:
497
            util.makedirs(self._vfs.dirname(self._path), self._createmode)
498
499
        except OSError:
            return None
500
501
        if self._createmode is not None:
            pre_existed = os.access(self._path, os.R_OK)
502
        con = sqlite3.connect(encoding.strfromlocal(self._path), timeout=30,
503
                              isolation_level=r"IMMEDIATE")
504
        con.text_factory = bytes
505
506
507
508
509
        if self._createmode is not None and not pre_existed:
            try:
                os.chmod(self._path, self._createmode & 0o666)
            except OSError:
                pass
510
511
        return con

512
513
514
515
516
517
518
    @util.propertycache
    def _con(self):
        if not self._valid:
            return None
        repo = self._repo()
        if repo is None:
            return None
519
520
521
        con = self._db()
        if con is None:
            return None
522
523
524
525
526
        cur = con.execute(_queryexist)
        if cur.fetchone() is None:
            self._valid = False
            return None
        meta = con.execute(_querymeta).fetchone()
527
        if meta is None or meta[0] != self._schemaversion:
528
529
            self._valid = False
            return None
530
        self._cachekey = self._ondiskcachekey = meta[1:]
531
532
533
        return con

    def save(self, repo):
534
535
536
537
        if self._cachekey is None:
            return
        if self._cachekey == self._ondiskcachekey and not self._new:
            return
538
539
540
        repo = repo.unfiltered()
        try:
            with repo.lock():
541
                if r'stablerange' in vars(repo):
542
                    repo.stablerange.save(repo)
543
544
545
546
547
                self._save(repo)
        except error.LockError:
            # Exceptionnally we are noisy about it since performance impact
            # is large We should address that before using this more
            # widely.
548
            msg = _(b'obshashrange cache: skipping save unable to lock repo\n')
549
550
551
            repo.ui.warn(msg)

    def _save(self, repo):
552
553
        if not self._new:
            return
554
555
        try:
            return self._trysave(repo)
556
        except (IOError, OSError, sqlite3.DatabaseError, sqlite3.OperationalError, sqlite3.IntegrityError) as exc:
557
558
559
560
            # Catch error that may arise under stress
            #
            # operational error catch read-only and locked database
            # IntegrityError catch Unique constraint error that may arise
561
            if r'_con' in vars(self):
562
563
                del self._con
            self._new.clear()
564
            repo.ui.log(b'evoext-cache',
565
                        b'error while saving new data: %s\n'
566
                        % forcebytestr(exc))
567
            repo.ui.debug(b'evoext-cache: error while saving new data: %s\n'
568
                          % forcebytestr(exc))
569
570

    def _trysave(self, repo):
571
572
        if self._con is None:
            util.unlinkpath(self._path, ignoremissing=True)
573
            if r'_con' in vars(self):
574
575
                del self._con

576
577
            con = self._db()
            if con is None:
578
                repo.ui.log(b'evoext-cache', b'unable to write obshashrange cache'
579
                            b' - cannot create database\n')
580
                return
581
582
583
584
            with con:
                for req in _sqliteschema:
                    con.execute(req)

585
586
587
                meta = [self._schemaversion] + list(self.emptykey)
                con.execute(_newmeta, meta)
                self._ondiskcachekey = self.emptykey
588
589
        else:
            con = self._con
590
        with con:
591
592
593
594
595
            meta = con.execute(_querymeta).fetchone()
            if meta[1:] != self._ondiskcachekey:
                # drifting is currently an issue because this means another
                # process might have already added the cache line we are about
                # to add. This will confuse sqlite
596
597
                msg = _(b'obshashrange cache: skipping write, '
                        b'database drifted under my feet\n')
598
                repo.ui.warn(msg)
599
600
                self._new.clear()
                self._valid = False
601
                if r'_con' in vars(self):
602
603
                    del self._con
                self._valid = False
604
                return
605
606
607
608
609
610
611
612
            data = ((rangeid[0], rangeid[1], self.get(rangeid)) for rangeid in self._new)
            con.executemany(_updateobshash, data)
            cachekey = self._fullcachekey
            con.execute(_clearmeta) # remove the older entry
            con.execute(_newmeta, cachekey)
            self._new.clear()
            self._valid = True
            self._ondiskcachekey = self._cachekey
613

614
615
616
617
618
@eh.wrapfunction(obsolete.obsstore, '_addmarkers')
def _addmarkers(orig, obsstore, *args, **kwargs):
    obsstore.rangeobshashcache.clear()
    return orig(obsstore, *args, **kwargs)

619
obsstorefilecache = localrepo.localrepository.obsstore
620
621


622
# obsstore is a filecache so we have do to some spacial dancing
623
@eh.wrapfunction(obsstorefilecache, 'func')
624
625
626
627
628
629
630
631
632
633
634
def obsstorewithcache(orig, repo):
    obsstore = orig(repo)
    obsstore.rangeobshashcache = _obshashcache(repo.unfiltered())
    return obsstore

@eh.reposetup
def setupcache(ui, repo):

    class obshashrepo(repo.__class__):
        @localrepo.unfilteredmethod
        def destroyed(self):
635
            if r'obsstore' in vars(self):
636
                self.obsstore.rangeobshashcache.clear()
637
638
639
640
641
642
643
644
            toplevel = not util.safehasattr(self, '_destroying')
            if toplevel:
                self._destroying = True
            try:
                super(obshashrepo, self).destroyed()
            finally:
                if toplevel:
                    del self._destroying
645

646
647
648
649
650
651
        @localrepo.unfilteredmethod
        def updatecaches(self, tr=None, **kwargs):
            if utility.shouldwarmcache(self, tr):
                self.obsstore.rangeobshashcache.update(self)
                self.obsstore.rangeobshashcache.save(self)
            super(obshashrepo, self).updatecaches(tr, **kwargs)
652

653
    repo.__class__ = obshashrepo
654

655
656
657
658
659
660
661
662
### wire protocol commands

def _obshashrange_v0(repo, ranges):
    """return a list of hash from a list of range

    The range have the id encoded as a node

    return 'wdirid' for unknown range"""
663
664
    getrev = compat.getgetrev(repo.changelog)
    ranges = [(getrev(n), idx) for n, idx in ranges]
665
666
667
668
669
    if ranges:
        maxrev = max(r for r, i in ranges)
        if maxrev is not None:
            repo.stablerange.warmup(repo, upto=maxrev)
    result = []
670
    repo.obsstore.rangeobshashcache.update(repo)
671
672
    for r in ranges:
        if r[0] is None:
673
            result.append(nodemod.wdirid)
674
675
676
677
678
        else:
            result.append(_obshashrange(repo, r))
    repo.obsstore.rangeobshashcache.save(repo)
    return result

679
@eh.addattr(localrepo.localpeer, 'evoext_obshashrange_v1')
680
681
682
683
def local_obshashrange_v0(peer, ranges):
    return _obshashrange_v0(peer._repo, ranges)


684
_indexformat = b'>I'
685
686
687
688
689
690
691
692
693
694
695
696
697
_indexsize = _calcsize(_indexformat)
def _encrange(node_rangeid):
    """encode a (node) range"""
    headnode, index = node_rangeid
    return headnode + _pack(_indexformat, index)

def _decrange(data):
    """encode a (node) range"""
    assert _indexsize < len(data), len(data)
    headnode = data[:-_indexsize]
    index = _unpack(_indexformat, data[-_indexsize:])[0]
    return (headnode, index)

698
@eh.addattr(wirepeer, 'evoext_obshashrange_v1')
699
700
def peer_obshashrange_v0(self, ranges):
    binranges = [_encrange(r) for r in ranges]
701
    encranges = encodelist(binranges)
702
    d = self._call(b"evoext_obshashrange_v1", ranges=encranges)
703
    try:
704
        return decodelist(d)
705
    except ValueError:
706
        self._abort(error.ResponseError(_(b"unexpected response:"), d))
707

708
@wireprotov1server.wireprotocommand(b'evoext_obshashrange_v1', b'ranges', b'pull')
709
def srv_obshashrange_v1(repo, proto, ranges):
710
    ranges = decodelist(ranges)
711
712
    ranges = [_decrange(r) for r in ranges]
    hashes = _obshashrange_v0(repo, ranges)
713
    return encodelist(hashes)
714

715
def _useobshashrange(repo):
716
    base = repo.ui.configbool(b'experimental', b'obshashrange')
717
    if base:
718
        maxrevs = repo.ui.configint(b'experimental', b'obshashrange.max-revs')
719
720
721
        if maxrevs is not None and maxrevs < len(repo.unfiltered()):
            base = False
    return base
722
723

def _canobshashrange(local, remote):
724
    return (_useobshashrange(local)
725
            and remote.capable(b'_evoext_obshashrange_v1'))
726
727
728
729

def _obshashrange_capabilities(orig, repo, proto):
    """wrapper to advertise new capability"""
    caps = orig(repo, proto)
730
    enabled = _useobshashrange(repo)
731
    if obsolete.isenabled(repo, obsolete.exchangeopt) and enabled:
732
        caps.append(b'_evoext_obshashrange_v1')
733
734
735
736
    return caps

@eh.extsetup
def obshashrange_extsetup(ui):
737
    extensions.wrapfunction(wireprotov1server, '_capabilities',
738
                            _obshashrange_capabilities)
739

740
741
742
743
744
745
746
747
748
749
##########################################
###  trigger discovery during exchange ###
##########################################

def _dopushmarkers(pushop):
    return (# we have any markers to push
            pushop.repo.obsstore
            # exchange of obsmarkers is enabled locally
            and obsolete.isenabled(pushop.repo, obsolete.exchangeopt)
            # remote server accept markers
750
            and b'obsolete' in pushop.remote.listkeys(b'namespaces'))
751

752
753
754
755
756
757
758
def _pushobshashrange(pushop, commonrevs):
    repo = pushop.repo.unfiltered()
    remote = pushop.remote
    missing = findmissingrange(pushop.ui, repo, remote, commonrevs)
    missing += pushop.outgoing.missing
    return missing

759
760
761
762
763
764
# available discovery method, first valid is used
# tuple (canuse, perform discovery))
obsdiscoveries = [
    (_canobshashrange, _pushobshashrange),
]

765
obsdiscovery_skip_message = b"""\
766
767
768
769
770
(skipping discovery of obsolescence markers, will exchange everything)
(controled by 'experimental.evolution.obsdiscovery' configuration)
"""

def usediscovery(repo):
771
    return repo.ui.configbool(b'experimental', b'evolution.obsdiscovery')
772

773
774
775
776
@eh.wrapfunction(exchange, '_pushdiscoveryobsmarkers')
def _pushdiscoveryobsmarkers(orig, pushop):
    if _dopushmarkers(pushop):
        repo = pushop.repo
777
        remote = pushop.remote
778
779
        obsexcmsg(repo.ui, b"computing relevant nodes\n")
        revs = list(repo.revs(b'::%ln', pushop.futureheads))
780
        unfi = repo.unfiltered()
781

782
783
784
785
786
        if not usediscovery(repo):
            # discovery disabled by user
            repo.ui.status(obsdiscovery_skip_message)
            return orig(pushop)

787
788
789
790
791
792
793
794
795
796
        # look for an obs-discovery protocol we can use
        discovery = None
        for candidate in obsdiscoveries:
            if candidate[0](repo, remote):
                discovery = candidate[1]
                break

        if discovery is None:
            # no discovery available, rely on core to push all relevants
            # obs markers.
797
798
            return orig(pushop)

799
        obsexcmsg(repo.ui, b"looking for common markers in %i nodes\n"
800
                           % len(revs))
801
        commonrevs = list(unfi.revs(b'::%ln', pushop.outgoing.commonheads))
802
803
        # find the nodes where the relevant obsmarkers mismatches
        nodes = discovery(pushop, commonrevs)
804
805

        if nodes:
806
            obsexcmsg(repo.ui, b"computing markers relevant to %i nodes\n"
807
808
809
                               % len(nodes))
            pushop.outobsmarkers = repo.obsstore.relevantmarkers(nodes)
        else:
810
            obsexcmsg(repo.ui, b"markers already in sync\n")
811
812
813
814
            pushop.outobsmarkers = []

@eh.extsetup
def _installobsmarkersdiscovery(ui):
815
    olddisco = exchange.pushdiscoverymapping[b'obsmarker']
816
817
818

    def newdisco(pushop):
        _pushdiscoveryobsmarkers(olddisco, pushop)
819
    exchange.pushdiscoverymapping[b'obsmarker'] = newdisco
820
821
822
823
824
825
826
827
828

def buildpullobsmarkersboundaries(pullop, bundle2=True):
    """small function returning the argument for pull markers call
    may to contains 'heads' and 'common'. skip the key for None.

    It is a separed function to play around with strategy for that."""
    repo = pullop.repo
    remote = pullop.remote
    unfi = repo.unfiltered()
829
830
831
832
833
834
    # Also exclude filtered revisions. Working on unfiltered repository can
    # give a bit more precise view of the repository. However it makes the
    # overall operation more complicated.
    filteredrevs = repo.changelog.filteredrevs
    # XXX probably not very efficient
    revs = unfi.revs(b'::(%ln - null) - %ld', pullop.common, filteredrevs)
835
    boundaries = {b'heads': pullop.pulledsubset}
836
    if not revs: # nothing common
837
        boundaries[b'common'] = [nodemod.nullid]
838
839
        return boundaries

840
841
842
    if not usediscovery(repo):
        # discovery disabled by users.
        repo.ui.status(obsdiscovery_skip_message)
843
        boundaries[b'common'] = [nodemod.nullid]
844
845
        return boundaries

846
    if bundle2 and _canobshashrange(repo, remote):
847
        obsexcmsg(repo.ui, b"looking for common markers in %i nodes\n"
848
                  % len(revs))
849
850
        missing = findmissingrange(repo.ui, repo, pullop.remote, revs)
        boundaries[b'missing'] = missing
851
852
853
854
855
856
        # using getattr since `limitedarguments` is missing
        # hg <= 5.0 (69921d02daaf)
        if getattr(pullop.remote, 'limitedarguments', False):
            # prepare for a possible fallback to common
            common = repo.set("heads(only(%ld, %ln))", revs, missing)
            boundaries[b'common'] = [c.node() for c in common]
857
    else:
858
        boundaries[b'common'] = [nodemod.nullid]
859
    return boundaries
860
861
862

# merge later for outer layer wrapping
eh.merge(stablerangecache.eh)