<commit_before>from twisted.internet import task
from base64 import b64encode
from Queue import Queue
from typing import Union
import random
import logging
from collections import defaultdict

from trustchain import TrustChain, TxBlock, CpBlock, Signature, Cons
from src.utils.utils import Replay, Handled, collate_cp_blocks, my_err_back, call_later
from src.utils.messages import SynMsg, SynAckMsg, AckMsg, SigMsg, CpMsg, ConsMsg


class RoundState:
    def __init__(self):
        self.received_cons = None
        self.received_sigs = []
        self.received_cps = []

    def new_cons(self, cons):
        # type: (Cons) -> None
        assert isinstance(cons, Cons)
        if self.received_cons is None:
            self.received_cons = cons
        else:
            # TODO eventually we need to store all received cons and check which are correctly and sufficiently signed
            assert cons == self.received_cons

    def new_sig(self, s):
        # type: (Signature) -> None
        assert isinstance(s, Signature)
        self.received_sigs.append(s)

    def new_cp(self, cp):
        # type: (CpBlock) -> None
        assert isinstance(cp, CpBlock)
        if len(self.received_cps) > 0:
            assert self.received_cps[0].round == cp.round
        self.received_cps.append(cp)


class TrustChainRunner:
    """
    We keep a queue of messages and handle them in order
    the handle function by itself essentially pushes the messages into the queue
    """
    def __init__(self, factory, msg_wrapper_f=lambda x: x, consensus_delay=0):
        self.tc = TrustChain()
        self.factory = factory
        self.recv_q = Queue()
        self.send_q = Queue()  # only for syn messages
        self.msg_wrapper_f = msg_wrapper_f
        self.consensus_delay = consensus_delay

        self.recv_lc = task.LoopingCall(self.process_recv_q)
        self.recv_lc.start(1, False).addErrback(my_err_back)

        self.send_lc = task.LoopingCall(self.process_send_q)
        self.send_lc.start(1, False).addErrback(my_err_back)

        # attributes below are states used for negotiating transaction
        self.tx_locked = False  # only process one transaction at a time, otherwise there'll be hash pointer collisions
        self.block_r = None  # type: TxBlock
        self.tx_id = -1  # type: int
        self.src = None  # type: str
        self.s_s = None  # type: Signature
        self.m = None  # type: str
        self.prev_r = None  # type: str

        # attributes below are states for building new CP blocks
        # TODO compare with my own chain and prune old rounds
        self.round_states = defaultdict(RoundState)

        self.bootstrap_lc = None

        random.seed()

    def reset_state(self):
        self.tx_locked = False
        self.tx_id = -1  # the id of the transaction that we're processing
        self.block_r = None
        self.src = None
        self.s_s = None
        self.m = None
        self.prev_r = None

    def assert_unlocked_state(self):
        assert not self.tx_locked
        assert self.block_r is None
        assert self.src is None
        assert self.tx_id == -1
        assert self.s_s is None
        assert self.m is None
        assert self.prev_r is None

    def assert_after_syn_state(self):
        assert self.tx_locked
        assert self.block_r is None
        assert self.src is not None
        assert self.tx_id != -1
        assert self.s_s is None
        assert self.m is not None
        assert self.prev_r is None

    def assert_full_state(self):
        assert self.tx_locked
        assert self.block_r is not None
        assert self.src is not None
        assert self.tx_id != -1
        assert self.s_s is not None
        assert self.m is not None
        assert self.prev_r is not None

    def update_state(self, lock, block, tx_id, src, s_s, m, prev_r):
        self.tx_locked = lock
        self.block_r = block
        self.tx_id = tx_id
        self.src = src
        self.s_s = s_s
        self.m = m
        self.prev_r = prev_r

    def sufficient_sigs(self, r):
        if len(self.round_states[r].received_sigs) > self.factory.config.t:
            return True
        return False

    def latest_promoters(self):
        r = self.tc.latest_round
        return self.tc.consensus[r].get_promoters(self.factory.config.n)

    def handle_cons_from_acs(self, msg):
        """
        This is only called after we get the output from ACS
        :param msg:
        :return:<commit_msg>Prune old round_states<commit_after>from twisted.internet import task
from base64 import b64encode
from Queue import Queue
from typing import Union
import random
import logging
from collections import defaultdict

from trustchain import TrustChain, TxBlock, CpBlock, Signature, Cons
from src.utils.utils import Replay, Handled, collate_cp_blocks, my_err_back, call_later
from src.utils.messages import SynMsg, SynAckMsg, AckMsg, SigMsg, CpMsg, ConsMsg


class RoundState:
    def __init__(self):
        self.received_cons = None
        self.received_sigs = []
        self.received_cps = []

    def new_cons(self, cons):
        # type: (Cons) -> None
        assert isinstance(cons, Cons)
        if self.received_cons is None:
            self.received_cons = cons
        else:
            # TODO eventually we need to store all received cons and check which are correctly and sufficiently signed
            assert cons == self.received_cons

    def new_sig(self, s):
        # type: (Signature) -> None
        assert isinstance(s, Signature)
        self.received_sigs.append(s)

    def new_cp(self, cp):
        # type: (CpBlock) -> None
        assert isinstance(cp, CpBlock)
        if len(self.received_cps) > 0:
            assert self.received_cps[0].round == cp.round
        self.received_cps.append(cp)


class TrustChainRunner:
    """
    We keep a queue of messages and handle them in order
    the handle function by itself essentially pushes the messages into the queue
    """
    def __init__(self, factory, msg_wrapper_f=lambda x: x, consensus_delay=0):
        self.tc = TrustChain()
        self.factory = factory
        self.recv_q = Queue()
        self.send_q = Queue()  # only for syn messages
        self.msg_wrapper_f = msg_wrapper_f
        self.consensus_delay = consensus_delay

        self.recv_lc = task.LoopingCall(self.process_recv_q)
        self.recv_lc.start(1, False).addErrback(my_err_back)

        self.send_lc = task.LoopingCall(self.process_send_q)
        self.send_lc.start(1, False).addErrback(my_err_back)

        self.collect_rubbish_lc = task.LoopingCall(self.collect_rubbish)
        self.collect_rubbish_lc.start(5, False).addErrback(my_err_back)

        self.bootstrap_lc = None

        # attributes below are states used for negotiating transaction
        self.tx_locked = False  # only process one transaction at a time, otherwise there'll be hash pointer collisions
        self.block_r = None  # type: TxBlock
        self.tx_id = -1  # type: int
        self.src = None  # type: str
        self.s_s = None  # type: Signature
        self.m = None  # type: str
        self.prev_r = None  # type: str

        # attributes below are states for building new CP blocks
        self.round_states = defaultdict(RoundState)

        random.seed()

    def reset_state(self):
        self.tx_locked = False
        self.tx_id = -1  # the id of the transaction that we're processing
        self.block_r = None
        self.src = None
        self.s_s = None
        self.m = None
        self.prev_r = None

    def assert_unlocked_state(self):
        assert not self.tx_locked
        assert self.block_r is None
        assert self.src is None
        assert self.tx_id == -1
        assert self.s_s is None
        assert self.m is None
        assert self.prev_r is None

    def assert_after_syn_state(self):
        assert self.tx_locked
        assert self.block_r is None
        assert self.src is not None
        assert self.tx_id != -1
        assert self.s_s is None
        assert self.m is not None
        assert self.prev_r is None

    def assert_full_state(self):
        assert self.tx_locked
        assert self.block_r is not None
        assert self.src is not None
        assert self.tx_id != -1
        assert self.s_s is not None
        assert self.m is not None
        assert self.prev_r is not None

    def update_state(self, lock, block, tx_id, src, s_s, m, prev_r):
        self.tx_locked = lock
        self.block_r = block
        self.tx_id = tx_id
        self.src = src
        self.s_s = s_s
        self.m = m
        self.prev_r = prev_r

    def sufficient_sigs(self, r):
        if len(self.round_states[r].received_sigs) > self.factory.config.t:
            return True
        return False

    def collect_rubbish(self):
        for k in self.round_states.keys():
            if k < self.tc.latest_round:
                logging.debug("TC: pruning key {}".format(k))
                del self.round_states[k]

    def latest_promoters(self):
        r = self.tc.latest_round
        return self.tc.consensus[r].get_promoters(self.factory.config.n)

    def handle_cons_from_acs(self, msg):
        """
        This is only called after we get the output from ACS
        :param msg:
        :return:
{'fraction_non_alphanumeric': 0.07569250590508911, 'fraction_numerical': 0.003113592441485935, 'mean_word_length': 2.2592722183344995, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 13, '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': '1866778', 'n_tokens_mistral': 2861, 'n_tokens_neox': 2760, 'n_words': 840}