Accessing begin and end in RcppParallel (exmaple calculating the mean of a vector)

Question

I'm having a problem while trying to learn RcppParallel. I tried to modify the code form https://rcppcore.github.io/RcppParallel/ for the vector summation to calculate the mean of a vector instead, to see if I understand the general principle.

My code is shown below. Using the function parallelVectorMean() on c(1,2,3,4,5) delivers inconsistent and most often incorrect results. I assume this has to do with me not understanding how to access begin and end correctly to scale my partial means accordingly during the join.

// [[Rcpp::depends(RcppParallel)]]
#include <RcppParallel.h>
#include <Rcpp.h>
using namespace RcppParallel;

struct Mean : public Worker
{
   // source vector
   const RVector<double> input;
   
   // accumulated value
   double value;
   
   // number of elements
   double num;
   
   // constructors
   Mean(const Rcpp::NumericVector input) : input(input), value(0), num(0) {}
   Mean(const Mean& mean, Split) : input(mean.input), value(0), num(0) {}
   
   // accumulate just the element of the range I've been asked to
   void operator()(std::size_t begin, std::size_t end) {
      num = (double) end - (double) begin;
      value += (std::accumulate(input.begin() + begin, input.begin() + end, 0.0) / num);
   }
   
   // join my value with that of another Mean
   void join(const Mean& rhs) {
      value = (num*value + rhs.num*rhs.value)/(num + rhs.num);
      num = num + rhs.num;
   }
};

// [[Rcpp::export]]
double parallelVectorMean(Rcpp::NumericVector x) {
   
   // declare the MeanBody instance
   Mean mean(x);
   
   // call parallel_reduce to start the work
   RcppParallel::parallelReduce(0, x.length(), mean);
   
   // return the computed mean
   return mean.value;
}

I'm looking forward to learning from you guys.

Answer 1

Accessing begin and end is fine, but your operator function's logic isn't correct.

Check this:

   void operator()(std::size_t begin, std::size_t end) {
      double temp_num = (double) end - (double) begin;
      double temp_value = (std::accumulate(input.begin() + begin, input.begin() + end, 0.0) / temp_num);
      value = (num*value + temp_num*temp_value)/(num + temp_num);
      num = num + temp_num;
   }

A thread may continue running on a new range without joining, so you have to account for that possibility.