1

我有一个矩阵,我想计算第i行和每隔一行之间的距离(比如说欧几里得)(即我想要成对距离矩阵的第i行)。

#include <Rcpp.h>
#include <cmath>
#include <algorithm>
#include <RcppParallel.h>
//#include <RcppArmadillo.h>
#include <queue>

using namespace std;
using namespace Rcpp;
using namespace RcppParallel;

// [[Rcpp::export]]
double dist_fun(NumericVector row1, NumericVector row2){
   double rval = 0;
   for (int i = 0; i < row1.length(); i++){
      rval += (row1[i] - row2[i]) * (row1[i] - row2[i]);
   }
   return rval;
}

// [[Rcpp::export]]
NumericVector dist_row(NumericMatrix mat, int i){
   NumericVector row(mat.nrow());

   NumericMatrix::Row row1 = mat.row(i - 1);
   for (int j = 0; j < mat.nrow(); j++){
      NumericMatrix::Row row2 = mat.row(j);

      row(j) = dist_fun(row1, row2);
   }
   return row;
}

// [[Rcpp::depends(RcppParallel)]]
struct JsDistance: public Worker {

   // input matrix to read from
   const NumericMatrix mat;
   int i;

   // output vector to write to
   NumericVector output;

   // initialize from Rcpp input and output matrixes (the RMatrix class
   // can be automatically converted to from the Rcpp matrix type)
   JsDistance(const NumericMatrix mat, int i, NumericVector output)
      : mat(mat), i(i), output(output) {}


   // function call operator that work for the specified range (begin/end)
    void operator()(std::size_t begin, std::size_t end) {
      NumericVector row1 = mat.row(i);
      for (std::size_t j = begin; j < end; j++) {

         NumericVector row2 = mat.row(j);
         output[j] = dist_fun(row1, row2);
      }
    }
};

// [[Rcpp::export]]
NumericVector parallel_dist_row(NumericMatrix mat, int i) {

   // allocate the matrix we will return
   NumericVector output(mat.nrow());

   // create the worker
   JsDistance JsDistance(mat, i, output);

   // call it with parallelFor
   parallelFor(0, mat.nrow(), JsDistance);
   return output;
}

使用 Rcpp 的顺序方式是上面写的函数“row_dist”。然而我想使用的矩阵非常大,所以我想并行化它。但是后来我会遇到一个我不太明白为什么的段错误。要触发错误,您可以运行以下代码:

library(Rcpp)
library(RcppParallel)

setThreadOptions(numThreads = 20)


set.seed(42)
X = matrix(rnorm(10000 * 400), 10000, 400)
sourceCpp("question.cpp")


start1 = proc.time()
print(dist_row(X, 2)[1:30])
print(proc.time() - start1)

start2 = proc.time()
print(parallel_dist_row(X, 2)[1:30])
print(proc.time() - start2)

在此处输入图像描述

有人可以给我一些关于我做错了什么的提示吗?在此先感谢您的时间!

==================================================== ======================

编辑:

inline double d(double a, double b){
   return fabs(a - b);
}

    // [[Rcpp::depends(RcppParallel)]
struct dtwDistance: public Worker {

  // Input matrix to read from must be of the RMatrix<T> form
  // if using Rcpp objects
  const RMatrix<double> mat;
  int i;

  // Output vector to write to must be of the RVector<T> form
  // if using Rcpp objects
  RVector<double> output;

  // initialize from Rcpp input and output matrixes (the RMatrix class
  // can be automatically converted to from the Rcpp matrix type)
  dtwDistance(const NumericMatrix mat, int i, NumericVector output)
    : mat(mat), i(i - 1), output(output) {}

  // Note the -1 ^^^^ to match results from prior function

  // Function call operator to iterate over a specified range (begin/end)
  void operator()(std::size_t begin, std::size_t end) {

    RMatrix<double>::Row row1 = mat.row(i);

    for (std::size_t j = begin; j < end; ++j) {

      RMatrix<double>::Row row2 = mat.row(j);

      size_t n = row1.length();
      size_t m = row2.length();
      NumericMatrix cost(n + 1, m + 1);

      for (int ii = 1; ii <= n; ii++){
        cost(i, 0) = numeric_limits<double>::infinity();
      }

      for (int jj = 1; jj <= m; jj++){
        cost(0, j) = numeric_limits<double>::infinity();
      }

      for (int ii = 1; ii <= n; ii++){
          for (int jj = 1; jj <= m; jj++){
            double dist = d(row1[ii - 1], row2[jj - 1]);
            cost(ii, jj) = dist + min(min(cost(ii - 1, jj), cost(ii, jj - 1)), cost(ii - 1, jj - 1));
            //cout << ii << ", " << jj << ", " << cost(ii, jj) << "\n";
          }
      }
      output[j] = cost(n, m);

    }
  }
};
// [[Rcpp::export]]
NumericVector parallel_dist_row_dtw(NumericMatrix mat, int i) {

   // allocate the matrix we will return
   //RMatrix<double> input(mat);
   NumericVector y(mat.nrow());
   //RVector<double> output(y);

   // create the worker
   dtwDistance dtwDistance(mat, i, y);

   // call it with parallelFor
   parallelFor(0, mat.nrow(), dtwDistance);
   return y;
}

