<commit_before>/** @file distributed_kpca_dev.h
 *
 *  @author <NAME> (<EMAIL>)
 */

#ifndef MLPACK_DISTRIBUTED_KPCA_DISTRIBUTED_KPCA_DEV_H
#define MLPACK_DISTRIBUTED_KPCA_DISTRIBUTED_KPCA_DEV_H

#include "core/gnp/distributed_dualtree_dfs_dev.h"
#include "core/metric_kernels/lmetric.h"
#include "core/table/memory_mapped_file.h"
#include "core/table/transform.h"
#include "mlpack/distributed_kpca/distributed_kpca.h"
#include "mlpack/series_expansion/random_feature.h"

namespace core {
namespace parallel {

class AddDensePoint:
  public std::binary_function <
  core::table::DensePoint,
  core::table::DensePoint,
    core::table::DensePoint > {
  public:
    const core::table::DensePoint operator()(
      const core::table::DensePoint &a,
      const core::table::DensePoint &b) const {

      core::table::DensePoint sum;
      sum.Init(a.length());
      arma::vec a_alias;
      arma::vec b_alias;
      arma::vec sum_alias;
      core::table::DensePointToArmaVec(a, &a_alias);
      core::table::DensePointToArmaVec(b, &b_alias);
      core::table::DensePointToArmaVec(sum, &sum_alias);
      sum_alias = a_alias + b_alias;
      return sum;
    }
};
}
}

namespace boost {
namespace mpi {

template<>
class is_commutative <
  core::parallel::AddDensePoint,
  core::table::DensePoint  > :
  public boost::mpl::true_ {

};
}
}

namespace core {
namespace table {
extern core::table::MemoryMappedFile *global_m_file_;
}
}

