<commit_before>import heapq
import math
import operator
import time
from collections import Callable, Counter
from functools import partial
from random import shuffle
from typing import Any, List, Dict, Tuple
from typing import Optional

from ledger.merkle_verifier import MerkleVerifier
from plenum.common.config_util import getConfig
from plenum.common.constants import POOL_LEDGER_ID, LedgerState, CONSISTENCY_PROOF, CATCH_UP_PREFIX
from plenum.common.ledger import Ledger
from plenum.common.ledger_info import LedgerInfo
from plenum.common.messages.node_messages import LedgerStatus, CatchupRep, \
    ConsistencyProof, f, CatchupReq
from plenum.common.metrics_collector import MetricsCollector, NullMetricsCollector, measure_time, MetricsName
from plenum.common.util import compare_3PC_keys, SortedDict, min_3PC_key
from plenum.server.has_action_queue import HasActionQueue
from plenum.server.quorums import Quorums
from stp_core.common.log import getlogger

logger = getlogger()


class LedgerManager(HasActionQueue):
    def __init__(self,
                 owner,
                 ownedByNode: bool = True,
                 postAllLedgersCaughtUp: Optional[Callable] = None,
                 preCatchupClbk: Optional[Callable] = None,
                 postCatchupClbk: Optional[Callable] = None,
                 ledger_sync_order: Optional[List] = None,
                 metrics: MetricsCollector = NullMetricsCollector()):
        # If ledger_sync_order is not provided (is None), it is assumed that
        # `postCatchupCompleteClbk` of the LedgerInfo will be used
        self.owner = owner
        self.ownedByNode = ownedByNode
        self.postAllLedgersCaughtUp = postAllLedgersCaughtUp
        self.preCatchupClbk = preCatchupClbk
        self.postCatchupClbk = postCatchupClbk
        self.ledger_sync_order = ledger_sync_order
        self.request_ledger_status_action_ids = dict()
        self.request_consistency_proof_action_ids = dict()
        self.metrics = metrics

        self.config = getConfig()
        # Needs to schedule actions. The owner of the manager has the
        # responsibility of calling its `_serviceActions` method periodically
        HasActionQueue.__init__(self)

        # Holds ledgers of different types with
        # their info like callbacks, state, etc
        self.ledgerRegistry = {}  # type: Dict[int, LedgerInfo]

        # Largest 3 phase key received during catchup.
        # This field is needed to discard any stashed 3PC messages or
        # ordered messages since the transactions part of those messages
        # will be applied when they are received through the catchup process
        self.last_caught_up_3PC = (0, 0)

    def __repr__(self):
        return self.owner.name

    @measure_time(MetricsName.SERVICE_LEDGER_MANAGER_TIME)
    def service(self):
        return self._serviceActions()

    def addLedger(self, iD: int, ledger: Ledger,
                  preCatchupStartClbk: Callable = None,
                  postCatchupStartClbk: Callable = None,
                  preCatchupCompleteClbk: Callable = None,
                  postCatchupCompleteClbk: Callable = None,
                  postTxnAddedToLedgerClbk: Callable = None):

        if iD in self.ledgerRegistry:
            logger.error("{} already present in ledgers "
                         "so cannot replace that ledger".format(iD))
            return

        self.ledgerRegistry[iD] = LedgerInfo(
            iD,
            ledger=ledger,
            preCatchupStartClbk=preCatchupStartClbk,
            postCatchupStartClbk=postCatchupStartClbk,
            preCatchupCompleteClbk=preCatchupCompleteClbk,
            postCatchupCompleteClbk=postCatchupCompleteClbk,
            postTxnAddedToLedgerClbk=postTxnAddedToLedgerClbk,
            verifier=MerkleVerifier(ledger.hasher)
        )

    def _cancel_request_ledger_statuses_and_consistency_proofs(self, ledger_id):
        if ledger_id in self.request_ledger_status_action_ids:
            action_id = self.request_ledger_status_action_ids.pop(ledger_id)
            self._cancel(aid=action_id)

        if ledger_id in self.request_consistency_proof_action_ids:
            action_id = self.request_consistency_proof_action_ids.pop(ledger_id)
            self._cancel(aid=action_id)

    def reask_for_ledger_status(self, ledger_id):
        self.request_ledger_status_action_ids.pop(ledger_id)
        ledgerInfo = self.getLedgerInfoByType(ledger_id)
        nodes = [node for node in self.owner.nodeReg if node not in ledgerInfo.ledgerStatusOk]
        self.owner.request_ledger_status_from_nodes(ledger_id, nodes)

    def reask_for_last_consistency_proof(self, ledger_id):
        self.request_consistency_proof_action_ids.pop(ledger_id)
        ledgerInfo = self.getLedgerInfoByType(ledger_id)
        recvdConsProof = ledgerInfo.recvdConsistencyProofs
        ledger_status = self.owner.build_ledger_status(ledger_id)
        nodes = [frm for frm in self.owner.nodeReg if
                 frm not in recvdConsProof and frm != self.owner.name]
        for frm in nodes:
                self.sendTo(ledger_status, frm)

    def request_CPs_if_needed(self, ledgerId):
        ledgerInfo = self.getLedgerInfoByType(ledgerId)
        if ledgerInfo.consistencyProofsTimer is None:
            return<commit_msg>INDY-1740: Fixed a bug in re-asking cancellation<commit_after>import heapq
