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

#ifndef CORE_TABLE_DISTRIBUTED_TABLE_H
#define CORE_TABLE_DISTRIBUTED_TABLE_H

#include <armadillo>
#include "boost/mpi.hpp"
#include "boost/mpi/collectives.hpp"
#include "boost/thread.hpp"
#include "boost/serialization/string.hpp"
#include "core/table/table.h"
#include "core/table/distributed_table_message.h"
#include "core/table/point_request_message.h"
#include "core/table/mailbox.h"

namespace core {
namespace table {
class DistributedTable: public boost::noncopyable {

    typedef core::tree::GeneralBinarySpaceTree < core::tree::BallBound <
    core::table::DensePoint > > TreeType;

  private:

    core::table::Table *owned_table_;

    std::vector<int> local_n_entries_;

    TreeType *global_tree_;

    boost::mpi::communicator *comm_;

    core::table::PointInbox point_inbox_;

    core::table::PointRequestMessageBox point_request_message_box_;

  public:

    int rank() const {
      return comm_->rank();
    }

    bool IsIndexed() const {
      return global_tree_ != NULL;
    }

    DistributedTable() {
      comm_ = NULL;
      owned_table_ = NULL;
      global_tree_ = NULL;
    }

    ~DistributedTable() {

      // Put a barrier so that the distributed tables get destructed
      // when all of the requests are fulfilled for all of the
      // distributed processes.
      comm_->barrier();

      // Terminate the point request message box.
      {
        comm_->isend(
          comm_->rank(),
          core::table::DistributedTableMessage::TERMINATE_POINT_REQUEST_MESSAGE_BOX, 0);
        boost::unique_lock<boost::mutex> point_request_message_box_lock(
          point_request_message_box_.mutex());
        point_request_message_box_.point_request_message_box_quitting().wait(
          point_request_message_box_lock);
      }

      // Wait until the point inbox is destroyed.
      {
        comm_->isend(
          comm_->rank(),
          core::table::DistributedTableMessage::TERMINATE_POINT_INBOX, 0);
        boost::unique_lock<boost::mutex> point_inbox_lock(
          point_inbox_.termination_mutex());
        point_inbox_.termination_cond().wait(point_inbox_lock);
      }

      // Put a barrier so that all processes are ready to destroy each
      // of their own tables and trees.
      comm_->barrier();

      // Delete the table.
      if(owned_table_ != NULL) {
        delete owned_table_;
        owned_table_ = NULL;
      }

      // Delete the tree.
      if(global_tree_ != NULL) {
        delete global_tree_;
        global_tree_ = NULL;
      }
    }

    const TreeType::BoundType &get_node_bound(TreeType * node) const {
      return node->bound();
    }

    TreeType::BoundType &get_node_bound(TreeType * node) {
      return node->bound();
    }

    TreeType *get_node_left_child(TreeType * node) {
      return node->left();
    }

    TreeType *get_node_right_child(TreeType * node) {
      return node->right();
    }

    bool node_is_leaf(TreeType * node) const {
      return node->is_leaf();
    }

    core::tree::AbstractStatistic *&get_node_stat(TreeType * node) {
      return node->stat();
    }

    int get_node_count(TreeType * node) const {
      return node->count();
    }

    TreeType *get_tree() {
      return global_tree_;
    }

    int n_attributes() const {
      return owned_table_->n_attributes();
    }

    int local_n_entries(int rank_in) const {
      return local_n_entries_[rank_in];
    }

    int local_n_entries() const {
      return owned_table_->n_entries();
    }

    void Init(
      const std::string & file_name,
      boost::mpi::communicator * communicator_in) {

      // Set the communicator and read the table.
      comm_ = communicator_in;
      owned_table_ = new core::table::Table();
      owned_table_->Init(file_name);

      // Allocate the vector for storing the number of entries for all
      // the tables in the world, and do an all-gather operation to
      // find out all the sizes.
      boost::mpi::all_gather(
        *comm_, owned_table_->n_entries(), local_n_entries_);

      // Initialize the mail boxes.
      point_inbox_.Init(comm_);
      point_request_message_box_.Init(comm_, owned_table_);

      // Detach the server threads for each distributed process.
      point_inbox_.Detach();
      point_request_message_box_.Detach();

      // Put a barrier to ensure that every process has started up the
      // mailboxes.
      comm_->barrier();
    }

    void Save(const std::string & file_name) const {

    }

    void IndexData(
      const core::metric_kernels::AbstractMetric & metric_in, int leaf_size) {

      // We need to build the top tree first.
      if(comm_->rank() == 0) {
      }
    }

