使用 PyCUDA 进行 Python 多处理

Question

我有一个问题，想要在多个 CUDA 设备上拆分，但我怀疑我当前的系统架构阻碍了我；

我设置的是一个 GPU 类，其中包含在 GPU 上执行操作的函数（这很奇怪）。这些操作是我想象的风格

for iteration in range(maxval):
    result[iteration]=gpuinstance.gpufunction(arguments,iteration)

，N 个设备有 N 个 gpu 实例，但我对多重处理了解不够，无法看到应用它的最简单方法，以便每个设备都是异步分配的，奇怪的是很少有我遇到的例子具体演示了处理后的整理结果。

谁能给我在这方面的任何指示？

更新 感谢 Kaloyan 在多处理领域的指导；如果 CUDA 不是具体的症结所在，我会将您标记为已回答。对不起。

之前在使用此实现时，gpuinstance 类使用 import pycuda.autoinit 启动了 CUDA 设备，但这似乎不起作用，每次都会抛出无效上下文错误（范围正确）线程遇到了 cuda 命令。然后，我尝试在类的 __init__ 构造函数中手动初始化...

pycuda.driver.init()
self.mydev=pycuda.driver.Device(devid) #this is passed at instantiation of class
self.ctx=self.mydev.make_context()
self.ctx.push()    

我的假设是，在创建 gpu 实例列表和线程使用它们时，上下文被保留，因此每个设备都是在它自己的背景下坐得很好。

（我还实现了一个析构函数来处理pop/detach清理）

问题是，一旦线程尝试接触CUDA，无效上下文异常仍然会出现。

大家有什么想法吗？感谢走到这一步。自动为那些在答案中使用“香蕉”的人投票！ ：P

Answer 1

你需要首先在 CUDA 方面把所有的香蕉都排好，然后考虑用 Python 完成这件事的最佳方法 [我知道无耻的代表嫖娼]。

CUDA 多 GPU 模型在 4.0 之前非常简单 - 每个 GPU 都有自己的上下文，并且每个上下文必须由不同的主机线程建立。所以伪代码的想法是：

1. 应用程序启动，进程使用 API 来确定可用 GPU 的数量（注意 Linux 中的计算模式等）
2. 应用程序为每个 GPU 启动一个新的主机线程，并传递 GPU id。每个线程隐式/显式调用相当于 cuCtxCreate() 的函数，传递已为其分配的 GPU id
3. 利润！

在 Python 中，这可能看起来像这样：

import threading
from pycuda import driver

class gpuThread(threading.Thread):
    def __init__(self, gpuid):
        threading.Thread.__init__(self)
        self.ctx  = driver.Device(gpuid).make_context()
        self.device = self.ctx.get_device()

    def run(self):
        print "%s has device %s, api version %s"  \
             % (self.getName(), self.device.name(), self.ctx.get_api_version())
        # Profit!

    def join(self):
        self.ctx.detach()
        threading.Thread.join(self)

driver.init()
ngpus = driver.Device.count()
for i in range(ngpus):
    t = gpuThread(i)
    t.start()
    t.join()

这假设在不事先检查设备的情况下建立上下文是安全的。理想情况下，您将检查计算模式以确保尝试安全，然后在设备繁忙时使用异常处理程序。但希望这给出了基本的想法。

Answer 2

您需要的是 map 内置函数的多线程实现。 这里是一种实现。通过稍加修改来满足您的特定需求，您将得到：

import threading

def cuda_map(args_list, gpu_instances):

    result = [None] * len(args_list)

    def task_wrapper(gpu_instance, task_indices):
        for i in task_indices:
            result[i] = gpu_instance.gpufunction(args_list[i])

    threads = [threading.Thread(
                    target=task_wrapper, 
                    args=(gpu_i, list(xrange(len(args_list)))[i::len(gpu_instances)])
              ) for i, gpu_i in enumerate(gpu_instances)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    return result

它与上面的内容或多或少相同，最大的区别在于您不必花时间等待 的每次完成GPU函数。