8

我有一个相当简单的问题，但我无法找到一个优雅的解决方案。

我有一个推力代码，它产生c包含值的相同大小的向量。假设这些c向量中的每一个都有一个索引。我想为每个向量位置获取c值最低的向量的索引：

例子：

C0 =     (0,10,20,3,40)
C1 =     (1,2 ,3 ,5,10)

我会得到一个向量，C其中包含具有最小值的向量的索引：

result = (0,1 ,1 ,0,1)

我曾考虑过使用推力 zip 迭代器来做这件事，但遇到了问题：我可以压缩所有c向量并实现一个任意转换，该转换采用一个元组并返回其最小值的索引，但是：

  1. 如何迭代元组的内容？
  2. 据我了解，元组最多只能存储10元素，并且可以存储的不仅仅是10 c向量。

然后我考虑过这样做：与其使用c单独的向量，不如将它们全部附加到一个向量C中，然后生成引用位置的键并按键执行稳定的排序，这会将来自同一位置的向量条目重新组合在一起。在将给出的示例中：

C =      (0,10,20,3,40,1,2,3,5,10)
keys =   (0,1 ,2 ,3,4 ,0,1,2,3,4 )
after stable sort by key:
output = (0,1,10,2,20,3,3,5,40,10)
keys =   (0,0,1 ,1,2 ,2,3,3,4 ,4 )

然后使用向量中的位置生成键，使用向量的索引压缩输出，c然后使用自定义函子按键执行归约，每个归约输出具有最低值的索引。在示例中：

input =  (0,1,10,2,20,3,3,5,40,10)
indexes= (0,1,0 ,1,0 ,1,0,1,0 ,1)
keys =   (0,0,1 ,1,2 ,2,3,3,4 ,4)
after reduce by keys on zipped input and indexes:
output = (0,1,1,0,1)

但是，如何编写这样的函子来进行按键归约操作呢？

4

3 回答 3

5

因为你的向量的长度必须相同。最好将它们连接在一起并将它们视为矩阵C。

然后你的问题变成了在行主矩阵中找到每列的最小元素的索引。可以如下解决。

  1. 将 row-major 更改为 col-major；
  2. 查找每列的索引。

在步骤 1 中，您建议使用stable_sort_by_key重新排列元素顺序，这不是一种有效的方法。由于给定矩阵的#row和#col，可以直接计算重排。总之，可以使用置换迭代器来完成，如下所示：

thrust::make_permutation_iterator(
    c.begin(),
    thrust::make_transform_iterator(
        thrust::make_counting_iterator((int) 0),
        (_1 % row) * col + _1 / row)
)

在第 2 步中，reduce_by_key可以完全按照您的意愿行事。在您的情况下，减少二元运算仿函数很容易，因为已经定义了对元组（压缩向量的元素）的比较以比较元组的第一个元素，并且它由推力支持为

thrust::minimum< thrust::tuple<float, int> >()

整个程序如下所示。Thrust 1.6.0+ 是必需的，因为我在花哨的迭代器中使用占位符。

#include <iterator>
#include <algorithm>

#include <thrust/device_vector.h>
#include <thrust/iterator/counting_iterator.h>
#include <thrust/iterator/transform_iterator.h>
#include <thrust/iterator/permutation_iterator.h>
#include <thrust/iterator/zip_iterator.h>
#include <thrust/iterator/discard_iterator.h>
#include <thrust/reduce.h>
#include <thrust/functional.h>

using namespace thrust::placeholders;

int main()
{

    const int row = 2;
    const int col = 5;
    float initc[] =
            { 0, 10, 20, 3, 40, 1, 2, 3, 5, 10 };
    thrust::device_vector<float> c(initc, initc + row * col);

    thrust::device_vector<float> minval(col);
    thrust::device_vector<int> minidx(col);

    thrust::reduce_by_key(
            thrust::make_transform_iterator(
                    thrust::make_counting_iterator((int) 0),
                    _1 / row),
            thrust::make_transform_iterator(
                    thrust::make_counting_iterator((int) 0),
                    _1 / row) + row * col,
            thrust::make_zip_iterator(
                    thrust::make_tuple(
                            thrust::make_permutation_iterator(
                                    c.begin(),
                                    thrust::make_transform_iterator(
                                            thrust::make_counting_iterator((int) 0), (_1 % row) * col + _1 / row)),
                            thrust::make_transform_iterator(
                                    thrust::make_counting_iterator((int) 0), _1 % row))),
            thrust::make_discard_iterator(),
            thrust::make_zip_iterator(
                    thrust::make_tuple(
                            minval.begin(),
                            minidx.begin())),
            thrust::equal_to<int>(),
            thrust::minimum<thrust::tuple<float, int> >()
    );

    std::copy(minidx.begin(), minidx.end(), std::ostream_iterator<int>(std::cout, " "));
    std::cout << std::endl;
    return 0;
}

