<commit_before># This file is part of beets.
# Copyright 2010, <NAME>.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
# 
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

"""Simple but robust implementation of generator/coroutine-based
pipelines in Python. The pipelines may be run either sequentially
(single-threaded) or in parallel (one thread per pipeline stage).

This implementation supports pipeline bubbles (indications that the
processing for a certain item should abort). To use them, yield the
BUBBLE constant from any stage coroutine except the last.

In the parallel case, the implementation transparently handles thread
shutdown when the processing is complete and when a stage raises an
exception.
"""
from __future__ import with_statement # for Python 2.5
import Queue
from threading import Thread, Lock
import sys

BUBBLE = '__PIPELINE_BUBBLE__'
POISON = '__PIPELINE_POISON__'

DEFAULT_QUEUE_SIZE = 16

def clear_queue(q):
    """Safely empty a queue."""
    # This very hacky approach to clearing the queue
    # compliments of <NAME>:
    # http://www.mail-archive.com/python-list@python.org/msg95322.html
    q.mutex.acquire()
    try:
       q.queue.clear()
       q.unfinished_tasks = 0
       q.not_full.notify()
       q.all_tasks_done.notifyAll()
    finally:
       q.mutex.release()

class PipelineError(object):
    """An indication that an exception occurred in the pipeline. The
    object is passed through the pipeline to shut down all threads
    before it is raised again in the main thread.
    """
    def __init__(self, exc_info):
        self.exc_info = exc_info

class PipelineThread(Thread):
    """Abstract base class for pipeline-stage threads."""
    def __init__(self):
        super(PipelineThread, self).__init__()
        self.abort_lock = Lock()
        self.abort_flag = False

    def abort(self):
        """Shut down the thread at the next chance possible.
        """
        with self.abort_lock:
            self.abort_flag = True
            # Empty the channel before poisoning it.

class FirstPipelineThread(PipelineThread):
    """The thread running the first stage in a parallel pipeline setup.
    The coroutine should just be a generator.
    """
    def __init__(self, coro, out_queue):
        super(FirstPipelineThread, self).__init__()
        self.coro = coro
        self.out_queue = out_queue
        
        self.abort_lock = Lock()
        self.abort_flag = False
    
    def run(self):
        while True:
            # Time to abort?
            with self.abort_lock:
                if self.abort_flag:
                    return
            
            # Get the value from the generator.
            try:
                msg = self.coro.next()
            except StopIteration:
                break
            except:
                self.out_queue.put(PipelineError(sys.exc_info()))
                return
            
            # Send it to the next stage.
            self.out_queue.put(msg)
            if msg is BUBBLE:
                continue
        
        # Generator finished; shut down the pipeline.
        self.out_queue.put(POISON)
    
class MiddlePipelineThread(PipelineThread):
    """A thread running any stage in the pipeline except the first or
    last.
    """
    def __init__(self, coro, in_queue, out_queue):
        super(MiddlePipelineThread, self).__init__()
        self.coro = coro
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):
        # Prime the coroutine.
        self.coro.next()<commit_msg>fixed issue in pipeline when the first coro sends the BUBBLE message<commit_after># This file is part of beets.
# Copyright 2010, <NAME>.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
# 
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

"""Simple but robust implementation of generator/coroutine-based
pipelines in Python. The pipelines may be run either sequentially
(single-threaded) or in parallel (one thread per pipeline stage).

This implementation supports pipeline bubbles (indications that the
processing for a certain item should abort). To use them, yield the
BUBBLE constant from any stage coroutine except the last.

In the parallel case, the implementation transparently handles thread
shutdown when the processing is complete and when a stage raises an
exception.
"""
from __future__ import with_statement # for Python 2.5
import Queue
from threading import Thread, Lock
import sys

BUBBLE = '__PIPELINE_BUBBLE__'
POISON = '__PIPELINE_POISON__'

DEFAULT_QUEUE_SIZE = 16

def clear_queue(q):
    """Safely empty a queue."""
    # This very hacky approach to clearing the queue
    # compliments of <NAME>:
    # http://www.mail-archive.com/python-list@python.org/msg95322.html
    q.mutex.acquire()
    try:
       q.queue.clear()
       q.unfinished_tasks = 0
       q.not_full.notify()
       q.all_tasks_done.notifyAll()
    finally:
       q.mutex.release()

class PipelineError(object):
    """An indication that an exception occurred in the pipeline. The
    object is passed through the pipeline to shut down all threads
    before it is raised again in the main thread.
    """
    def __init__(self, exc_info):
        self.exc_info = exc_info

class PipelineThread(Thread):
    """Abstract base class for pipeline-stage threads."""
    def __init__(self):
        super(PipelineThread, self).__init__()
        self.abort_lock = Lock()
        self.abort_flag = False

    def abort(self):
        """Shut down the thread at the next chance possible.
        """
        with self.abort_lock:
            self.abort_flag = True
            # Empty the channel before poisoning it.

class FirstPipelineThread(PipelineThread):
    """The thread running the first stage in a parallel pipeline setup.
    The coroutine should just be a generator.
    """
    def __init__(self, coro, out_queue):
        super(FirstPipelineThread, self).__init__()
        self.coro = coro
        self.out_queue = out_queue
        
        self.abort_lock = Lock()
        self.abort_flag = False
    
    def run(self):
        while True:
            # Time to abort?
            with self.abort_lock:
                if self.abort_flag:
                    return
            
            # Get the value from the generator.
            try:
                msg = self.coro.next()
            except StopIteration:
                break
            except:
                self.out_queue.put(PipelineError(sys.exc_info()))
                return
            
            # Send it to the next stage.
            if msg is BUBBLE:
                continue
            self.out_queue.put(msg)

        # Generator finished; shut down the pipeline.
        self.out_queue.put(POISON)
    
class MiddlePipelineThread(PipelineThread):
    """A thread running any stage in the pipeline except the first or
    last.
    """
    def __init__(self, coro, in_queue, out_queue):
        super(MiddlePipelineThread, self).__init__()
        self.coro = coro
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):
        # Prime the coroutine.
        self.coro.next()
{'fraction_non_alphanumeric': 0.0729153836679394, 'fraction_numerical': 0.0034487005788890255, 'mean_word_length': 2.5427574171029668, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, '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': '3789789', 'n_tokens_mistral': 2203, 'n_tokens_neox': 2158, 'n_words': 828}