    void get(
      int requested_rank, int point_id,
      core::table::DensePoint * entry) {

      // If owned by the process, just return the point. Otherwise, we
      // need to send an MPI request to the process holding the
      // required resource.
      if(comm_->rank() == requested_rank) {
        owned_table_->get(point_id, entry);
      }<commit_msg>Starting distributed tree building.<commit_after>/** @file distributed_table.h
 *
 *  @author <NAME> (<EMAIL>)
 */

#ifndef CORE_TABLE_DISTRIBUTED_TABLE_H
#define CORE_TABLE_DISTRIBUTED_TABLE_H

#include <armadillo>
#include "boost/mpi.hpp"
#include "boost/mpi/collectives.hpp"
#include "boost/thread.hpp"
#include "boost/serialization/string.hpp"
#include "core/table/table.h"
#include "core/table/distributed_table_message.h"
#include "core/table/point_request_message.h"
#include "core/table/mailbox.h"

namespace core {
namespace table {
class DistributedTable: public boost::noncopyable {

    typedef core::tree::GeneralBinarySpaceTree < core::tree::BallBound <
    core::table::DensePoint > > TreeType;

  private:

    core::table::Table *owned_table_;

    std::vector<int> local_n_entries_;

    TreeType *global_tree_;

    boost::mpi::communicator *comm_;

    core::table::PointInbox point_inbox_;

    core::table::PointRequestMessageBox point_request_message_box_;

  public:

    int rank() const {
      return comm_->rank();
    }

    bool IsIndexed() const {
      return global_tree_ != NULL;
    }

    DistributedTable() {
      comm_ = NULL;
      owned_table_ = NULL;
      global_tree_ = NULL;
    }

    ~DistributedTable() {

      // Put a barrier so that the distributed tables get destructed
      // when all of the requests are fulfilled for all of the
      // distributed processes.
      comm_->barrier();

      // Terminate the point request message box.
      {
        comm_->isend(
          comm_->rank(),
          core::table::DistributedTableMessage::TERMINATE_POINT_REQUEST_MESSAGE_BOX, 0);
        boost::unique_lock<boost::mutex> point_request_message_box_lock(
          point_request_message_box_.mutex());
        point_request_message_box_.point_request_message_box_quitting().wait(
          point_request_message_box_lock);
      }

      // Wait until the point inbox is destroyed.
      {
        comm_->isend(
          comm_->rank(),
          core::table::DistributedTableMessage::TERMINATE_POINT_INBOX, 0);
        boost::unique_lock<boost::mutex> point_inbox_lock(
          point_inbox_.termination_mutex());
        point_inbox_.termination_cond().wait(point_inbox_lock);
      }

      // Put a barrier so that all processes are ready to destroy each
      // of their own tables and trees.
      comm_->barrier();

      // Delete the table.
      if(owned_table_ != NULL) {
        delete owned_table_;
        owned_table_ = NULL;
      }

      // Delete the tree.
      if(global_tree_ != NULL) {
        delete global_tree_;
        global_tree_ = NULL;
      }
    }

    const TreeType::BoundType &get_node_bound(TreeType * node) const {
      return node->bound();
    }

    TreeType::BoundType &get_node_bound(TreeType * node) {
      return node->bound();
    }

    TreeType *get_node_left_child(TreeType * node) {
      return node->left();
    }

    TreeType *get_node_right_child(TreeType * node) {
      return node->right();
    }

    bool node_is_leaf(TreeType * node) const {
      return node->is_leaf();
    }

    core::tree::AbstractStatistic *&get_node_stat(TreeType * node) {
      return node->stat();
    }

    int get_node_count(TreeType * node) const {
      return node->count();
    }

    TreeType *get_tree() {
      return global_tree_;
    }

    int n_attributes() const {
      return owned_table_->n_attributes();
    }

    int local_n_entries(int rank_in) const {
      return local_n_entries_[rank_in];
    }

    int local_n_entries() const {
      return owned_table_->n_entries();
    }

    void Init(
      const std::string & file_name,
      boost::mpi::communicator * communicator_in) {

      // Set the communicator and read the table.
      comm_ = communicator_in;
      owned_table_ = new core::table::Table();
      owned_table_->Init(file_name);

      // Allocate the vector for storing the number of entries for all
      // the tables in the world, and do an all-gather operation to
      // find out all the sizes.
      boost::mpi::all_gather(
        *comm_, owned_table_->n_entries(), local_n_entries_);

      // Initialize the mail boxes.
      point_inbox_.Init(comm_);
      point_request_message_box_.Init(comm_, owned_table_);

      // Detach the server threads for each distributed process.
      point_inbox_.Detach();
      point_request_message_box_.Detach();

      // Put a barrier to ensure that every process has started up the
      // mailboxes.
      comm_->barrier();
    }

    void Save(const std::string & file_name) const {

    }

    void IndexData(
      const core::metric_kernels::AbstractMetric & metric_in,
      double sample_probability_in) {

      // For each process, select a subset of indices to send to the
      // master node.
      std::vector<int> sampled_indices;
      for(int i = 0; i < this->n_entries(); i++) {
        if(core::math::Random() <= sample_probability_in) {
          sampled_indices.push_back(i);
        }
      }

      // For the master node,
      if(comm_->rank() == 0) {

        // Find out the list of points sampled so that we can build a
        // sampled table to build the tree from.
        std::vector< std::vector<int> > list_of_sampled_indices;
        int total_num_sampled_points = 0;
        boost::mpi::gather(
          *comm_, sampled_indices, list_of_sampled_indices, 0);

        // Gather all the necessary data to build the tree.
        for(unsigned int i = 0; i < list_of_sampled_indices.size(); i++) {
          total_num_sample_points += list_of_sampled_indices[i].size();
        }
        core::table::Table sampled_table;
        sampled_table.Init(this->n_attributes(), total_num_sample_points);
        for(int i = 0; i < list_of_sampled_indices.size(); i++) {
          const std::vector<int> &sampled_indices_per_process =
            list_of_sampled_indices[i];
          for(int j = 0; j < sampled_indices_per_process.size(); j++) {

          }
        }


        // Broadcast the top tree to all the other processes.

      }

      // For the other nodes,
      else {
        boost::mpi::reduce(
          *comm_, random_indices.size(), boost::mpi::sum<int>(), 0);
      }
    }

    void get(
      int requested_rank, int point_id,
      core::table::DensePoint * entry) {

      // If owned by the process, just return the point. Otherwise, we
      // need to send an MPI request to the process holding the
      // required resource.
      if(comm_->rank() == requested_rank) {
        owned_table_->get(point_id, entry);
      }
{'fraction_non_alphanumeric': 0.10906600462289187, 'fraction_numerical': 0.0011129184145193048, 'mean_word_length': 2.6689698492462313, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 70, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1091739', 'n_tokens_mistral': 3609, 'n_tokens_neox': 3463, 'n_words': 840}