<commit_before># coding: utf8
from __future__ import print_function, absolute_import

import inspect
import random
import re
import sys
import traceback
from cStringIO import StringIO
from collections import deque
from decimal import Decimal
from pickle import Unpickler, BUILD
from cPickle import dumps

from twisted import internet
from twisted.internet import reactor
from twisted.internet.task import LoopingCall
from txzmq import ZmqEndpoint

from spinoff.actor import Ref, Uri, Node
from spinoff.actor._actor import _VALID_NODEID_RE, _validate_nodeid
from spinoff.actor.events import Events, DeadLetter
from spinoff.util.logging import Logging, logstring


# TODO: use shorter messages outside of testing
PING = 'ping'
PONG = 'pong'


_VALID_ADDR_RE = re.compile('tcp://%s' % (_VALID_NODEID_RE.pattern,))


class ConnectedNode(object):
    def __init__(self, state, last_seen):
        self.state = state
        self.last_seen = last_seen
        self.queue = deque()


class Hub(Logging):
    """Handles traffic between actors on different nodes.

    The wire-transport implementation is specified/overridden by the `incoming` and `outgoing` parameters.

    """
    __doc_HEARTBEAT_INTERVAL__ = (
        "Time on seconds after which to send out a heartbeat signal to all known nodes. Regular messages can be "
        "subsituted by the framework for heartbeats to save network bandwidth.")
    HEARTBEAT_INTERVAL = 1.0

    __doc_MAX_SILENCE_BETWEEN_HEARTBEATS__ = (
        "Maximum length of silence in seconds between two consecutive heartbeat signals from a node after which to "
        "consider the node as temporarily not available, put it in the 'radio-silence' state, and start queueing all "
        "messages posted to it by actors.")
    MAX_SILENCE_BETWEEN_HEARTBEATS = 5.0

    __doc_TIME_TO_KEEP_HOPE__ = (
        "Time in seconds after which the 'radio-silence' state transforms into a 'silently-hoping' state wherein the "
        "is considered to have gone offline for an extended duration. All currently queued and any future messages to"
        "that node will immediately be turned into `DeadLetter` events.")
    TIME_TO_KEEP_HOPE = 55.0

    __doc_QUEUE_ITEM_LIFETIME__ = (
        "Time in seconds for which to keep a queued message alive, after which the message is turned into a "
        "`DeadLetter` event. The default value is chosen such that exactly when the target node goes into the "
        "'silently-hoping' visibility state, all messages to it are discarded.")
    QUEUE_ITEM_LIFETIME = MAX_SILENCE_BETWEEN_HEARTBEATS + TIME_TO_KEEP_HOPE

    node = None

    def __init__(self, incoming, outgoing, node, reactor=reactor):
        if not node or not isinstance(node, str):
            raise TypeError("The 'node' argument to Hub must be a str")
        _validate_nodeid(node)
        self.node = node
        self.addr = 'tcp://' + node if node else None

        self.reactor = reactor

        self.outgoing = outgoing
        incoming.gotMessage = self._got_message
        incoming.addEndpoints([ZmqEndpoint('bind', self.addr)])

        self.connections = {}

        l1 = LoopingCall(self._send_heartbeat)
        l1.clock = reactor
        l1.start(self.HEARTBEAT_INTERVAL)

        l2 = LoopingCall(self._purge_old_items_in_queue)
        l2.clock = reactor
        l2.start(1.0)

    _guardian = None

    def logstate(self):
        return {str(self.reactor.seconds()): True}

    def _get_guardian(self):
        return self._guardian

    def _set_guardian(self, guardian):
        if self._guardian:
            raise RuntimeError("Hub already bound to a Guardian")
        self._guardian = guardian

    guardian = property(_get_guardian, _set_guardian)

    @logstring(u"⇝")
    def send(self, msg, to_remote_actor_pointed_to_by):
        ref = to_remote_actor_pointed_to_by
        # self.dbg(u"%r → %r" % (msg, ref))

        nodeid = ref.uri.node

        # assert addr and addr != self.node, "TODO: remote-ref pointing to the local node detected"
        # TODO: if it's determined that it makes sense to allow the existence of such refs, enable the following block
        if not nodeid or nodeid == self.node:
            cell = self.guardian.lookup_cell(ref.uri)
            if not cell:
                ref.is_local = True  # next time, just put it straight to DeadLetters
                Events.log(DeadLetter(ref, msg))
                return
            ref._cell = cell
            ref.is_local = True
            ref << msg
            return

        addr = ref.uri.root.url

        conn = self.connections.get(addr)
        if not conn:
            self.dbg("%s set from not-known => %s" % (addr, 'radiosilence',))
            conn = self.connections[addr] = ConnectedNode(
                state='radiosilence',
                # so last_seen checks would mark the node as `silentlyhoping` in `TIME_TO_KEEP_HOPE` seconds from now;
                # conceptually speaking, this means the node seems to have always existed but went out of view exactly
                # now, and before sending can start "again", it needs to come back:
                last_seen=self.reactor.seconds() - self.MAX_SILENCE_BETWEEN_HEARTBEATS
            )
            self._connect(addr, conn)

        if conn.state == 'visible':
            self.outgoing.sendMsg((addr, dumps((ref.uri.path, msg), protocol=2)))
        else:
            if conn.queue is None:
                Events.log(DeadLetter(ref, msg))
            else:
                conn.queue.append(((ref, msg), self.reactor.seconds()))

    @logstring(u"⇜")
    def _got_message(self, sender_addr, msg):
        if msg in (PING, PONG):
            self.dbg(u"❤ %s ← %s" % (msg, sender_addr,))
        else:
            path, msg_ = self._loads(msg)
            self.dbg(u"%r ← %s   → %s" % (msg_, sender_addr, path))
            cell = self.guardian.lookup_cell(Uri.parse(path))
            if not cell:
                Events.log(DeadLetter(Ref(None, Uri.parse(path)), msg_))
            else:
                cell.receive(msg_)  # XXX: force_async=True perhaps?

        if sender_addr not in self.connections:
            assert msg == PING, "initial message sent to another node should be PING"
            self.dbg("%s went not-known => %s" % (sender_addr, 'reverse-radiosilence',))
            conn = self.connections[sender_addr] = ConnectedNode(
                state='reverse-radiosilence',
                last_seen=self.reactor.seconds())
            self._connect(sender_addr, conn)
        else:
            conn = self.connections[sender_addr]
            conn.last_seen = self.reactor.seconds()

            prevstate = conn.state
            conn.state = 'reverse-radiosilence' if msg == PING else 'visible'
            if prevstate != conn.state:
                self.dbg("%s went %s => %s" % (sender_addr, prevstate, conn.state))
            if prevstate != 'visible' and conn.state == 'visible':
                while conn.queue:
                    (ref, queued_msg), _ = conn.queue.popleft()
                    assert ref.uri.root.url == sender_addr
                    self.outgoing.sendMsg((sender_addr, dumps((ref.uri.path, queued_msg), protocol=2)))

    def _connect(self, addr, conn):
        assert _valid_addr(addr)
        # self.dbg("...connecting to %s" % (addr,))
        self.outgoing.addEndpoints([ZmqEndpoint('connect', addr)])
        # send one heartbeat immediately for better latency
        self.dbg(u"►► ❤ → %s" % (addr,))
        self._heartbeat_once(addr, PING if conn.state == 'radiosilence' else PONG)

    @logstring(u"❤")
    def _send_heartbeat(self):
        try:
            # self.dbg("→ %r" % (list(self.connections),))
            t = self.reactor.seconds()
            consider_dead_from = t - self.MAX_SILENCE_BETWEEN_HEARTBEATS
            consider_lost_from = consider_dead_from - self.TIME_TO_KEEP_HOPE
            # self.dbg("consider_dead_from", consider_dead_from, "consider_lost_from", consider_lost_from)

            for addr, conn in self.connections.items():
                # self.dbg("%s last seen at %ss" % (addr, conn.last_seen))
                if conn.state == 'silentlyhoping':
                    # self.dbg("silently hoping...")
                    self._heartbeat_once(addr, PING)
                elif conn.last_seen < consider_lost_from:
                    self.dbg("%s went %s => %s after %ds of silence" % (addr, conn.state, 'silentlyhoping', (t - conn.last_seen)))
                    conn.state = 'silentlyhoping'
                    for (ref, msg), _ in conn.queue:
                        # self.dbg("dropping %r" % (msg,))
                        Events.log(DeadLetter(ref, msg))
                    conn.queue = None
                    self._heartbeat_once(addr, PING)
                elif conn.last_seen < consider_dead_from:
                    if conn.state != 'radiosilence':
                        self.dbg("%s went %s => %s" % (addr, conn.state, 'radiosilence',))
                        conn.state = 'radiosilence'
                    self._heartbeat_once(addr, PING)
                else:
                    # self.dbg("%s still %s; not seen for %s" % (addr, conn.state, '%ds' % (t - conn.last_seen) if conn.last_seen is not None else 'eternity',))
                    self._heartbeat_once(addr, PONG)
            # self.dbg(u"%s ✓" % (self.reactor.seconds(),))
        except Exception:
            self.panic("failed to send heartbeat:\n", traceback.format_exc())

    @logstring(u"⇝ ❤")
    def _heartbeat_once(self, addr, signal):
        assert _valid_addr(addr)
        self.log(u"%s →" % (signal,), addr)
        self.outgoing.sendMsg((addr, signal))

    @logstring("PURGE")
    def _purge_old_items_in_queue(self):
        for conn in self.connections.values():
            try:
                keep_until = self.reactor.seconds() - self.QUEUE_ITEM_LIFETIME
                # self.dbg(conn.queue, "keep_until = %r, QUEUE_ITEM_LIFETIME = %r" % (keep_until, self.QUEUE_ITEM_LIFETIME,))
                q = conn.queue
                while q:
                    (ref, msg), timestamp = q[0]
                    if timestamp >= keep_until:
                        break
                    else:
                        q.popleft()
                        Events.log(DeadLetter(ref, msg))
            except Exception:
                self.panic("failed to clean queue:\n", traceback.format_exc())

    def _loads(self, data):
        return IncomingMessageUnpickler(self, StringIO(data)).load()

    def __repr__(self):
        return '<%s>' % (self.node,)


