<commit_before>'''
Created on 2015/11/30

@author: hubo

Process events multi-threaded or multi-processed
'''
from vlcp.event.runnable import RoutineContainer
import multiprocessing
import threading
import socket
import errno
import sys
from vlcp.event.event import Event, withIndices
from vlcp.event.core import POLLING_IN, PollEvent
import functools
import traceback
if sys.version_info[0] >= 3:
    from queue import Full, Queue, Empty
else:
    from Queue import Full, Queue, Empty

@withIndices('connector', 'type')
class ConnectorControlEvent(Event):
    ADDMATCHERS = 'addmatcher'
    REMOVEMATCHERS = 'removematcher'
    SETMATCHERS = 'setmatchers'
    STOPRECEIVE = 'stopreceive'
    STARTRECEIVE = 'startreceive'
import os

class _Pipe(object):
    "Make a pipe looks like a socket"
    def __init__(self, fd, canread = True):
        self.canread = canread
        self.canwrite = not canread
        self.fd = fd
    def setblocking(self, blocking):
        import fcntl
        if blocking:
            fcntl.fcntl(self.fd, fcntl.F_SETFL, 0)
        else:
            fcntl.fcntl(self.fd, fcntl.F_SETFL, os.O_NONBLOCK)
    def fileno(self):
        return self.fd
    def send(self, data):
        os.write(self.fd, data)
    def recv(self, size):
        return os.read(self.fd, size)
    def close(self):
        if self.fd:
            os.close(self.fd)
            self.fd = None
    def __del__(self):
        self.close()

class Connector(RoutineContainer):
    def __init__(self, worker_start, matchers = (), scheduler = None, mp = True, inputlimit = 0, allowcontrol = True):
        '''
        @param worker_start: func(queuein, queueout), queuein is the input queue, queueout is the
               output queue. For queuein, each object is (event, matcher) tuple; For queueout, each
               object is a tuple of events to send. Every object in queuein must have a response in
               queueout.
        @param matcheres: match events to be processed by connector.
        @param scheduler: bind to specified scheduler
        @param mp: use multiprocessing if possible. For windows, multi-threading is always used.
        @param inputlimit: input queue size limit. 0 = infinite.
        @param allowcontrol: if True, the connector accepts ConnectorControlEvent for connector configuration.
        '''
        RoutineContainer.__init__(self, scheduler, True)
        self.worker_start = worker_start
        self.matchers = set(matchers)
        self.mp = mp
        self.inputlimit = inputlimit
        self.allowcontrol = allowcontrol
        self.stopreceive = False
        self.jobs = 0
    @staticmethod
    def connector_pipe(queuein, pipeout, worker_start):
        try:
            queueout = multiprocessing.Queue()
            worker_start(queuein, queueout)
            while True:
                try:
                    events = queueout.get()
                    if events is None:
                        break
                    pipeout.send(events)
                except EOFError:
                    break
                except OSError as exc:
                    if exc.args[0] == errno.EINTR:
                        continue
                    else:
                        break
        finally:
            pipeout.close()
    @staticmethod
    def connector_socket(queuein, pipeout, worker_start):
        try:
            queueout = Queue()
            worker_start(queuein, queueout)
            while True:
                try:
                    events = queueout.get()
                    if events is None:
                        break
                    pipeout[0].put(events)
                    while True:
                        try:
                            events = queueout.get(False)
                            pipeout[0].put(events)
                        except Empty:
                            break
                    pipeout[1].send(b'\x00')
                except EOFError:
                    break
                except OSError as exc:
                    if exc.args[0] == errno.EINTR:
                        continue
                    else:
                        break
                except:
                    pass
        finally:
            pipeout[1].close()
    def enqueue(self, queue, event, matcher):
        queue.put((event, matcher), False)
        event.canignore = True
        self.jobs += 1
        if self.jobs == 1:
            self.scheduler.setPollingDaemon(self.pipein, False)
    def sendevents(self, events):
        for e in events:
            self.scheduler.emergesend(e)
        self.jobs -= 1
        if self.jobs == 0:
            self.scheduler.setPollingDaemon(self.pipein, True)
    def _createobjs(self, fork, mp):
        if mp:
            queue = multiprocessing.Queue(self.inputlimit)
        else:
            queue = Queue(self.inputlimit)
        if mp:
            pipein, pipeout = multiprocessing.Pipe()
            process = multiprocessing.Process(target=self.connector_pipe, args=(queue, pipeout, self.worker_start))
            outqueue = None<commit_msg>reset signal handler in sub process<commit_after>'''
Created on 2015/11/30

@author: hubo

Process events multi-threaded or multi-processed
'''
from vlcp.event.runnable import RoutineContainer
import multiprocessing
import threading
import socket
import signal
import errno
import sys
from vlcp.event.event import Event, withIndices
from vlcp.event.core import POLLING_IN, PollEvent
import functools
import traceback
if sys.version_info[0] >= 3:
    from queue import Full, Queue, Empty
