python - 使用 Python 包装的 C++ 类进行并行计算 : performing operations in every element of a vector

Question

考虑 C++ 类。第一个是分支类:

class Branch{
  map<string,double> properties;
}  

即分支 对象仅由存储在 map 中的其属性来表征。每个属性都有一个名称并与一个 double 值相关联。第二个是树类，由许多分支组成:

class Tree{
  vector<*Branch> tree;

  void addBranch(int index); //adds a branch to tree vector at index
  void removeBranch(int index); //removes branch at index and and its descendents
  double getProperty(int index, string name);//gets value of property name of branch at index
  void addProperty(int index, string name, double value);
  void setProperty(int index, string name, double value);
}

现在假设树类是使用 cython 包装的。然后，在 Python 中我们可以操作 PyTree 对象，添加和删除分支以及操作每个分支的属性。考虑以下 python 程序:

tree=PyTree()
for i in range(TotalTime):
  k=random.random()
  if k>0.1:
    tree.addBranch(random_index) #it is not important how we get the index
    tree.addProperty(random_index,'prop1',1)
    tree.addProperty(random_index,'prop2',1)
  k=random.random()
  if k>0.9:
    tree.removeBranch(random_index)
  for j in range(NumberOfBranches): #it's not important how we get the number of branches
    operation1(j,'prop1') # assume this functions were defined 
    operation2(j,'prop2')

在这个程序中，我随机添加和删除分支。每个分支都有两个属性 prop1 和 prop2。有一个 operation1 函数在 'prop1' 上执行涉及 getProperty 和 setProperty 函数的操作，还有一个 operation2 函数在 'prop2' 上做同样的事情。

我想要的是让一个处理器(或线程)执行每个操作。由于程序不断调用外部 C++ 库，我应该使用 threading 而不是 multiprocessor 吗？

我应该如何实现并行化？我试着用这个来激励自己 https://www.quantstart.com/articles/parallelising-python-with-threading-and-multiprocessing ，但是当我同时使用 threading 或 multiprocessor 时，我得到一个较慢的程序...

Answer 1

我建议使用threading - 因为主要工作是在您包装的 C++ 函数中完成的，所以您应该能够释放 GIL 并让所有内容可靠地并行工作。

一个好的通用规则是尽可能少地创建新线程(这可能是一个有点慢的操作)，然后向它们提供数据直到完成。您在评论中说您根本不关心操作的执行顺序(好!)。考虑到这一点，我建议将包含操作的 lambda 函数放入 Queue 中，并让线程选择它们并运行时:

下面的代码然后进入你的 for i in range(TotalTime): 循环代替 for j in range(NumberOfBranches)::

q = Queue()

def thread_func():
    """A function for each thread to run.

    It waits to get an item off the queue and then runs that item.
    None is used to indicate that we're done. If we get None, we 
    put it pack on the Queue to ensure every thread terminates"""
    while True:
        f = q.get()
        if f is None:
            q.put(None)
            return     
        f()

# make and start the threads.
# "no_threads" should be something similar to the number of cores you have.
# 4-8 might be a good number to try?
threads = [ threading.Thread(target=thread_func) for n in range(no_threads) ]
[ t.start() for t in  threads ]

# put the required operations on the Queue
for j in (NumberOfBranches):
    # note the awkward syntax to 
    # ensure we capture j: http://stackoverflow.com/a/7514158/4657412
    q.put(lambda j=j: operation1(x,"prop1"))
    q.put(lambda j=j: operation2(x,"prop2"))

q.put(None) # to terminate

# wait for threads to finish
[ t.join() for t in threads ]

为了让线程真正并行工作，您需要确保 GIL 在您的 Cython 包装器中发布:

def operation1(a,b):
   with nogil:
      cplusplus_operation1(a,b)

---

我对使用multiprocessing有两个顾虑

1. 它不是非常便携(它在 Windows 上的工作方式不同)。
2. 如果 operation1/2 修改了 C++ 数据，那么您可能会发现修改后的数据不会在您没有特别努力的情况下跨进程共享(这会破坏执行操作的意义!)