3

我有一个相对简单的 python 多处理脚本，它设置了一个工作池，dataframe通过自定义管理器将输出附加到熊猫。我发现当我在池上调用 close()/join() 时，并不是所有由 apply_async 提交的任务都已完成。

这是一个简化的示例，它提交了 1000 个作业，但只完成了一半，导致断言错误。我是否忽略了一些非常简单的事情，或者这可能是一个错误？

from pandas import DataFrame
from multiprocessing.managers import BaseManager, Pool

class DataFrameResults:
    def __init__(self):
        self.results = DataFrame(columns=("A", "B")) 

    def get_count(self):
        return self.results["A"].count()

    def register_result(self, a, b):
        self.results = self.results.append([{"A": a, "B": b}], ignore_index=True)

class MyManager(BaseManager): pass

MyManager.register('DataFrameResults', DataFrameResults)

def f1(results, a, b):
    results.register_result(a, b)

def main():
    manager = MyManager()
    manager.start()
    results = manager.DataFrameResults()

    pool = Pool(processes=4)

    for (i) in range(0, 1000):
        pool.apply_async(f1, [results, i, i*i])
    pool.close()
    pool.join()

    print results.get_count()
    assert results.get_count() == 1000

if __name__ == "__main__":
    main()
4

1 回答 1

3

[编辑]你看到的问题是因为这段代码：

self.results = self.results.append(...)

这不是原子的。所以在某些情况下，线程会在读取之后self.results（或在追加时）但在它可以将新帧分配给之前被中断self.results-> 这个实例将丢失。

正确的解决方案是等待使用结果对象来获取结果，然后将它们全部附加到主线程中。

于 2012-08-15T14:26:08.017 回答