Counting the number of positives values in a numeric vector. serial and multithreaded versions.

count_positive.cpp

#include <Rcpp.h>
#include <thread>
#include <future>

using namespace Rcpp ;

typedef NumericVector::iterator Iterator ;
  
struct count_positive {
  int operator()(Iterator first, Iterator last){
    int res = std::count_if( first, last, 
      [](double x){ return x > 0.0 ; } 
    );    
    return res ;
  }
} ;

// [[Rcpp::export]]
int count_positive_serial(NumericVector data){
  return count_positive()(data.begin(), data.end());
}
  
typedef std::packaged_task<int(Iterator,Iterator)> Task ;

// [[Rcpp::export]]
int count_positive_threaded(NumericVector data, int nthreads){
  int n = data.size() ;
  int chunk_size = n / nthreads ; 
  
  std::vector<std::future<int>> futures(nthreads-1) ;
  std::vector<std::thread> threads(nthreads-1) ;
  
  Iterator it = data.begin() ;
  for( int i=0; i<nthreads-1; i++){
    count_positive counter ;        
    Task task(counter) ;
    futures[i] = task.get_future();
    threads[i] = std::thread( std::move(task), it, it + chunk_size ) ;
    it += chunk_size ;
  }
  
  int result = count_positive()(it, data.end()); 
  
  for( int i=0; i<nthreads-1; i++){
    threads[i].join() ;
  }
  for( int i=0; i<nthreads-1; i++){
    result += futures[i].get() ;  
  }
                            
  return result ;
}



/*** R
  require(microbenchmark)
  
  bench <- function(n = 1e8 ){
    vec <- rnorm( n )
    microbenchmark( 
      R         = sum(vec > 0), 
      serial    = count_positive_serial( vec ), 
      threads_2 = count_positive_threaded( vec, 2 ), 
      threads_4 = count_positive_threaded( vec, 4 ), 
      threads_8 = count_positive_threaded( vec, 8 ), 
      times = 10, 
      unit = "us"
    )
  }
  
  for( n in 5:9 ){
    print( bench( 10^n ) )
  }
  
  
*/