我需要计算的距离是动态时间扭曲距离。我如上所述实现了它。然而在运行时,它会给出一个“堆栈不平衡”的警告。并且运行几次后会出现段错误。我想知道现在有什么问题。

为了触发这个问题,我做了:

library(Rcpp)
library(RcppParallel)
setThreadOptions(numThreads = 4)
sourceCpp("scripts/chisq_dtw.cpp")
set.seed(42)
X = matrix(rnorm(1000), 100, 10)

parallel_dist_row_dtw(X, 1)
parallel_dist_row_dtw(X, 2)
parallel_dist_row_dtw(X, 3)
parallel_dist_row_dtw(X, 4)
parallel_dist_row_dtw(X, 5)
4

1 回答 1

4

问题是您没有通过RMatrix<T>和使用围绕 R 对象的线程安全包装器RVector<T>。这些类很重要,因为在后台线程上执行并行化,这是调用 R 或 Rcpp API 不安全的区域。官方文档Safe Accessors部分强调了这一点。

特别是,我们有:

为了提供对 R 向量和矩阵底层数组的安全和方便的访问,RcppParallel 引入了几个访问器类:

RVector<T>— 包装各种类型的 R 向量

RMatrix<T>— 包装各种类型的 R 矩阵(也包括RowColumn类)

要为 Rcpp 向量或矩阵创建线程安全访问器,只需构造它的实例RVectorRMatrix使用它。


代码修复

因此,您可以通过切换*MatrixRMatrix<T>*Vector来修复您的工作RVector<T>

struct JsDistance: public Worker {

  // Input matrix to read from must be of the RMatrix<T> form
  // if using Rcpp objects
  const RMatrix<double> mat;
  int i;

  // Output vector to write to must be of the RVector<T> form
  // if using Rcpp objects
  RVector<double> output;

  // initialize from Rcpp input and output matrixes (the RMatrix class
  // can be automatically converted to from the Rcpp matrix type)
  JsDistance(const NumericMatrix mat, int i, NumericVector output)
    : mat(mat), i(i - 1), output(output) {}

  // Note the -1 ^^^^ to match results from prior function

  // Function call operator to iterate over a specified range (begin/end)
  void operator()(std::size_t begin, std::size_t end) {

    RMatrix<double>::Row row1 = mat.row(i);

    for (std::size_t j = begin; j < end; ++j) {

      RMatrix<double>::Row row2 = mat.row(j);

      double rval = 0;
      for (unsigned int k = 0; k < row1.length(); ++k) {
        rval += (row1[k] - row2[k]) * (row1[k] - row2[k]);
      }

      output[j] = rval;

    }
  }
};

特别是,这里使用的数据类型RMatrix<double>甚至是用于访问矩阵的形式。

此外,在并行化版本中,缺少一条i-1语句。为了解决这个问题,我选择在JSDistance.


测试

set.seed(42)
X = matrix(rnorm(10000 * 400), 10000, 400)

start1 = proc.time()

print(dist_row(X, 2)[1:30])
# [1] 811.8873   0.0000 799.8153 810.1442 720.3232 730.6083 797.8441 781.8066 827.1511 834.1863 842.9392 850.2476 724.5842 673.1428 775.0994
# [16] 805.5752 804.9281 774.9770 799.7669 870.3187 815.1129 934.7581 726.1554 804.2097 758.4943 772.8931 806.6026 715.8257 847.8980 831.7555

print(proc.time() - start1)
# user  system elapsed 
# 0.22    0.00    0.23 

start2 = proc.time()

print(parallel_dist_row(X, 2)[1:30])
# [1] 811.8873   0.0000 799.8153 810.1442 720.3232 730.6083 797.8441 781.8066 827.1511 834.1863 842.9392 850.2476 724.5842 673.1428 775.0994
# [16] 805.5752 804.9281 774.9770 799.7669 870.3187 815.1129 934.7581 726.1554 804.2097 758.4943 772.8931 806.6026 715.8257 847.8980 831.7555

print(proc.time() - start2)
#   user  system elapsed 
#   0.28    0.00    0.06 

all.equal(parallel_dist_row(X, 2), dist_row(X, 2))
# [1] TRUE
于 2017-08-10T17:58:37.807 回答