<commit_before># coding=utf-8
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import struct
from queue import PriorityQueue
from typing import Callable, Optional

from google.protobuf.json_format import MessageToJson
from pyqrllib.pyqrllib import bin2hstr  # noqa
from twisted.internet.protocol import Protocol, connectionDone

from qrl.core import config
from qrl.core.OutgoingMessage import OutgoingMessage
from qrl.core.misc import logger, ntp
from qrl.core.p2p.IPMetadata import IPMetadata
from qrl.core.p2p.p2pObservable import P2PObservable
from qrl.generated import qrllegacy_pb2, qrl_pb2


# Rename to p2p channel
class P2PProtocol(Protocol):
    def __init__(self):
        self._buffer = bytes()

        # Need to use composition instead of inheritance here
        self._observable = P2PObservable(self)

        self.last_rate_limit_update = 0
        self.rate_limit = config.user.peer_rate_limit
        self.in_counter = 0
        self.out_counter = 0

        self.bytes_sent = 0
        self.outgoing_queue = PriorityQueue(maxsize=config.user.p2p_q_size)

        self._connected_at = ntp.getTime()
        self._valid_message_count = 0

    @property
    def peer(self):
        return IPMetadata(self.transport.getPeer().host, self.transport.getPeer().port)

    @property
    def host(self):
        return IPMetadata(self.transport.getHost().host, self.transport.getHost().port)

    @property
    def connected_at(self):
        return self._connected_at

    @property
    def valid_message_count(self):
        return self._valid_message_count

    @property
    def connection_time(self):
        return ntp.getTime() - self._connected_at

    @property
    def peer_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.peer_manager

    @property
    def p2pchain_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.p2pchain_manager

    @property
    def tx_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.tx_manager

    def register(self, message_type, func: Callable):
        self._observable.register(message_type, func)

    def connectionMade(self):
        if self.factory.add_connection(self):
            self.peer_manager.new_channel(self)
            self.p2pchain_manager.new_channel(self)
            self.tx_manager.new_channel(self)

            self.send_peer_list()
            self.send_version_request()
        else:
            self.loseConnection()

    def connectionLost(self, reason=connectionDone):
        logger.debug('%s disconnected. remainder connected: %d', self.peer, self.factory.num_connections)

        self.factory.remove_connection(self)
        if self.peer_manager:
            self.peer_manager.remove_channel(self)

    def dataReceived(self, data: bytes) -> None:
        self._buffer += data
        total_read = len(self._buffer)

        if total_read > config.dev.max_bytes_out:
            logger.warning('Disconnecting peer %s', self.peer)
            logger.warning('Buffer Size %s', len(self._buffer))
            self.loseConnection()
            return

        read_bytes = [0]

        msg = None
        for msg in self._parse_buffer(read_bytes):
            self.update_counters()
            self.in_counter += 1
            if self.in_counter > self.rate_limit:
                logger.warning("Rate Limit hit by %s %s", self.peer.ip, self.peer.port)
                self.peer_manager.ban_channel(self)

            if self._valid_message_count < config.dev.trust_min_msgcount * 2:
                # Avoid overflows
                self._valid_message_count += 1

            self._observable.notify(msg)

        if msg is not None and read_bytes[0] and msg.func_name != qrllegacy_pb2.LegacyMessage.P2P_ACK:
            p2p_ack = qrl_pb2.P2PAcknowledgement(bytes_processed=read_bytes[0])
            msg = qrllegacy_pb2.LegacyMessage(func_name=qrllegacy_pb2.LegacyMessage.P2P_ACK,
                                              p2pAckData=p2p_ack)
            self.send(msg)

    def update_counters(self):
        time_diff = ntp.getTime() - self.last_rate_limit_update
        if time_diff > 60:
            self.out_counter = 0
            self.in_counter = 0
            self.last_rate_limit_update = ntp.getTime()
            return

    def send_next(self):
        self.update_counters()
        if self.out_counter == self.rate_limit:
            return

        if self.bytes_sent < config.dev.max_bytes_out:
            outgoing_bytes = self.get_bytes_from_q()

            if outgoing_bytes:
                self.bytes_sent += len(outgoing_bytes)
                self.transport.write(outgoing_bytes)

    def get_bytes_from_q(self):
        outgoing_bytes = b''
        while not self.outgoing_queue.empty():
            outgoing_msg = self.outgoing_queue.get()[2]
            if not outgoing_msg.is_expired():
                wrapped_message = self._wrap_message(outgoing_msg.message)
                if wrapped_message is not None:
                    if len(wrapped_message) + len(outgoing_bytes) > config.dev.max_bytes_out:
                        self.outgoing_queue.put((outgoing_msg.priority, outgoing_msg.timestamp, outgoing_msg))
                        break
                    outgoing_bytes += wrapped_message

                    self.out_counter += 1
                    if self.out_counter == self.rate_limit:
                        break

        return outgoing_bytes

    def send(self, message: qrllegacy_pb2.LegacyMessage):
        priority = self.factory.p2p_msg_priority[message.func_name]
        outgoing_msg = OutgoingMessage(priority, message)
        if self.outgoing_queue.full():
            return
        self.outgoing_queue.put((outgoing_msg.priority, outgoing_msg.timestamp, outgoing_msg))
        self.send_next()

    def loseConnection(self):
        self.transport.loseConnection()

    ###################################################
    ###################################################
    ###################################################
    ###################################################
    # Low-level serialization/connections/etc
    # FIXME: This is a temporary refactoring, it will be completely replaced before release

    @staticmethod<commit_msg>Ranged rate_limit<commit_after># coding=utf-8
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import struct
from queue import PriorityQueue
from typing import Callable, Optional

