void
all_gatherv_test(const mpi::communicator& comm, Generator generator,
                 std::string kind)
{
  typedef typename Generator::result_type value_type;
  using boost::mpi::all_gatherv;

  std::vector<value_type> myvalues, expected, values;
  std::vector<int>        sizes;
  for(int r = 0; r < comm.size(); ++r) {
    value_type value = generator(r);
    sizes.push_back(r+1);
    for (int k=0; k < r+1; ++k) {
      expected.push_back(value);
      if(comm.rank() == r) {
        myvalues.push_back(value);
      }
    }
  }
  if (comm.rank() == 0) {
    std::cout << "Gathering " << kind << "...";
    std::cout.flush();
  }
  
  mpi::all_gatherv(comm, myvalues, values, sizes);
  
  BOOST_CHECK(values == expected);
  
  if (comm.rank() == 0 && values == expected)
    std::cout << "OK." << std::endl;
  
  (comm.barrier)();
}

std::string run_model(Properties& props, boost::mpi::communicator& world) {

	try {
		repast::relogo::SimulationRunner runner(&world);

		if (world.rank() == 0) {
			std::string time;
			repast::timestamp(time);
			std::cout << "Start Time: " << time << std::endl;
		}

		repast::Timer timer;
		timer.start();
		runner.run<ZombieObserver, repast::relogo::Patch>(props);

		if (world.rank() == 0) {
			std::string time;
			repast::timestamp(time);
			std::cout << "End Time: " << time << "\nElapsed Time: "
					<< timer.stop() << std::endl;
		}

	} catch (std::exception& exp) {
		// catch any exception (e.g. if data files couldn't be opened) and
		// print out the errors.
		std::cerr << "ERROR: " << exp.what() << std::endl;
	}
	return props.getProperty(OUTPUT_KEY);
}

void TwoParticleGF::compute(bool clear, const boost::mpi::communicator & comm)
{
    if (Status < Prepared) throw (exStatusMismatch());
    if (Status >= Computed) return;
    if (!Vanishing) {
        // Create a "skeleton" class with pointers to part that can call a compute method
        pMPI::mpi_skel<ComputeAndClearWrap> skel;
        bool fill_container = m_data_.NBosonic() > 0 && m_data_.NFermionic() > 0; 
        skel.parts.reserve(parts.size());
        for (size_t i=0; i<parts.size(); i++) { 
            skel.parts.push_back(ComputeAndClearWrap(&m_data_, parts[i], clear, fill_container, 1));
            };
        std::map<pMPI::JobId, pMPI::WorkerId> job_map = skel.run(comm, true); // actual running - very costly
        int rank = comm.rank();
        int comm_size = comm.size(); 

        // Start distributing data
        //DEBUG(comm.rank() << getIndex(0) << getIndex(1) << getIndex(2) << getIndex(3) << " Start distributing data");
        comm.barrier();
         

        if (!clear) { 
            for (size_t p = 0; p<parts.size(); p++) {
                boost::mpi::broadcast(comm, parts[p]->NonResonantTerms, job_map[p]);
                boost::mpi::broadcast(comm, parts[p]->ResonantTerms, job_map[p]);
                if (rank == job_map[p]) { 
                    parts[p]->Status = TwoParticleGFPart::Computed;
                     };
                };
            comm.barrier();
        }
    };
    Status = Computed;
}

void runMaster(mpi::communicator world, int size, int grid_dimension)
{
  // Start timer and go.
  boost::chrono::system_clock::time_point start = boost::chrono::system_clock::now();

  // Send
  Matrix      A(Size(size, size));
  Matrix result(Size(size, size));

  for(int row = 0; row < A.size.rows; ++row){
    for(int col = 0; col < A.size.cols; ++col){
      A.data[row][col] = (row % 11) + (col % 11);
    }
  }
  //cout << A << endl;
  //cout << "\nProduct:\n" << A*A << endl;

  // Do sequential
  if (grid_dimension == 0)
    A.square(result);

  // Else parallel
  else{
    // Split matrix up and send to slaves
    int slave_id = 1;
    int sub_matrix_sizes = size / grid_dimension;

    for(int i = 0; i < size; i += sub_matrix_sizes){
      for(int j = 0; j < size; j += sub_matrix_sizes){
        MatrixCrossSection cs = getCrossSection( A, i, j, sub_matrix_sizes);
        world.send(slave_id, 0, cs);
        slave_id ++;
      }
    }

    // Recieve
    std::vector<Matrix> saved;
    int num_slaves = world.size() -1;

    for(int i = 1; i <= num_slaves; ++i){
      Matrix r;
      world.recv(i, 0, r);
      result.insertSubMatrix(r);
    }
  }

  // Done
  boost::chrono::duration<double> sec = boost::chrono::system_clock::now() - start;
  cout << sec.count() << endl;

  // Print Result
  //cout << "\nResult:\n" << result << endl;

  //assert ( result == A*A);
}