class HubWithNoRemoting(object):
    """A dummy hub used during networkless testing and in production when no remoting should be available.

    All it does is imitate the interface of the real `Hub`, and report attempts to send remote messages as
    `RuntimeError`s.

    """
    # to be compatible with Hub:
    guardian = None
    node = None

    def send(self, ref, msg):
        raise RuntimeError("Attempt to send a message to a remote ref but remoting is not available")


class IncomingMessageUnpickler(Unpickler, Logging):
    """Unpickler for attaching a `Hub` instance to all deserialized `Ref`s."""

    def __init__(self, hub, file):
        Unpickler.__init__(self, file)
        self.hub = hub

    # called by `Unpickler.load` before an uninitalized object is about to be filled with members;
    def _load_build(self):
        """See `pickle.py` in Python's source code."""
        # if the ctor. function (penultimate on the stack) is the `Ref` class...
        if isinstance(self.stack[-2], Ref):
            state = self.stack[-1]
            # Ref.__setstate__ will know it's a remote ref if the state is a tuple
            self.stack[-1] = (state, self.hub)
            self.load_build()  # continue with the default implementation

            # detect our own refs sent back to us
            ref = self.stack[-1]
            if ref.uri.node == self.hub.node:
                ref.is_local = True
                ref._cell = self.hub.guardian.lookup_cell(ref.uri)
                # self.dbg(("dead " if not ref._cell else "") + "local ref detected")
                del ref._hub
        else:
            self.load_build()<commit_msg>Only set Ref._hub in remoting.IncomingMessageUnpickler during TESTING<commit_after># coding: utf8
