0_Simple__simpleCallback

Posted 2020-10-15 爨爨爨好

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

? 源代码：没有用到的部分被注释起来了

 1 /*multithreading.h*/
 2 #ifndef MULTITHREADING_H
 3     #define MULTITHREADING_H
 4 
 5 #include <windows.h>
 6 
 7 struct CUTBarrier                       // 线程墙
 8 {
 9     CRITICAL_SECTION criticalSection;   // Windows 中有关线程的结构
10     HANDLE barrierEvent;
11     int releaseCount;
12     int count;
13 };
14 
15     #ifdef __cplusplus
16 extern "C" {
17     #endif
18 
19 HANDLE cutStartThread(unsigned (WINAPI * function)(void *), void *data);
20 
21 //void cutEndThread(HANDLE thread);
22 
23 //void cutWaitForThreads(const HANDLE *threads, int num);
24 
25 CUTBarrier cutCreateBarrier(int releaseCount);
26 
27 void cutIncrementBarrier(CUTBarrier *barrier);
28 
29 void cutWaitForBarrier(CUTBarrier *barrier);
30 
31 //void cutDestroyBarrier(CUTBarrier *barrier);
32 
33     #ifdef __cplusplus
34 }
35     #endif
36 
37 #endif //MULTITHREADING_H

 1 /*multithreading.cpp*/
 2 #include "multithreading.h"
 3 
 4 // 创建新线程
 5 HANDLE cutStartThread(unsigned (WINAPI * func)(void *), void *data)// 注意函数指针的形式
 6 {
 7     return CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)func, data, 0, NULL);
 8 }
 9 
