3

我最近发布了一个问题Using multiprocessing for find network paths，很高兴@unutbu 提供了一个简洁的解决方案

然而，我在执行test_workers()（利用多处理）功能时遇到了困难。N代码运行，但在我的网络中有大量节点挂起G

使用 Mac OS X Lion 10.7.5 -- python 2.7 运行，当 N>500 时挂起。logging 带来以下消息，之后它会挂起

[DEBUG/MainProcess] doing self._thread.start()
[DEBUG/MainProcess] starting thread to feed data to pipe
[DEBUG/MainProcess] ... done self._thread.start()

通过 VMware fusion 在 Windows 7 上运行有助于更大的网络，但最终会在 N> 20,000 个节点周围出现图表（理想情况下，我希望在 N = 500,000 的网络上使用它）。悬挂点来自窗户一侧的消息：

[DEBUG/MainProcess] starting thread to feed data to pipe
[DEBUG/MainProcess] ... done self._thread.start()[DEBUG/MainProcess] telling queue thread to quit
Traceback (most recent call last):
      File "C:\Users\Scott\Desktop\fp_test.py", line 75, in <module>
    Traceback (most recent call last):
          File "C:\Python27\lib\multiprocessing\queues.py", line 264, in _feed
    test_workers()
    MemoryError

我想知道是否有人对为什么会发生这种情况有任何想法？如果有任何建议如何让它适用于更大的网络？

非常感谢您提出的任何建议。

@unutbu 的代码：

import networkx as nx
import multiprocessing as mp
import random
import sys
import itertools as IT
import logging
logger = mp.log_to_stderr(logging.DEBUG)


def worker(inqueue, output):
    result = []
    count = 0
    for pair in iter(inqueue.get, sentinel):
        source, target = pair
        for path in nx.all_simple_paths(G, source = source, target = target,
                                        cutoff = None):
            result.append(path)
            count += 1
            if count % 10 == 0:
                logger.info('{c}'.format(c = count))
    output.put(result)

def test_workers():
    result = []
    inqueue = mp.Queue()
    for source, target in IT.product(sources, targets):
        inqueue.put((source, target))
    procs = [mp.Process(target = worker, args = (inqueue, output))
             for i in range(mp.cpu_count())]
    for proc in procs:
        proc.daemon = True
        proc.start()
    for proc in procs:    
        inqueue.put(sentinel)
    for proc in procs:
        result.extend(output.get())
    for proc in procs:
        proc.join()
    return result

def test_single_worker():
    result = []
    count = 0
    for source, target in IT.product(sources, targets):
        for path in nx.all_simple_paths(G, source = source, target = target,
                                        cutoff = None):
            result.append(path)
            count += 1
            if count % 10 == 0:
                logger.info('{c}'.format(c = count))

    return result

sentinel = None

seed = 1
m = 1
N = 1340//m
G = nx.gnm_random_graph(N, int(1.7*N), seed)
random.seed(seed)
sources = [random.randrange(N) for i in range(340//m)]
targets = [random.randrange(N) for i in range(1000//m)]
output = mp.Queue()

if __name__ == '__main__':
    test_workers()
    # test_single_worker()
    # assert set(map(tuple, test_workers())) == set(map(tuple, test_single_worker()))
4

1 回答 1

2

您遇到了logging模块的死锁。

该模块保留了一些线程锁以允许跨线程进行安全日志记录，但是当当前进程被分叉时它不能很好地发挥作用。例如，请参阅此处以了解正在发生的事情。

解决方案是删除logging调用或使用 plain prints 代替。

无论如何，作为一般规则，避免使用线程+分叉。并始终检查哪些模块在幕后使用线程。

请注意，在 windows 上它可以正常工作，因为 windows 没有fork，因此不存在锁定克隆和后续死锁的问题。在这种情况下，MemoryError表明该进程正在消耗过多的 RAM。您可能不得不重新考虑算法以使用更少的 RAM，但这与您在 OSX 上遇到的问题完全不同

于 2012-12-30T15:23:24.487 回答