c++ - 键/值数组的双调排序

Question

我正在尝试修改对 cl_int 数组进行排序的 the Intel's Bitonic Sorting 算法，以对 cl_int2 数组进行排序(基于键 - 即 cl_int2.x )。

英特尔的示例包含一个简单的主机代码和一个 OpenCL 内核，该内核在一次排序操作(多 channel )期间被多次调用。内核一次加载 4 个数组项作为 cl_int4 并对它们进行操作。

我没有修改主机代码算法，只修改了设备代码。 内核函数的变化列表:

• 将第一个内核的参数类型从 int4* 修改为 int8*(以加载四个键值对)
• 仅使用 .even 元素的 theArray 组件来比较值 ( < )
• 创建“pseudomask ” ( int4 ) 并基于此创建 mask 作为 pseudomask.xxyyzzww (以捕获值)

尽管我修改后的内核的输出完全按照第一个组件 ( cl_int2 ) 排序 cl_int2.x 数组，但值 ( cl_int2.y ) 不正确——一个项目的值在接下来的 4 或 8 个项目中重复，然后使用并重复新值...

我确定有一个微不足道的错误，但我无法找到它。

Diff of the original Intel code and my modified version .

编辑:当每个键 ( cl_int2 ) 是唯一的时，cl_int2.x 数组被完美排序。

---

示例输入:http://pastebin.com/92qB1csT

示例输出:http://pastebin.com/dsU97Npn

