Worker.h

Go to the documentation of this file.

//
// Class Worker
//   A worker MPI entity consists of a processor group that runs a
//   simulation of type Sim_t. The main loop in run() accepts new jobs from the
//   master process runs the simulation and reports back the results.
//
//   @see Pilot
//   @see Poller
//   @see MPIHelper.h
//
//   @tparam Sim_T type of simulation to run
//
// Copyright (c) 2010 - 2013, Yves Ineichen, ETH Zürich
// All rights reserved
//
// Implemented as part of the PhD thesis
// "Toward massively parallel multi-objective optimization with application to
// particle accelerators" (https://doi.org/10.3929/ethz-a-009792359)
//
// This file is part of OPAL.
//
// OPAL 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.
//
// You should have received a copy of the GNU General Public License
// along with OPAL. If not, see <https://www.gnu.org/licenses/>.
//
#ifndef __WORKER_H__
#define __WORKER_H__
 
#include <iostream>
 
#include "boost/smart_ptr.hpp"
 
#include "Pilot/Poller.h"
 
#include "Comm/types.h"
#include "Util/Types.h"
#include "Util/MPIHelper.h"
#include "Util/CmdArguments.h"
 
template <class Sim_t>
class Worker : protected Poller {
 
public:
 
    Worker(Expressions::Named_t constraints,
           std::string simName,
           Comm::Bundle_t comms,
           CmdArguments_t args)
        : Poller(comms.worker)
        , cmd_args_(args)
    {
        constraints_     = constraints;
        simulation_name_ = simName;
        pilot_rank_      = comms.master_local_pid;
        is_idle_         = true;
        coworker_comm_   = comms.coworkers;
 
        leader_pid_      = 0;
        MPI_Comm_size(coworker_comm_, &num_coworkers_);
    }
 
    Worker(Expressions::Named_t objectives,
           Expressions::Named_t constraints,
           std::string simName,
           Comm::Bundle_t comms,
           CmdArguments_t args,
           const std::map<std::string, std::string> &userVariables,
           bool isOptimizer = true)
        : Poller(comms.worker)
        , cmd_args_(args)
        , userVariables_(userVariables)
    {
        objectives_      = objectives;
        constraints_     = constraints;
        simulation_name_ = simName;
        pilot_rank_      = comms.master_local_pid;
        is_idle_         = true;
        coworker_comm_   = comms.coworkers;
 
        leader_pid_      = 0;
        MPI_Comm_size(coworker_comm_, &num_coworkers_);
 
        if (!isOptimizer) return;
        int my_local_pid = 0;
        MPI_Comm_rank(coworker_comm_, &my_local_pid);
 
        // distinction between leader and coworkers
        if(my_local_pid == leader_pid_)
            run();
        else
            runCoWorker();
    }
 
    ~Worker()
    {}
 
 
protected:
    typedef boost::scoped_ptr<Sim_t> SimPtr_t;
 
    bool is_idle_;
    MPI_Comm coworker_comm_;
 
    Expressions::Named_t objectives_;
    Expressions::Named_t constraints_;
 
 
    void runCoWorker() {
 
        MPI_Request stop_req;
        size_t stop_value = 0;
 
        MPI_Irecv(&stop_value, 1, MPI_UNSIGNED_LONG, leader_pid_,
                  MPI_ANY_TAG, coworker_comm_, &stop_req);
        is_running_ = true;
 
        while(is_running_) {
 
            //FIXME: bcast blocks after our leader stopped working
            // Either we create a new class implementing a coworker in the
            // same manner as the worker (poll loop). Anyway there is no way
            // around removing the Bcast and adding another tag in the poll
            // loop above in order to be able to exit cleanly.
            if(stop_req != MPI_REQUEST_NULL) {
                MPI_Status status;
                int flag = 0;
                MPI_Test(&stop_req, &flag, &status);
 
                if(flag) {
 
                    if(status.MPI_TAG == MPI_COWORKER_NEW_JOB_TAG) {
                        Param_t params;
                        MPI_Bcast_params(params, leader_pid_, coworker_comm_);
 
                        try {
                            SimPtr_t sim(new Sim_t(objectives_, constraints_,
                                                   params, simulation_name_, coworker_comm_,
                                                   cmd_args_, userVariables_));
 
                            sim->run();
                        } catch(OptPilotException &ex) {
                            std::cout << "Exception while running simulation: "
                                      << ex.what() << std::endl;
                        }
                        MPI_Irecv(&stop_value, 1, MPI_UNSIGNED_LONG, leader_pid_,
                                  MPI_ANY_TAG, coworker_comm_, &stop_req);
                    }
 
                    if(status.MPI_TAG == MPI_STOP_TAG) {
                        is_running_ = false;
                        break;
                    }
                }
            }
        }
    }
 
 
protected:
 