void runSlave(mpi::communicator world)
{
  // Recieve
  MatrixCrossSection cs;
  world.recv(0, 0, cs);

  Matrix subMatrix(Size(cs.row_data.size(), cs.row_data.size()));
  cs.calculateVectorProduct(subMatrix);

  world.send(0, 0, subMatrix);
}

void
printit(const boost::mpi::communicator& comm,
        const std::vector<T>& things,
        const std::string& caption)
{
    if (!caption.empty() && comm.rank() == 0) {
        std::cout << caption << std::endl;
        std::cout.flush();
    }
    for (int p = 0; p < comm.size(); ++p) {
        if (comm.rank() == p) {
            std::cout << p << ": ";
            std::copy(things.begin(), things.end(),
                      std::ostream_iterator<T>(std::cout, ","));
            std::cout << std::endl;
            std::cout.flush();
        }
        comm.barrier();
    }

    comm.barrier();
    size_t global_size;
    boost::mpi::reduce(comm, things.size(), global_size, std::plus<size_t>(), 0);
    if (comm.rank() == 0) {
        if (!caption.empty()) {
            std::cout << caption;
        }
        std::cout << "Number of things: " << global_size << std::endl;
    }
    comm.barrier();
    std::cout << comm.rank() << ": leaving printit()" << std::endl;
}

static void slave(mpi::communicator world) {

  int work;
  int result;

  while (1) {

    mpi::status status = world.recv(mpi::any_source,mpi::any_tag,work);
    if (status.tag() == DIETAG) {
      return;
    }
      do_work(work,result);
      world.send(0,0,result);
  }
}

void bi::MarginalSISHandler<B,A,S>::handleAdapterSamples(
    boost::mpi::communicator child, boost::mpi::status status) {
  typedef typename temp_host_matrix<real>::type matrix_type;

  static const int N = B::NP;

  /* add samples */
  boost::optional<int> n = status.template count<real>();
  if (n) {
    matrix_type Z(N + T, *n / (N + T));
    child.recv(status.source(), status.tag(), Z.buf(), *n);

    for (int j = 0; j < Z.size2(); ++j) {
      adapter.add(subrange(column(Z,j), 0, N), subrange(column(Z,j), N, T));
    }
  }

  /* send new proposal if necessary */
  if (adapter.stop(t)) {
    adapter.adapt(t);
    BOOST_AUTO(q, adapter.get(t));
    BOOST_AUTO(iter, node.children.begin());
    for (; iter != node.children.end(); ++iter) {
      node.requests.push_front(iter->isend(0, MPI_TAG_ADAPTER_PROPOSAL, q));
    }
    ///@todo Serialize q into archive just once, then send to all. This may
    ///be how broadcast is already implemented in Boost.MPI.
  }
}

void
broadcast_test(const mpi::communicator& comm, const T& bc_value,
               std::string const& kind)
{
  for (int root = 0; root < comm.size(); ++root) {
    broadcast_test(comm, bc_value, kind, root);
  }
}

    void init(const mpi::communicator& comm)
    {
        PetscErrorCode ierr;
        PetscInt lo, hi;
        ierr = VecCreate(comm,&x); // CHKERRQ(ierr);
        ierr = VecSetSizes(x, PETSC_DECIDE, 5*comm.size()); // CHKERRQ(ierr);
        ierr = VecSetFromOptions(x); // CHKERRQ(ierr);

        ierr = VecGetOwnershipRange(x, &lo, &hi);
        for (PetscInt i = lo; i <= hi; ++i) {
            std::complex<double> v(i, 5*comm.size() - i - 1);
            ierr = VecSetValue(x, i, v, INSERT_VALUES);
        }

        ierr = VecAssemblyBegin(x); // CHKERRQ(ierr);
        ierr = VecAssemblyEnd(x); // CHKERRQ(ierr);
    }