from __future__ import print_function, absolute_import

import inspect
import random
import re
import sys
import traceback
from cStringIO import StringIO
from collections import deque
from decimal import Decimal
from pickle import Unpickler, BUILD
from cPickle import dumps

from twisted import internet
from twisted.internet import reactor
from twisted.internet.task import LoopingCall
from txzmq import ZmqEndpoint

from spinoff.actor import _actor
from spinoff.actor import Ref, Uri, Node
from spinoff.actor._actor import _VALID_NODEID_RE, _validate_nodeid
from spinoff.actor.events import Events, DeadLetter
from spinoff.util.logging import Logging, logstring


# TODO: use shorter messages outside of testing
PING = 'ping'
PONG = 'pong'


_VALID_ADDR_RE = re.compile('tcp://%s' % (_VALID_NODEID_RE.pattern,))


class ConnectedNode(object):
    def __init__(self, state, last_seen):
        self.state = state
        self.last_seen = last_seen
        self.queue = deque()


class Hub(Logging):
    """Handles traffic between actors on different nodes.

    The wire-transport implementation is specified/overridden by the `incoming` and `outgoing` parameters.

    """
    __doc_HEARTBEAT_INTERVAL__ = (
        "Time on seconds after which to send out a heartbeat signal to all known nodes. Regular messages can be "
        "subsituted by the framework for heartbeats to save network bandwidth.")
    HEARTBEAT_INTERVAL = 1.0

    __doc_MAX_SILENCE_BETWEEN_HEARTBEATS__ = (
        "Maximum length of silence in seconds between two consecutive heartbeat signals from a node after which to "
        "consider the node as temporarily not available, put it in the 'radio-silence' state, and start queueing all "
        "messages posted to it by actors.")
    MAX_SILENCE_BETWEEN_HEARTBEATS = 5.0

    __doc_TIME_TO_KEEP_HOPE__ = (
        "Time in seconds after which the 'radio-silence' state transforms into a 'silently-hoping' state wherein the "
        "is considered to have gone offline for an extended duration. All currently queued and any future messages to"
        "that node will immediately be turned into `DeadLetter` events.")
    TIME_TO_KEEP_HOPE = 55.0

    __doc_QUEUE_ITEM_LIFETIME__ = (
        "Time in seconds for which to keep a queued message alive, after which the message is turned into a "
        "`DeadLetter` event. The default value is chosen such that exactly when the target node goes into the "
        "'silently-hoping' visibility state, all messages to it are discarded.")
    QUEUE_ITEM_LIFETIME = MAX_SILENCE_BETWEEN_HEARTBEATS + TIME_TO_KEEP_HOPE

    node = None

    def __init__(self, incoming, outgoing, node, reactor=reactor):
        if not node or not isinstance(node, str):
            raise TypeError("The 'node' argument to Hub must be a str")
        _validate_nodeid(node)
        self.node = node
        self.addr = 'tcp://' + node if node else None

        self.reactor = reactor

        self.outgoing = outgoing
        incoming.gotMessage = self._got_message
        incoming.addEndpoints([ZmqEndpoint('bind', self.addr)])

        self.connections = {}

        l1 = LoopingCall(self._send_heartbeat)
        l1.clock = reactor
        l1.start(self.HEARTBEAT_INTERVAL)

        l2 = LoopingCall(self._purge_old_items_in_queue)
        l2.clock = reactor
        l2.start(1.0)

    _guardian = None

    def logstate(self):
        return {str(self.reactor.seconds()): True}

    def _get_guardian(self):
        return self._guardian

    def _set_guardian(self, guardian):
        if self._guardian:
            raise RuntimeError("Hub already bound to a Guardian")
        self._guardian = guardian

    guardian = property(_get_guardian, _set_guardian)

    @logstring(u"⇝")
    def send(self, msg, to_remote_actor_pointed_to_by):
        ref = to_remote_actor_pointed_to_by
        # self.dbg(u"%r → %r" % (msg, ref))

        nodeid = ref.uri.node

        # assert addr and addr != self.node, "TODO: remote-ref pointing to the local node detected"
        # TODO: if it's determined that it makes sense to allow the existence of such refs, enable the following block
        if not nodeid or nodeid == self.node:
            cell = self.guardian.lookup_cell(ref.uri)
            if not cell:
                ref.is_local = True  # next time, just put it straight to DeadLetters
                Events.log(DeadLetter(ref, msg))
                return
            ref._cell = cell
            ref.is_local = True
            ref << msg
            return

        addr = ref.uri.root.url

        conn = self.connections.get(addr)
        if not conn:
            self.dbg("%s set from not-known => %s" % (addr, 'radiosilence',))
            conn = self.connections[addr] = ConnectedNode(
                state='radiosilence',
                # so last_seen checks would mark the node as `silentlyhoping` in `TIME_TO_KEEP_HOPE` seconds from now;
                # conceptually speaking, this means the node seems to have always existed but went out of view exactly
                # now, and before sending can start "again", it needs to come back:
                last_seen=self.reactor.seconds() - self.MAX_SILENCE_BETWEEN_HEARTBEATS
            )
            self._connect(addr, conn)

        if conn.state == 'visible':
            self.outgoing.sendMsg((addr, dumps((ref.uri.path, msg), protocol=2)))
        else:
            if conn.queue is None:
                Events.log(DeadLetter(ref, msg))
            else:
                conn.queue.append(((ref, msg), self.reactor.seconds()))

    @logstring(u"⇜")
    def _got_message(self, sender_addr, msg):
        if msg in (PING, PONG):
            self.dbg(u"❤ %s ← %s" % (msg, sender_addr,))
        else:
            path, msg_ = self._loads(msg)
            self.dbg(u"%r ← %s   → %s" % (msg_, sender_addr, path))
            cell = self.guardian.lookup_cell(Uri.parse(path))
            if not cell:
                Events.log(DeadLetter(Ref(None, Uri.parse(path)), msg_))
            else:
                cell.receive(msg_)  # XXX: force_async=True perhaps?

        if sender_addr not in self.connections:
            assert msg == PING, "initial message sent to another node should be PING"
            self.dbg("%s went not-known => %s" % (sender_addr, 'reverse-radiosilence',))
            conn = self.connections[sender_addr] = ConnectedNode(
                state='reverse-radiosilence',
                last_seen=self.reactor.seconds())
            self._connect(sender_addr, conn)
        else:
            conn = self.connections[sender_addr]
            conn.last_seen = self.reactor.seconds()

            prevstate = conn.state
            conn.state = 'reverse-radiosilence' if msg == PING else 'visible'
            if prevstate != conn.state:
                self.dbg("%s went %s => %s" % (sender_addr, prevstate, conn.state))
            if prevstate != 'visible' and conn.state == 'visible':
                while conn.queue:
                    (ref, queued_msg), _ = conn.queue.popleft()
                    assert ref.uri.root.url == sender_addr
                    self.outgoing.sendMsg((sender_addr, dumps((ref.uri.path, queued_msg), protocol=2)))

    def _connect(self, addr, conn):
        assert _valid_addr(addr)
        # self.dbg("...connecting to %s" % (addr,))
        self.outgoing.addEndpoints([ZmqEndpoint('connect', addr)])
        # send one heartbeat immediately for better latency
        self.dbg(u"►► ❤ → %s" % (addr,))
        self._heartbeat_once(addr, PING if conn.state == 'radiosilence' else PONG)

    @logstring(u"❤")
    def _send_heartbeat(self):
        try:
            # self.dbg("→ %r" % (list(self.connections),))
            t = self.reactor.seconds()
            consider_dead_from = t - self.MAX_SILENCE_BETWEEN_HEARTBEATS
            consider_lost_from = consider_dead_from - self.TIME_TO_KEEP_HOPE
            # self.dbg("consider_dead_from", consider_dead_from, "consider_lost_from", consider_lost_from)

            for addr, conn in self.connections.items():
                # self.dbg("%s last seen at %ss" % (addr, conn.last_seen))
                if conn.state == 'silentlyhoping':
                    # self.dbg("silently hoping...")
                    self._heartbeat_once(addr, PING)
                elif conn.last_seen < consider_lost_from:
                    self.dbg("%s went %s => %s after %ds of silence" % (addr, conn.state, 'silentlyhoping', (t - conn.last_seen)))
                    conn.state = 'silentlyhoping'
                    for (ref, msg), _ in conn.queue:
                        # self.dbg("dropping %r" % (msg,))
                        Events.log(DeadLetter(ref, msg))
                    conn.queue = None
                    self._heartbeat_once(addr, PING)
                elif conn.last_seen < consider_dead_from:
                    if conn.state != 'radiosilence':
                        self.dbg("%s went %s => %s" % (addr, conn.state, 'radiosilence',))
                        conn.state = 'radiosilence'
                    self._heartbeat_once(addr, PING)
                else:
                    # self.dbg("%s still %s; not seen for %s" % (addr, conn.state, '%ds' % (t - conn.last_seen) if conn.last_seen is not None else 'eternity',))
                    self._heartbeat_once(addr, PONG)
            # self.dbg(u"%s ✓" % (self.reactor.seconds(),))
        except Exception:
            self.panic("failed to send heartbeat:\n", traceback.format_exc())

    @logstring(u"⇝ ❤")
    def _heartbeat_once(self, addr, signal):
        assert _valid_addr(addr)
        self.log(u"%s →" % (signal,), addr)
        self.outgoing.sendMsg((addr, signal))

    @logstring("PURGE")
    def _purge_old_items_in_queue(self):
        for conn in self.connections.values():
            try:
                keep_until = self.reactor.seconds() - self.QUEUE_ITEM_LIFETIME
                # self.dbg(conn.queue, "keep_until = %r, QUEUE_ITEM_LIFETIME = %r" % (keep_until, self.QUEUE_ITEM_LIFETIME,))
                q = conn.queue
                while q:
                    (ref, msg), timestamp = q[0]
                    if timestamp >= keep_until:
                        break
                    else:
                        q.popleft()
                        Events.log(DeadLetter(ref, msg))
            except Exception:
                self.panic("failed to clean queue:\n", traceback.format_exc())

    def _loads(self, data):
        return IncomingMessageUnpickler(self, StringIO(data)).load()

    def __repr__(self):
        return '<%s>' % (self.node,)


