我想在共享内存机器上使用异步并行性并行处理时态图(本质上是^{multiprocessing
模块的Pool.apply_async()
。时间图由5个单元(快照)图组成。对于每个单位图,我执行多个计算代价高昂的矩阵运算
首先考虑一个简单的顺序例子:
#------------------------------------
# Constants
#------------------------------------
NV = 100 # No. of vertices
NE = 25 # No. of edges
NG = 5 # No. of unit graphs
#------------------------------------
# Generate random time-varying graph
#------------------------------------
Gt = gen_time_graph(NV, NE, NG)
# Snapshot index
k = 0
# for each unit graph
for Gk in Gt:
# Temporal adjacency matrix
Atk = adj_mtrx(Gk)
# Temporal weight matrix
# ...
# Temporal eigenvector centrality
# ...
k += 1
它工作完美无瑕。接下来,我尝试将每个矩阵操作分配给池中的一个工作者:
#------------------------------------
# Constants
#------------------------------------
NV = 100 # No. of vertices
NE = 25 # No. of edges
NG = 5 # No. of unit graphs
NP = 2 # No. of processes
#------------------------------------
# Generate random time-varying graph
#------------------------------------
Gt = gen_time_graph(NV, NE, NG)
# Snapshot index
k = 0
if __name__ == '__main__':
with Pool(processes=NP) as pool:
# for each unit graph
for Gk in Gt:
# Temporal adjacency matrix
Atk = pool.apply_async( adj_mtrx, (Gk) ).get()
# Temporal weight matrix
# ...
# Temporal eigenvector centrality
# ...
k += 1
但是,在此代码崩溃,出现以下错误:
multiprocessing.pool.RemoteTraceback:
"""
Traceback (most recent call last):
File "/usr/lib/python3.8/multiprocessing/pool.py", line 125, in worker
result = (True, func(*args, **kwds))
TypeError: adj_mtrx() takes 1 positional argument but 100 were given
"""
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "./aggr_vs_time_dat_par_mini.py", line 100, in <module>
Atk = pool.apply_async( adj_mtrx, (Gk) ).get()
File "/usr/lib/python3.8/multiprocessing/pool.py", line 771, in get
raise self._value
TypeError: adj_mtrx() takes 1 positional argument but 100 were given
我需要帮助调试这个问题。看起来,图Gk
被Pool
分解,并作为一组顶点传递给函数。另外,如果您能对我的通用并行化方法与Pool.apply_async()
from multiprocessing
进行评论,我将不胜感激
您可以找到以下最小工作示例的所有必要代码:
import networkx as nx
import random as rnd
import numpy as np
from multiprocessing import Pool
# Generates random graph
def gen_rnd_graph(nv, ne):
# Create random list of sources
Vsrc = [rnd.randint(0,nv-1) for iter in range(ne)]
# Create random list of sinks
Vsnk = [rnd.randint(0,nv-1) for iter in range(ne)]
# Create random list of edge weights
U = [rnd.random() for iter in range(ne)]
# Create list of tuples {Vsrc, Vsnk, U}
T = list(zip(Vsrc,Vsnk,U))
# Create graph
G = nx.Graph()
# Create list of vertices
V = list(range(nv))
# Add nodes to graph
G.add_nodes_from(V)
# Add edges between random vertices with random edge weights
G.add_weighted_edges_from(T)
return G
# Generates time-varying graph
def gen_time_graph(nv, ne, ng):
# Initialise list of graphs
l = []
for i in range(ng):
gi = gen_rnd_graph(nv, ne)
l.append(gi)
return l
# Computes adjacency matrix for snaphot of time-varying graph
def adj_mtrx(Gk):
# no. of vertices
n = Gk.number_of_nodes()
# adjacency matrix
Ak = np.zeros([n,n])
# for each vertex
for i in range(n):
for j in range(n):
if Gk.has_edge(i,j): Ak[i,j] = 1
return Ak
#------------------------------------
# Constants
#------------------------------------
NV = 100 # No. of vertices
NE = 25 # No. of edges
NG = 5 # No. of unit graphs
NP = 2 # No. of processes
#------------------------------------
# Generate random time-varying graph
#------------------------------------
Gt = gen_time_graph(NV, NE, NG)
# Snapshot index
k = 0
if __name__ == '__main__':
with Pool(processes=NP) as pool:
# for each unit graph
for Gk in Gt:
# Temporal adjacency matrix
Atk = pool.apply_async( adj_mtrx, (Gk) ).get()
k += 1
从^{} 的文档中,函数的签名是
因此,您需要将
Gk
作为元组传递,即(Gk,)
:背景
您的函数将
*Gk
作为输入检索,从而生成节点列表:1和0长度元组
有关创建0和1元素元组的更多详细信息:How to create a tuple with only one element或直接在python documentation中的节中
基本上,您可以使用
()
创建长度为0
的元组,使用(Gk,)
创建长度为1的元组,对于任何数量较大的元素,您可以使用(x_1, ..., x_n)
或(x_1, ..., x_n,)
*
-运算符*
-运算符可用于使用任意数量的参数。见python documentation和section before。类似地,您可以对任意数量的关键字参数使用**
。有关更多详细信息,请查看本问题中列出的What does the star operator mean, in a function call?和重复项相关问题 更多 >
编程相关推荐