如何为并发编写进度计数器/时间

Question

我有一个非常简单的Clock类，它将记录时间、任务数量，并允许代码的其他部分调用它的静态方法Clock.increment()，以便用户可以获得进度的反馈。但是，多进程为每个进程创建单独的副本，即使我在主进程中初始化类，子进程也无法访问它。这是我的Clock类:

class Counter(object):
    '''
    This counter needs to be initiated
    '''
    startTime = time.time()
    currentProgress = 0
    def __init__(self, totalTask):
        # self.startTime = time.time()
        Counter.totalTask = totalTask
        print("Counter initiated")
    def increment():
        Counter.currentProgress += 1
        Counter.expectedTime = ((time.time() - Counter.startTime) / Counter.currentProgress)*(Counter.totalTask - Counter.currentProgress)
        print("Progress: "+str(Counter.currentProgress)+" / "+ str(Counter.totalTask) + " : " + str(float(Counter.currentProgress) / float(Counter.totalTask)*100)+"%")
        print("Expected finish in: " + str(Counter.expectedTime/3600.0) + " hrs")
    increment = staticmethod(increment)

我这样称呼它:

if __name__ == "__main__":
    example = [1,2,3,4,5]
    counter = Counter(len(example))
    p = Pool(processes=2)
    p.map(conprintNum, example)
def conprintNum(num):
    print(num)
    Counter.increment()

这行不通。我不想将总任务号手工编码到Clock类，但这可能是最后的手段(违反所有良好的编程实践)。无论如何，我可以使这个工作在并发?我也尝试过单例模式，但没有，它并不能真正解决问题。

Answer 1

您可以使用multiprocessing.BaseManager在所有子进程之间共享Counter实例:

import time
from functools import partial
from multiprocessing import Pool
from multiprocessing.managers import BaseManager
if __name__ == "__main__":
    example = [1,2,3,4,5]
    # Create our custom manager, register the Counter object with it,
    # start it, and then create our shared Counter instance.
    m = BaseManager()
    m.register('Counter', Counter)
    m.start()
    counter = m.Counter(len(example))
    p = Pool(processes=2)
    # We create a partial so that it's easier to pass the counter instance
    # along with every value in our example iterable.
    func = partial(conprintNum, counter)
    p.map(func, example)

输出:

Counter initiated
1
Progress: 1 / 5 : 20.0%
Expected finish in: 4.0926668379e-05 hrs
2
Progress: 2 / 5 : 40.0%
Expected finish in: 1.61524613698e-05 hrs
3
Progress: 3 / 5 : 60.0%
Expected finish in: 7.86887274848e-06 hrs
4
Progress: 4 / 5 : 80.0%
Expected finish in: 3.15326783392e-06 hrs
5
Progress: 5 / 5 : 100.0%
Expected finish in: 0.0 hrs

编辑:

正如在评论中指出的那样，这里有一个竞争条件，其中多个进程可以同时进入increment方法，导致输出不按照您希望的方式显示。如果我们在increment:

中添加sleep调用，可以更清楚地看到这一点

def increment():
    Counter.currentProgress += 1
    time.sleep(random.randint(1,6)) # Artificially delay execution.
    Counter.expectedTime = ((time.time() - Counter.startTime) / Counter.currentProgress)*(Counter.totalTask - Counter.currentProgress)
    print("Progress: "+str(Counter.currentProgress)+" / "+ str(Counter.totalTask) + " : " + str(float(Counter.currentProgress) / float(Counter.totalTask)*100)+"%")
    print("Expected finish in: " + str(Counter.expectedTime/3600.0) + " hrs")

现在，你得到这样的输出:

Counter initiated
1
2
3
4
Progress: 4 / 5 : 80.0%
Expected finish in: 7.12237589889e-05 hrs
Progress: 4 / 5 : 80.0%
Expected finish in: 7.15903441111e-05 hrs
5
Progress: 5 / 5 : 100.0%
Expected finish in: 0.0 hrs
Progress: 5 / 5 : 100.0%
Expected finish in: 0.0 hrs
Progress: 5 / 5 : 100.0%
Expected finish in: 0.0 hrs

显然这不好。这很容易避免，但是，通过在Lock块中调用增量:

def conprintNum(counter, lock, num):
    print(num)
    with lock:
        counter.increment()
if __name__ == "__main__":
    example = [1,2,3,4,5]
    m = SyncManager()  # SyncManager, rather than BaseManager
    m.register('Counter', Counter)
    m.start()
    lock = m.Lock()  # SyncManager comes with a shared Lock implementation.
    counter = m.Counter(len(example))
    p = Pool(processes=4)
    func = partial(conprintNum, counter, lock)
    p.map(func, example)