gronod/headphones
1
0
Fork 0
mirror of https://github.com/rembo10/headphones.git synced 2026-07-20 16:03:59 +01:00
Code Issues Packages Projects Releases Wiki Activity

InRough update of the beets lib to 1.0b15

Browse Source
This commit is contained in:
rembo10
2012-07-28 23:45:08 +05:30
parent c1edd5085a
commit d245428ca2
17 changed files with 2786 additions and 1263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/beets/util/pipeline.py
+31 -30
View File
@@ -8,7 +8,7 @@
# 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.
@@ -30,7 +30,8 @@ up a bottleneck stage by dividing its work among multiple threads.
To do so, pass an iterable of coroutines to the Pipeline constructor
in place of any single coroutine.
"""
from __future__ import with_statement # for Python 2.5
from __future__ import print_function
import Queue
from threading import Thread, Lock
import sys
@@ -177,23 +178,23 @@ class FirstPipelineThread(PipelineThread):
self.coro = coro
self.out_queue = out_queue
self.out_queue.acquire()
self.abort_lock = Lock()
self.abort_flag = False
def run(self):
try:
while True:
with self.abort_lock:
if self.abort_flag:
return
# Get the value from the generator.
try:
msg = self.coro.next()
except StopIteration:
break
# Send messages to the next stage.
for msg in _allmsgs(msg):
with self.abort_lock:
@@ -207,7 +208,7 @@ class FirstPipelineThread(PipelineThread):
# Generator finished; shut down the pipeline.
self.out_queue.release()
class MiddlePipelineThread(PipelineThread):
"""A thread running any stage in the pipeline except the first or
last.
@@ -223,7 +224,7 @@ class MiddlePipelineThread(PipelineThread):
try:
# Prime the coroutine.
self.coro.next()
while True:
with self.abort_lock:
if self.abort_flag:
@@ -233,14 +234,14 @@ class MiddlePipelineThread(PipelineThread):
msg = self.in_queue.get()
if msg is POISON:
break
with self.abort_lock:
if self.abort_flag:
return
# Invoke the current stage.
out = self.coro.send(msg)
# Send messages to next stage.
for msg in _allmsgs(out):
with self.abort_lock:
@@ -251,7 +252,7 @@ class MiddlePipelineThread(PipelineThread):
except:
self.abort_all(sys.exc_info())
return
# Pipeline is shutting down normally.
self.out_queue.release()
@@ -273,12 +274,12 @@ class LastPipelineThread(PipelineThread):
with self.abort_lock:
if self.abort_flag:
return
# Get the message from the previous stage.
msg = self.in_queue.get()
if msg is POISON:
break
with self.abort_lock:
if self.abort_flag:
return
@@ -308,7 +309,7 @@ class Pipeline(object):
self.stages.append((stage,))
else:
self.stages.append(stage)
def run_sequential(self):
"""Run the pipeline sequentially in the current thread. The
stages are run one after the other. Only the first coroutine
@@ -319,7 +320,7 @@ class Pipeline(object):
# "Prime" the coroutines.
for coro in coros[1:]:
coro.next()
# Begin the pipeline.
for out in coros[0]:
msgs = _allmsgs(out)
@@ -329,7 +330,7 @@ class Pipeline(object):
out = coro.send(msg)
next_msgs.extend(_allmsgs(out))
msgs = next_msgs
def run_parallel(self, queue_size=DEFAULT_QUEUE_SIZE):
"""Run the pipeline in parallel using one thread per stage. The
messages between the stages are stored in queues of the given
@@ -354,11 +355,11 @@ class Pipeline(object):
threads.append(
LastPipelineThread(coro, queues[-1], threads)
)
# Start threads.
for thread in threads:
thread.start()
# Wait for termination. The final thread lasts the longest.
try:
# Using a timeout allows us to receive KeyboardInterrupt
@@ -371,7 +372,7 @@ class Pipeline(object):
for thread in threads:
thread.abort()
raise
finally:
# Make completely sure that all the threads have finished
# before we return. They should already be either finished,
@@ -388,25 +389,25 @@ class Pipeline(object):
# Smoke test.
if __name__ == '__main__':
import time
# Test a normally-terminating pipeline both in sequence and
# in parallel.
def produce():
for i in range(5):
print 'generating %i' % i
print('generating %i' % i)
time.sleep(1)
yield i
def work():
num = yield
while True:
print 'processing %i' % num
print('processing %i' % num)
time.sleep(2)
num = yield num*2
def consume():
while True:
num = yield
time.sleep(1)
print 'received %i' % num
print('received %i' % num)
ts_start = time.time()
Pipeline([produce(), work(), consume()]).run_sequential()
ts_seq = time.time()
@@ -414,21 +415,21 @@ if __name__ == '__main__':
ts_par = time.time()
Pipeline([produce(), (work(), work()), consume()]).run_parallel()
ts_end = time.time()
print 'Sequential time:', ts_seq - ts_start
print 'Parallel time:', ts_par - ts_seq
print 'Multiply-parallel time:', ts_end - ts_par
print
print('Sequential time:', ts_seq - ts_start)
print('Parallel time:', ts_par - ts_seq)
print('Multiply-parallel time:', ts_end - ts_par)
print()
# Test a pipeline that raises an exception.
def exc_produce():
for i in range(10):
print 'generating %i' % i
print('generating %i' % i)
time.sleep(1)
yield i
def exc_work():
num = yield
while True:
print 'processing %i' % num
print('processing %i' % num)
time.sleep(3)
if num == 3:
raise Exception()
@@ -438,5 +439,5 @@ if __name__ == '__main__':
num = yield
#if num == 4:
# raise Exception()
print 'received %i' % num
print('received %i' % num)
Pipeline([exc_produce(), exc_work(), exc_consume()]).run_parallel(1)