<commit_before>import inspect
import os
import zmq
import sys
import time
import itertools
import logbook
from setproctitle import setproctitle
from signal import SIGHUP, SIGINT

from collections import OrderedDict, Counter

from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \
    CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \

from zipline.utils.protocol_utils import ndict

INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES

CONTROLLER_TRANSITIONS = frozenset([
    (-1            , INIT),
    (INIT          , SOURCES_READY),
    (SOURCES_READY , RUNNING),

    (INIT          , TERMINATE), # pseudo failure mode
    (SOURCES_READY , TERMINATE), # pseudo failure mode
    (RUNNING       , TERMINATE),
])

class UnknownChatter(Exception):
    def __init__(self, name):
        self.named = name
    def __str__(self):
        return """Component calling itself "%s" talking on unexpected channel""" % self.named


log = logbook.Logger('Controller')

# The scalars determining the timing of the monitor behavior for
# the system.

PARAMETERS = ndict(dict(
    # time Monitor will wait for a heartbeat, in seconds
    GENERATIONAL_PERIOD        = 10,
    # time Component will wait for GO and for a heartbeat before
    # timing out.
    MAX_COMPONENT_WAIT         = 20,
    ALLOWED_SKIPPED_HEARTBEATS = 10,
    ALLOWED_INVALID_HEARTBEATS = 3,
    PRESTART_HEARBEATS         = 3,
    SOURCES_START_HEARTBEATS   = 3,
    SYSTEM_TIMEOUT             = 50,
))

class Controller(object):
    """
    A N to M messaging system for inter component communication.

    :param pub_socket: Socket to publish messages, the starting
                       point of :func message_listener: .

    :param route_socket: Socket to listen for status updates for
                         the individual components.
                         :func message_sender: .

    """

    # Turn on debug for verbose logging of the system.
    debug = True
    period = PARAMETERS.GENERATIONAL_PERIOD

    def __init__(self, pub_socket, route_socket):

        self.nosignals  = False
        self.context    = None
        self.zmq        = None
        self.zmq_poller = None

        self.running = False
        self.alive = False
        self.tracked = set()
        self.finished = set()

        self.responses = set()

        self.ctime    = 0
        self.tic      = time.time()
        self.freeform = False
        self._state   = -1

        self.associated = []

        self.pub_socket   = pub_socket
        self.route_socket = route_socket

        self.error_replay = OrderedDict()

        self.missed_beats = Counter()

        # if we are inside a test, we want to skip signalling
        # back to the parent process.
        self.inside_test = 'nose' in inspect.stack()[-1][1]



    def init_zmq(self):
        self.zmq        = zmq
        self.context    = self.zmq.Context()
        self.zmq_poller = self.zmq.Poller
        return

    def manage(self, topology):
        """
        Give the controller a set set of components to manage and
        a set of state transitions for the entire system.
        """
        # A freeform topology is where we heartbeat with anything
        # that shows up.
        if topology == 'freeform':
            self.freeform = True
            self.topology = frozenset([])
        else:
            self.freeform = False
            self.topology = frozenset(topology)
        self.alive = True

    @property
    def state(self):
        #log.info('returned %s' % self._state)
        return self._state

    @state.setter
    def state(self, new):
        old = self._state

        if (old, new) in CONTROLLER_TRANSITIONS:
            self._state = new
            log.info("State Transition : %s -> %s" % (old, self._state))
        else:
            raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new))

    def run(self):
        self.running = True
        self.init_zmq(self.zmq_flavor)
        setproctitle('Monitor')

        self.state = CONTROL_STATES.INIT

        # Interpreter SIDE EFFECT
        # -----------------------
        # The last breathe of the interpreter will assume that we've
        # failed unless we specify otherwise.
        log.info('registering exit function')
        sys.exitfunc = self.signal_interrupt
        # We overload this if ( and only if ) the topology exits
        # cleanly. This prevents failure modes where the monitor
        # dies.

        try:
            return self._poll() # use a python loop
        except KeyboardInterrupt:
            log.info('Shutdown event loop')

    def log_status(self):
        """
        Snapshot of the tracked components at every period.
        """
        #log.info("Tracking component : %s" % ([c for c in self.tracked],))
        pass

    def replay_errors(self):<commit_msg>minor fix<commit_after>import inspect
import os
import zmq
import sys
import time
import itertools
import logbook
from setproctitle import setproctitle
from signal import SIGHUP, SIGINT