namespace mlpack {
namespace distributed_kpca {

template<typename DistributedTableType, typename KernelType>
DistributedTableType *DistributedKpca <
DistributedTableType, KernelType >::query_table() {
  return query_table_;
}

template<typename DistributedTableType, typename KernelType>
DistributedTableType *DistributedKpca <
DistributedTableType, KernelType >::reference_table() {
  return reference_table_;
}

template<typename DistributedTableType, typename KernelType>
void DistributedKpca<DistributedTableType, KernelType>::Compute(
  const mlpack::distributed_kpca::DistributedKpcaArguments <
  DistributedTableType > &arguments_in,
  mlpack::distributed_kpca::KpcaResult *result_out) {

  // Barrier so that every process is here.
  world_->barrier();

  boost::mpi::timer timer;

  // The kernel.
  KernelType kernel;
  kernel.Init(arguments_in.bandwidth_);

  // The local kernel sum.
  core::monte_carlo::MeanVariancePairVector local_kernel_sum;
  local_kernel_sum.Init(arguments_in.query_table_->local_table()->n_entries());

  // Call the computation.
  bool all_done = false;
  const int num_random_fourier_features = 20;
  int num_iterations = 0;
  do {

    // The master generates a set of random Fourier features and do a
    // broadcast.
    std::vector <
    core::table::DensePoint > random_variates(num_random_fourier_features);
    std::vector< arma::vec > random_variate_aliases(
      num_random_fourier_features);
    if(world_->rank() == 0) {
      for(int i = 0; i < num_random_fourier_features; i++) {

        // Draw a random Fourier feature.
        kernel.DrawRandomVariate(
          arguments_in.reference_table_->n_attributes(), & random_variates[i]);
      }
    }
    boost::mpi::broadcast(*world_, random_variates, 0);
    for(unsigned int i = 0; i < num_random_fourier_features; i++) {
      core::table::DensePointToArmaVec(
        random_variates[i], &(random_variate_aliases[i]));
    }

    // Each process computes the sum of the projections of the local
    // reference set and does an all-reduction to compute the global
    // sum projections.
    core::table::DensePoint local_reference_sum;
    mlpack::series_expansion::RandomFeature::SumTransform(
      *(arguments_in.reference_table_->local_table()),
      random_variate_aliases, &local_reference_sum);
    core::table::DensePoint global_reference_sum;
    boost::mpi::all_reduce(
      *world_, local_reference_sum, global_reference_sum,
      core::parallel::AddDensePoint());

    // Each process computes the local projection of each query point
    // and adds up.
    arma::vec global_reference_sum_alias;
    core::table::DensePointToArmaVec(
      global_reference_sum, &global_reference_sum_alias);
    for(int i = 0;
        i < arguments_in.query_table_->local_table()->n_entries(); i++) {
      arma::vec query_point;
      arguments_in.query_table_->local_table()->get(i, &query_point);
      arma::vec query_point_projected;
      mlpack::series_expansion::RandomFeature::Transform(
        query_point, random_variate_aliases, &query_point_projected);
      local_kernel_sum[i].push_back(
        arma::dot(query_point_projected, global_reference_sum_alias));
    }
    num_iterations++;
  }
  while(num_iterations < 20);

  // Barrier so that every process is done.
  world_->barrier();
  if(world_->rank() == 0) {
    printf("Spent %g seconds in computation.\n", timer.elapsed());
  }
}

template<typename DistributedTableType, typename KernelType>
void DistributedKpca<DistributedTableType, KernelType>::Init(
  boost::mpi::communicator &world_in,
  mlpack::distributed_kpca::DistributedKpcaArguments <
  DistributedTableType > &arguments_in) {

  world_ = &world_in;<commit_msg>I think it works.<commit_after>/** @file distributed_kpca_dev.h
 *
 *  @author <NAME> (<EMAIL>)
 */

#ifndef MLPACK_DISTRIBUTED_KPCA_DISTRIBUTED_KPCA_DEV_H
#define MLPACK_DISTRIBUTED_KPCA_DISTRIBUTED_KPCA_DEV_H

#include "core/gnp/distributed_dualtree_dfs_dev.h"
#include "core/metric_kernels/lmetric.h"
#include "core/table/memory_mapped_file.h"
#include "core/table/transform.h"
#include "mlpack/distributed_kpca/distributed_kpca.h"
#include "mlpack/series_expansion/random_feature.h"

namespace core {
namespace parallel {

class AddDensePoint:
  public std::binary_function <
  core::table::DensePoint,
  core::table::DensePoint,
    core::table::DensePoint > {
  public:
    const core::table::DensePoint operator()(
      const core::table::DensePoint &a,
      const core::table::DensePoint &b) const {

      core::table::DensePoint sum;
      sum.Init(a.length());
      arma::vec a_alias;
      arma::vec b_alias;
      arma::vec sum_alias;
      core::table::DensePointToArmaVec(a, &a_alias);
      core::table::DensePointToArmaVec(b, &b_alias);
      core::table::DensePointToArmaVec(sum, &sum_alias);
      sum_alias = a_alias + b_alias;
      return sum;
    }
};
}
}

namespace boost {
namespace mpi {

template<>
class is_commutative <
  core::parallel::AddDensePoint,
  core::table::DensePoint  > :
  public boost::mpl::true_ {

};
}
}

namespace core {
namespace table {
extern core::table::MemoryMappedFile *global_m_file_;
}
}

namespace mlpack {
namespace distributed_kpca {

template<typename DistributedTableType, typename KernelType>
DistributedTableType *DistributedKpca <
DistributedTableType, KernelType >::query_table() {
  return query_table_;
}

template<typename DistributedTableType, typename KernelType>
DistributedTableType *DistributedKpca <
DistributedTableType, KernelType >::reference_table() {
  return reference_table_;
}

template<typename DistributedTableType, typename KernelType>
void DistributedKpca<DistributedTableType, KernelType>::Compute(
  const mlpack::distributed_kpca::DistributedKpcaArguments <
  DistributedTableType > &arguments_in,
  mlpack::distributed_kpca::KpcaResult *result_out) {

  // Barrier so that every process is here.
  world_->barrier();

  // The MPI timer.
  boost::mpi::timer timer;

  // The kernel.
  KernelType kernel;
  kernel.Init(arguments_in.bandwidth_);

  // The local kernel sum.
  core::monte_carlo::MeanVariancePairVector local_kernel_sum;
  local_kernel_sum.Init(arguments_in.query_table_->local_table()->n_entries());

  // Call the computation.
  const int num_random_fourier_features = 20;
  int num_iterations = 0;
  bool all_query_converged = true;
  do {

    // The master generates a set of random Fourier features and do a
    // broadcast.
    std::vector <
    core::table::DensePoint > random_variates(num_random_fourier_features);
    std::vector< arma::vec > random_variate_aliases(
      num_random_fourier_features);
    if(world_->rank() == 0) {
      for(int i = 0; i < num_random_fourier_features; i++) {

        // Draw a random Fourier feature.
        kernel.DrawRandomVariate(
          arguments_in.reference_table_->n_attributes(), & random_variates[i]);
      }
    }
    boost::mpi::broadcast(*world_, random_variates, 0);
    for(int i = 0; i < num_random_fourier_features; i++) {
      core::table::DensePointToArmaVec(
        random_variates[i], &(random_variate_aliases[i]));
    }

    // Each process computes the sum of the projections of the local
    // reference set and does an all-reduction to compute the global
    // sum projections.
    core::table::DensePoint local_reference_sum;
    mlpack::series_expansion::RandomFeature::SumTransform(
      *(arguments_in.reference_table_->local_table()),
      random_variate_aliases, &local_reference_sum);
    core::table::DensePoint global_reference_sum;
    boost::mpi::all_reduce(
      *world_, local_reference_sum, global_reference_sum,
      core::parallel::AddDensePoint());

    // Each process computes the local projection of each query point
    // and adds up.
    arma::vec global_reference_sum_alias;
    core::table::DensePointToArmaVec(
      global_reference_sum, &global_reference_sum_alias);

    bool all_local_query_converged = true;
    int converged_count = 0;
    for(int i = 0;
        i < arguments_in.query_table_->local_table()->n_entries(); i++) {
      arma::vec query_point;
      arguments_in.query_table_->local_table()->get(i, &query_point);
      arma::vec query_point_projected;
      mlpack::series_expansion::RandomFeature::Transform(
        query_point, random_variate_aliases, &query_point_projected);
      for(int j = 0; j < num_random_fourier_features; j++) {
        double contribution =
          query_point_projected[j] * global_reference_sum_alias[j] +
          query_point_projected[j + num_random_fourier_features] *
          global_reference_sum_alias[j + num_random_fourier_features];
        local_kernel_sum[i].push_back(contribution);
      }

      double left_hand_side =
        1.64 * sqrt(local_kernel_sum[i].sample_mean_variance());
      double right_hand_side =
        arguments_in.relative_error_ * local_kernel_sum[i].sample_mean();
      bool query_point_converged = (left_hand_side <= right_hand_side);
      if(query_point_converged) {
        converged_count++;
      }
      all_local_query_converged = all_local_query_converged &&
                                  query_point_converged;
    }
    printf("Process %d has %d converged queries.\n", world_->rank(),
           converged_count);
    num_iterations++;

    // Do an all-reduction to find out we are all done.
    boost::mpi::all_reduce(
      *world_, all_local_query_converged,
      all_query_converged, std::logical_and<bool>());
  }
  while(! all_query_converged);

  // Barrier so that every process is done.
  world_->barrier();
  if(world_->rank() == 0) {
    printf("Spent %g seconds in computation.\n", timer.elapsed());
  }
}

template<typename DistributedTableType, typename KernelType>
void DistributedKpca<DistributedTableType, KernelType>::Init(
  boost::mpi::communicator &world_in,
  mlpack::distributed_kpca::DistributedKpcaArguments <
  DistributedTableType > &arguments_in) {

  world_ = &world_in;
{'fraction_non_alphanumeric': 0.11319261213720316, 'fraction_numerical': 0.0021987686895338612, 'mean_word_length': 4.404467680608365, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2774923', 'n_tokens_mistral': 3806, 'n_tokens_neox': 3646, 'n_words': 686}