else:
    from Queue import Full, Queue, Empty

@withIndices('connector', 'type')
class ConnectorControlEvent(Event):
    ADDMATCHERS = 'addmatcher'
    REMOVEMATCHERS = 'removematcher'
    SETMATCHERS = 'setmatchers'
    STOPRECEIVE = 'stopreceive'
    STARTRECEIVE = 'startreceive'
import os

class _Pipe(object):
    "Make a pipe looks like a socket"
    def __init__(self, fd, canread = True):
        self.canread = canread
        self.canwrite = not canread
        self.fd = fd
    def setblocking(self, blocking):
        import fcntl
        if blocking:
            fcntl.fcntl(self.fd, fcntl.F_SETFL, 0)
        else:
            fcntl.fcntl(self.fd, fcntl.F_SETFL, os.O_NONBLOCK)
    def fileno(self):
        return self.fd
    def send(self, data):
        os.write(self.fd, data)
    def recv(self, size):
        return os.read(self.fd, size)
    def close(self):
        if self.fd:
            os.close(self.fd)
            self.fd = None
    def __del__(self):
        self.close()

class Connector(RoutineContainer):
    def __init__(self, worker_start, matchers = (), scheduler = None, mp = True, inputlimit = 0, allowcontrol = True):
        '''
        @param worker_start: func(queuein, queueout), queuein is the input queue, queueout is the
               output queue. For queuein, each object is (event, matcher) tuple; For queueout, each
               object is a tuple of events to send. Every object in queuein must have a response in
               queueout.
        @param matcheres: match events to be processed by connector.
        @param scheduler: bind to specified scheduler
        @param mp: use multiprocessing if possible. For windows, multi-threading is always used.
        @param inputlimit: input queue size limit. 0 = infinite.
        @param allowcontrol: if True, the connector accepts ConnectorControlEvent for connector configuration.
        '''
        RoutineContainer.__init__(self, scheduler, True)
        self.worker_start = worker_start
        self.matchers = set(matchers)
        self.mp = mp
        self.inputlimit = inputlimit
        self.allowcontrol = allowcontrol
        self.stopreceive = False
        self.jobs = 0
    @staticmethod
    def wrap_signal(func):
        def f(*args, **kwargs):
            signal.signal(signal.SIGINT, signal.SIG_DFL)
            signal.signal(signal.SIGTERM, signal.SIG_DFL)
            return func(*args, **kwargs)
        return f
    @staticmethod
    def connector_pipe(queuein, pipeout, worker_start):
        try:
            queueout = multiprocessing.Queue()
            worker_start(queuein, queueout)
            while True:
                try:
                    events = queueout.get()
                    if events is None:
                        break
                    pipeout.send(events)
                except EOFError:
                    break
                except OSError as exc:
                    if exc.args[0] == errno.EINTR:
                        continue
                    else:
                        break
        finally:
            pipeout.close()
    @staticmethod
    def connector_socket(queuein, pipeout, worker_start):
        try:
            queueout = Queue()
            worker_start(queuein, queueout)
            while True:
                try:
                    events = queueout.get()
                    if events is None:
                        break
                    pipeout[0].put(events)
                    while True:
                        try:
                            events = queueout.get(False)
                            pipeout[0].put(events)
                        except Empty:
                            break
                    pipeout[1].send(b'\x00')
                except EOFError:
                    break
                except OSError as exc:
                    if exc.args[0] == errno.EINTR:
                        continue
                    else:
                        break
                except:
                    pass
        finally:
            pipeout[1].close()
    def enqueue(self, queue, event, matcher):
        queue.put((event, matcher), False)
        event.canignore = True
        self.jobs += 1
        if self.jobs == 1:
            self.scheduler.setPollingDaemon(self.pipein, False)
    def sendevents(self, events):
        for e in events:
            self.scheduler.emergesend(e)
        self.jobs -= 1
        if self.jobs == 0:
            self.scheduler.setPollingDaemon(self.pipein, True)
    def _createobjs(self, fork, mp):
        if mp:
            queue = multiprocessing.Queue(self.inputlimit)
        else:
            queue = Queue(self.inputlimit)
        if mp:
            pipein, pipeout = multiprocessing.Pipe()
            process = multiprocessing.Process(target=self.wrap_signal(self.connector_pipe), args=(queue, pipeout, self.worker_start))
            outqueue = None
{'fraction_non_alphanumeric': 0.07996201329534663, 'fraction_numerical': 0.0049382716049382715, 'mean_word_length': 1.6823739174732553, 'pattern_counts': {'":': 0, '<': 3, '<?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': '2192117', 'n_tokens_mistral': 2978, 'n_tokens_neox': 2833, 'n_words': 801}