'Run the pipeline in parallel using one thread per stage. The
messages between the stages are stored in queues of the given
size.'

def run_parallel(self, queue_size=DEFAULT_QUEUE_SIZE):

    queue_count = (len(self.stages) - 1)
    queues = [CountedQueue(queue_size) for i in range(queue_count)]
    threads = []
    for coro in self.stages[0]:
        threads.append(FirstPipelineThread(coro, queues[0], threads))
    for i in range(1, queue_count):
        for coro in self.stages[i]:
            threads.append(MiddlePipelineThread(coro, queues[(i - 1)], queues[i], threads))
    for coro in self.stages[(-1)]:
        threads.append(LastPipelineThread(coro, queues[(-1)], threads))
    for thread in threads:
        thread.start()
    try:
        while threads[(-1)].is_alive():
            threads[(-1)].join(1)
    except BaseException:
        for thread in threads:
            thread.abort()
        raise
    finally:
        for thread in threads[:(-1)]:
            thread.join()
    for thread in threads:
        exc_info = thread.exc_info
        if exc_info:
            six.reraise(exc_info[0], exc_info[1], exc_info[2])