    int leader_pid_;
    int num_coworkers_;
    int pilot_rank_;
    std::string simulation_name_;
    CmdArguments_t cmd_args_;
 
    const std::map<std::string, std::string> userVariables_;
 
    void notifyCoWorkers(int tag) {
 
        for(int i=0; i < num_coworkers_; i++) {
            if(i == leader_pid_) continue;
 
            size_t dummy = 0;
            MPI_Send(&dummy, 1, MPI_UNSIGNED_LONG, i, tag, coworker_comm_);
        }
    }
 
    void setupPoll() {
        size_t dummy = 1;
        MPI_Send(&dummy, 1, MPI_UNSIGNED_LONG, pilot_rank_,
                 MPI_WORKER_STATUSUPDATE_TAG, comm_m);
    }
 
    void prePoll()
    {}
 
    void postPoll()
    {}
 
    void onStop() {
        if(num_coworkers_ > 1)
            notifyCoWorkers(MPI_STOP_TAG);
    }
 
    virtual bool onMessage(MPI_Status status, size_t recv_value) {
 
        if(status.MPI_TAG == MPI_WORK_JOBID_TAG) {
 
            is_idle_ = false;
            size_t job_id = recv_value;
 
            // get new job
            Param_t params;
            MPI_Recv_params(params, (size_t)pilot_rank_, comm_m);
 
            // and forward to coworkers (if any)
            if(num_coworkers_ > 1) {
                notifyCoWorkers(MPI_COWORKER_NEW_JOB_TAG);
                MPI_Bcast_params(params, leader_pid_, coworker_comm_);
            }
 
            //XXX we need to know if we want EVAL or DERIVATIVE
            //reqVarContainer_t reqVars;
            //MPI_Recv_reqvars(reqVars, (size_t)pilot_rank_, comm_m);
 
            reqVarContainer_t requested_results;
            try {
                SimPtr_t sim(new Sim_t(objectives_, constraints_,
                                       params, simulation_name_, coworker_comm_, cmd_args_,
                                       userVariables_));
 
                // run simulation in a "blocking" fashion
                sim->run();
                sim->collectResults();
                sim->cleanUp();
                requested_results = sim->getResults();
            } catch(OptPilotException &ex) {
                std::cout << "Exception while running simulation: "
                          << ex.what() << std::endl;
            }
 
            MPI_Send(&job_id, 1, MPI_UNSIGNED_LONG, pilot_rank_,
                     MPI_WORKER_FINISHED_TAG, comm_m);
 
            size_t dummy = 0;
            MPI_Recv(&dummy, 1, MPI_UNSIGNED_LONG, pilot_rank_,
                     MPI_WORKER_FINISHED_ACK_TAG, comm_m, &status);
 
            MPI_Send_reqvars(requested_results, (size_t)pilot_rank_, comm_m);
 
            is_idle_ = true;
            return true;
 
        } else {
            std::stringstream os;
            os << "Unexpected MPI_TAG: " << status.MPI_TAG;
            std::cout << "(Worker) Error: " << os.str() << std::endl;
            throw OptPilotException("Worker::onMessage", os.str());
        }
    }
};
 
#endif