# 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.

# 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.

from __future__ import absolute_import

import hashlib

import heapq

import os

import sqlite3

import struct

import weakref

from mercurial import (

    encoding,

    error,

    exchange,

    extensions,

    localrepo,

    node,

    obsolete,

    scmutil,

    store,

    util,

)

from mercurial.i18n import _

from mercurial.utils.stringutil import forcebytestr

from . import (

    compat,

    exthelper,

    obscache,

    utility,

    stablerange,

    stablerangecache,

)

from mercurial import wireprototypes, wireprotov1server

from mercurial.wireprotov1peer import wirepeer

from mercurial.wireprototypes import encodelist, decodelist

_pack = struct.pack

_unpack = struct.unpack

_calcsize = struct.calcsize

eh = exthelper.exthelper()

obsexcmsg = utility.obsexcmsg

# Config

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)

##################################

###  Code performing discovery ###

##################################

def findmissingrange(ui, local, remote, probeset,

                     initialsamplesize=100,

                     fullsamplesize=200):

    missing = set()

    starttime = util.timer()

    heads = local.revs(b'heads(%ld)', probeset)

    local.stablerange.warmup(local)

    rangelength = local.stablerange.rangelength

    subranges = local.stablerange.subranges

    # 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:

        entry = (h, 0)

        addentry(entry)

    local.obsstore.rangeobshashcache.update(local)

    querycount = 0

    compat.progress(ui, _(b"comparing obsmarker with other"), querycount,

                    unit=_(b"queries"))

    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:

            ui.debug(b"query %i; add more sample (target %i, current %i)\n"

                     % (querycount, samplesize, len(sample)))

            # we need more sample !

            needed = samplesize - len(sample)

            sliceme = []

            heapify(sliceme)

            for entry in sample:

                if 1 < rangelength(local, entry):

                    heappush(sliceme, (-rangelength(local, entry), entry))

            while sliceme and 0 < needed:

                _key, target = heappop(sliceme)

                for new in subranges(local, target):

                    # XXX we could record hierarchy to optimise drop

                    if addentry(new):

                        if 1 < len(new):

                            heappush(sliceme, (-rangelength(local, new), new))

                        needed -= 1

                        if needed <= 0:

                            break

        # no longer the first interation

        samplesize = fullsamplesize

        nbsample = len(sample)

        maxsize = max([rangelength(local, r) for r in sample])

        ui.debug(b"query %i; sample size is %i, largest range %i\n"

                 % (querycount, nbsample, maxsize))

        nbreplies = 0

        replies = list(_queryrange(ui, local, remote, sample))

        sample = []

        n = local.changelog.node

        for entry, remotehash in replies:

            nbreplies += 1

            if remotehash == _obshashrange(local, entry):

                continue

            elif 1 == rangelength(local, entry):

                missing.add(n(entry[0]))

            else:

                for new in subranges(local, entry):

                    addentry(new)

        assert nbsample == nbreplies

        querycount += 1

        compat.progress(ui, _(b"comparing obsmarker with other"), querycount,

                        unit=_(b"queries"))

    compat.progress(ui, _(b"comparing obsmarker with other"), None)

    local.obsstore.rangeobshashcache.save(local)

    duration = util.timer() - starttime

    logmsg = (b'obsdiscovery, %d/%d mismatch'

              b' - %d obshashrange queries in %.4f seconds\n')

    logmsg %= (len(missing), len(probeset), querycount, duration)

    ui.log(b'evoext-obsdiscovery', logmsg)

    ui.debug(logmsg)

    return sorted(missing)

def _queryrange(ui, repo, remote, allentries):

    #  question are asked with node

    n = repo.changelog.node

    noderanges = [(n(entry[0]), entry[1]) for entry in allentries]

    replies = remote.evoext_obshashrange_v1(noderanges)

    result = []

    for idx, entry in enumerate(allentries):

        result.append((entry, replies[idx]))

    return result

##############################

### Range Hash computation ###

##############################

@eh.command(

    b'debugobshashrange',

    [

        (b'', b'rev', [], b'display obshash for all (rev, 0) range in REVS'),

        (b'', b'subranges', False, b'display all subranges'),

    ],

    _(b''))

def debugobshashrange(ui, repo, **opts):

    """display the ::REVS set topologically sorted in a stable way

    """

    s = node.short

    revs = scmutil.revrange(repo, opts['rev'])

    # prewarm depth cache

    if revs:

        repo.stablerange.warmup(repo, max(revs))

    cl = repo.changelog

    rangelength = repo.stablerange.rangelength

    depthrev = repo.stablerange.depthrev

    if opts['subranges']:

        ranges = stablerange.subrangesclosure(repo, repo.stablerange, revs)

    else:

        ranges = [(r, 0) for r in revs]

    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')

    ui.status(headertemplate % headers)

    repo.obsstore.rangeobshashcache.update(repo)

    for r in ranges:

        d = (r[0],

             s(cl.node(r[0])),

             r[1],

             rangelength(repo, r),

             depthrev(repo, r[0]),

             node.short(_obshashrange(repo, r)))

        ui.status(linetemplate % d)

    repo.obsstore.rangeobshashcache.save(repo)

