使用 openmp 并行化 python 中的“for 循环”迭代

Question

我想使用 OpenMP 线程或 python 中的类似技术并行化“for 循环”迭代。 代码如下所示。 “idexs”迭代 1024 次，它所做的只是选择一个索引 (i) 并在 self._storage[i] 处进行数组访问并将所有信息存储在data.

我可以使用 Python 中的 OpenMP 技术来加速此操作吗？

代码：

 def _encode_sample(self, idxes):
        obses_t, actions, rewards, obses_tp1, dones = [], [], [], [], []
        for i in idxes:
            data = self._storage[i]
            obs_t, action, reward, obs_tp1, done = data
            obses_t.append(np.array(obs_t, copy=False))
            actions.append(np.array(action, copy=False))
            rewards.append(reward)
            obses_tp1.append(np.array(obs_tp1, copy=False))
            dones.append(done)
        return np.array(obses_t), np.array(actions), np.array(rewards), np.array(obses_tp1), np.array(dones)

我希望 for 循环 像这样执行：

Answer 1

我用 Ray 回答，因为你允许“OpenMP 线程或类似线程”。

Ray 使并行化 python 变得非常容易——你可以用下面的代码做你想做的事（假设你的机器上有四个内核）：

import ray

def _encode_sample(self, idxes):
    split_idxes = np.array_split(idxes, 4)
    for subrange in split_idxes:
        futures.append(_encode_sample_helper.remote(self.storage, subrange))

    obses_t, actions, rewards, obses_tp1, dones = [], [], [], [], []
    outputs = ray.get(futures)

    for a, b, c, d, e in outputs:
        obses_t.extend(a)
        actions.extend(b)
        rewards.extend(c)
        obses_tp1.extend(d)
        dones.extend(e)

    return np.array(obses_t), np.array(actions), np.array(rewards), np.array(obses_tp1), np.array(dones)


@ray.remote(num_cpus=1)
def _encode_sample_helper(storage, subrange):
    obses_t, actions, rewards, obses_tp1, dones = [], [], [], [], []
    for i in subrange:
        data = storage[i]
        obs_t, action, reward, obs_tp1, done = data
        obses_t.append(np.array(obs_t, copy=False))
        actions.append(np.array(action, copy=False))
        rewards.append(reward)
        obses_tp1.append(np.array(obs_tp1, copy=False))
        dones.append(done)
    return np.array(obses_t), np.array(actions), np.array(rewards), np.array(obses_tp1), np.array(dones)

Answer 2

与其使用自定义模块，不如使用标准模块。更少的开销。使用线程模块。

下面包含前 250 个项目的逻辑。为每组项目更改此逻辑。第一个参数是线程函数，第二个参数是参数列表。

import threading

def thread_logic(start, end):
    // logic

thread1 = threading.Thread(target=thread_logic, args=(1,250,))
thread1.start()

在性能方面你不会在这里获得太多，因为创建新线程时有很多开销。还要小心将 Python 中的类添加到源代码中。由于开销，这也会降低性能。总有一天线程会超过执行开销，您将受益于线程化。换句话说，您可能需要更多的数据才能看到好处。我建议记录每个函数的性能 (cProfiling) 以查看何时发生。运行带线程和不带线程的 cProfile，你会看到它是否值得运行它。

使用此链接中的方法3。最后的报告会告诉你一个函数被调用了多少次和性能信息（即：执行该函数的平均时间）

https://www.geeksforgeeks.org/profiling-in-python/

在 Python 中，他们以不同的方式使用术语“并行”和“并发”（当大多数人将这些术语视为同义词时），因此在 Python 的上下文中，您正在并发运行这些线程，而不是并行运行。所以我不知道最大并发线程数是可能的。

事实上，Python 进程不能并行运行线程，但它可以 在 I/O 绑定期间通过上下文切换同时运行它们 操作。这个限制实际上是由 GIL 强制执行的。巨蟒 全局解释器锁 (GIL) 防止同一进程内的线程 同时执行。