剩下的两个问题可能会影响性能。

  1. 必须输出最小值，这不是必需的；
  2. reduce_by_key是为具有不同长度的段而设计的，它可能不是减少相同长度段的最快算法。

编写自己的内核可能是获得最高性能的最佳解决方案。

于 2013-07-18T08:34:30.353 回答
4

一种可能的想法，基于此处的矢量化排序想法

  1. 假设我有这样的向量：

    values:    C =      ( 0,10,20, 3,40, 1, 2, 3, 5,10)
keys:      K =      ( 0, 1, 2, 3, 4, 0, 1, 2, 3, 4)
segments:  S =      ( 0, 0, 0, 0, 0, 1, 1, 1, 1, 1)

  2. 将 K 和 S 压​​缩在一起以创建 KS

  3. stable_sort_by_key 使用 C 作为键，KS 作为值：

    stable_sort_by_key(C.begin(), C.end(), KS_begin);

  4. 将重新排序的 C 和 K 向量压缩在一起，以创建 CK

  5. stable_sort_by_key 使用重新排序的 S 作为键，CK 作为值：

    stable_sort_by_key(S.begin(), S.end(), CK_begin);

  6. 使用置换迭代器或跨步范围迭代器访问新重新排序的 K 向量的每个第 N 个元素 (0, N, 2N, ...)，以检索每个段中最小元素的索引向量，其中N 是段的长度。

我还没有真正实现这一点，现在这只是一个想法。也许由于我尚未观察到的某种原因它不起作用。

segments( S) 和keys( K) 实际上是行和列索引。

而且您的问题对我来说似乎很奇怪，因为您的标题提到了“查找最大值索引”，但您的大部分问题似乎都指的是“最低值”。无论如何，通过更改我的算法的第 6 步，您可以找到任一值。

于 2013-07-18T02:59:02.050 回答
1

我有兴趣测试以前哪种方法更快。因此，我在下面的代码中实现了 Robert Crovella 的想法，为了完整起见，它还报告了 Eric 的方法。

#include <iterator>
#include <algorithm>

#include <thrust/random.h>
#include <thrust/device_vector.h>
#include <thrust/iterator/counting_iterator.h>
#include <thrust/iterator/transform_iterator.h>
#include <thrust/iterator/permutation_iterator.h>
#include <thrust/iterator/zip_iterator.h>
#include <thrust/iterator/discard_iterator.h>
#include <thrust/reduce.h>
#include <thrust/functional.h>
#include <thrust/sort.h>

#include "TimingGPU.cuh"

using namespace thrust::placeholders;

template <typename Iterator>
class strided_range
{
    public:

    typedef typename thrust::iterator_difference<Iterator>::type difference_type;

    struct stride_functor : public thrust::unary_function<difference_type,difference_type>
    {
        difference_type stride;

        stride_functor(difference_type stride)
            : stride(stride) {}

        __host__ __device__
        difference_type operator()(const difference_type& i) const
        { 
            return stride * i;
        }
    };

    typedef typename thrust::counting_iterator<difference_type>                   CountingIterator;
    typedef typename thrust::transform_iterator<stride_functor, CountingIterator> TransformIterator;
    typedef typename thrust::permutation_iterator<Iterator,TransformIterator>     PermutationIterator;

    // type of the strided_range iterator
    typedef PermutationIterator iterator;

    // construct strided_range for the range [first,last)
    strided_range(Iterator first, Iterator last, difference_type stride)
        : first(first), last(last), stride(stride) {}

    iterator begin(void) const
    {
        return PermutationIterator(first, TransformIterator(CountingIterator(0), stride_functor(stride)));
    }

    iterator end(void) const
    {
        return begin() + ((last - first) + (stride - 1)) / stride;
    }

    protected:
    Iterator first;
    Iterator last;
    difference_type stride;
};


/**************************************************************/
/* CONVERT LINEAR INDEX TO ROW INDEX - NEEDED FOR APPROACH #1 */
/**************************************************************/
template< typename T >
struct mod_functor {
    __host__ __device__ T operator()(T a, T b) { return a % b; }
};

