<commit_before># -*- coding: utf8 -*-
import gevent
from gevent.queue import Queue
from gevent.event import AsyncResult
from ethereum import slogging

from raiden.messages import decode, Ack, BaseError, Secret
from raiden.utils import isaddress, sha3, pex

log = slogging.get_logger(__name__)  # pylint: disable=invalid-name


class RaidenProtocol(object):
    """ Encode the message into a packet and send it.

    Each message received is stored by hash and if it is received twice the
    previous answer is resent.

    Repeat sending messages until an acknowledgment is received or the maximum
    number of retries is hitted.
    """

    try_interval = 1.
    max_tries = 5
    max_message_size = 1200

    def __init__(self, transport, discovery, raiden):
        self.transport = transport
        self.discovery = discovery
        self.raiden = raiden
        self.queued_messages = Queue()
        self.stop_event = AsyncResult()

        self.number_of_tries = dict()  # msg hash: count_tries
        self.sent_acks = dict()  # msghash: Ack

        gevent.spawn(self._send_queued_messages)

    def _send_queued_messages(self):
        timeout = 0.1
        countdown_to_send = self.try_interval / timeout
        countdown = 0

        stop = None
        while stop is None:
            # blocks waiting for data in queue, don't remove data from
            # the queue just now because sending can still fail.
            receiver_address, message = self.queued_messages.peek()

            data = message.encode()
            host_port = self.discovery.get(receiver_address)

            # msghash is removed from the `number_of_tries` once a Ack is
            # received, resend until we receive it or give up
            msghash = sha3(data)
            self.number_of_tries[msghash] = 0

            log.info('SENDING {} > {} : [{}] {}'.format(
                pex(self.raiden.address),
                pex(receiver_address),
                pex(msghash),
                message,
            ))

            # loop should iterate as fast as possible checking for acks
            while msghash in self.number_of_tries:
                if self.number_of_tries[msghash] > self.max_tries:
                    # free send_and_wait...
                    del self.number_of_tries[msghash]

                    # FIXME: there was no connectivity or other network error?
                    # for now just losing the packet but better error handler
                    # needs to be added.
                    log.error('DEACTIVATED MSG resents {} {}'.format(
                        pex(receiver_address),
                        message,
                    ))

                # only send a message again after some time (don't DOS!)
                if countdown == 0:
                    self.number_of_tries[msghash] += 1
                    self.transport.send(self.raiden, host_port, data)
                    countdown = countdown_to_send

                gevent.sleep(timeout)
                countdown -= 1

            # consume last sent message
            self.queued_messages.get()
            stop = self.stop_event.wait(timeout)

    def send(self, receiver_address, message):
        if not isaddress(receiver_address):
            raise ValueError('Invalid address {}'.format(pex(receiver_address)))

        if isinstance(message, (Ack, BaseError)):
            raise ValueError('Do not use send for Ack messages or Erorrs')

        if len(message.encode()) > self.max_message_size:
            raise ValueError('message size excedes the maximum {}'.format(self.max_message_size))

        self.queued_messages.put((receiver_address, message))

    def send_and_wait(self, receiver_address, message):
        """Sends a message and wait for the response ack."""
        self.send(receiver_address, message)

        data = message.encode()
        msghash = sha3(data)
        # FIXME: add error handling
        while msghash in self.number_of_tries:
            gevent.sleep(0.1)

    def send_ack(self, receiver_address, message):
        if not isaddress(receiver_address):
            raise ValueError('Invalid address {}'.format(pex(receiver_address)))

        if not isinstance(message, (Ack, BaseError)):
            raise ValueError('Use send_Ack only for Ack messages or Erorrs')<commit_msg>Add class definition of short delay time<commit_after># -*- coding: utf8 -*-
import gevent
from gevent.queue import Queue
from gevent.event import AsyncResult
from ethereum import slogging

from raiden.messages import decode, Ack, BaseError, Secret
from raiden.utils import isaddress, sha3, pex

log = slogging.get_logger(__name__)  # pylint: disable=invalid-name


class RaidenProtocol(object):
    """ Encode the message into a packet and send it.

    Each message received is stored by hash and if it is received twice the
    previous answer is resent.

    Repeat sending messages until an acknowledgment is received or the maximum
    number of retries is hitted.
    """

    try_interval = 1.
    max_tries = 5
    max_message_size = 1200
    short_delay = .1

    def __init__(self, transport, discovery, raiden):
        self.transport = transport
        self.discovery = discovery
        self.raiden = raiden
        self.queued_messages = Queue()
        self.stop_event = AsyncResult()

        self.number_of_tries = dict()  # msg hash: count_tries
        self.sent_acks = dict()  # msghash: Ack

        gevent.spawn(self._send_queued_messages)

    def _send_queued_messages(self):
        countdown_to_send = self.try_interval / self.short_delay
        countdown = 0

        stop = None
        while stop is None:
            # blocks waiting for data in queue, don't remove data from
            # the queue just now because sending can still fail.
            receiver_address, message = self.queued_messages.peek()

            data = message.encode()
            host_port = self.discovery.get(receiver_address)

            # msghash is removed from the `number_of_tries` once a Ack is
            # received, resend until we receive it or give up
            msghash = sha3(data)
            self.number_of_tries[msghash] = 0

            log.info('SENDING {} > {} : [{}] {}'.format(
                pex(self.raiden.address),
                pex(receiver_address),
                pex(msghash),
                message,
            ))

            # loop should iterate as fast as possible checking for acks
            while msghash in self.number_of_tries:
                if self.number_of_tries[msghash] > self.max_tries:
                    # free send_and_wait...
                    del self.number_of_tries[msghash]

                    # FIXME: there was no connectivity or other network error?
                    # for now just losing the packet but better error handler
                    # needs to be added.
                    log.error('DEACTIVATED MSG resents {} {}'.format(
                        pex(receiver_address),
                        message,
                    ))

                # only send a message again after some time (don't DOS!)
                if countdown == 0:
                    self.number_of_tries[msghash] += 1
                    self.transport.send(self.raiden, host_port, data)
                    countdown = countdown_to_send

                gevent.sleep(self.short_delay)
                countdown -= 1

            # consume last sent message
            self.queued_messages.get()
            stop = self.stop_event.wait(self.short_delay)

    def send(self, receiver_address, message):
        if not isaddress(receiver_address):
            raise ValueError('Invalid address {}'.format(pex(receiver_address)))

        if isinstance(message, (Ack, BaseError)):
            raise ValueError('Do not use send for Ack messages or Erorrs')

        if len(message.encode()) > self.max_message_size:
            raise ValueError('message size excedes the maximum {}'.format(self.max_message_size))

        self.queued_messages.put((receiver_address, message))

    def send_and_wait(self, receiver_address, message):
        """Sends a message and wait for the response ack."""
        self.send(receiver_address, message)

        data = message.encode()
        msghash = sha3(data)
        # FIXME: add error handling
        while msghash in self.number_of_tries:
            gevent.sleep(self.short_delay)

    def send_ack(self, receiver_address, message):
        if not isaddress(receiver_address):
            raise ValueError('Invalid address {}'.format(pex(receiver_address)))

        if not isinstance(message, (Ack, BaseError)):
            raise ValueError('Use send_Ack only for Ack messages or Erorrs')
{'fraction_non_alphanumeric': 0.07925434583714547, 'fraction_numerical': 0.0040027447392497716, 'mean_word_length': 2.072733661278988, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '782219', 'n_tokens_mistral': 2459, 'n_tokens_neox': 2297, 'n_words': 705}