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

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

# Import System Modules
import sys
import time
import socket
import logging
from threading import Lock
try:
    from queue import Queue
except ImportError:
    from Queue import Queue
# Import Package Modules
from . import DEFAULTS, SERVER_SOFTWARE, NullHandler
from .worker import get_method

# Setup Logging
log = logging.getLogger('Rocket.Errors.ThreadPool')
log.addHandler(NullHandler())

class ThreadPool:
    """The ThreadPool class is a container class for all the worker threads. It
    manages the number of actively running threads."""
    queue = None
    threads = set()

    def __init__(self,
                 method,
                 min_threads=DEFAULTS['MIN_THREADS'],
                 max_threads=DEFAULTS['MAX_THREADS'],
                 app_info=None,
                 server_software=SERVER_SOFTWARE,
                 timeout_queue=None):

        log.debug("Initializing ThreadPool.")
        self.check_for_dead_threads = 0
        self.resize_lock = Lock()
        self.queue = Queue()

        self.worker_class = W = get_method(method)
        self.min_threads = min_threads
        self.max_threads = max_threads
        self.timeout_queue = timeout_queue
        self.stop_server = False
        self.grow_threshold = int(max_threads/10) + 2

        if not isinstance(app_info, dict):
            app_info = dict()

        app_info.update(max_threads=max_threads,
                        min_threads=min_threads)

        W.app_info = app_info
        W.pool = self
        W.server_software = server_software
        W.queue = self.queue
        W.wait_queue = self.timeout_queue
        W.timeout = max_threads * 0.2 if max_threads != 0 else 2

        self.threads = set([self.worker_class() for x in range(min_threads)])

    def start(self):
        self.stop_server = False
        log.debug("Starting threads.")
        for thread in self.threads:
            thread.daemon = True
            thread._pool = self
            thread.start()

    def stop(self):
        log.debug("Stopping threads.")
        self.stop_server = True

        # Prompt the threads to die
        for t in self.threads:
            self.queue.put(None)

        # Give them the gun
        for t in self.threads:
            t.kill()

        # Wait until they pull the trigger
        for t in self.threads:
            t.join()

        # Clean up the mess
        self.resize_lock.acquire()
        self.bring_out_your_dead()
        self.resize_lock.release()

    def bring_out_your_dead(self):
        # Remove dead threads from the pool
        # Assumes resize_lock is acquired from calling thread

        dead_threads = [t for t in self.threads if not t.is_alive()]
        for t in dead_threads:
            log.debug("Removing dead thread: %s." % t.getName())
            self.threads.remove(t)
        self.check_for_dead_threads -= len(dead_threads)

    def grow(self, amount=None):
        # Assumes resize_lock is acquired from calling thread
        if self.stop_server:
            return

        if not amount:
            amount = self.max_threads

        amount = min([amount, self.max_threads - len(self.threads)])
<commit_msg>Fix to use Python 2.5 threading API.<commit_after># -*- coding: utf-8 -*-

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

# Import System Modules
import sys
import time
import socket
import logging
from threading import Lock
try:
    from queue import Queue
except ImportError:
    from Queue import Queue
# Import Package Modules
from . import DEFAULTS, SERVER_SOFTWARE, NullHandler
from .worker import get_method

# Setup Logging
log = logging.getLogger('Rocket.Errors.ThreadPool')
log.addHandler(NullHandler())

class ThreadPool:
    """The ThreadPool class is a container class for all the worker threads. It
    manages the number of actively running threads."""
    queue = None
    threads = set()

    def __init__(self,
                 method,
                 min_threads=DEFAULTS['MIN_THREADS'],
                 max_threads=DEFAULTS['MAX_THREADS'],
                 app_info=None,
                 server_software=SERVER_SOFTWARE,
                 timeout_queue=None):

        log.debug("Initializing ThreadPool.")
        self.check_for_dead_threads = 0
        self.resize_lock = Lock()
        self.queue = Queue()

        self.worker_class = W = get_method(method)
        self.min_threads = min_threads
        self.max_threads = max_threads
        self.timeout_queue = timeout_queue
        self.stop_server = False
        self.grow_threshold = int(max_threads/10) + 2

        if not isinstance(app_info, dict):
            app_info = dict()

        app_info.update(max_threads=max_threads,
                        min_threads=min_threads)

        W.app_info = app_info
        W.pool = self
        W.server_software = server_software
        W.queue = self.queue
        W.wait_queue = self.timeout_queue
        W.timeout = max_threads * 0.2 if max_threads != 0 else 2

        self.threads = set([self.worker_class() for x in range(min_threads)])

    def start(self):
        self.stop_server = False
        log.debug("Starting threads.")
        for thread in self.threads:
            thread.daemon = True
            thread._pool = self
            thread.start()

    def stop(self):
        log.debug("Stopping threads.")
        self.stop_server = True

        # Prompt the threads to die
        for t in self.threads:
            self.queue.put(None)

        # Give them the gun
        for t in self.threads:
            t.kill()

        # Wait until they pull the trigger
        for t in self.threads:
            t.join()

        # Clean up the mess
        self.resize_lock.acquire()
        self.bring_out_your_dead()
        self.resize_lock.release()

    def bring_out_your_dead(self):
        # Remove dead threads from the pool
        # Assumes resize_lock is acquired from calling thread

        dead_threads = [t for t in self.threads if not t.isAlive()]
        for t in dead_threads:
            log.debug("Removing dead thread: %s." % t.getName())
            self.threads.remove(t)
        self.check_for_dead_threads -= len(dead_threads)

    def grow(self, amount=None):
        # Assumes resize_lock is acquired from calling thread
        if self.stop_server:
            return

        if not amount:
            amount = self.max_threads

        amount = min([amount, self.max_threads - len(self.threads)])

{'fraction_non_alphanumeric': 0.08227079538554949, 'fraction_numerical': 0.004250151791135397, 'mean_word_length': 2.1153664302600474, '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': '618158', 'n_tokens_mistral': 1888, 'n_tokens_neox': 1822, 'n_words': 532}