from collections import OrderedDict, Counter

from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \
    CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \

from zipline.utils.protocol_utils import ndict

INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES

CONTROLLER_TRANSITIONS = frozenset([
    (-1            , INIT),
    (INIT          , SOURCES_READY),
    (SOURCES_READY , RUNNING),

    (INIT          , TERMINATE), # pseudo failure mode
    (SOURCES_READY , TERMINATE), # pseudo failure mode
    (RUNNING       , TERMINATE),
])

class UnknownChatter(Exception):
    def __init__(self, name):
        self.named = name
    def __str__(self):
        return """Component calling itself "%s" talking on unexpected channel""" % self.named


log = logbook.Logger('Controller')

# The scalars determining the timing of the monitor behavior for
# the system.

PARAMETERS = ndict(dict(
    # time Monitor will wait for a heartbeat, in seconds
    GENERATIONAL_PERIOD        = 10,
    # time Component will wait for GO and for a heartbeat before
    # timing out.
    MAX_COMPONENT_WAIT         = 20,
    ALLOWED_SKIPPED_HEARTBEATS = 10,
    ALLOWED_INVALID_HEARTBEATS = 3,
    PRESTART_HEARBEATS         = 3,
    SOURCES_START_HEARTBEATS   = 3,
    SYSTEM_TIMEOUT             = 50,
))

class Controller(object):
    """
    A N to M messaging system for inter component communication.

    :param pub_socket: Socket to publish messages, the starting
                       point of :func message_listener: .

    :param route_socket: Socket to listen for status updates for
                         the individual components.
                         :func message_sender: .

    """

    # Turn on debug for verbose logging of the system.
    debug = True
    period = PARAMETERS.GENERATIONAL_PERIOD

    def __init__(self, pub_socket, route_socket):

        self.nosignals  = False
        self.context    = None
        self.zmq        = None
        self.zmq_poller = None

        self.running = False
        self.alive = False
        self.tracked = set()
        self.finished = set()

        self.responses = set()

        self.ctime    = 0
        self.tic      = time.time()
        self.freeform = False
        self._state   = -1

        self.associated = []

        self.pub_socket   = pub_socket
        self.route_socket = route_socket

        self.error_replay = OrderedDict()

        self.missed_beats = Counter()

        # if we are inside a test, we want to skip signalling
        # back to the parent process.
        self.inside_test = 'nose' in inspect.stack()[-1][1]



    def init_zmq(self):
        self.zmq        = zmq
        self.context    = self.zmq.Context()
        self.zmq_poller = self.zmq.Poller
        return

    def manage(self, topology):
        """
        Give the controller a set set of components to manage and
        a set of state transitions for the entire system.
        """
        # A freeform topology is where we heartbeat with anything
        # that shows up.
        if topology == 'freeform':
            self.freeform = True
            self.topology = frozenset([])
        else:
            self.freeform = False
            self.topology = frozenset(topology)
        self.alive = True

    @property
    def state(self):
        #log.info('returned %s' % self._state)
        return self._state

    @state.setter
    def state(self, new):
        old = self._state

        if (old, new) in CONTROLLER_TRANSITIONS:
            self._state = new
            log.info("State Transition : %s -> %s" % (old, self._state))
        else:
            raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new))

    def run(self):
        self.running = True
        self.init_zmq()
        setproctitle('Monitor')

        self.state = CONTROL_STATES.INIT

        # Interpreter SIDE EFFECT
        # -----------------------
        # The last breathe of the interpreter will assume that we've
        # failed unless we specify otherwise.
        log.info('registering exit function')
        sys.exitfunc = self.signal_interrupt
        # We overload this if ( and only if ) the topology exits
        # cleanly. This prevents failure modes where the monitor
        # dies.

        try:
            return self._poll() # use a python loop
        except KeyboardInterrupt:
            log.info('Shutdown event loop')

    def log_status(self):
        """
        Snapshot of the tracked components at every period.
        """
        #log.info("Tracking component : %s" % ([c for c in self.tracked],))
        pass

    def replay_errors(self):
{'fraction_non_alphanumeric': 0.08502524992270431, 'fraction_numerical': 0.0032979490879109555, 'mean_word_length': 2.0602333648691267, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '6780106', 'n_tokens_mistral': 2921, 'n_tokens_neox': 2721, 'n_words': 846}