void DocumentDecoder::manageTranslators(
	boost::mpi::communicator comm,
	NistXmlCorpus &testset
) {
	namespace mpi = boost::mpi;

	mpi::request reqs[2];
	int stopped = 0;

	NumberedOutputDocument translation;
	reqs[0] = comm.irecv(mpi::any_source, TAG_COLLECT, translation);
	reqs[1] = comm.irecv(mpi::any_source, TAG_STOP_COLLECTING);

	NistXmlCorpus::const_iterator it = testset.begin();
	uint docno = 0;
	for(int i = 0; i < comm.size() && it != testset.end(); ++i, ++docno, ++it) {
		LOG(logger_, debug, "S: Sending document " << docno << " to translator " << i);
		comm.send(i, TAG_TRANSLATE, std::make_pair(docno, *(*it)->asMMAXDocument()));
	}

	for(;;) {
		std::pair<mpi::status, mpi::request *> wstat = mpi::wait_any(reqs, reqs + 2);
		if(wstat.first.tag() == TAG_STOP_COLLECTING) {
			stopped++;
			LOG(logger_, debug, "C: Received STOP_COLLECTING from translator "
				<< wstat.first.source() << ", now " << stopped << " stopped translators.");
			if(stopped == comm.size()) {
				reqs[0].cancel();
				return;
			}
			*wstat.second = comm.irecv(mpi::any_source, TAG_STOP_COLLECTING);
		} else {
			LOG(logger_, debug, "C: Received translation of document " <<
				translation.first << " from translator " << wstat.first.source());
			reqs[0] = comm.irecv(mpi::any_source, TAG_COLLECT, translation);
			if(it != testset.end()) {
				LOG(logger_, debug, "S: Sending document " << docno <<
					" to translator " << wstat.first.source());
				comm.send(wstat.first.source(), TAG_TRANSLATE,
					std::make_pair(docno, *(*it)->asMMAXDocument()));
				++docno; ++it;
			} else {
				LOG(logger_, debug,
					"S: Sending STOP_TRANSLATING to translator " << wstat.first.source());
				comm.send(wstat.first.source(), TAG_STOP_TRANSLATING);
			}
			testset[translation.first]->setTranslation(translation.second);
		}
	}
}

void reduce_and_check(const boost::mpi::communicator &comm, bool local_value) {
  if(comm.rank() == 0) {
    bool total;
    boost::mpi::reduce(comm, local_value, total, std::logical_and<bool>(), 0);
    BOOST_CHECK(total);
  } else {
    boost::mpi::reduce(comm, local_value, std::logical_and<bool>(), 0);
  }
}

void 
random_scattered_vector(const boost::mpi::communicator& comm, 
                        const int& global_size,
                        std::vector<I>& local_values)
{
  int me(comm.rank());
  int nproc(comm.size());

  std::vector< std::vector<I> > toscatter(nproc);

  if (me == 0) {
    for (int i = 0; i < global_size; ++i) {
      boost::random::uniform_int_distribution<> dist(0, nproc-1);
      int p(dist(gen));
      toscatter[p].push_back(i);
    }
  }
  scatter(comm, toscatter, local_values, 0);
}

void mpi_send_workaround(int dest, int tag, const T& value,
                         boost::mpi::communicator & comm)
{
    // serialize T into a string
    std::ostringstream oss;
    boost::archive::text_oarchive oa(oss);
    oa << value;
    // send the string
    comm.send(dest, tag, oss.str());
}