/********/
/* MAIN */
/********/
int main()
{
    /***********************/
    /* SETTING THE PROBLEM */
    /***********************/
    const int Nrows = 200;
    const int Ncols = 200;

    // --- Random uniform integer distribution between 10 and 99
    thrust::default_random_engine rng;
    thrust::uniform_int_distribution<int> dist(10, 99);

    // --- Matrix allocation and initialization
    thrust::device_vector<float> d_matrix(Nrows * Ncols);
    for (size_t i = 0; i < d_matrix.size(); i++) d_matrix[i] = (float)dist(rng);

    TimingGPU timerGPU;

    /******************/
    /* APPROACH NR. 1 */
    /******************/
    timerGPU.StartCounter();

    thrust::device_vector<float>    d_min_values(Ncols);
    thrust::device_vector<int>      d_min_indices_1(Ncols);

    thrust::reduce_by_key(
            thrust::make_transform_iterator(
                    thrust::make_counting_iterator((int) 0),
                    _1 / Nrows),
            thrust::make_transform_iterator(
                    thrust::make_counting_iterator((int) 0),
                    _1 / Nrows) + Nrows * Ncols,
            thrust::make_zip_iterator(
                    thrust::make_tuple(
                            thrust::make_permutation_iterator(
                                    d_matrix.begin(),
                                    thrust::make_transform_iterator(
                                            thrust::make_counting_iterator((int) 0), (_1 % Nrows) * Ncols + _1 / Nrows)),
                            thrust::make_transform_iterator(
                                    thrust::make_counting_iterator((int) 0), _1 % Nrows))),
            thrust::make_discard_iterator(),
            thrust::make_zip_iterator(
                    thrust::make_tuple(
                            d_min_values.begin(),
                            d_min_indices_1.begin())),
            thrust::equal_to<int>(),
            thrust::minimum<thrust::tuple<float, int> >()
    );

    printf("Timing for approach #1 = %f\n", timerGPU.GetCounter());

    /******************/
    /* APPROACH NR. 2 */
    /******************/
    timerGPU.StartCounter();

    // --- Computing row indices vector
    thrust::device_vector<int> d_row_indices(Nrows * Ncols);
    thrust::transform(thrust::make_counting_iterator(0), thrust::make_counting_iterator(Nrows * Ncols), thrust::make_constant_iterator(Ncols), d_row_indices.begin(), thrust::divides<int>() );

    // --- Computing column indices vector
    thrust::device_vector<int> d_column_indices(Nrows * Ncols);
    thrust::transform(thrust::make_counting_iterator(0), thrust::make_counting_iterator(Nrows * Ncols), thrust::make_constant_iterator(Ncols), d_column_indices.begin(), mod_functor<int>());

    // --- int and float iterators
    typedef thrust::device_vector<int>::iterator        IntIterator;
    typedef thrust::device_vector<float>::iterator      FloatIterator;

    // --- Relevant tuples of int and float iterators
    typedef thrust::tuple<IntIterator, IntIterator>     IteratorTuple1;
    typedef thrust::tuple<FloatIterator, IntIterator>   IteratorTuple2;

    // --- zip_iterator of the relevant tuples
    typedef thrust::zip_iterator<IteratorTuple1>        ZipIterator1;
    typedef thrust::zip_iterator<IteratorTuple2>        ZipIterator2;

    // --- zip_iterator creation
    ZipIterator1 iter1(thrust::make_tuple(d_column_indices.begin(), d_row_indices.begin()));

    thrust::stable_sort_by_key(d_matrix.begin(), d_matrix.end(), iter1);

    ZipIterator2 iter2(thrust::make_tuple(d_matrix.begin(), d_row_indices.begin()));

    thrust::stable_sort_by_key(d_column_indices.begin(), d_column_indices.end(), iter2);

    typedef thrust::device_vector<int>::iterator Iterator;

    // --- Strided access to the sorted array
    strided_range<Iterator> d_min_indices_2(d_row_indices.begin(), d_row_indices.end(), Nrows);

    printf("Timing for approach #2 = %f\n", timerGPU.GetCounter());

    printf("\n\n");
    std::copy(d_min_indices_2.begin(), d_min_indices_2.end(), std::ostream_iterator<int>(std::cout, " "));
    std::cout << std::endl;

    return 0;
}

在大小矩阵的情况下测试两种方法2000x2000，这是在 Kepler K20c 卡上的结果：

Eric's             :  8.4s
Robert Crovella's  : 33.4s
于 2015-04-24T07:19:09.277 回答