artis-test/src/tests/mpi/cluster/main.cpp

/**
 * @file tests/mpi/cluster/main.cpp
 * @author The PARADEVS Development Team
 * See the AUTHORS or Authors.txt file
 */

/*
 * PARADEVS - the multimodeling and simulation environment
 * This file is a part of the PARADEVS environment
 *
 * Copyright (C) 2013-2015 ULCO http://www.univ-litoral.fr
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <paradevs/common/RootCoordinator.hpp>

#include <boost/mpi/environment.hpp>
#include <boost/mpi/communicator.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>

#include <paradevs/common/time/DoubleTime.hpp>
#include <paradevs/kernel/pdevs/mpi/LogicalProcessor.hpp>
#include <paradevs/kernel/pdevs/multithreading/Coordinator.hpp>
#include <tests/mpi/cluster/graph_manager.hpp>

#include <chrono>
#include <fstream>

using namespace paradevs::tests::mpi::cluster;
using namespace paradevs::common;
using namespace boost::mpi;
using namespace std::chrono;

std::vector < std::vector < int > > clusters;
std::vector < std::vector < std::pair < int, int > > > graphs;
std::vector < std::vector < std::pair < int, int > > > inputs;
std::vector < std::vector < std::pair < int, int > > > outputs;
std::vector < std::vector < std::pair <
			      std::pair < int, int >,
			      std::pair < int, int > > > > parents;

void read_graphs(const std::string& path)
{
  for (int index = 0; index < clusters.size(); ++index) {
    std::stringstream ss;
    ss << path << "graphe/graphe_" << index << ".txt";
    std::ifstream graphFile(ss.str());

    graphs.push_back(std::vector < std::pair < int, int > >());
    while (not graphFile.eof()) {
      std::string str;
      std::vector < std::string > strs;
      
      std::getline(graphFile, str);
      if (str.size() > 0) { 
	boost::split(strs, str, boost::is_any_of(" "));
	graphs[graphs.size() - 1].
	  push_back(std::make_pair(boost::lexical_cast < int >(strs[0]),
				   boost::lexical_cast < int >(strs[1])));
      }
    }
    graphFile.close();
  }
}

void read_inputs(const std::string& path)
{
  for (int index = 0; index < clusters.size(); ++index) {
    std::stringstream ss;
    ss << path << "input_edges/input_edges_" << index << ".txt";
    std::ifstream inputsFile(ss.str());

    inputs.push_back(std::vector < std::pair < int, int > >());
    while (not inputsFile.eof()) {
      std::string str;
      std::vector < std::string > strs;
      
      std::getline(inputsFile, str);
      if (str.size() > 0) { 
	boost::split(strs, str, boost::is_any_of(" "));
	inputs[inputs.size() - 1].
	  push_back(std::make_pair(boost::lexical_cast < int >(strs[0]),
				   boost::lexical_cast < int >(strs[1])));
      }
    }
    inputsFile.close();
  }
}

void read_outputs(const std::string& path)
{
  for (int index = 0; index < clusters.size(); ++index) {
    std::stringstream ss;
    ss << path << "output_edges/output_edges_" << index << ".txt";
    std::ifstream outputsFile(ss.str());

    outputs.push_back(std::vector < std::pair < int, int > >());
    while (not outputsFile.eof()) {
      std::string str;
      std::vector < std::string > strs;
      
      std::getline(outputsFile, str);
      if (str.size() > 0) { 
	boost::split(strs, str, boost::is_any_of(" "));
	outputs[outputs.size() - 1].
	  push_back(std::make_pair(boost::lexical_cast < int >(strs[0]),
				   boost::lexical_cast < int >(strs[1])));
      }
    }
    outputsFile.close();
  }
}

void read_parents(const std::string& path)
{
  for (int index = 0; index < clusters.size(); ++index) {
    std::stringstream ss;
    ss << path << "parent_connection/parent_connection_" << index
       << ".txt";
    std::ifstream parentsFile(ss.str());

    parents.push_back(std::vector < std::pair < std::pair < int, int >,
		      std::pair < int, int > > >());
    while (not parentsFile.eof()) {
      std::string str;
      std::vector < std::string > strs;
      
      std::getline(parentsFile, str);
      if (str.size() > 0) { 
	boost::split(strs, str, boost::is_any_of(" "));
	parents[parents.size() - 1].
	  push_back(std::make_pair(
	     std::make_pair(boost::lexical_cast < int >(strs[0]),
			    boost::lexical_cast < int >(strs[1])),
	     std::make_pair(boost::lexical_cast < int >(strs[2]),
			    boost::lexical_cast < int >(strs[3]))));
      }
    }
    parentsFile.close();
  }
}

void read(const std::string& path)
{
  std::stringstream ss;
  ss << path << "partition.txt";
  std::ifstream clusterFile(ss.str());

  while (not clusterFile.eof()) {
    std::string str;
    std::vector < std::string > strs;

    std::getline(clusterFile, str);
    boost::split(strs, str, boost::is_any_of(" "));

    if (str.size() > 0) {
      clusters.push_back(std::vector < int >());
      for (std::vector < std::string >::const_iterator it = strs.begin();
	   it != strs.end(); ++it) {
	if (it->size() > 0) {
	  clusters[clusters.size() - 1].
	    push_back(boost::lexical_cast < int >(*it));
	}
      }
    }
  }
  clusterFile.close();
  read_graphs(path);
  read_inputs(path);
  read_outputs(path);
  read_parents(path);
}

void example_simple(int argc, char *argv[])
{
    environment env(argc, argv);
    communicator world;

    if (world.rank() == 0) {
        paradevs::tests::mpi::cluster::RootGraphManagerParameters parameters;

	parameters.parents = parents;
        paradevs::common::RootCoordinator <
            DoubleTime,
	    paradevs::pdevs::multithreading::Coordinator <
                DoubleTime,
                paradevs::tests::mpi::cluster::RootGraphManager,
                paradevs::common::NoParameters,
                paradevs::tests::mpi::cluster::RootGraphManagerParameters >
            > rc(0, 100, "root", paradevs::common::NoParameters(), parameters);

        steady_clock::time_point t1 = steady_clock::now();

	rc.run();

        steady_clock::time_point t2 = steady_clock::now();

        duration < double > time_span = duration_cast <
            duration < double > >(t2 - t1);

        std::cout << "time = " << time_span.count() << std::endl;

    } else {
        paradevs::tests::mpi::cluster::SubGraphManagerParameters parameters;
        std::stringstream ss;

	parameters.indexes = clusters[world.rank() - 1];
	parameters.inputs = inputs[world.rank() - 1];
	parameters.outputs = outputs[world.rank() - 1];
	parameters.internals = graphs[world.rank() - 1];
        ss << "S" << world.rank();
	paradevs::pdevs::mpi::Coordinator <
	  DoubleTime,
	  paradevs::tests::mpi::cluster::SubGraphManager,
                paradevs::common::NoParameters,
                paradevs::tests::mpi::cluster::SubGraphManagerParameters >
	model(ss.str(), paradevs::common::NoParameters(), parameters);
	paradevs::pdevs::mpi::LogicalProcessor <
	  DoubleTime > LP(&model, world.rank(), 0);
	
	model.set_logical_processor(&LP);
	LP.loop();
    }
}

int main(int argc, char *argv[])
{
  if (argc == 2) {
    read(argv[1]);
    example_simple(argc, argv);
    return 0;
  } else {
    return -1;
  }
}