(正确排序的数组:http://pastebin.com/Nb56BuQK)

修改后的内核代码(注释):

// Copyright (c) 2009-2011 Intel Corporation
// https://software.intel.com/en-us/articles/bitonic-sorting

// Modified to sort int2 key-value array

__kernel void BitonicSort(__global int8* theArray,
                         const uint stage,
                         const uint passOfStage,
                         const uint dir)
{
    size_t i = get_global_id(0);
    int8 srcLeft, srcRight, mask;
    int4 pseudomask;
    int4 imask10 = (int4)(0,  0, -1, -1);
    int4 imask11 = (int4)(0, -1,  0, -1);

    if(stage > 0)
    {
        if(passOfStage > 0)    // upper level pass, exchange between two fours,
        {
            size_t r = 1 << (passOfStage - 1);
            size_t lmask = r - 1;
            size_t left = ((i>>(passOfStage-1)) << passOfStage) + (i & lmask);
            size_t right = left + r;

            srcLeft = theArray[left];
            srcRight = theArray[right];
            pseudomask = srcLeft.even < srcRight.even;
            mask = pseudomask.xxyyzzww;

            int8 imin = (srcLeft & mask) | (srcRight & ~mask);
            int8 imax = (srcLeft & ~mask) | (srcRight & mask);

            if( ((i>>(stage-1)) & 1) ^ dir )
            {
                theArray[left]  = imin;
                theArray[right] = imax;
            }
            else
            {
                theArray[right] = imin;
                theArray[left]  = imax;
            }
        }
        else    // last pass, sort inside one four
        {
            srcLeft = theArray[i];
            srcRight = srcLeft.s45670123;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask10;
            mask = pseudomask.xxyyzzww;

            if(((i >> stage) & 1) ^ dir)
            {
                srcLeft = (srcLeft & mask) | (srcRight & ~mask);

                srcRight = srcLeft.s23016745;
                pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
                mask = pseudomask.xxyyzzww;

                theArray[i] = (srcLeft & mask) | (srcRight & ~mask);
            }
            else
            {
                srcLeft = (srcLeft & ~mask) | (srcRight & mask);

                srcRight = srcLeft.s23016745;
                pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
                mask = pseudomask.xxyyzzww;

                theArray[i] = (srcLeft & ~mask) | (srcRight & mask);
            }
        }
    }
    else    // first stage, sort inside one four
    {
        /*
         *  To convert this code to int2 sorter, do this:
         *      1. instead of loading int4, load int8 (key,value, key,value, ...)
         *      2. when there is a vector swizzling, replace component index with two consecutive indices:
         *           srcLeft.yxwz  ->  srcLeft.s23016745
         *         use this rewrite rule:
         *           x  y  z  w
         *           01 23 45 67
         *      3. replace comparison operands with only their keys swizzled:
         *           mask = srcLeft < srcRight;    ->    pseudomask = srcLeft.even < srcRight.even; mask = pseudomask.xxyyzzww;
         */

        //  make bitonic sequence out of 4.
        int4 imask0 = (int4)(0, -1, -1,  0); // -1 in comparison = true (all bits set - two's complement)
        srcLeft = theArray[i];
        srcRight = srcLeft.s23016745;

        /*
         * This XOR mask flips bits, so that in `mask` are the following
         * results (remember that srcRight is srcLeft with swapped component pairs):
         *
         *      [ left.x<left.y, left.x<left.y,    left.w<left.z, left.w<left.z  ]
         *  or: [ left.x<left.y, left.x<left.y,    left.z>left.w, left.z>left.w  ]
         */
        pseudomask = (srcLeft.even < srcRight.even) ^ imask0;
        mask = pseudomask.xxyyzzww;

        if( dir )
            srcLeft = (srcLeft & mask) | (srcRight & ~mask);  // make sure the numbers are sorted like this:
        else
            srcLeft = (srcLeft & ~mask) | (srcRight & mask);

        /*
         *  Now the pairs of numbers in `srcLeft` are sorted according to the specified `dir`ection.
         *  If dir == true, then
         *    The components `x` and `y` are swapped so that `x` < `y`. Moreover `z` and `w` are swapped so that `z` > `w`. This resembles up-hill: /\
         *  else
         *    The components `x` and `y` are swapped so that `x` > `y`. Moreover `z` and `w` are swapped so that `z` < `w`. This resembles down-hill: \/
         *
         *  This swapping is achieved by creating `srcLeft`, which is in normal order, and `srcRight`, which has component pairs switched (xyzw -> yxwz).
         *  Then the `mask` is created. The mask bits are redundant because it applies to vector component pairs (so in order to implement key-value sorting,
         *  I have to increase the length of masks!).
         *
         *  The non-ordered component pairs in `srcLeft` are masked out by `mask` while the inverted `mask` is applied to the (pair-wise switched) `srcRight`.
         *
         *  This (the previous) first flipping just makes a 4-bitonic sequence.
         */


        /*
         *  This second step just sorts the bitonic sequence
         */
        srcRight = srcLeft.s45670123; // inverts the bitonic sequence

        // [ left.a<left.c, left.b<left.d,    left.a<left.c, left.b<left.d ]
        pseudomask = (srcLeft.even < srcRight.even) ^ imask10;  // imask10 = (noflip, noflip,  flip, flip)
        mask = pseudomask.xxyyzzww;

        // even or odd (The output of this thread is sorted monotonic sequence. The monotonicity changes and thus preparing bitonic sequence for the next pass.).
        if((i & 1) ^ dir)
        {
            // this sorts the bitonic sequence, hence splitting it
            srcLeft = (srcLeft & mask) | (srcRight & ~mask);

            srcRight = srcLeft.s23016745;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
            mask = pseudomask.xxyyzzww;

            theArray[i] = (srcLeft & mask) | (srcRight & ~mask);
        }
        else
        {
            srcLeft = (srcLeft & ~mask) | (srcRight & mask);

            srcRight = srcLeft.s23016745;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
            mask = pseudomask.xxyyzzww;

            theArray[i] = (srcLeft & ~mask) | (srcRight & mask);
        }
    }
}

主机端代码:

void ExecuteSortKernel(cl_kernel kernel, cl_command_queue queue, cl_mem cl_input_buffer, cl_int arraySize, cl_uint sortAscending)
{
    cl_int numStages = 0;

    cl_int stage;
    cl_int passOfStage;

    for (cl_int temp = arraySize; temp > 2; temp >>= 1)
        numStages++;

    clSetKernelArg(kernel, 0, sizeof(cl_mem), (void *) &cl_input_buffer);
    clSetKernelArg(kernel, 3, sizeof(cl_uint), (void *) &sortAscending);

    for (stage = 0; stage < numStages; stage++) {
        clSetKernelArg(kernel, 1, sizeof(cl_uint), (void *) &stage);

        for (passOfStage = stage; passOfStage >= 0; passOfStage--) {
            clSetKernelArg(kernel, 2, sizeof(cl_uint), (void *) &passOfStage);

            // set work-item dimensions
            size_t gsz = arraySize / (2*4);
            size_t global_work_size[1] = { passOfStage ? gsz : gsz << 1 };    //number of quad items in input array

            // execute kernel
            clEnqueueNDRangeKernel(queue, kernel, 1, NULL, global_work_size, NULL, 0, NULL, NULL);
        }
    }
}

Answer 1

我终于解决了这个问题!

棘手的部分在于原始英特尔代码处理加载的 4 元组中相邻对的相等值的方式 — 它没有明确处理它!

错误存在于所有其他 stage 的最后一个 passOfStage(即 passOfStage = 0)中的第一个 stage 和中。这些代码部分在一个 4 元组(由 cl_int8 数组 theArray 表示)内交换各个 2 元组。

让我们以这个摘录为例(对于 4 元组中的相等相邻 2 元组，它不能正常工作):

imask0     = (int4)(0, -1, -1,  0);
srcLeft    = theArray[i];  // int8
srcRight   = srcLeft.s23016745;
pseudomask = (srcLeft.even < srcRight.even) ^ imask0;
mask       = pseudomask.xxyyzzww;
result     = (srcLeft & mask) | (srcRight & ~mask);

想象一下当我们使用这个(未固定的)代码和 srcLeft.even = (int4)(7,7, 5,5) 时会发生什么。操作 srcLeft.even < srcRight.even 会产生 (int4)(0,0,0,0) ，然后我们用 imask0 屏蔽这个结果，我们会得到...... pseudomask = (int4)(0,-1,-1,0) – 即 imask 本身。然而，这是错误的。

形成此模式需要 pseudomask 的值:(int4)(a,a, b,b)(其中 a 和 b 可以是 0 或 -1)。这意味着进行以下比较以形成正确的 mask 就足够了:quasimask = srcLeft.s07 < srcRight.s07。然后正确的掩码将被创建为 mask = quasimask.xxxxyyyy 。前 2 个 x es 掩码 4 元组的第一个 2 元组中的第一个键值对(4 元组 = theArray 中的一个元素)。由于我们想要对相应的二元组进行位掩码(由 imask0 指定为 0 – -1 对)，我们添加了另一个 xx 。我们对 4 元组中的第二个 2 元组进行类似的位掩码，这给我们留下了 yyyy 。

使用 imask11 进行位移的可视化示例

srcLeft:                        x  y  z  w
                                <  <  <  <
srcRight [relative to srcLeft]: y  x  w  z
^ imask0:                       0 -1  0  1
------------------------------------------
(srcLeft<srcRight)^imask0:      x  x  z  z

固定的、功能齐全的版本(我已经评论了固定的部分):

__kernel void BitonicSort(__global int8* theArray,
                         const uint stage,
                         const uint passOfStage,
                         const uint dir)
{
    size_t i = get_global_id(0);
    int8 srcLeft, srcRight, mask;
    int4 pseudomask;
    int4 imask10 = (int4)(0,  0, -1, -1);
    int4 imask11 = (int4)(0, -1,  0, -1);

    if(stage > 0)
    {
        if(passOfStage > 0)    // upper level pass, exchange between two fours
        {
            size_t r = 1 << (passOfStage - 1);
            size_t lmask = r - 1;
            size_t left = ((i>>(passOfStage-1)) << passOfStage) + (i & lmask);
            size_t right = left + r;

            srcLeft = theArray[left];
            srcRight = theArray[right];
            pseudomask = srcLeft.even < srcRight.even;
            mask = pseudomask.xxyyzzww; // here we interchange individual components, so no mask is applied and hence no 2 pairs must contain the same bit-pattern

            int8 imin = (srcLeft & mask) | (srcRight & ~mask);
            int8 imax = (srcLeft & ~mask) | (srcRight & mask);

            if( ((i>>(stage-1)) & 1) ^ dir )
            {
                theArray[left]  = imin;
                theArray[right] = imax;
            }
            else
            {
                theArray[right] = imin;
                theArray[left]  = imax;
            }
        }
        else    // last pass, sort inside one four
        {
            srcLeft = theArray[i];
            srcRight = srcLeft.s45670123;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask10;
            mask = pseudomask.xxyyxxyy;

            if(((i >> stage) & 1) ^ dir)
            {
                srcLeft = (srcLeft & mask) | (srcRight & ~mask);

                srcRight = srcLeft.s23016745;
                pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
                mask = pseudomask.xxxxzzzz; // the 0th and 1st elements must contain the exact same value (as well as 2nd and 3rd)

                theArray[i] = (srcLeft & mask) | (srcRight & ~mask);
            }
            else
            {
                srcLeft = (srcLeft & ~mask) | (srcRight & mask);

                srcRight = srcLeft.s23016745;
                pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
                mask = pseudomask.xxxxzzzz; // the 0th and 1st elements must contain the exact same value (as well as 2nd and 3rd)

                theArray[i] = (srcLeft & ~mask) | (srcRight & mask);
            }
        }
    }
    else    // first stage, sort inside one four
    {
        int4 imask0 = (int4)(0, -1, -1,  0);
        srcLeft = theArray[i];
        srcRight = srcLeft.s23016745;

        pseudomask = (srcLeft.even < srcRight.even) ^ imask0;
        mask = pseudomask.xxxxwwww; // the 0th and 1st elements must contain the exact same value (as well as 2nd and 3rd)

        if( dir )
            srcLeft = (srcLeft & mask) | (srcRight & ~mask);
        else
            srcLeft = (srcLeft & ~mask) | (srcRight & mask);


        srcRight = srcLeft.s45670123;
        pseudomask = (srcLeft.even < srcRight.even) ^ imask10;
        mask = pseudomask.xxyyxxyy; // the 0th and 2nd elements must contain the exact same value (as well as 1st and 3rd)

        if((i & 1) ^ dir)
        {
            srcLeft = (srcLeft & mask) | (srcRight & ~mask);

            srcRight = srcLeft.s23016745;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
            mask = pseudomask.xxxxzzzz; // the 0th and 1st elements must contain the exact same value (as well as 2nd and 3rd)

            theArray[i] = (srcLeft & mask) | (srcRight & ~mask);
        }
        else
        {
            srcLeft = (srcLeft & ~mask) | (srcRight & mask);

            srcRight = srcLeft.s23016745;
            pseudomask = (srcLeft.even < srcRight.even) ^ imask11;
            mask = pseudomask.xxxxzzzz; // the 0th and 1st elements must contain the exact same value (as well as 2nd and 3rd)

            theArray[i] = (srcLeft & ~mask) | (srcRight & mask);
        }
    }
}