def _obshashrange(repo, rangeid):

    """return the obsolete hash associated to a range"""

    cache = repo.obsstore.rangeobshashcache

    cl = repo.changelog

    obshash = cache.get(rangeid)

    if obshash is not None:

        return obshash

    pieces = []

    nullid = node.nullid

    if repo.stablerange.rangelength(repo, rangeid) == 1:

        rangenode = cl.node(rangeid[0])

        tmarkers = repo.obsstore.relevantmarkers([rangenode])

        pieces = []

        for m in tmarkers:

            mbin = obsolete._fm1encodeonemarker(m)

            pieces.append(mbin)

        pieces.sort()

    else:

        for subrange in repo.stablerange.subranges(repo, rangeid):

            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:

        obshash = node.nullid

    elif len(pieces) != 1 or obshash is None:

        sha = hashlib.sha1()

        for p in pieces:

            sha.update(p)

        obshash = sha.digest()

    cache[rangeid] = obshash

    return obshash

### sqlite caching

_sqliteschema = [

    r"""CREATE TABLE obshashrange(rev     INTEGER NOT NULL,

                                 idx     INTEGER NOT NULL,

                                 obshash BLOB    NOT NULL,

                                 PRIMARY KEY(rev, idx));""",

    r"CREATE INDEX range_index ON obshashrange(rev, idx);",

    r"""CREATE TABLE meta(schemaversion INTEGER NOT NULL,

                         tiprev        INTEGER NOT NULL,

                         tipnode       BLOB    NOT NULL,

                         nbobsmarker   INTEGER NOT NULL,

                         obssize       BLOB    NOT NULL,

                         obskey        BLOB    NOT NULL

                        );""",

]

_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)

            VALUES (?,?,?,?,?,?);"""

_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"

_reset = r"DELETE FROM obshashrange;"

_delete = r"DELETE FROM obshashrange WHERE (rev = ? AND idx = ?);"

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)

    node_to_proceed = set()

    marker_to_proceed = set(known_markers)

    obsstore = repo.obsstore

    while node_to_proceed or marker_to_proceed:

        while marker_to_proceed:

            m = marker_to_proceed.pop()

            # check successors and parent

            if m[1]:

                relevant = (m[1], )

            else: # prune case

                relevant = ((m[0], ), m[5])

            for l in relevant:

                if l is None:

                    continue

                for n in l:

                    if n not in affected_nodes:

                        node_to_proceed.add(n)

                    affected_nodes.add(n)

        # marker_to_proceed is now empty:

        if node_to_proceed:

            n = node_to_proceed.pop()

            markers = set()

            markers.update(obsstore.successors.get(n, ()))

            markers.update(obsstore.predecessors.get(n, ()))

            markers -= known_markers

            marker_to_proceed.update(markers)

            known_markers.update(markers)

    return affected_nodes

# if there is that many new obsmarkers, reset without analysing them

RESET_ABOVE = 10000

class _obshashcache(obscache.dualsourcecache):

    _schemaversion = 3

    _cachename = b'evo-ext-obshashrange' # used for error message

    _filename = b'evoext_obshashrange_v2.sqlite'

    def __init__(self, repo):

        super(_obshashcache, self).__init__()

        self._vfs = repo.vfs

        self._path = repo.cachevfs.join(self._filename)

        self._new = set()

        self._valid = True

        self._repo = weakref.ref(repo.unfiltered())

        # cache status

        self._ondiskcachekey = None

        self._data = {}

        self._createmode = store._calcmode(self._vfs)

    def clear(self, reset=False):

        super(_obshashcache, self).clear(reset=reset)

        self._data.clear()

        self._new.clear()

        if reset:

            self._valid = False

        if r'_con' in vars(self):

            del self._con

    def get(self, rangeid):

        # revision should be covered by the tiprev

        #

        # XXX there are issue with cache warming, we hack around it for now

        if not getattr(self, '_updating', False):

            if self._cachekey[0] < rangeid[0]:

                msg = (b'using unwarmed obshashrangecache (%s %s)'

                       % (rangeid[0], self._cachekey[0]))

                raise error.ProgrammingError(msg)

        value = self._data.get(rangeid)

        if value is None and self._con is not None:

            nrange = (rangeid[0], rangeid[1])

            try:

                obshash = self._con.execute(_queryobshash, nrange).fetchone()