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-2016 ULCO http://www.univ-littoral.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 < std::pair < int, 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 (unsigned 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 (unsigned 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 (unsigned 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 (unsigned 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 < std::pair < int, int > >());
            for (std::vector < std::string >::const_iterator it = strs.begin();
                 it != strs.end();) {
                if (it->size() > 0) {
                    int index = boost::lexical_cast < int >(*it);

                    ++it;
                    clusters[clusters.size() - 1].
                        push_back(std::make_pair(
                                      index, boost::lexical_cast < int >(*it)));
                }
                ++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;
    }
}