diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h index 6ec29f1070..a17d68cbe5 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h +++ b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h @@ -26,12 +26,11 @@ namespace table { extern MemoryMappedFile *global_m_file_; +template class DistributedTable: public boost::noncopyable { public: - typedef core::tree::GenMetricTree < core::table::DensePoint > TreeSpecType; - typedef core::tree::GeneralBinarySpaceTree TreeType; typedef core::table::Table TableType; @@ -55,6 +54,28 @@ class DistributedTable: public boost::noncopyable { private: + void AssignLeafNodes_( + const core::metric_kernels::AbstractMetric &metric_in, + const std::vector &top_leaf_nodes, + std::vector *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 *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 point_assignments; + AssignLeafNodes_(metric_in, top_leaf_nodes, &point_assignments); } void get( diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc deleted file mode 100644 index 72487285ab..0000000000 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc +++ /dev/null @@ -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 -#include - -typedef core::tree::GenKdTree< mlpack::kde::KdeStatistic > TreeSpecType; -typedef core::tree::GeneralBinarySpaceTree < core::tree::GenKdTree > TreeType; -typedef core::table::Table 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 *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 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(); - 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; -} diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/table.h b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/table.h index 8153e06611..9c8b3320bc 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/table.h +++ b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/table.h @@ -7,7 +7,8 @@ #define CORE_TABLE_TABLE_H #include -#include "boost/utility.hpp" +#include +#include #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 new_from_old_; - TreeType *tree_; + boost::interprocess::offset_ptr 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( diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/ball_bound.h b/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/ball_bound.h index ea22b15e38..94e4072617 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/ball_bound.h +++ b/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); + } }; }; }; diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/tree.test.cc b/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/tree.test.cc index ccd4ade3c4..1b7c2fc5fd 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/tree/tree.test.cc +++ b/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 > TableType; + core::tree::GenMetricTree > TableType; // Call the tests. core::tree::TestTree tree_test; diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/distributed_kde.cc b/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/distributed_kde.cc index 89a69c7d83..4e9db1826f 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/distributed_kde.cc +++ b/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 TreeSpecType; +typedef core::tree::GenMetricTree TreeSpecType; typedef core::tree::GeneralBinarySpaceTree < TreeSpecType > TreeType; typedef core::table::Table TableType; -core::table::DistributedTable *InitDistributedTable( +core::table::DistributedTable *InitDistributedTable( boost::mpi::communicator &world, boost::mpi::communicator &table_outbox_group) { - std::pair< core::table::DistributedTable *, std::size_t > + std::pair< core::table::DistributedTable *, std::size_t > distributed_table_pair = - core::table::global_m_file_->UniqueFind(); - core::table::DistributedTable *distributed_table = + core::table::global_m_file_->UniqueFind < + core::table::DistributedTable > (); + core::table::DistributedTable *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 > (); 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 *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 *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 *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 *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 *, std::size_t > distributed_table_pair = - core::table::global_m_file_->UniqueFind(); + core::table::global_m_file_->UniqueFind < + core::table::DistributedTable > (); distributed_table = distributed_table_pair.first; // The main computation loop.