import math
import operator
import time
from collections import Callable, Counter
from functools import partial
from random import shuffle
from typing import Any, List, Dict, Tuple
from typing import Optional

from ledger.merkle_verifier import MerkleVerifier
from plenum.common.config_util import getConfig
from plenum.common.constants import POOL_LEDGER_ID, LedgerState, CONSISTENCY_PROOF, CATCH_UP_PREFIX
from plenum.common.ledger import Ledger
from plenum.common.ledger_info import LedgerInfo
from plenum.common.messages.node_messages import LedgerStatus, CatchupRep, \
    ConsistencyProof, f, CatchupReq
from plenum.common.metrics_collector import MetricsCollector, NullMetricsCollector, measure_time, MetricsName
from plenum.common.util import compare_3PC_keys, SortedDict, min_3PC_key
from plenum.server.has_action_queue import HasActionQueue
from plenum.server.quorums import Quorums
from stp_core.common.log import getlogger

logger = getlogger()


class LedgerManager(HasActionQueue):
    def __init__(self,
                 owner,
                 ownedByNode: bool = True,
                 postAllLedgersCaughtUp: Optional[Callable] = None,
                 preCatchupClbk: Optional[Callable] = None,
                 postCatchupClbk: Optional[Callable] = None,
                 ledger_sync_order: Optional[List] = None,
                 metrics: MetricsCollector = NullMetricsCollector()):
        # If ledger_sync_order is not provided (is None), it is assumed that
        # `postCatchupCompleteClbk` of the LedgerInfo will be used
        self.owner = owner
        self.ownedByNode = ownedByNode
        self.postAllLedgersCaughtUp = postAllLedgersCaughtUp
        self.preCatchupClbk = preCatchupClbk
        self.postCatchupClbk = postCatchupClbk
        self.ledger_sync_order = ledger_sync_order
        self.request_ledger_status_action_ids = dict()
        self.request_consistency_proof_action_ids = dict()
        self.metrics = metrics

        self.config = getConfig()
        # Needs to schedule actions. The owner of the manager has the
        # responsibility of calling its `_serviceActions` method periodically
        HasActionQueue.__init__(self)

        # Holds ledgers of different types with
        # their info like callbacks, state, etc
        self.ledgerRegistry = {}  # type: Dict[int, LedgerInfo]

        # Largest 3 phase key received during catchup.
        # This field is needed to discard any stashed 3PC messages or
        # ordered messages since the transactions part of those messages
        # will be applied when they are received through the catchup process
        self.last_caught_up_3PC = (0, 0)

    def __repr__(self):
        return self.owner.name

    @measure_time(MetricsName.SERVICE_LEDGER_MANAGER_TIME)
    def service(self):
        return self._serviceActions()

    def addLedger(self, iD: int, ledger: Ledger,
                  preCatchupStartClbk: Callable = None,
                  postCatchupStartClbk: Callable = None,
                  preCatchupCompleteClbk: Callable = None,
                  postCatchupCompleteClbk: Callable = None,
                  postTxnAddedToLedgerClbk: Callable = None):

        if iD in self.ledgerRegistry:
            logger.error("{} already present in ledgers "
                         "so cannot replace that ledger".format(iD))
            return

        self.ledgerRegistry[iD] = LedgerInfo(
            iD,
            ledger=ledger,
            preCatchupStartClbk=preCatchupStartClbk,
            postCatchupStartClbk=postCatchupStartClbk,
            preCatchupCompleteClbk=preCatchupCompleteClbk,
            postCatchupCompleteClbk=postCatchupCompleteClbk,
            postTxnAddedToLedgerClbk=postTxnAddedToLedgerClbk,
            verifier=MerkleVerifier(ledger.hasher)
        )

    def _cancel_request_ledger_statuses_and_consistency_proofs(self, ledger_id):
        if ledger_id in self.request_ledger_status_action_ids:
            action_id = self.request_ledger_status_action_ids.pop(ledger_id)
            self._cancel(aid=action_id)

        if ledger_id in self.request_consistency_proof_action_ids:
            action_id = self.request_consistency_proof_action_ids.pop(ledger_id)
            self._cancel(aid=action_id)

    def reask_for_ledger_status(self, ledger_id):
        self.request_ledger_status_action_ids.pop(ledger_id, None)
        ledgerInfo = self.getLedgerInfoByType(ledger_id)
        nodes = [node for node in self.owner.nodeReg if node not in ledgerInfo.ledgerStatusOk]
        self.owner.request_ledger_status_from_nodes(ledger_id, nodes)

    def reask_for_last_consistency_proof(self, ledger_id):
        self.request_consistency_proof_action_ids.pop(ledger_id, None)
        ledgerInfo = self.getLedgerInfoByType(ledger_id)
        recvdConsProof = ledgerInfo.recvdConsistencyProofs
        ledger_status = self.owner.build_ledger_status(ledger_id)
        nodes = [frm for frm in self.owner.nodeReg if
                 frm not in recvdConsProof and frm != self.owner.name]
        for frm in nodes:
                self.sendTo(ledger_status, frm)

    def request_CPs_if_needed(self, ledgerId):
        ledgerInfo = self.getLedgerInfoByType(ledgerId)
        if ledgerInfo.consistencyProofsTimer is None:
            return
{'fraction_non_alphanumeric': 0.05851512030708735, 'fraction_numerical': 0.0016852354648441157, 'mean_word_length': 3.0172997367431367, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '644870', 'n_tokens_mistral': 3288, 'n_tokens_neox': 3178, 'n_words': 725}