2

我目前正在使用 networkx 函数 *all_simple_paths* 为给定的一组源节点和目标节点查找网络 G 中的所有路径。

在更大/更密集的网络上，这个过程非常密集。

我想知道是否可以在这个问题上使用多处理，以及是否有人对如何通过创建池等来实现它有任何想法。

import networkx as nx

G = nx.complete_graph(8)
sources = [1,2]
targets = [5,6,7]

for target in targets:
    for source in sources:
        for path in nx.all_simple_paths(G, source=source, target=target, cutoff=None):
            print(path)

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

4

2 回答 2

3

对于最简单的情况，您的路径似乎彼此没有关系，除了是同一个图表的一部分，因此不会有任何锁定问题。

我要做的是您可以使用该模块通过using a和该方法multiprocessing在每个循环上启动一个新进程。targetsPoolmap

def create_graph_from_target( target )
    for source in sources:
        for path in nx.all_simple_paths(G, source=source, target=target, cutoff=None):
            print(path)

from multiprocessing import Pool
p = Pool( processes=4 )

p.map( create_graph_from_target, targets )
p.close()
p.join()
于 2012-12-21T16:39:48.037 回答
2

这是一个使用工作进程集合的版本。每个工作人员source, target从队列中获取对，并在列表中收集路径。找到所有路径后，将结果放入输出队列，并由主进程进行整理。

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()))

test_workers使用多处理，test_single_worker使用单个进程。

运行test.py不会引发 AssertionError，因此看起来两个函数都返回相同的结果（至少对于我运行的有限测试而言）。

以下是 timeit 结果：

% python -mtimeit -s'import test as t' 't.test_workers()'
10 loops, best of 3: 6.71 sec per loop

% python -mtimeit -s'import test as t' 't.test_single_worker()'
10 loops, best of 3: 12.2 sec per loop

因此，在这种情况下，test_workers 能够在 2 核系统上实现比 test_single_worker 快 1.8 倍的速度。希望代码也能很好地解决您的实际问题。我很想知道结果。

一些兴趣点：

  • 调用pool.apply_async一个短暂的函数非常慢，因为太多时间花在传递参数上，并通过队列而不是使用 CPU 进行有用的计算。
  • 最好将结果收集到一个列表中并将完整的结果放入output队列中，而不是一次放入output一个结果。放入队列中的每个对象都被腌制，腌制一个大列表比腌制许多小列表更快。
  • 我认为仅从一个进程进行打印更安全，因此打印语句不会相互影响（导致输出错误）。
于 2012-12-21T17:55:57.457 回答