使用CUDA Thrust确定每个矩阵列的最小元素及其位置

Question

使用CUDA Thrust确定每个矩阵列的最小元素及其位置

8

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

我有一段产生包含值的相同大小的 c 向量的Thrust代码。假设这些 c 向量中的每个向量都有一个索引。我希望对于每个向量位置，获取值最低的 c 向量的索引：

例如：

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

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

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

我考虑使用thrust zip迭代器来完成，但遇到了问题：我可以将所有 c 向量进行压缩，并实现一个任意转换，该转换接受一个元组并返回其最低值的索引，但是：

如何迭代元组的内容？
据我所知，元组只能存储多达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)

然而，如何为按键缩减操作编写这样的函数对象呢？

- Namux

你实际上正在尝试查找行主元矩阵中每列最小元素的索引。 - kangshiyin

3个回答

4

有一个可能的想法，基于向量化排序的想法在这里

Suppose I have vectors like this:

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)

zip together K and S to create KS
stable_sort_by_key using C as the keys, and KS as the values:
```
stable_sort_by_key(C.begin(), C.end(), KS_begin);
```
zip together the reordered C and K vectors, to create CK
stable_sort_by_key using the reordered S as the keys, and CK as the values:
```
stable_sort_by_key(S.begin(), S.end(), CK_begin);
```
use a permutation iterator or a strided range iterator to access every Nth element (0, N, 2N, ...) of the newly re-ordered K vector, to retrieve a vector of the indices of the min element in each segment, where N is the length of the segments.

我实际上还没有实现这个想法，现在它只是一个想法。也许有一些我尚未观察到的原因导致它无法正常工作。

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

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

- Robert Crovella

谢谢您的回复。它确实有效，除了第4步和第5步，需要将C和S压缩，并按K作为键进行排序。关于标题，您是正确的，我已经编辑过了 :) - Namux

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

- Vitality

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- kangshiyin · Accepted Answer

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

然后，您的问题变成了查找行主矩阵中每列最小元素的索引。可以按以下方式解决：

1. 将行主转换为列主； 2. 找到每个列的指数。

在步骤1中，您建议使用stable_sort_by_key重新排列元素顺序，这不是一种有效的方法。由于给定矩阵的#row和#col可以直接计算出重排列，因此可以使用排列迭代器在Thrust中完成它：

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

在第二步中，reduce_by_key可以完全达到您的要求。在您的情况下，减少二元操作函数很容易，因为已经定义了元组（您的压缩向量的元素）的比较来比较元组的第一个元素，并且它由thrust支持。

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;
}

还有两个问题可能会影响性能。

需要输出最小值，但这并不是必需的；
reduce_by_key 是为长度不同的片段设计的算法，对于长度相同的片段来说，它可能不是最快的归约算法。

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