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 生产者-消费者 - 图1

图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 生产者-消费者主机端完整代码