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 49018b8d5e..dc80849e03 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 @@ -213,8 +213,33 @@ class DistributedTable: public boost::noncopyable { void IndexData( const core::metric_kernels::AbstractMetric & metric_in, + boost::mpi::communicator &table_outbox_group_comm, double sample_probability_in) { + // Each process generates a random subset of the data points to + // send to the master. This is a MPI gather operation. + TableType sampled_table; + std::vector sampled_indices; + for(int i = 0; i < owned_table_->n_entries(); i++) { + if(core::math::Random() <= sample_probability_in) { + sampled_indices.push_back(i); + } + } + // Send the number of points chosen in this process to the + // master so that the master can allocate the appropriate amount + // of space to receive all the points. + int total_num_samples = 0; + if(table_outbox_group_comm.rank() == 0) { + boost::mpi::reduce( + table_outbox_group_comm, sampled_indices.size(), + boost::mpi::sum(), 0); + } + else { + boost::mpi::reduce( + table_outbox_group_comm, sampled_indices(), + boost::mpi::sum(), 0); + } + } diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/CMakeLists.txt b/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/CMakeLists.txt index cfcf7c6235..f298b7e637 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/CMakeLists.txt +++ b/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/CMakeLists.txt @@ -1,6 +1,6 @@ cmake_minimum_required(VERSION 2.8) -# the test for distributed_kde.test.cc +# the test for distributed_kde add_executable(distributed_kde EXCLUDE_FROM_ALL distributed_kde.cc @@ -38,16 +38,3 @@ target_link_libraries(prefix_scan ${TRILINOS_LIBS} ${Boost_LIBRARIES} ) - -# the main driver for parallel_kde -add_executable(parallel_kde - EXCLUDE_FROM_ALL - parallel_kde.cc - ) - -# link dependencies of parallel_kde -target_link_libraries(parallel_kde - core - ${TRILINOS_LIBS} - ${Boost_LIBRARIES} - ) diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/parallel_kde.cc b/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/parallel_kde.cc deleted file mode 100644 index 31e0a397f3..0000000000 --- a/fastlib/trunk/contrib/dongryel/thesis_research/mlpack/distributed_kde/parallel_kde.cc +++ /dev/null @@ -1,110 +0,0 @@ -#include -#include -#include -#include "boost/mpi.hpp" -#include "boost/serialization/string.hpp" -#include "core/math/math_lib.h" -#include -#include - -class Particle { - private: - std::vector position_; - - friend class boost::serialization::access; - - public: - - template - void serialize(Archive &ar, const unsigned int version) { - ar & position_; - } - - const std::vector &position() const { - return position_; - } - - Particle() { - position_.resize(3); - for(int i = 0; i < position_.size(); i++) { - position_[i] = 0; - } - } - - void AddParticle(const Particle &particle_in) { - for(unsigned int i = 0; i < 3; i++) { - position_[i] += particle_in.position()[i]; - } - } - - void Randomize() { - for(unsigned int i = 0; i < 3; i++) { - position_[i] = core::math::RandInt(0, 3); - } - } - - void Print() const { - std::cout << position_[0] << " " << position_[1] << - " " << position_[2] << "\n"; - } -}; - -int main(int argc, char *argv[]) { - boost::mpi::environment env(argc, argv); - boost::mpi::communicator world; - - if(world.size() <= 1) { - std::cout << "Please specify a process number greater than 1.\n"; - exit(0); - } - - Particle generated_particle; - - // Broadcast the skeleton from the master to the slaves. - broadcast(world, boost::mpi::skeleton(generated_particle), 0); - - boost::mpi::content c = boost::mpi::get_content(generated_particle); - - // The master process. - if(world.rank() == 0) { - std::vector num_messages(world.size(), 0); - for(int i = 1; i < world.size(); i++) { - num_messages[i] = core::math::RandInt(3, 8); - - // Send the number of messages to receive per each processor. - world.isend(i, 0, num_messages[i]); - printf("Processor 0 is sending processor %d %d message.\n", i, - num_messages[i]); - } - for(int i = 1; i < world.size(); i++) { - for(int j = 0; j < num_messages[i]; j++) { - generated_particle.Randomize(); - std::cout << "Master generated and sending to " << i << ": "; - generated_particle.Print(); - world.isend(i, 1, c); - } - } - } - - // The slave process. - else { - int num_messages_to_receive; - boost::mpi::request main_request = - world.irecv(0, 0, num_messages_to_receive); - main_request.wait(); - - Particle received_particle; - for(int j = 0; j < num_messages_to_receive; j++) { - // Receive from the master. - boost::mpi::request receive_request = world.irecv(0, 1, c); - receive_request.wait(); - std::cout << "Process " << world.rank() << " received: "; - generated_particle.Print(); - received_particle.AddParticle(generated_particle); - } - std::cout << "Process " << world.rank() << " got: "; - received_particle.Print(); - } - - return 0; -}