mercurial/revlogutils/deltas.py

51348:d58e14262587

# If a revision's full text is that much bigger than a base candidate full
# text's, it is very unlikely that it will produce a valid delta. We no longer
# consider these candidates.
LIMIT_BASE2TEXT = 500


class _BaseDeltaSearch(abc.ABC):
    """perform the search of a good delta for a single revlog revision

            snapshot_cache = SnapshotCache()
        self.snapshot_cache = snapshot_cache

        self.tested = {nullrev}

        self.current_group = None
        self._init_group()

    def is_good_delta_info(self, deltainfo):
        """Returns True if the given delta is good.

    This search variant is to be used in case where we should not store delta.
    """

    def _init_group(self):
        pass

    def next_group(self, good_delta=None):
        pass

    This search variant is to be used when the format does not allow for delta
    against arbitrary bases.
    """

    def _init_group(self):
        self.current_group = [self.target_rev - 1]

    def next_group(self, good_delta=None):
        self.current_group = None


class _DeltaSearch(_BaseDeltaSearch):
    """Generic delta search variants

            self.cachedelta is not None
            and self.cachedelta[2] > DELTA_BASE_REUSE_NO
        ):
            # Assume what we received from the server is a good choice
            # build delta will reuse the cache
            good = yield (self.cachedelta[0],)
            if good is not None:
                yield None
                return
        groups = self._raw_groups()

                break

        # If sparse revlog is enabled, we can try to refine the available
        # deltas
        if not self.revlog.delta_config.sparse_revlog:
            yield None
            return

        # if we have a refinable value, try to refine it
        if (
            good is not None
            and good not in (self.p1, self.p2)
            and self.revlog.issnapshot(good)
        ):
            # refine snapshot down
            previous = None
            while previous != good:
                previous = good
                base = self.revlog.deltaparent(good)

                children = tuple(
                    sorted(c for c in self.snapshot_cache.snapshots[good])
                )
                good = yield children

        yield None

    def _raw_groups(self):
        """Provides group of revision to be tested as delta base

        deltachain = lambda rev: self.revlog._deltachain(rev)[0]

        # exclude already lazy tested base if any
        parents = [p for p in (self.p1, self.p2) if p != nullrev]

        if (
            not self.revlog.delta_config.delta_both_parents
            and len(parents) == 2
        ):
            parents.sort()

        elif len(parents) > 0:
            # Test all parents (1 or 2), and keep the best candidate
            yield parents

        if sparse and parents:
            # See if we can use an existing snapshot in the parent chains to
            # use as a base for a new intermediate-snapshot
            #
            # search for snapshot in parents delta chain map: snapshot-level:
            # snapshot-rev

            yield tuple(sorted(full))

        if not sparse:
            # other approach failed try against prev to hopefully save us a
            # fulltext.
            yield (prev,)


class SnapshotCache:
    __slots__ = ('snapshots', '_start_rev', '_end_rev')

51349:cc806f20d756

# If a revision's full text is that much bigger than a base candidate full
# text's, it is very unlikely that it will produce a valid delta. We no longer
# consider these candidates.
LIMIT_BASE2TEXT = 500

### stage of the search, used for debug and to select and to adjust some logic.
# initial stage, next step is unknown
_STAGE_UNSPECIFIED = "unspecified"
# trying the cached delta
_STAGE_CACHED = "cached"
# trying delta based on parents
_STAGE_PARENTS = "parents"
# trying to build a valid snapshot of any level
_STAGE_SNAPSHOT = "snapshot"
# trying to build a delta based of the previous revision
_STAGE_PREV = "prev"
# trying to build a full snapshot
_STAGE_FULL = "full"


class _BaseDeltaSearch(abc.ABC):
    """perform the search of a good delta for a single revlog revision

            snapshot_cache = SnapshotCache()
        self.snapshot_cache = snapshot_cache

        self.tested = {nullrev}

        self.current_stage = _STAGE_UNSPECIFIED
        self.current_group = None
        self._init_group()

    def is_good_delta_info(self, deltainfo):
        """Returns True if the given delta is good.

    This search variant is to be used in case where we should not store delta.
    """

    def _init_group(self):
        self.current_stage = _STAGE_FULL

    def next_group(self, good_delta=None):
        pass

    This search variant is to be used when the format does not allow for delta
    against arbitrary bases.
    """

    def _init_group(self):
        self.current_stage = _STAGE_PREV
        self.current_group = [self.target_rev - 1]

    def next_group(self, good_delta=None):
        self.current_stage = _STAGE_FULL
        self.current_group = None


class _DeltaSearch(_BaseDeltaSearch):
    """Generic delta search variants

            self.cachedelta is not None
            and self.cachedelta[2] > DELTA_BASE_REUSE_NO
        ):
            # Assume what we received from the server is a good choice
            # build delta will reuse the cache
            self.current_stage = _STAGE_CACHED
            good = yield (self.cachedelta[0],)
            if good is not None:
                yield None
                return
        groups = self._raw_groups()

                break

        # If sparse revlog is enabled, we can try to refine the available
        # deltas
        if not self.revlog.delta_config.sparse_revlog:
            self.current_stage = _STAGE_FULL
            yield None
            return

        # if we have a refinable value, try to refine it
        if (
            good is not None
            and good not in (self.p1, self.p2)
            and self.revlog.issnapshot(good)
        ):
            self.current_stage = _STAGE_SNAPSHOT
            # refine snapshot down
            previous = None
            while previous != good:
                previous = good
                base = self.revlog.deltaparent(good)

                children = tuple(
                    sorted(c for c in self.snapshot_cache.snapshots[good])
                )
                good = yield children

        self.current_stage = _STAGE_FULL
        yield None

    def _raw_groups(self):
        """Provides group of revision to be tested as delta base

        deltachain = lambda rev: self.revlog._deltachain(rev)[0]

        # exclude already lazy tested base if any
        parents = [p for p in (self.p1, self.p2) if p != nullrev]

        self.current_stage = _STAGE_PARENTS
        if (
            not self.revlog.delta_config.delta_both_parents
            and len(parents) == 2
        ):
            parents.sort()

        elif len(parents) > 0:
            # Test all parents (1 or 2), and keep the best candidate
            yield parents

        if sparse and parents:
            self.current_stage = _STAGE_SNAPSHOT
            # See if we can use an existing snapshot in the parent chains to
            # use as a base for a new intermediate-snapshot
            #
            # search for snapshot in parents delta chain map: snapshot-level:
            # snapshot-rev

            yield tuple(sorted(full))

        if not sparse:
            # other approach failed try against prev to hopefully save us a
            # fulltext.
            self.current_stage = _STAGE_PREV
            yield (prev,)


class SnapshotCache:
    __slots__ = ('snapshots', '_start_rev', '_end_rev')