当集合 {B} 的任何线程正在运行时阻塞一个线程 R

Question

我的应用程序有一个运行时间长、CPU 消耗量大的线程（下面是 Runner 类的示例），而线程池需要快速构建一个页面然后返回它（例如 Builder 类）。 Runner 正在进行后台媒体编码，对时间不敏感。 Builder.build_stuff() 运行速度很快并且对时间敏感，所以如果Builder 试图运行，我想阻止Runner 开始任何新工作。

我最初使用单个threading.Lock() 来阻止Runner.do_work，但这限制了多个Builders 一次运行。我在下面做了一个简化的例子来展示我想出的策略；但我不是 100% 确定这是最好的解决方案，或者没有现有的结构可以做到这一点。

import threading
runner = Runner()

"""Thread that is always running"""
def run():
    while True:
        runner.do_work()
running_work = threading.Thread(target=run).start()

"""Builders are created for a short duration by an outside pool of threads"""
class Builder:
    def __del__(self):
        runner.outside_release()

    def __init__(self):
        runner.outside_acquire()
        self.build_stuff()

    def build_stuff(self):
        """Do build some stuff here"""
        return """thing that was built"""

class Runner:
    def __init__(self):
        self.building_flag = 0
        self.building_lock = threading.Condition(threading.Lock())

    def outside_acquire(self):
        with runner.building_lock:
            runner.building_flag += 1
        self.build_stuff()

    def outside_release(self):
        with runner.building_lock:
            runner.building_flag -= 1
            runner.building_lock.notify()

    def do_work(self):
        with self.building_lock:
            while self.building_flag:
                self.building_lock.wait()
        """Do some work here"""

Answer 1

首先，您已经提到了这一点，但我会为其他阅读此内容的人再说一遍：中断长时间运行的线程中的进程并重新启动它比将长时间运行的进程分成块并检查要困难得多如果你应该经常停下来。

我发现您的示例有点令人困惑且不完整，因此我制作了一个单独的示例来演示如何解决所描述的问题。我使用了一个简单的列表，使用threading.Lock 保护来跟踪任何正在运行的构建器，并使用threading.Event 来通知跑步者何时停止和恢复工作。最重要的部分是构建器的 run 方法中的“工作”之前和之后重新开始。

from threading import Thread, Lock, Event
from time import sleep

class Builder(Thread):

    running = [] #list to keep track of any running builders
    running_lock = Lock() #mutex for running builder list

    def __init__(self, work, can_work_event):
        super().__init__()
        self.work = work
        self.runner_can_work = can_work_event


    def run(self):
        #before we do our work, add ourselves to running list and tell runner not to work
        self.runner_can_work.clear() #runner cannot start new work now
        with Builder.running_lock: #append and remove are not likely thread safe
            Builder.running.append(self) #add self to running builder list

        print(f'builder doing {self.work}')
        sleep(1)

        #this is not robust against builders crashing. Perhaps a better solution would
        #  keep track of thread id's, and periodically clean out id's of crashed threads
        #  from the Builder.running list

        #del isn't a reliable way ot determine when a task is done. Do this at the end
        #  of the work you intend to perform.

        with Builder.running_lock: #when builder is done with work
            Builder.running.remove(self) #remove self from list
            if not Builder.running: #no more builders are in the list
                self.runner_can_work.set() #allow runner to begin new work

class Runner(Thread):

    def __init__(self, work_to_do, can_work_event):
        super().__init__()
        self.work = work_to_do
        self.can_work = can_work_event

    def run(self):
        for chunk in self.work:
            self.can_work.wait() #wait on any running Builders
            print(f'runner working on {chunk}')
            sleep(2) #heavy computation

#example demonstration

runner_can_work = Event() #event used to notify runner of ability to work
runner_can_work.set() #set event to initially true (default is false)

r = Runner(range(10), runner_can_work) # range = dummy work to do
b1 = Builder('work 1', runner_can_work)
b2 = Builder('work 2', runner_can_work)
b3 = Builder('work 3', runner_can_work)
b4 = Builder('work 4', runner_can_work)
b5 = Builder('work 5', runner_can_work)

r.start()
sleep(3)
b1.start()
sleep(4)
b2.start()
b3.start()
sleep(3)
b4.start()
b5.start()

for t in (r,b1,b2,b3,b4,b5): t.join()
print('done')