<commit_before>from threading import Thread, Lock
from collections import deque

_task_queues = {}

class TaskQueue():

  def __init__(self):
    self.tasks = deque()
    self.running = False
    self.lock = Lock()
  
  def add_task(self, fn, args=[]):
    self.tasks.append((fn, args))
    self.lock.acquire()
    if not self.running:
      self.run()
    self.lock.release()

  def run(self):
    if self.running:
      return
    self.running = True
    def task_loop():
      while 1:
        self.lock.acquire()
        try:
          task = self.tasks.popleft()
        except IndexError:
          self.running = False
          return
        finally:
          self.lock.release()
        task[0](*task[1])
    Thread(target=task_loop).start()

  @staticmethod
  def instance(name):
    try:
      return _task_queues[name]
    except KeyError:
      _task_queues[name] = TaskQueue()
      return _task_queues[name]<commit_msg>added threadpool capabilities to task queue<commit_after>from threading import Thread, Lock
from collections import deque
import logging
import traceback

_task_queues = {}

class TaskQueue():

  def __init__(self, thread_count=1):
    self.tasks = deque()
    self.running = False
    self.lock = Lock()
    self.threads = set()
    self.thread_count = thread_count
  
  def add_task(self, fn, args=[]):
    self.tasks.append((fn, args))
    self.lock.acquire()
    self.run()
    self.lock.release()

  def run(self):
    if len(self.threads) >= self.thread_count:
      return
    thread = None
    def task_loop():
      while 1:
        self.lock.acquire()
        try:
          task = self.tasks.popleft()
        except IndexError:
          self.threads.remove(thread)
          return
        except Exception as e:
          logging.error(traceback.format_exc())
        finally:
          self.lock.release()
        task[0](*task[1])
    thread = Thread(target=task_loop)
    thread.daemon = True
    self.threads.add(thread)
    thread.start()

  @staticmethod
  def instance(name, thread_count=1):
    try:
      return _task_queues[name]
    except KeyError:
      _task_queues[name] = TaskQueue(thread_count)
      return _task_queues[name]
{'fraction_non_alphanumeric': 0.11126310989512084, 'fraction_numerical': 0.00364797081623347, 'mean_word_length': 2.471518987341772, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, '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': '7169084', 'n_tokens_mistral': 698, 'n_tokens_neox': 691, 'n_words': 157}