10 /*// 结束一个线程
11 void cutEndThread(HANDLE thread)
12 {
13     WaitForSingleObject(thread, INFINITE);
14     CloseHandle(thread);
15 }
16 
17 // 结束多个线程
18 void cutWaitForThreads(const HANDLE *threads, int num)
19 {
20     WaitForMultipleObjects(num, threads, true, INFINITE);
21 
22     for (int i = 0; i < num; i++)
23         CloseHandle(threads[i]);
24 }
25 */
26 // 创建线程墙
27 CUTBarrier cutCreateBarrier(int releaseCount)
28 {
29     CUTBarrier barrier;
30 
31     InitializeCriticalSection(&barrier.criticalSection);
32     barrier.barrierEvent = CreateEvent(NULL, TRUE, FALSE, TEXT("BarrierEvent"));
33     barrier.count = 0;
34     barrier.releaseCount = releaseCount;
35 
36     return barrier;
37 }
38 
39 // 线程墙判断县城工作是否已经全部结束
40 void cutIncrementBarrier(CUTBarrier *barrier)
41 {
42     int myBarrierCount;
43     EnterCriticalSection(&barrier->criticalSection);
44     myBarrierCount = ++barrier->count;
45     LeaveCriticalSection(&barrier->criticalSection);
46 
47     if (myBarrierCount >= barrier->releaseCount)// 发出的线程已经全部结束
48         SetEvent(barrier->barrierEvent);
49 }
50 
51 // 回收线程墙
52 void cutWaitForBarrier(CUTBarrier *barrier)
53 {
54     WaitForSingleObject(barrier->barrierEvent, INFINITE);
55 }
56 
57 /*// 销毁线程墙
58 void cutDestroyBarrier(CUTBarrier *barrier)
59 {
60 }
61 */

  1 /*simpleCallback.cu*/
  2 #include <stdio.h>
  3 #include <cuda_runtime.h>
  4 #include "device_launch_parameters.h"
  5 #include <helper_functions.h>
  6 #include <helper_cuda.h>
  7 #include "multithreading.h"
  8 
  9 const int N_workloads  = 8;
 10 const int block = 512;
 11 const int element = 100000;
 12 
 13 CUTBarrier thread_barrier;
 14 
 15 struct heterogeneous_workload   // 用于分配工作的结构
 16 {
 17     int id;                     // 主函数分配，工作编号
 18     int cudaDeviceID;           // 主函数分配，执行工作的设备号
 19     int *h_data;                // 起始函数分配，主机指针
 20     int *d_data;                // 起始函数分配，设备指针
 21     cudaStream_t stream;        // 起始函数分配，工作流号（一个工作使用一条流）
 22     bool success;               // 起始函数分配，由回调函数检查工作是否完成的标志
 23 };
 24 
 25 __global__ void incKernel(int *data, int N)// 将 data 中所有元素递增总线程个数次
 26 {
 27     int idx = blockIdx.x * blockDim.x + threadIdx.x;
 28 
 29     if (idx < N)
 30         data[idx]++;
 31 }
 32 
 33 unsigned WINAPI postprocess(void *void_arg)
 34 {
 35     heterogeneous_workload *workload = (heterogeneous_workload *)void_arg;
 36     cudaSetDevice(workload->cudaDeviceID);
 37     
 38     // 检查GPU计算结果
 39     getLastCudaError("Kernel execution failed");
 40     workload->success = true;
 41     for (int i = 0; i < N_workloads; ++i)
 42         workload->success &= (workload->h_data[i] == workload->id + i + 1);
 43 
 44     // 回收工作
 45     cudaFree(workload->d_data);
 46     cudaFreeHost(workload->h_data);
 47     cudaStreamDestroy(workload->stream);
 48 
 49     // 向线程墙发送工作完成的信号
 50     cutIncrementBarrier(&thread_barrier);
 51 
 52     return 0;
 53 }
 54 
 55 void CUDART_CB myStreamCallback(cudaStream_t stream, cudaError_t status, void *data)
 56 {
 57     // 回调函数，调用 postprocess 函数
 58     cutStartThread(postprocess, data);
 59 }
 60 
 61 unsigned WINAPI launch(void *void_arg)
 62 {
 63     heterogeneous_workload *workload = (heterogeneous_workload *) void_arg;
 64 
 65     // 初始化工作参数
 66     cudaSetDevice(workload->cudaDeviceID);
 67     cudaStreamCreate(&workload->stream);
 68     cudaMalloc(&workload->d_data, element * sizeof(int));
 69     cudaHostAlloc(&workload->h_data, element * sizeof(int), cudaHostAllocPortable);
 70     for (int i=0; i < element; ++i)
 71         workload->h_data[i] = workload->id + i;
 72 
 73     // 每个CPU线程对应一条CUDA流，分别调度流中的工作，可以并行运行，不阻塞CPU线程
 74     cudaMemcpyAsync(workload->d_data, workload->h_data, element * sizeof(int), cudaMemcpyHostToDevice, workload->stream);
 75     
 76     incKernel << <(element + block - 1) / block, block, 0, workload->stream >> > (workload->d_data, element);
 77     
 78     cudaMemcpyAsync(workload->h_data, workload->d_data, element * sizeof(int), cudaMemcpyDeviceToHost, workload->stream);
 79     
 80     // 回调函数，调用主机函数放入CUDA流中。在这里用于检查GPU结果和回收内存（相当于以前主函数中的收尾工作）
 81     cudaStreamAddCallback(workload->stream, myStreamCallback, workload, 0);
 82 
 83     return 0;
 84 }
 85 
 86 int main(int argc, char **argv)
 87 {
 88     printf("\n\tStart.\n");
 89     int i;
 90     bool success = true;
 91 
 92     // 创建工作表
 93     heterogeneous_workload *workloads = (heterogeneous_workload *) malloc(N_workloads * sizeof(heterogeneous_workload));
 94     
 95     // 创建线程墙，以便所有工作结束后回收
 96     thread_barrier = cutCreateBarrier(N_workloads);
 97     
 98     // 分配任务
 99     for (i = 0; i < N_workloads; ++i)
100     {
101         workloads[i].id = i; 
102         workloads[i].cudaDeviceID = 0;          // 将任务全部分配给 0 号设备
103         cutStartThread(launch, &workloads[i]);
104     }
105 
106     // 回收线程墙
107     cutWaitForBarrier(&thread_barrier);
108     printf("\n%d workloads all finished.\n",N_workloads);
109 
110     // 检查每项工作的 success 分量是否为 true
111     for (i = 0; i < N_workloads; success &= workloads[i].success, ++i);
112     printf("\n\t%s\n", success ? "Correct." : "Failure.");
113 
114     free(workloads);
115 
116     getchar();
117     return success;
118 }

 

? 输出结果：

    Start
8 workloads all finished.
    Success

 

? 涨姿势

● 回调函数的使用。

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

1 #define CUDART_CB __stdcall
2 #define CUDARTAPI  __stdcall
3 
4 // cudaStreamCallback_t 的定义
5 typedef void (CUDART_CB *cudaStreamCallback_t)(cudaStream_t stream, cudaError_t status, void *userData);
6 
7 // cudaStreamAddCallback 的定义
8 extern __host__ cudaError_t CUDARTAPI cudaStreamAddCallback(cudaStream_t stream, cudaStreamCallback_t callback, void *userData, unsigned int flags);

 

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

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