void
report_features(mpi::communicator const& comm) {
  if (comm.rank() == 0) {
    std::cout << "Assuming working MPI_Improbe:" <<
#if defined(BOOST_MPI_USE_IMPROBE)
      "yes" << '\n';
#else
      "no"  << '\n';
#endif
  }
}

    /// Reduce the results of the measures, and reports some statistics
    void collect_results(boost::mpi::communicator const & c) {

     uint64_t nmeasures_tot;
     MCSignType sum_sign_tot;
     boost::mpi::reduce(c, nmeasures, nmeasures_tot, std::plus<uint64_t>(), 0);
     boost::mpi::reduce(c, sum_sign, sum_sign_tot, std::plus<MCSignType>(), 0);

     report(3) << "[Node "<<c.rank()<<"] Acceptance rate for all moves:\n" << AllMoves.get_statistics(c);
     report(3) << "[Node "<<c.rank()<<"] Simulation lasted: " << double(Timer) << " seconds" << std::endl;
     report(3) << "[Node "<<c.rank()<<"] Number of measures: " << nmeasures  << std::endl;
     report(3) << "[Node "<<c.rank()<<"] Average sign: " << sum_sign / double(nmeasures) << std::endl << std::endl << std::flush;
  
     if (c.rank()==0) {
      sign_av = sum_sign_tot / double(nmeasures_tot);
      report(2) << "Total number of measures: " << nmeasures_tot << std::endl;
      report(2) << "Average sign: " << sign_av << std::endl << std::endl << std::flush;
     }
     boost::mpi::broadcast(c, sign_av, 0);
     AllMeasures.collect_results(c);

    }

boost::mpi::status mpi_recv_workaround(int source, int tag, T& value,
                                       boost::mpi::communicator & comm)
{
    // receive a string
    std::string s;
    boost::mpi::status st = comm.recv(source, tag, s);
    // serialize into T
    std::istringstream iss(s);
    boost::archive::text_iarchive ia(iss);
    ia >> value;
    return st;
}

void generate_data(mpi::communicator local, mpi::communicator world)
{
  using std::srand;
  using std::rand;

  // The rank of the collector within the world communicator
  int master_collector = local.size();

  srand(time(0) + world.rank());

  // Send out several blocks of random data to the collectors.
  int num_data_blocks = rand() % 3 + 1;
  for (int block = 0; block < num_data_blocks; ++block) {
    // Generate some random data
    int num_samples = rand() % 1000;
    std::vector<int> data;
    for (int i = 0; i < num_samples; ++i) {
      data.push_back(rand());
    }

    // Send our data to the master collector process.
    std::cout << "Generator #" << local.rank() << " sends some data..."
              << std::endl;
    world.send(master_collector, msg_data_packet, data);
  }

  // Wait for all of the generators to complete
  (local.barrier)();

  // The first generator will send the message to the master collector
  // indicating that we're done.
  if (local.rank() == 0)
    world.send(master_collector, msg_finished);
}

void DocumentDecoder::translate() {
	namespace mpi = boost::mpi;

	mpi::request reqs[2];
	reqs[1] = communicator_.irecv(0, TAG_STOP_TRANSLATING);
	NumberedInputDocument input;
	for(;;) {
		reqs[0] = communicator_.irecv(0, TAG_TRANSLATE, input);
		std::pair<mpi::status, mpi::request *> wstat = mpi::wait_any(reqs, reqs + 2);
		if(wstat.first.tag() == TAG_STOP_TRANSLATING) {
			LOG(logger_, debug, "T: Received STOP_TRANSLATING.");
			reqs[0].cancel();
			communicator_.send(0, TAG_STOP_COLLECTING);
			return;
		} else {
			NumberedOutputDocument output;
			LOG(logger_, debug, "T: Received document " << input.first << " for translation.");
			output.first = input.first;
			output.second = runDecoder(input);
			LOG(logger_, debug, "T: Sending translation of document " << input.first << " to collector.");
			communicator_.send(0, TAG_COLLECT, output);
		}
	}
}

void
broadcast_test(const mpi::communicator& comm, const T& bc_value,
               std::string const& kind, int root) {
  using boost::mpi::broadcast;
  
  T value;
  if (comm.rank() == root) {
    value = bc_value;
    std::cout << "Broadcasting " << kind << " from root " << root << "...";
    std::cout.flush();
  }
  
  
  broadcast(comm, value, root);
  BOOST_CHECK(value == bc_value);
  if (comm.rank() == root) {
    if (value == bc_value) {
      std::cout << "OK." << std::endl;
    } else {
      std::cout << "FAIL." << std::endl;
    }
  }
  comm.barrier();
}