Getting there, believe it.

fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h

@@ -26,12 +26,11 @@ namespace table {
 
 extern MemoryMappedFile *global_m_file_;
 
+template<typename TreeSpecType>
 class DistributedTable: public boost::noncopyable {
 
   public:
 
-    typedef core::tree::GenMetricTree < core::table::DensePoint > TreeSpecType;
-
     typedef core::tree::GeneralBinarySpaceTree <TreeSpecType> TreeType;
 
     typedef core::table::Table<TreeSpecType> TableType;
@@ -55,6 +54,28 @@ class DistributedTable: public boost::noncopyable {
 
   private:
 
+    void AssignLeafNodes_(
+      const core::metric_kernels::AbstractMetric &metric_in,
+      const std::vector<TreeType *> &top_leaf_nodes,
+      std::vector<int> *point_assignments) {
+
+      // Loop through each point and find the closest leaf node.
+      for(int i = 0; i < owned_table_->n_entries(); i++) {
+        core::table::DensePoint point;
+        owned_table_->get(i, &point);
+
+        // Loop through each leaf node.
+        for(unsigned int j = 0; j < top_leaf_nodes.size(); j++) {
+          const typename TreeType::BoundType &leaf_node_bound =
+            top_leaf_nodes[j]->bound();
+
+          // Compute the squared mid-distance.
+          double squared_mid_distance = leaf_node_bound.MidDistanceSq(
+                                          metric_in, point);
+        }
+      }
+    }
+
     void SelectSubset_(
       double sample_probability_in, std::vector<int> *sampled_indices_out) {
 
@@ -160,11 +181,11 @@ class DistributedTable: public boost::noncopyable {
       }
     }
 
-    const TreeType::BoundType &get_node_bound(TreeType * node) const {
+    const typename TreeType::BoundType &get_node_bound(TreeType * node) const {
       return node->bound();
     }
 
-    TreeType::BoundType &get_node_bound(TreeType * node) {
+    typename TreeType::BoundType &get_node_bound(TreeType * node) {
       return node->bound();
     }
 
@@ -180,7 +201,7 @@ class DistributedTable: public boost::noncopyable {
       return node->is_leaf();
     }
 
-    TreeSpecType::StatisticType &get_node_stat(TreeType * node) {
+    typename TreeSpecType::StatisticType &get_node_stat(TreeType * node) {
       return node->stat();
     }
 
@@ -299,7 +320,7 @@ class DistributedTable: public boost::noncopyable {
           metric_in, leaf_size, table_outbox_group_comm.size());
 
         // Broadcast the leaf nodes.
-        sampled_table_->get_leaf_nodes(
+        sampled_table.get_leaf_nodes(
           sampled_table.get_tree(), &top_leaf_nodes);
         boost::mpi::broadcast(table_outbox_group_comm, top_leaf_nodes, 0);
       }
@@ -309,8 +330,9 @@ class DistributedTable: public boost::noncopyable {
         boost::mpi::broadcast(table_outbox_group_comm, top_leaf_nodes, 0);
       }
 
-      // Assign each point to one of the leaf nodes, and do a
-      // prefix-sum style sending to re-distribute the data.
+      // Assign each point to one of the leaf nodes.
+      std::vector<int> point_assignments;
+      AssignLeafNodes_(metric_in, top_leaf_nodes, &point_assignments);
     }
 
     void get(

fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc

@@ -1,260 +0,0 @@
-/** @file distributed_table_test.cc
- *
- *  @author Dongryeol Lee (dongryel@cc.gatech.edu)
- */
-#include "core/metric_kernels/lmetric.h"
-#include "core/table/distributed_table.h"
-#include "core/table/mailbox.h"
-#include "core/tree/gen_kdtree.h"
-#include "mlpack/kde/kde_dualtree.h"
-#include <boost/archive/text_oarchive.hpp>
-#include <boost/archive/text_iarchive.hpp>
-
-typedef core::tree::GenKdTree< mlpack::kde::KdeStatistic > TreeSpecType;
-typedef core::tree::GeneralBinarySpaceTree < core::tree::GenKdTree > TreeType;
-typedef core::table::Table<TreeType> TableType;
-
-bool CheckDistributedTableIntegrity(
-  const core::table::DistributedTable &table_in,
-  const boost::mpi::communicator &world,
-  const boost::mpi::communicator &table_outbox_group,
-  const boost::mpi::communicator &table_inbox_group) {
-  for(int i = 0; i < world.size(); i++) {
-    printf(
-      "Process %d thinks Process %d owns %d points of dimensionality %d.\n",
-      world.rank(), i, table_in.local_n_entries(i % (world.size() / 3)),
-      table_in.n_attributes());
-  }
-  return true;
-}
-
-core::table::DistributedTable *InitDistributedTable(
-  boost::mpi::communicator &world,
-  boost::mpi::communicator &table_outbox_group) {
-
-  std::pair< core::table::DistributedTable *, std::size_t >
-  distributed_table_pair =
-    core::table::global_m_file_->UniqueFind<core::table::DistributedTable>();
-  core::table::DistributedTable *distributed_table =
-    distributed_table_pair.first;
-
-  if(distributed_table == NULL) {
-    printf("Process %d: TableOutbox.\n", world.rank());
-
-    // Each process generates its own random data, dumps it to the file,
-    // and read its own file back into its own distributed table.
-    core::table::Table<TreeType> random_dataset;
-    const int num_dimensions = 5;
-    int num_points = core::math::RandInt(10, 20);
-    random_dataset.Init(5, num_points);
-    for(int j = 0; j < num_points; j++) {
-      core::table::DensePoint point;
-      random_dataset.get(j, &point);
-      for(int i = 0; i < num_dimensions; i++) {
-        point[i] = core::math::Random(0.1, 1.0);
-      }
-    }
-    printf("Process %d generated %d points...\n", world.rank(), num_points);
-    std::stringstream file_name_sstr;
-    file_name_sstr << "random_dataset_" << table_outbox_group.rank() << ".csv";
-    std::string file_name = file_name_sstr.str();
-    random_dataset.Save(file_name);
-
-    std::stringstream distributed_table_name_sstr;
-    distributed_table_name_sstr << "distributed_table_" << world.rank() << "\n";
-    distributed_table = core::table::global_m_file_->UniqueConstruct <
-                        core::table::DistributedTable > ();
-    distributed_table->Init(
-      file_name, table_outbox_group);
-    printf(
-      "Process %d read in %d points...\n",
-      world.rank(), distributed_table->local_n_entries());
-  }
-  return distributed_table;
-}
-
-void TableOutboxProcess(
-  core::table::DistributedTable *distributed_table,
-  boost::mpi::communicator &world,
-  boost::mpi::intercommunicator &outbox_to_inbox_comm,
-  boost::mpi::intercommunicator &outbox_to_computation_comm) {
-
-  printf("Process %d: TableOutbox.\n", world.rank());
-  distributed_table->RunOutbox(
-    outbox_to_inbox_comm, outbox_to_computation_comm);
-}
-
-void TableInboxProcess(
-  core::table::DistributedTable *distributed_table,
-  boost::mpi::communicator &world,
-  boost::mpi::intercommunicator &inbox_to_outbox_comm,
-  boost::mpi::intercommunicator &inbox_to_computation_comm) {
-  printf("Process %d: TableInbox.\n", world.rank());
-
-  distributed_table->RunInbox(
-    inbox_to_outbox_comm, inbox_to_computation_comm);
-}
-
-void ComputationProcess(
-  core::table::DistributedTable *distributed_table,
-  boost::mpi::communicator &world,
-  boost::mpi::communicator &local_group_comm,
-  boost::mpi::intercommunicator &computation_to_outbox_comm,
-  boost::mpi::intercommunicator &computation_to_inbox_comm) {
-
-  printf("Process %d: Computation.\n", world.rank());
-
-  // Do a test where each computation process requests a random point
-  // from a randomly chosen process.
-  int num_points = core::math::RandInt(10, 30);
-  for(int n = 0; n < num_points; n++) {
-    core::table::DenseConstPoint point;
-    int random_request_rank = core::math::RandInt(
-                                0, computation_to_outbox_comm.remote_size());
-    int random_request_point_id =
-      core::math::RandInt(
-        0, distributed_table->local_n_entries(random_request_rank));
-    printf("Computation Process %d is requesting point %d from Table Outbox "
-           "Process %d\n",
-           local_group_comm.rank(), random_request_point_id,
-           random_request_rank);
-    distributed_table->get(
-      computation_to_outbox_comm, computation_to_inbox_comm,
-      random_request_rank, random_request_point_id, &point);
-
-    // Print the point.
-    point.Print();
-
-    // Tell the inbox that we are done using the point.
-    distributed_table->UnlockPointinTableInbox();
-  }
-
-  // Barrier so that all computation groups are here, at which outbox
-  // and inboxes are terminated.
-  printf("Notifying all mailboxes that Computation group %d is done!\n",
-         local_group_comm.rank());
-  for(int i = 0; i < computation_to_outbox_comm.remote_size(); i++) {
-    computation_to_outbox_comm.isend(
-      i, core::table::DistributedTableMessage::TERMINATE_TABLE_OUTBOX,
-      0);
-    computation_to_inbox_comm.isend(
-      i, core::table::DistributedTableMessage::TERMINATE_TABLE_INBOX,
-      0);
-  }
-}
-
-int main(int argc, char *argv[]) {
-
-  // Initialize boost MPI.
-  boost::mpi::environment env(argc, argv);
-  boost::mpi::communicator world;
-
-  if(world.size() <= 1 || world.size() % 3 != 0) {
-    std::cout << "Please specify a process number greater than 1 and "
-              "a multiple of 3.\n";
-    return 0;
-  }
-
-  // Delete the teporary files and put a barrier.
-  std::stringstream temporary_file_name;
-  temporary_file_name << "tmp_file" << world.rank();
-  remove(temporary_file_name.str().c_str());
-  world.barrier();
-
-  // Initialize the memory allocator.
-  int membership_key = world.rank() % 3;
-  core::table::global_m_file_ = new core::table::MemoryMappedFile();
-  core::table::global_m_file_->Init(
-    std::string("tmp_file"), world.rank(),
-    (int) floor(world.rank() / 3), 5000000);
-
-  // Seed the random number.
-  srand(time(NULL) + world.rank());
-
-  if(world.rank() == 0) {
-    printf("%d processes are present...\n", world.size());
-  }
-
-  // Split the world communicator into three groups: the first group
-  // that sends stuffs to other processes, the second group that
-  // receives stuffs from other processes, and the third group that
-  // does the computation.
-  boost::mpi::communicator local_group_comm = world.split(membership_key);
-
-  // Build the intercommunicator between the table outbox group and
-  // the table inbox group and the computation group.
-  boost::mpi::intercommunicator *first_inter_comm = NULL;
-  boost::mpi::intercommunicator *second_inter_comm = NULL;
-  if(membership_key == 0) {
-    first_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 1);
-    second_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 2);
-  }
-  else if(membership_key == 1) {
-    first_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 0);
-    second_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 2);
-  }
-  else {
-    first_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 0);
-    second_inter_comm = new boost::mpi::intercommunicator(
-      local_group_comm, 0, world, 1);
-  }
-
-  printf("Rank: %d %d %d %d %d\n", world.rank(),
-         first_inter_comm->local_size(), first_inter_comm->remote_size(),
-         second_inter_comm->local_size(), second_inter_comm->remote_size());
-
-  // Declare the distributed table.
-  core::table::DistributedTable *distributed_table = NULL;
-
-  // Wait until the memory allocator is in synch.
-  world.barrier();
-
-  // Read the distributed table once per each compute node, and put a
-  // barrier.
-  if(membership_key == 0) {
-    distributed_table =
-      InitDistributedTable(world, local_group_comm);
-  }
-  world.barrier();
-
-  // Attach the distributed table for all the processes and put a
-  // barrier.
-  std::pair< core::table::DistributedTable *, std::size_t >
-  distributed_table_pair =
-    core::table::global_m_file_->UniqueFind<core::table::DistributedTable>();
-  distributed_table = distributed_table_pair.first;
-
-  // Check the integrity of the distributed table.
-  if(membership_key == 0) {
-    CheckDistributedTableIntegrity(
-      *distributed_table, world,
-      *first_inter_comm, *second_inter_comm);
-  }
-
-  // The main computation loop.
-  if(membership_key == 0) {
-    TableOutboxProcess(
-      distributed_table, world, *first_inter_comm, *second_inter_comm);
-  }
-  else if(membership_key == 1) {
-    TableInboxProcess(
-      distributed_table, world, *first_inter_comm, *second_inter_comm);
-  }
-  else {
-    ComputationProcess(
-      distributed_table, world, local_group_comm,
-      *first_inter_comm, *second_inter_comm);
-  }
-
-  // Free the intercommunicators.
-  world.barrier();
-  delete first_inter_comm;
-  delete second_inter_comm;
-
-  return 0;
-}

fastlib/trunk/contrib/dongryel/thesis_research/core/table/table.h

@@ -7,7 +7,8 @@
 #define CORE_TABLE_TABLE_H
 
 #include <armadillo>
-#include "boost/utility.hpp"
+#include <boost/interprocess/offset_ptr.hpp>
+#include <boost/utility.hpp>
 #include "core/csv_parser/dataset_reader.h"
 #include "core/metric_kernels/abstract_metric.h"
 #include "core/tree/general_spacetree.h"
@@ -136,7 +137,7 @@ class Table: public boost::noncopyable {
 
     std::vector<int> new_from_old_;
 
-    TreeType *tree_;
+    boost::interprocess::offset_ptr<TreeType> tree_;
 
   public:
 
@@ -155,10 +156,10 @@ class Table: public boost::noncopyable {
     ~Table() {
       if(tree_) {
         if(core::table::global_m_file_) {
-          RecursiveDeallocate_(tree_);
+          RecursiveDeallocate_(tree_.get());
         }
         else {
-          delete tree_;
+          delete tree_.get();
         }
       }
       tree_ = NULL;
@@ -201,7 +202,7 @@ class Table: public boost::noncopyable {
     }
 
     TreeType *get_tree() {
-      return tree_;
+      return tree_.get();
     }
 
     void get_leaf_nodes(

fastlib/trunk/contrib/dongryel/thesis_research/core/tree/ball_bound.h

@@ -224,6 +224,12 @@ class BallBound {
       const core::table::DensePoint& point) const {
       return metric.Distance(center_, point);
     }
+
+    double MidDistanceSq(
+      const core::metric_kernels::AbstractMetric &metric,
+      const core::table::DensePoint& point) const {
+      return metric.DistanceSq(center_, point);
+    }
 };
 };
 };

fastlib/trunk/contrib/dongryel/thesis_research/core/tree/tree.test.cc

@@ -28,10 +28,10 @@ class TestTree {
       typename TableType::TreeIterator node_it =
         table.get_node_iterator(node);
       do {
-        core::table::DenseConstPoint point;
+        core::table::DensePoint point;
         int point_id;
         node_it.Next(&point, &point_id);
-        core::table::DenseConstPoint compare_point;
+        core::table::DensePoint compare_point;
         table.get(point_id, &compare_point);
 
         for(int i = 0; i < point.length(); i++) {
@@ -106,8 +106,8 @@ class TestTree {
       core::metric_kernels::LMetric<2> l2_metric;
       reordered_table.IndexData(l2_metric, 20);
       for(int i = 0; i < reordered_table.n_entries(); i++) {
-        core::table::DenseConstPoint reordered_point;
-        core::table::DenseConstPoint original_point;
+        core::table::DensePoint reordered_point;
+        core::table::DensePoint original_point;
         reordered_table.get(i, &reordered_point);
         original_table.get(i, &original_point);
         for(int j = 0; j < reordered_table.n_attributes(); j++) {
@@ -131,7 +131,7 @@ BOOST_AUTO_TEST_CASE(TestCaseKde) {
 
   // Tree type: hard-coded for a metric tree.
   typedef core::table::Table <
-  core::tree::GenMetricTree<core::table::DensePoint> > TableType;
+  core::tree::GenMetricTree<core::tree::AbstractStatistic> > TableType;
 
   // Call the tests.
   core::tree::TestTree<TableType> tree_test;

fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/distributed_kde.cc

@@ -9,20 +9,22 @@
 #include "core/table/distributed_table.h"
 #include "core/table/mailbox.h"
 #include "core/tree/gen_kdtree.h"
+#include "core/tree/gen_metric_tree.h"
 #include "mlpack/kde/kde_dualtree.h"
 
-typedef core::tree::GenKdTree<mlpack::kde::KdeStatistic> TreeSpecType;
+typedef core::tree::GenMetricTree<mlpack::kde::KdeStatistic> TreeSpecType;
 typedef core::tree::GeneralBinarySpaceTree < TreeSpecType > TreeType;
 typedef core::table::Table<TreeType> TableType;
 
-core::table::DistributedTable *InitDistributedTable(
+core::table::DistributedTable<TreeSpecType> *InitDistributedTable(
   boost::mpi::communicator &world,
   boost::mpi::communicator &table_outbox_group) {
 
-  std::pair< core::table::DistributedTable *, std::size_t >
+  std::pair< core::table::DistributedTable<TreeSpecType> *, std::size_t >
   distributed_table_pair =
-    core::table::global_m_file_->UniqueFind<core::table::DistributedTable>();
-  core::table::DistributedTable *distributed_table =
+    core::table::global_m_file_->UniqueFind <
+    core::table::DistributedTable<TreeSpecType> > ();
+  core::table::DistributedTable<TreeSpecType> *distributed_table =
     distributed_table_pair.first;
 
   if(distributed_table == NULL) {
@@ -50,7 +52,7 @@ core::table::DistributedTable *InitDistributedTable(
     std::stringstream distributed_table_name_sstr;
     distributed_table_name_sstr << "distributed_table_" << world.rank() << "\n";
     distributed_table = core::table::global_m_file_->UniqueConstruct <
-                        core::table::DistributedTable > ();
+                        core::table::DistributedTable<TreeSpecType> > ();
     distributed_table->Init(
       file_name, table_outbox_group);
     printf(
@@ -61,7 +63,7 @@ core::table::DistributedTable *InitDistributedTable(
 }
 
 void TableOutboxProcess(
-  core::table::DistributedTable *distributed_table,
+  core::table::DistributedTable<TreeSpecType> *distributed_table,
   boost::mpi::communicator &world,
   boost::mpi::intercommunicator &outbox_to_inbox_comm,
   boost::mpi::intercommunicator &outbox_to_computation_comm) {
@@ -72,7 +74,7 @@ void TableOutboxProcess(
 }
 
 void TableInboxProcess(
-  core::table::DistributedTable *distributed_table,
+  core::table::DistributedTable<TreeSpecType> *distributed_table,
   boost::mpi::communicator &world,
   boost::mpi::intercommunicator &inbox_to_outbox_comm,
   boost::mpi::intercommunicator &inbox_to_computation_comm) {
@@ -83,7 +85,7 @@ void TableInboxProcess(
 }
 
 void ComputationProcess(
-  core::table::DistributedTable *distributed_table,
+  core::table::DistributedTable<TreeSpecType> *distributed_table,
   boost::mpi::communicator &world,
   boost::mpi::communicator &local_group_comm,
   boost::mpi::intercommunicator &computation_to_outbox_comm,
@@ -192,7 +194,7 @@ int main(int argc, char *argv[]) {
   }
 
   // Declare the distributed table.
-  core::table::DistributedTable *distributed_table = NULL;
+  core::table::DistributedTable<TreeSpecType> *distributed_table = NULL;
 
   // Wait until the memory allocator is in synch.
   world.barrier();
@@ -207,9 +209,10 @@ int main(int argc, char *argv[]) {
 
   // Attach the distributed table for all the processes and put a
   // barrier.
-  std::pair< core::table::DistributedTable *, std::size_t >
+  std::pair< core::table::DistributedTable<TreeSpecType> *, std::size_t >
   distributed_table_pair =
-    core::table::global_m_file_->UniqueFind<core::table::DistributedTable>();
+    core::table::global_m_file_->UniqueFind <
+    core::table::DistributedTable<TreeSpecType> > ();
   distributed_table = distributed_table_pair.first;
 
   // The main computation loop.