0_Simple__simpleCallback

▶ 学习回调函数的基本概念，并在CUDA的任务流中插入基于CPU的主机函数，作为回调函数使用。

▶ 源代码（合并了 3 个源文件，删掉了没有用到的部分）

// simpleCallback.cu
#include <stdio.h>
#include <windows.h>
#include <cuda_runtime.h>
#include "device_launch_parameters.h"
#include <helper_functions.h>
#include <helper_cuda.h>

#define N_WORKLOAD  8
#define BLOCK       512
#define ELEMENT     100000

struct CUTBarrier                       // 线程墙
{
    CRITICAL_SECTION criticalSection;   // Windows 中有关线程的结构
    HANDLE barrierEvent;
    int releaseCount;
    int count;
};

CUTBarrier thread_barrier;

struct heterogeneous_workload   // 用于分配工作的结构
{
    int id;                     // 工作编号
    int cudaDeviceID;           // 执行工作的设备号
    int *h_data;                
    int *d_data;                
    cudaStream_t stream;        // 使用的流号（一个工作使用一条流）
    bool success;               // 检查结果是否正确的标志
};

HANDLE cutStartThread(unsigned (WINAPI * func)(void *), void *data) // 创建新线程，注意函数指针的形式
{
    return CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)func, data, 0, NULL);
}

CUTBarrier cutCreateBarrier(int releaseCount)                       // 创建线程墙
{
    CUTBarrier barrier;
    InitializeCriticalSection(&barrier.criticalSection);
    barrier.barrierEvent = CreateEvent(NULL, TRUE, FALSE, TEXT("BarrierEvent"));
    barrier.count = 0;
    barrier.releaseCount = releaseCount;
    return barrier;
}

void cutIncrementBarrier(CUTBarrier *barrier)                       // 线程墙判断线程工作是否已经全部结束
{
    int myBarrierCount;
    EnterCriticalSection(&barrier->criticalSection);
    myBarrierCount = ++barrier->count;
    LeaveCriticalSection(&barrier->criticalSection);
    if (myBarrierCount >= barrier->releaseCount)                    // 发出的线程已经全部结束
        SetEvent(barrier->barrierEvent);
}

void cutWaitForBarrier(CUTBarrier *barrier)                         // 回收线程墙
{
    WaitForSingleObject(barrier->barrierEvent, INFINITE);
}

__global__ void incKernel(int *data, int N)                         // 将 data 中所有元素递增总线程个数次
{
    int idx = blockIdx.x * blockDim.x + threadIdx.x;

    if (idx < N)
        data[idx]++;
}

unsigned WINAPI postprocess(void *void_arg)
{
    heterogeneous_workload *workload = (heterogeneous_workload *)void_arg;
    cudaSetDevice(workload->cudaDeviceID);
    
    getLastCudaError("Kernel execution failed");                            // 检查GPU计算结果
    workload->success = true;
    for (int i = 0; i < N_WORKLOAD; ++i)
        workload->success &= (workload->h_data[i] == workload->id + i + 1);
   
    cudaFree(workload->d_data);
    cudaFreeHost(workload->h_data);
    cudaStreamDestroy(workload->stream);

    printf("Workload %d finished!
", workload->id);                        // 回调函数工作完成
    cutIncrementBarrier(&thread_barrier);                                   // 向线程墙发送工作完成的信号
    return 0;
}

void CUDART_CB myStreamCallback(cudaStream_t stream, cudaError_t status, void *data)// 回调函数，参数格式固定
{    
    cutStartThread(postprocess, data);                                      // 调用函数 postprocess 完成结果检查和内存释放
}

unsigned WINAPI launch(void *void_arg)
{
    heterogeneous_workload *workload = (heterogeneous_workload *)void_arg;  // 初始化工作参数  
    cudaSetDevice(workload->cudaDeviceID);                                  
    cudaStreamCreate(&workload->stream);
    cudaMalloc(&workload->d_data, ELEMENT * sizeof(int));
    cudaHostAlloc(&workload->h_data, ELEMENT * sizeof(int), cudaHostAllocPortable);
    for (int i = 0; i < ELEMENT; ++i)
        workload->h_data[i] = workload->id + i;

    // 每个 CPU 线程对应一条 CUDA 流，分别调度流的工作，可以并行，不阻塞 CPU 线程
    cudaMemcpyAsync(workload->d_data, workload->h_data, ELEMENT * sizeof(int), cudaMemcpyHostToDevice, workload->stream);
    incKernel << <(ELEMENT + BLOCK - 1) / BLOCK, BLOCK, 0, workload->stream >> > (workload->d_data, ELEMENT);
    cudaMemcpyAsync(workload->h_data, workload->d_data, ELEMENT * sizeof(int), cudaMemcpyDeviceToHost, workload->stream);
    
    cudaStreamAddCallback(workload->stream, myStreamCallback, workload, 0); // 回调函数，调用主机函数放入 CUDA 流中，在这里用于检查 GPU 结果和回收内存
    return 0;
}

int main(int argc, char **argv)
{
    printf("	Start.
");

    heterogeneous_workload *workloads = (heterogeneous_workload *)malloc(N_WORKLOAD * sizeof(heterogeneous_workload));  // 创建工作表    
    thread_barrier = cutCreateBarrier(N_WORKLOAD);                          // 创建线程墙，以便所有工作结束后回收    
    
    for (int i = 0; i < N_WORKLOAD; ++i)                                    // 分配任务
    {
        workloads[i].id = i;
        workloads[i].cudaDeviceID = 0;                                      // 将任务全部分配给 0 号设备
        cutStartThread(launch, &workloads[i]);
    }
    
    cutWaitForBarrier(&thread_barrier);                                     // 回收线程
    printf("
%d workloads all finished.
", N_WORKLOAD);    
    
    int success = 1;
    for (int i = 0; i < N_WORKLOAD; success &= workloads[i].success, ++i);  // 检查正确性
    printf("
	%s
", success ? "Correct." : "Failure.");

    free(workloads);
    getchar();
    return success;
}

● 输出结果

        Start.
Work 2 finished!
Work 3 finished!
Work 7 finished!
Work 1 finished!
Work 5 finished!
Work 4 finished!
Work 6 finished!
Work 0 finished!

8 workloads all finished.

        Correct.

▶ 涨姿势

● 回调函数的使用：首先在 cuda_runtime_api.h 中给出了能作为回调函数的主机函数格式，然后给出了回调函数的定义。回调函数需要给出流编号，回调函数指针，回调函数需要的参数，以及一个标志（不太清楚其意义，可能与回调函数是否等待流中所有其他任务是否完成后再开始有关）

#define CUDART_CB __stdcall
#define CUDARTAPI  __stdcall

// cudaStreamCallback_t 的定义
typedef void (CUDART_CB *cudaStreamCallback_t)(cudaStream_t stream, cudaError_t status, void *userData);

// cudaStreamAddCallback 的定义
extern __host__ cudaError_t CUDARTAPI cudaStreamAddCallback(cudaStream_t stream, cudaStreamCallback_t callback, void *userData, unsigned int flags);

● 有关线程创建的一些参数

//winnt.h
typedef void* HANDLE;       // HANDLE 原来就是 void*
//minwindef.h
typedef unsigned long DWORD;// DWORD 原来就是 unsigned long