from google.protobuf.json_format import MessageToJson
from pyqrllib.pyqrllib import bin2hstr  # noqa
from twisted.internet.protocol import Protocol, connectionDone

from qrl.core import config
from qrl.core.OutgoingMessage import OutgoingMessage
from qrl.core.misc import logger, ntp
from qrl.core.p2p.IPMetadata import IPMetadata
from qrl.core.p2p.p2pObservable import P2PObservable
from qrl.generated import qrllegacy_pb2, qrl_pb2

OUT_FACTOR = 0.9
IN_FACTOR = 1.1


# Rename to p2p channel
class P2PProtocol(Protocol):
    def __init__(self):
        self._buffer = bytes()

        # Need to use composition instead of inheritance here
        self._observable = P2PObservable(self)

        self.last_rate_limit_update = 0
        self.rate_limit = config.user.peer_rate_limit
        self.in_counter = 0
        self.out_counter = 0

        self.bytes_sent = 0
        self.outgoing_queue = PriorityQueue(maxsize=config.user.p2p_q_size)

        self._connected_at = ntp.getTime()
        self._valid_message_count = 0

    @property
    def peer(self):
        return IPMetadata(self.transport.getPeer().host, self.transport.getPeer().port)

    @property
    def host(self):
        return IPMetadata(self.transport.getHost().host, self.transport.getHost().port)

    @property
    def connected_at(self):
        return self._connected_at

    @property
    def valid_message_count(self):
        return self._valid_message_count

    @property
    def connection_time(self):
        return ntp.getTime() - self._connected_at

    @property
    def peer_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.peer_manager

    @property
    def p2pchain_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.p2pchain_manager

    @property
    def tx_manager(self):
        # FIXME: this is breaking encapsulation
        return self.factory._qrl_node.tx_manager

    def register(self, message_type, func: Callable):
        self._observable.register(message_type, func)

    def connectionMade(self):
        if self.factory.add_connection(self):
            self.peer_manager.new_channel(self)
            self.p2pchain_manager.new_channel(self)
            self.tx_manager.new_channel(self)

            self.send_peer_list()
            self.send_version_request()
        else:
            self.loseConnection()

    def connectionLost(self, reason=connectionDone):
        logger.debug('%s disconnected. remainder connected: %d', self.peer, self.factory.num_connections)

        self.factory.remove_connection(self)
        if self.peer_manager:
            self.peer_manager.remove_channel(self)

    def dataReceived(self, data: bytes) -> None:
        self._buffer += data
        total_read = len(self._buffer)

        if total_read > config.dev.max_bytes_out:
            logger.warning('Disconnecting peer %s', self.peer)
            logger.warning('Buffer Size %s', len(self._buffer))
            self.loseConnection()
            return

        read_bytes = [0]

        msg = None
        for msg in self._parse_buffer(read_bytes):
            self.update_counters()
            self.in_counter += 1
            if self.in_counter > self.rate_limit * IN_FACTOR:
                logger.warning("Rate Limit hit by %s %s", self.peer.ip, self.peer.port)
                self.peer_manager.ban_channel(self)

            if self._valid_message_count < config.dev.trust_min_msgcount * 2:
                # Avoid overflows
                self._valid_message_count += 1

            self._observable.notify(msg)

        if msg is not None and read_bytes[0] and msg.func_name != qrllegacy_pb2.LegacyMessage.P2P_ACK:
            p2p_ack = qrl_pb2.P2PAcknowledgement(bytes_processed=read_bytes[0])
            msg = qrllegacy_pb2.LegacyMessage(func_name=qrllegacy_pb2.LegacyMessage.P2P_ACK,
                                              p2pAckData=p2p_ack)
            self.send(msg)

    def update_counters(self):
        time_diff = ntp.getTime() - self.last_rate_limit_update
        if time_diff > 60:
            self.out_counter = 0
            self.in_counter = 0
            self.last_rate_limit_update = ntp.getTime()
            return

    def send_next(self):
        self.update_counters()
        if self.out_counter >= self.rate_limit * OUT_FACTOR:
            return

        if self.bytes_sent < config.dev.max_bytes_out:
            outgoing_bytes = self.get_bytes_from_q()

            if outgoing_bytes:
                self.bytes_sent += len(outgoing_bytes)
                self.transport.write(outgoing_bytes)

    def get_bytes_from_q(self):
        outgoing_bytes = b''
        while not self.outgoing_queue.empty():
            outgoing_msg = self.outgoing_queue.get()[2]
            if not outgoing_msg.is_expired():
                wrapped_message = self._wrap_message(outgoing_msg.message)
                if wrapped_message is not None:
                    if len(wrapped_message) + len(outgoing_bytes) > config.dev.max_bytes_out:
                        self.outgoing_queue.put((outgoing_msg.priority, outgoing_msg.timestamp, outgoing_msg))
                        break
                    outgoing_bytes += wrapped_message

                    self.out_counter += 1
                    if self.out_counter >= self.rate_limit * OUT_FACTOR:
                        break

        return outgoing_bytes

    def send(self, message: qrllegacy_pb2.LegacyMessage):
        priority = self.factory.p2p_msg_priority[message.func_name]
        outgoing_msg = OutgoingMessage(priority, message)
        if self.outgoing_queue.full():
            return
        self.outgoing_queue.put((outgoing_msg.priority, outgoing_msg.timestamp, outgoing_msg))
        self.send_next()

    def loseConnection(self):
        self.transport.loseConnection()

    ###################################################
    ###################################################
    ###################################################
    ###################################################
    # Low-level serialization/connections/etc
    # FIXME: This is a temporary refactoring, it will be completely replaced before release

    @staticmethod
{'fraction_non_alphanumeric': 0.1096302003081664, 'fraction_numerical': 0.007087827426810477, 'mean_word_length': 2.513125845737483, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6517056', 'n_tokens_mistral': 3821, 'n_tokens_neox': 3640, 'n_words': 739}