<commit_before>from Queue import Empty
import logging
import multiprocessing
import random
from threading import Thread
import time
from time import time as _time
import uuid
import datetime
from motorway.messages import Message
from motorway.mixins import GrouperMixin, SendMessageMixin
from motorway.utils import set_timeouts_on_socket, get_connections_block
import zmq
from setproctitle import setproctitle

logger = logging.getLogger(__name__)


class Intersection(GrouperMixin, SendMessageMixin, object):

    fields = []

    def __init__(self):
        self.messages_processed = 0
        self.process_uuid = str(uuid.uuid4())
        self.receive_port = None
        self.send_socks = {}
        self.send_grouper = None
        self.controller_sock = None
        self.message_batch_start = datetime.datetime.now()  # This is used to time how much time messages take

    def _process(self, receive_sock, controller_sock=None):

        try:
            if getattr(self.process, 'batch_process', None):
                poller = zmq.Poller()
                poller.register(receive_sock, zmq.POLLIN)
                value = []
                end_time = _time() + self.process.wait
                while end_time > _time() and len(value) < self.process.limit:
                    socks = dict(poller.poll(timeout=1000))
                    if socks.get(receive_sock) == zmq.POLLIN:
                        value.append(receive_sock.recv_json())
                self.messages_processed += len(value)
            else:
                poller = zmq.Poller()
                poller.register(receive_sock, zmq.POLLIN)
                socks = dict(poller.poll(timeout=1000))
                if socks.get(receive_sock) == zmq.POLLIN:
                    value = receive_sock.recv_json()
                else:
                    return
                self.messages_processed += 1
            if value:
                if isinstance(value, list):
                    message = [Message.from_message(m, controller_sock, process_name=self.process_uuid) for m in value]
                else:
                    message = Message.from_message(value, controller_sock, process_name=self.process_uuid)
                try:
                    self.message_batch_start = datetime.datetime.now()
                    for generated_message in self.process(message):
                        if generated_message is not None and self.send_socks:
                            self.send_message(generated_message, self.process_uuid)
                except Exception as e:
                    logger.error(str(e), exc_info=True)
                    if isinstance(message, list):
                        [m.fail() for m in message]
                    else:
                        message.fail()

        except Empty:  # Didn't receive anything from ZMQ
            pass

    def ack(self, message):
        message.ack(time_consumed=datetime.datetime.now() - self.message_batch_start)
        self.message_batch_start = datetime.datetime.now()

    def fail(self, message, **kwargs):
        message.fail(**kwargs)

    def process(self, message):
        """
        This function is called continuously by the intersection.

        :yield: :class:`motorway.messages.Message` instance
        :param message: :class:`motorway.messages.Message` instance or :func:`list` if using
            :func:`motorway.decorators.batch_process`

        """
        raise NotImplementedError()

    @classmethod
    def run(cls, input_stream, output_stream=None, refresh_connection_stream=None, grouper_cls=None):
        self = cls()
        process_name = multiprocessing.current_process().name
        logger.info("Running %s" % process_name)
        logger.debug("%s listening on %s and pushing to %s" % (process_name, input_stream, output_stream))
        setproctitle("data-pipeline: %s" % process_name)
        context = zmq.Context()

        # Create Thread Factories :-)
<commit_msg>motorway/intersection: Use same method of timing as ramp - discarding any queuing time<commit_after>from Queue import Empty
import logging
import multiprocessing
import random
from threading import Thread
import time
from time import time as _time
import uuid
import datetime
from motorway.messages import Message
from motorway.mixins import GrouperMixin, SendMessageMixin
from motorway.utils import set_timeouts_on_socket, get_connections_block
import zmq
from setproctitle import setproctitle

logger = logging.getLogger(__name__)


class Intersection(GrouperMixin, SendMessageMixin, object):

    fields = []

    def __init__(self):
        self.messages_processed = 0
        self.process_uuid = str(uuid.uuid4())
        self.receive_port = None
        self.send_socks = {}
        self.send_grouper = None
        self.controller_sock = None
        self.message_batch_start = datetime.datetime.now()  # This is used to time how much time messages take

    def _process(self, receive_sock, controller_sock=None):

        try:
            if getattr(self.process, 'batch_process', None):
                poller = zmq.Poller()
                poller.register(receive_sock, zmq.POLLIN)
                value = []
                end_time = _time() + self.process.wait
                while end_time > _time() and len(value) < self.process.limit:
                    socks = dict(poller.poll(timeout=1000))
                    if socks.get(receive_sock) == zmq.POLLIN:
                        value.append(receive_sock.recv_json())
                self.messages_processed += len(value)
            else:
                poller = zmq.Poller()
                poller.register(receive_sock, zmq.POLLIN)
                socks = dict(poller.poll(timeout=1000))
                if socks.get(receive_sock) == zmq.POLLIN:
                    value = receive_sock.recv_json()
                else:
                    return
                self.messages_processed += 1
            if value:
                if isinstance(value, list):
                    message = [Message.from_message(m, controller_sock, process_name=self.process_uuid) for m in value]
                else:
                    message = Message.from_message(value, controller_sock, process_name=self.process_uuid)
                try:
                    start_time = datetime.datetime.now()
                    for generated_message in self.process(message):
                        if generated_message is not None and self.send_socks:
                            self.send_message(generated_message, self.process_uuid, time_consumed=datetime.datetime.now() - start_time)
                        start_time = datetime.datetime.now()
                except Exception as e:
                    logger.error(str(e), exc_info=True)
                    if isinstance(message, list):
                        [m.fail() for m in message]
                    else:
                        message.fail()

        except Empty:  # Didn't receive anything from ZMQ
            pass

    def ack(self, message):
        message.ack()

    def fail(self, message, **kwargs):
        message.fail(**kwargs)

    def process(self, message):
        """
        This function is called continuously by the intersection.

        :yield: :class:`motorway.messages.Message` instance
        :param message: :class:`motorway.messages.Message` instance or :func:`list` if using
            :func:`motorway.decorators.batch_process`

        """
        raise NotImplementedError()

    @classmethod
    def run(cls, input_stream, output_stream=None, refresh_connection_stream=None, grouper_cls=None):
        self = cls()
        process_name = multiprocessing.current_process().name
        logger.info("Running %s" % process_name)
        logger.debug("%s listening on %s and pushing to %s" % (process_name, input_stream, output_stream))
        setproctitle("data-pipeline: %s" % process_name)
        context = zmq.Context()

        # Create Thread Factories :-)

{'fraction_non_alphanumeric': 0.08526355327407037, 'fraction_numerical': 0.0027544760235382495, 'mean_word_length': 2.090135396518375, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '2193407', 'n_tokens_mistral': 2212, 'n_tokens_neox': 2077, 'n_words': 518}