共享内存复杂可写数据结构

Question

我有一个在大型图形结构上运行的算法，我想使其成为多线程以获得更好的性能。我看过的方法都不太符合我想要的：我希望图表存在于所有进程都可以读取和写入的共享内存中（使用锁来防止竞争条件）。本质上，我想要像 C 语言中的 OpenMP 一样的东西，其中所有内存都可以由每个线程访问。

我首先查看了线程模块，但 GIL 意味着性能提升微不足道。

我继续尝试多处理模块，正如我在该主题上找到的大多数帖子所建议的那样（例如，如何在多个进程之间共享字典？以及python multiprocessing 中的共享内存对象）。这有两个主要问题。

首先，多重处理似乎不适用于复杂的对象。考虑以下玩具问题：我有一个整数列表，想要将它们全部乘以 10，然后以任意顺序输出所有数字。我可以使用以下代码：

def multiply_list():
    manager = Manager()
    output = manager.list()
    threads = []

    for v in range(10):
        output.append(v)
    print([str(v) for v in output])

    def process(inputs, start, end):
        while start < end:
            inputs[start] *= 10
            start += 1

    t1 = Process(target=process,
        args = (output, 0, 5))
    t2 = Process(target=process,
        args = (output, 5, 10))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    print([str(v) for v in output])

与输出：

['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
['0', '10', '20', '30', '40', '50', '60', '70', '80', '90']

但是，如果我有一个对象列表，并修改这些对象：

class Container(object):
    def __init__(self, value):
        self.value = value
    def __str__(self):
        return "C" + str(self.value)

def multiply_containers():
    manager = Manager()
    output = manager.list()
    threads = []

    for v in range(10):
        output.append(Container(v))
    print([str(v) for v in output])

    def process(inputs, start, end):
        while start < end:
            inputs[start].value *= 10
            start += 1

    t1 = Process(target=process,
        args = (output, 0, 5))
    t2 = Process(target=process,
        args = (output, 5, 10))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    print([str(v) for v in output])

没有变化。

['C0', 'C1', 'C2', 'C3', 'C4', 'C5', 'C6', 'C7', 'C8', 'C9']
['C0', 'C1', 'C2', 'C3', 'C4', 'C5', 'C6', 'C7', 'C8', 'C9']

另一个问题是，我链接的 SO 帖子建议尝试写入数据结构会复制它，这是我不想要的。

为了阐明算法本身，第一步（构建图表）的工作原理如下：我有一个句子列表，它们是单词序列。我想构建一个有向图，其中每个顶点都是一个单词，出边指向某个句子中紧随其后的每个单词。例如，如果我的输入是“戴帽子的猫”和“房子里的猫”，我的输出图将是 => cat => in => the => hat, house（即“the”有两个出边，一个到“帽子”，一个到“房子”）。我还跟踪一些辅助信息，例如每个句子或单词的常见程度。每个顶点都有一个入边和出边列表以及一些属性。

我找到了一个可能有效的模块（http://poshmodule.sourceforge.net/posh/html/），但我不确定是否有“规范”或推荐的方法来执行此类操作。

谢谢！

Answer 1

这是示例代码（有效），它使用单独的管理器进程来控制对共享数据结构的访问，并基于您的示例代码以及问题Sharing object (class instance) in python using Managers，@freakish说可能是评论中存在重复的问题——我不清楚是否是这样，但总体方法似乎可以解决您的问题。

from multiprocessing import Lock, Manager, Process
from multiprocessing.managers import BaseManager

class Container(object):
    def __init__(self, value):
        self.value = value
    def __str__(self):
        return "C" + str(self.value)
    def multiply(self, factor):  # added method
        self.value *= factor

def process(inputs, start, end):
    for i in range(start, end):
        inputs.apply(i, 'multiply', (10,))

class ListProxy(object):
    def __init__(self):
        self.nl = []
    def append(self, x):
        self.nl.append(x)
    def __getitem__(self, key):
        return self.nl[key]
    def __iter__(self):
        return iter(self.nl)
    def apply(self, i, method, args, **kwargs):
        getattr(self.nl[i], method)(*args, **kwargs)

class ListManager(BaseManager):
    pass

ListManager.register('ListProxy', ListProxy,
                     exposed=['append', '__getitem__', '__iter__', 'apply'])

def main():
    manager = ListManager()
    manager.start()
    output = manager.ListProxy()

    for v in range(10):
        output.append(Container(v))
    print([str(v) for v in output])

    t1 = Process(target=process, args=(output, 0, 5))
    t2 = Process(target=process, args=(output, 5, 10))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    print([str(v) for v in output])

if __name__ == '__main__':
    main()