class HubWithNoRemoting(object):
    """A dummy hub used during networkless testing and in production when no remoting should be available.

    All it does is imitate the interface of the real `Hub`, and report attempts to send remote messages as
    `RuntimeError`s.

    """
    # to be compatible with Hub:
    guardian = None
    node = None

    def send(self, ref, msg):
        raise RuntimeError("Attempt to send a message to a remote ref but remoting is not available")


class IncomingMessageUnpickler(Unpickler, Logging):
    """Unpickler for attaching a `Hub` instance to all deserialized `Ref`s."""

    def __init__(self, hub, file):
        Unpickler.__init__(self, file)
        self.hub = hub

    # called by `Unpickler.load` before an uninitalized object is about to be filled with members;
    def _load_build(self):
        """See `pickle.py` in Python's source code."""
        # if the ctor. function (penultimate on the stack) is the `Ref` class...
        if isinstance(self.stack[-2], Ref):
            if _actor.TESTING:
                state = self.stack[-1]
                # Ref.__setstate__ will know it's a remote ref if the state is a tuple
                self.stack[-1] = (state, self.hub)
                self.load_build()  # continue with the default implementation

            # detect our own refs sent back to us
            ref = self.stack[-1]
            if ref.uri.node == self.hub.node:
                ref.is_local = True
                ref._cell = self.hub.guardian.lookup_cell(ref.uri)
                # self.dbg(("dead " if not ref._cell else "") + "local ref detected")
                del ref._hub
        else:
            self.load_build()
{'fraction_non_alphanumeric': 0.0995293353891098, 'fraction_numerical': 0.0018664286293921934, 'mean_word_length': 2.2146863179861747, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, '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': '1328450', 'n_tokens_mistral': 7408, 'n_tokens_neox': 7108, 'n_words': 2089}