我的书签
添加书签
移除书签4.4 生产者-消费者
很多OpenCL应用中，前一个内核的输出可能就会作为下一个内核的输入。换句话说，第一个内核是生产者，第二个内核是消费者。很多应用中生产者和消费者是并发工作的，生产者只将产生的数据交给消费者。OpenCL 2.0中提供管道内存对象，用来帮助生产者-消费者这样的应用。管道所提供的潜在功能性帮助，无论生产者-消费者内核是串行执行或并发执行。
本节中，我们将使用管道创建一个生产者-消费者应用，其中生产者和消费者分别用内核构成，这两个内核使用的是本章前两个例子：卷积和直方图。卷积内核将会对图像进行处理，然后使用管道将输出图像传入直方图内核中(如图4.5所示)。为了描述额外的功能，展示管道如何使用处理单元提高应用效率。本节的例子我们将使用多设备完成。卷积内核将执行在GPU设备上，直方图内核将执行在CPU设备上。多个设备上执行内核可以保证两个内核能够并发执行，其中管道就用来传输生产者需要的数据(且为消费者需要的数据)。对于管道对象的详细描述将在第6章展开。那么现在，让我们来了解一下本节例子的一些基本需求。
管道内存中的数据(称为packets)组织为先入先出(FIFO)结构。管道对象的内存在全局内存上开辟，所以可以被多个内核同时访问。这里需要注意的是，管道上存储的数据，主机端无法访问。
内核中管道属性可能是只读(read_only)或只写(write_only)，不过不能是读写。如果管道对象没有指定是只读或只写，那么编译器将默认其为只读。管道在内核的参数列表中，通过使用关键字pipe进行声明，后跟数据访问类型，和数据包的数据类型。例如，pipe __read_only float *input将会创建一个只读管道，该管道中包含的数据为单精度浮点类型。
图4.5 生产者内核将滤波后生成的像素点，通过管道传递给消费者内核，让消费者内核产生直方图：(a)为原始图像;(b)为滤波后图像;(c)为生成的直方图。
为了访问管道，OpenCL C提供内置函数read_pipe()和write_pipe()：
1. int read_pipe(pipe gentype p, gentype *ptr);
2. int write_pipe(pipe gentype p, const gentype *ptr);
当一个工作项调用read_pipe()(程序清单4.10，第16行)，一个包将从管道p中读取到ptr中。如果包读取正常，该函数返回0；如果管道为空，则该函数返回一个负值。write_pipe()(程序清单4.9，第50行)与读取类似，会将ptr上的包写入到管道p中。如果包写入正常，该函数返回0；如果管道已满，则该函数返回一个负值
程序清单4.9和4.10展示了我们应用中内核的实现。当我们指定目标消费者内核运行在CPU时，那么只有一个工作项去创建直方图。同样，当我们显式的指定一个CPU，我们需要之间将直方图的结果存放在全局内存中(第8章将对这样的权衡做更细化的讨论)。
1. __constant sampler_t sampler =
2. CLK_NORMALIZED_COORDS_FALSE |
3. CLK_FILTER_NEAREST |
4. CLK_ADDRESS_CLAMP_TO_EDGE;
6. __kernel
7. void producerKernel(
8. image2d_t __read_only inputImage,
9. pipe __write_only float *outputPipe,
10. __constant float *filter,
11. int filterWidth)
12. {
13. /* Store each work-item's unique row and column */
14. int column = get_global_id(0);
15. int row = get_global_id(1);
17. /* Half the width of the filter is needed for indexing
18. * memory later*/
19. int halfWidth = (int)(filterWidth / 2);
21. /* Used to hold the value of the output pixel */
22. float sum = 0.0f;
24. /* Iterator for the filter */
25. int filterIdx = 0;
27. /* Each work-item iterates around its local area on the basis of the
28. * size of the filter */
29. int2 coords; // Coordinates for accessing the image
31. /* Iterate the filter rows */
32. for (int i = -halfWidth; i <= halfWidth; i++)
33. {
34. coords.y = row + i;
35. /* Iterate over the filter columns */
36. for (int j = -halfWidth; j <= halfWidth; j++)
37. {
38. coords.x = column + j;
40. /* Read a pixel from the image. A single channel image
41. * stores the pixel in the x coordinate of the returned
42. * vector. */
43. float4 pixel;
44. pixel = read_imagef(inputImage, sampler, coords);
45. sum += pixel.x * filter[filterIdx++];
46. }
47. }
49. /* Write the output pixel to the pipe */
50. write_pipe(outputPipe, &sum);
51. }
程序清单4.9 卷积内核(生产者)
1. __kernel
2. void consumerKernel(
3. pipe __read_only float *inputPipe,
4. int totalPixels,
5. __global int *histogram)
6. {
7. int pixelCnt;
8. float pixel;
10. /* Loop to process all pixels from the producer kernel */
11. for (pixelCnt = 0; pixelCnt < totalPixels; pixelCnt++)
12. {
13. /* Keep trying to read a pixel from the pipe
14. * until one becomes available */
15. while(read_pipe(inputPipe, &pixel));
17. /* Add the pixel value to the histogram */
18. histogram[(int)pixel]++;
19. }
20. }
程序清单4.10 卷积内核(消费者)
虽然，存储在管道中的数据不能被主机访问，不过在主机端还是需要使用对应的API创建对应的管道对象。其创建API如下所示：
1. cl_pipe clCreatePipe(
2. cl_context context,
3. cl_mem_flags flags,
4. cl_uint pipe_packet_size,
5. cl_uint pipe_max_packets,
6. const cl_pipe_properties *properties,
7. cl_int *errcode_ret)
我们需要考虑两个内核不是并发的情况；因此，我们就需要创建足够大的管道对象能存放下图像元素数量个包：
1. cl_mem pipe = clCreatepipe(context, 0, sizeof(float), imageRows * imageCols, NULL, &status);
利用多个设备的话，就需要在主机端多加几步。当创建上下文对象时，需要提供两个设备(一个CPU设备，一个GPU设备)，并且每个设备都需要有自己的命令队列。另外，程序对象需要产生两个内核。加载内核是，需要分别入队其各自的命令队列：生产者(卷积)内核需要入队GPU命令队列，消费者(直方图)内核需要入队CPU命令队列。完整的代码在程序清单4.11中。
1. /* System includes */
2. #include <stdio.h>
3. #include <stdlib.h>
4. #include <string.h>
6. /* OpenCL includes */
7. #include <CL/cl.h>
9. /* Utility functions */
10. #include "utils.h"
11. #include "bmp-utils.h"
13. /* Filter for the convolution */
14. static float gaussianBlurFilter[25] = {
15. 1.0f / 273.0f, 4.0f / 273.0f, 7.0f / 273.0f, 4.0f / 273.0f, 1.0f / 273.0f,
16. 4.0f / 273.0f, 16.0f / 273.0f, 26.0f / 273.0f, 16.0f / 273.0f, 4.0f / 273.0f,
17. 7.0f / 273.0f, 26.0f / 273.0f, 41.0f / 273.0f, 26.0f / 273.0f, 7.0f / 273.0f,
18. 4.0f / 273.0f, 16.0f / 273.0f, 26.0f / 273.0f, 16.0f / 273.0f, 4.0f / 273.0f,
19. 1.0f / 273.0f, 4.0f / 273.0f, 7.0f / 273.0f, 4.0f / 273.0f, 1.0f / 273.0f
20. };
21. static const int filterWidth = 5;
22. static const int filterSize = 25 * sizeof(float);
24. /* Number of histogram bins */
25. static const int HIST_BINS = 256;
27. int main(int argc, char *argv[])
28. {
29. /* Host data */
30. float *hInputImage = NULL;
31. int *hOutputHistogram = NULL;
33. /* Allocate space for the input image and read the
34. * data from dist */
35. int imageRows;
36. int imageCols;
37. hInputImage = readBmpFloat("../../Images/cat.bmp", &imageRows, &imageCols);
38. const int imageElements = imageRows * imageCols;
39. const size_t imageSize = imageElements * sizeof(float);
41. /* Allocate space for the histogram on the host */
42. const int histogramSize = HIST_BINS * sizeof(int);
43. hOutputHistogram = (int *)malloc(histogramSize);
44. if (!hOutputHistogram){ exit(-1); }
46. /* Use this to check the output of each API call */
47. cl_int status;
49. /* Get the first platform */
50. cl_platform_id platform;
51. status = clGetPlatformIDs(1, &platform, NULL);
52. check(status);
54. /* Get the devices */
55. cl_device_id devices[2];
56. cl_device_id gpuDevice;
57. cl_device_id cpuDevice;
58. status = clGetDeviceIDs(platform, CL_DEVICE_TYPE_CPU, 1, &gpuDevice, NULL);
59. status = clGetDeviceIDs(platform, CL_DEVICE_TYPE_GPU, 1, &cpuDevice, NULL);
60. check(status);
61. devices[0] = gpuDevice;
62. devices[1] = cpuDevice;
64. /* Create a context and associate it with the devices */
65. cl_context context;
66. context = clCreateContext(NULL, 2, devices, NULL, NULL, &status);
67. check(status);
69. /* Create the command-queues */
70. cl_command_queue gpuQueue;
71. cl_command_queue cpuQueue;
72. gpuQueue = clCreateCommandQueue(context, gpuDevice, 0, &status);
73. check(status);
74. cpuQueue = clCreateCommandQueue(context, cpuDevice, 0, &status);
75. check(status);
77. /* The image desriptor describes how the data will be stored
78. * in memory. This descriptor initializes a 2D image with no pitch*/
79. cl_image_desc desc;
80. desc.image_type = CL_MEM_OBJECT_IMAGE2D;
81. desc.image_width = imageCols;
82. desc.image_height = imageRows;
83. desc.image_depth = 0;
84. desc.image_array_size = 0;
85. desc.image_row_pitch = 0;
86. desc.image_slice_pitch = 0;
87. desc.num_mip_levels = 0;
88. desc.num_samples = 0;
89. desc.buffer = NULL;
91. /* The image format descibes the properties of each pixel */
92. cl_image_format format;
93. format.image_channel_order = CL_R; // single channel
94. format.image_channel_data_type = CL_FLOAT;
96. /* Create the input image and initialize it using a
97. * pointer to the image data on the host. */
98. cl_mem inputImage;
99. inputImage = clCreateImage(context, CL_MEM_READ_ONLY, &format, &desc, NULL, NULL);
101. /* Create a buffer object for the ouput histogram */
102. cl_mem ouputHistogram;
103. outputHisrogram = clCreateBuffer(context, CL_MEM_WRITE_ONLY, &format, &desc, NULL, NULL);
105. /* Create a buffer for the filter */
106. cl_mem filter;
107. filter = clCreateBuffer(context, cl_MEM_READ_ONLY, filterSize, NULL, &status);
108. check(status);
110. cl_mem pipe;
111. pipe = clCreatePipe(context, 0, sizeof(float), imageRows * imageCols, NULL, &status);
113. /* Copy the host image data to the GPU */
114. size_t origin[3] = {0,0,0}; // Offset within the image to copy from
115. size_t region[3] = {imageCols, imageRows, 1}; // Elements to per dimension
116. status = clEnqueueWriteImage(gpuQueue, inputImage, CL_TRUE, origin, region, 0, 0, hInputImage, 0, NULL, NULL);
117. check(status);
119. /* Write the filter to the GPU */
120. status = clEnqueueWriteBuffer(gpuQueue, filter, CL_TRUE, 0, filterSize, gaussianBlurFilter, 0, NULL, NULL);
121. check(status);
123. /* Initialize the output istogram with zeros */
124. int zero = 0;
125. status = clEnqueueFillBuffer(cpuQueue, outputHistogram, &zero, sizeof(int), 0, histogramSize, 0, NULL, NULL);
126. check(status);
128. /* Create a program with source code */
129. char *programSource = readFile("producer-consumer.cl");
130. size_t programSourceSize = strlen(programSource);
131. cl_program program = clCreateProgramWithSource(context, 1, (const char**)&programSource, &programSourceLen, &status);
132. check(status);
134. /* Build (compile) the program for the devices */
135. status = clBuildProgram(program, 2, devices, NULL, NULL, NULL);
136. if (status != CL_SUCCESS)
137. {
138. printCompilerError(program, gpuDevice);
139. exit(-1);
140. }
142. /* Create the kernel */
143. cl_kernel producerKernel;
144. cl_kernel consumerKernel;
145. producerKernel = clCreateKernel(program, "producerKernel", &status);
146. check(status);
147. consumerKernel = clCreateKernel(program, "consumerKernel", &status);
148. check(status);
150. /* Set the kernel arguments */
151. status = clSetKernelArg(producerKernel, 0, sizeof(cl_mem), &inputImage);
152. status |= clSetKernelArg(producerKernel, 1, sizeof(cl_mem), &pipe);
153. status |= clSetKernelArg(producerKernel, 2, sizeof(int), &filterWidth);
154. check(status);
156. status |= clSetKernelArg(consumerKernel, 0, sizeof(cl_mem), &pipe);
157. status |= clSetKernelArg(consumerKernel, 1, sizeof(int), &imageElements);
158. status |= clSetKernelArg(consumerKernel, 2, sizeof(cl_mem), &outputHistogram);
159. check(status);
161. /* Define the index space and work-group size */
162. size_t producerGlobalSize[2];
163. producerGlobalSize[0] = imageCols;
164. producerGlobalSize[1] = imageRows;
166. size_t producerLocalSize[2];
167. producerLocalSize[0] = 8;
168. producerLocalSize[1] = 8;
170. size_t consumerGlobalSize[1];
171. consumerGlobalSize[0] = 1;
173. size_t consumerLocalSize[1];
174. consumerLocalSize[0] = 1;
176. /* Enqueue the kernels for execution */
177. status = clEnqueueNDRangeKernel(gpuQueue, producerKernel, 2, NULL, producerGlobalSize, producerLocalSize, 0, NULL, NULL);
179. status = clEnqueueNDRangeKernel(cpuQueue, consumerKernel, 2, NULL, consumerGlobalSize, consumerLocalSize, 0, NULL, NULL);
181. /* Read the output histogram buffer to the host */
182. status = clEnqueueReadBuffer(cpuQueue, outputHistogram, CL_TRUE, 0, histogramSize, hOutputHistogram, 0, NULL, NULL);
183. check(status);
185. /* Free OpenCL resources */
186. clReleaseKernel(producerKernel);
187. clReleaseKernel(consumerKernel);
188. clReleaseProgram(program);
189. clReleaseCommandQueue(gpuQueue);
190. clReleaseCommandQueue(cpuQueue);
191. clReleaseMemObject(inputImage);
192. clReleaseMemObject(outputHistogram);
193. clReleaseMemObject(filter);
194. clReleaseMemObject(pipe);
195. clReleaseContext(context);
197. /* Free host resources */
198. free(hInputImage);
199. free(hOutputHistogram);
200. free(programSource);
202. return 0;
203. }
程序清单4.11 生产者-消费者主机端完整代码
