123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196 |
- // Copyright (C) 2005-2006 Matthias Troyer
- // Use, modification and distribution is subject to the Boost Software
- // License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
- // http://www.boost.org/LICENSE_1_0.txt)
- // An example of a parallel Monte Carlo simulation using some nodes to produce
- // data and others to aggregate the data
- #include <iostream>
- #include <boost/mpi.hpp>
- #include <boost/random/parallel.hpp>
- #include <boost/random.hpp>
- #include <boost/foreach.hpp>
- #include <iostream>
- #include <cstdlib>
- namespace mpi = boost::mpi;
- enum {sample_tag, sample_skeleton_tag, sample_broadcast_tag, quit_tag};
- void calculate_samples(int sample_length)
- {
- int num_samples = 100;
- std::vector<double> sample(sample_length);
- // setup communicator by splitting
- mpi::communicator world;
- mpi::communicator calculate_communicator = world.split(0);
- unsigned int num_calculate_ranks = calculate_communicator.size();
-
- // the master of the accumulaion ranks is the first of them, hence
- // with a rank just one after the last calculation rank
- int master_accumulate_rank = num_calculate_ranks;
-
- // the master of the calculation ranks sends the skeleton of the sample
- // to the master of the accumulation ranks
- if (world.rank()==0)
- world.send(master_accumulate_rank,sample_skeleton_tag,mpi::skeleton(sample));
-
- // next we extract the content of the sample vector, to be used in sending
- // the content later on
-
- mpi::content sample_content = mpi::get_content(sample);
-
- // now intialize the parallel random number generator
-
- boost::lcg64 engine(
- boost::random::stream_number = calculate_communicator.rank(),
- boost::random::total_streams = calculate_communicator.size()
- );
-
- boost::variate_generator<boost::lcg64&,boost::uniform_real<> >
- rng(engine,boost::uniform_real<>());
-
- for (unsigned int i=0; i<num_samples/num_calculate_ranks+1;++i) {
-
- // calculate sample by filling the vector with random numbers
- // note that std::generate will not work since it takes the generator
- // by value, and boost::ref cannot be used as a generator.
- // boost::ref should be fixed so that it can be used as generator
-
- BOOST_FOREACH(double& x, sample)
- x = rng();
-
- // send sample to accumulation ranks
- // Ideally we want to do this as a broadcast with an inter-communicator
- // between the calculation and accumulation ranks. MPI2 should support
- // this, but here we present an MPI1 compatible solution.
-
- // send content of sample to first (master) accumulation process
-
- world.send(master_accumulate_rank,sample_tag,sample_content);
-
- // gather some results from all calculation ranks
-
- double local_result = sample[0];
- std::vector<double> gathered_results(calculate_communicator.size());
- mpi::all_gather(calculate_communicator,local_result,gathered_results);
- }
-
- // we are done: the master tells the accumulation ranks to quit
- if (world.rank()==0)
- world.send(master_accumulate_rank,quit_tag);
- }
- void accumulate_samples()
- {
- std::vector<double> sample;
- // setup the communicator for all accumulation ranks by splitting
- mpi::communicator world;
- mpi::communicator accumulate_communicator = world.split(1);
- bool is_master_accumulate_rank = accumulate_communicator.rank()==0;
- // the master receives the sample skeleton
-
- if (is_master_accumulate_rank)
- world.recv(0,sample_skeleton_tag,mpi::skeleton(sample));
-
- // and broadcasts it to all accumulation ranks
- mpi::broadcast(accumulate_communicator,mpi::skeleton(sample),0);
-
- // next we extract the content of the sample vector, to be used in receiving
- // the content later on
-
- mpi::content sample_content = mpi::get_content(sample);
-
- // accumulate until quit is called
- double sum=0.;
- while (true) {
-
-
- // the accumulation master checks whether we should quit
- if (world.iprobe(0,quit_tag)) {
- world.recv(0,quit_tag);
- for (int i=1; i<accumulate_communicator.size();++i)
- accumulate_communicator.send(i,quit_tag);
- std::cout << sum << "\n";
- break; // We're done
- }
- // the otehr accumulation ranks check whether we should quit
- if (accumulate_communicator.iprobe(0,quit_tag)) {
- accumulate_communicator.recv(0,quit_tag);
- std::cout << sum << "\n";
- break; // We're done
- }
-
- // check whether the master accumulation rank has received a sample
- if (world.iprobe(mpi::any_source,sample_tag)) {
- BOOST_ASSERT(is_master_accumulate_rank);
-
- // receive the content
- world.recv(mpi::any_source,sample_tag,sample_content);
-
- // now we need to braodcast
- // the problam is we do not have a non-blocking broadcast that we could
- // abort if we receive a quit message from the master. We thus need to
- // first tell all accumulation ranks to start a broadcast. If the sample
- // is small, we could just send the sample in this message, but here we
- // optimize the code for large samples, so that the overhead of these
- // sends can be ignored, and we count on an optimized broadcast
- // implementation with O(log N) complexity
- for (int i=1; i<accumulate_communicator.size();++i)
- accumulate_communicator.send(i,sample_broadcast_tag);
-
- // now broadcast the contents of the sample to all accumulate ranks
- mpi::broadcast(accumulate_communicator,sample_content,0);
-
- // and handle the sample by summing the appropriate value
- sum += sample[0];
- }
-
- // the other accumulation ranks wait for a mesage to start the broadcast
- if (accumulate_communicator.iprobe(0,sample_broadcast_tag)) {
- BOOST_ASSERT(!is_master_accumulate_rank);
-
- accumulate_communicator.recv(0,sample_broadcast_tag);
-
- // receive broadcast of the sample contents
- mpi::broadcast(accumulate_communicator,sample_content,0);
-
- // and handle the sample
-
- // and handle the sample by summing the appropriate value
- sum += sample[accumulate_communicator.rank()];
- }
- }
- }
- int main(int argc, char** argv)
- {
- mpi::environment env(argc, argv);
- mpi::communicator world;
- // half of the processes generate, the others accumulate
- // the sample size is just the number of accumulation ranks
- if (world.rank() < world.size()/2)
- calculate_samples(world.size()-world.size()/2);
- else
- accumulate_samples();
- return 0;
- }
|