<commit_before># -*- coding: utf-8 -*-

# This file is part of the Rocket Web Server
# Copyright (c) 2009 <NAME>

# Import System Modules
import time
import logging
import select
from threading import Thread

# Import Package Modules
from . import IS_JYTHON, THREAD_STOP_CHECK_INTERVAL, NullHandler

# Setup Polling if supported (but it doesn't work well on Jython)
if hasattr(select, 'poll') and not IS_JYTHON:
    try:
        poll = select.epoll()
    except:
        poll = select.poll()
else:
    poll = None

class Monitor(Thread):
    # Monitor worker class.

    def __init__(self,
                 monitor_queue,
                 active_queue,
                 timeout,
                 *args,
                 **kwargs):

        Thread.__init__(self, *args, **kwargs)

        # Instance Variables
        self.monitor_queue = monitor_queue
        self.active_queue = active_queue
        self.timeout = timeout

        self.connections = set()
        self.active = False

    def run(self):
        self.name = self.getName()
        self.log = logging.getLogger('Rocket.Monitor')
        self.log.addHandler(NullHandler())

        self.active = True
        poll_dict = dict()
        conn_list = list()
        list_changed = False

        if __debug__:
            self.log.debug('Entering monitor loop.')

        # Enter thread main loop
        while self.active:
            # Move the queued connections to the selection pool
            while not self.monitor_queue.empty() or not len(self.connections):
                if __debug__:
                    self.log.debug('In "receive timed-out connections" loop.')

                c = self.monitor_queue.get()

                if c is None:
                    # A non-client is a signal to die
                    if __debug__:
                        self.log.debug('Received a death threat.')
                    return

                self.log.debug('Received a timed out connection.')

                if __debug__:
                    assert(c not in self.connections)

                if IS_JYTHON:
                    # Jython requires a socket to be in Non-blocking mode in
                    # order to select on it.
                    c.setblocking(False)

                if __debug__:
                    self.log.debug('Adding connection to monitor list.')

                if poll:
                    poll.register(c)
                    poll_dict.update({c.fileno():c})

                self.connections.add(c)
                list_changed = True

            # Wait on those connections
            self.log.debug('Blocking on connections')
            if poll:
                readable = [poll_dict[x[0]] for x in poll.poll(THREAD_STOP_CHECK_INTERVAL)]
            else:
                if list_changed:
                    conn_list = list(self.connections)
                    list_changed = False

                readable = select.select(conn_list,
                                         [],
                                         [],
                                         THREAD_STOP_CHECK_INTERVAL)[0]

            # If we have any readable connections, put them back
            for r in readable:
                if __debug__:
                    self.log.debug('Restoring readable connection')

                if IS_JYTHON:
                    # Jython requires a socket to be in Non-blocking mode in
                    # order to select on it, but the rest of the code requires
                    # that it be in blocking mode.
                    r.setblocking(True)

                r.start_time = time.time()
                self.active_queue.put(r)
                
                if poll:
                    poll.unregister(r)
                    del poll_dict[r.fileno()]
                    
                self.connections.remove(r)
                list_changed = True

            # If we have any stale connections, kill them off.
            if self.timeout:
                now = time.time()
                stale = set()
                for c in self.connections:
                    if (now - c.start_time) >= self.timeout:
                        stale.add(c)

                for c in stale:
                    if __debug__:
                        # "EXPR and A or B" kept for Py2.4 compatibility
                        data = (c.client_addr, c.server_port, c.ssl and '*' or '')
                        self.log.debug('Flushing stale connection: %s:%i%s' % data)
                    if poll:
                        poll.unregister(r)
                        del poll_dict[r.fileno()]

                    self.connections.remove(c)
                    list_changed = True
                    
                    try:
                        c.close()
                    finally:
                        del c

    def stop(self):
        self.active = False

        if __debug__:
            self.log.debug('Flushing waiting connections')
            
        for c in self.connections:
            try:
                c.close()
            finally:
                del c<commit_msg>Removed (e)poll support in Monitor.  It exposes some weird (read buggy) behavior for not much of a speed gain.<commit_after># -*- coding: utf-8 -*-

# This file is part of the Rocket Web Server
# Copyright (c) 2009 <NAME>

# Import System Modules
import time
import logging
import select
from threading import Thread

# Import Package Modules
from . import IS_JYTHON, THREAD_STOP_CHECK_INTERVAL, NullHandler

class Monitor(Thread):
    # Monitor worker class.

    def __init__(self,
                 monitor_queue,
                 active_queue,
                 timeout,
                 *args,
                 **kwargs):

        Thread.__init__(self, *args, **kwargs)

        # Instance Variables
        self.monitor_queue = monitor_queue
        self.active_queue = active_queue
        self.timeout = timeout

        self.connections = set()
        self.active = False

    def run(self):
        self.name = self.getName()
        self.log = logging.getLogger('Rocket.Monitor')
        self.log.addHandler(NullHandler())

        self.active = True
        conn_list = list()
        list_changed = False

        if __debug__:
            self.log.debug('Entering monitor loop.')

        # Enter thread main loop
        while self.active:
            # Move the queued connections to the selection pool
            while not self.monitor_queue.empty() or not len(self.connections):
                if __debug__:
                    self.log.debug('In "receive timed-out connections" loop.')

                c = self.monitor_queue.get()

                if c is None:
                    # A non-client is a signal to die
                    if __debug__:
                        self.log.debug('Received a death threat.')
                    return

                self.log.debug('Received a timed out connection.')

                if __debug__:
                    assert(c not in self.connections)

                if IS_JYTHON:
                    # Jython requires a socket to be in Non-blocking mode in
                    # order to select on it.
                    c.setblocking(False)

                if __debug__:
                    self.log.debug('Adding connection to monitor list.')

                self.connections.add(c)
                list_changed = True

            # Wait on those connections
            self.log.debug('Blocking on connections')
            if list_changed:
                conn_list = list(self.connections)
                list_changed = False

            readable = select.select(conn_list,
                                     [],
                                     [],
                                     THREAD_STOP_CHECK_INTERVAL)[0]

            # If we have any readable connections, put them back
            for r in readable:
                if __debug__:
                    self.log.debug('Restoring readable connection')

                if IS_JYTHON:
                    # Jython requires a socket to be in Non-blocking mode in
                    # order to select on it, but the rest of the code requires
                    # that it be in blocking mode.
                    r.setblocking(True)

                r.start_time = time.time()
                self.active_queue.put(r)
                
                self.connections.remove(r)
                list_changed = True

            # If we have any stale connections, kill them off.
            if self.timeout:
                now = time.time()
                stale = set()
                for c in self.connections:
                    if (now - c.start_time) >= self.timeout:
                        stale.add(c)

                for c in stale:
                    if __debug__:
                        # "EXPR and A or B" kept for Py2.4 compatibility
                        data = (c.client_addr, c.server_port, c.ssl and '*' or '')
                        self.log.debug('Flushing stale connection: %s:%i%s' % data)

                    self.connections.remove(c)
                    list_changed = True
                    
                    try:
                        c.close()
                    finally:
                        del c

    def stop(self):
        self.active = False

        if __debug__:
            self.log.debug('Flushing waiting connections')
            
        for c in self.connections:
            try:
                c.close()
            finally:
                del c
{'fraction_non_alphanumeric': 0.07285252061371464, 'fraction_numerical': 0.0017743450579271475, 'mean_word_length': 1.1807009558488848, 'pattern_counts': {'":': 0, '<': 5, '<?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': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3662834', 'n_tokens_mistral': 2474, 'n_tokens_neox': 2252, 'n_words': 698}