worker.hpp

Go to the documentation of this file.

//This file is part of Bertini 2.
//
//bertini2/parallel/worker.hpp 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.
//
//bertini2/parallel/worker.hpp 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 bertini2/parallel/worker.hpp.  If not, see <http://www.gnu.org/licenses/>.
//
// Copyright(C) Bertini2 Development Team

 
#pragma once
 
#ifdef BERTINI2_HAVE_MPI
 
#include "bertini2/parallel/path_result.hpp"
#include "bertini2/parallel/mpi_utils.hpp"
#include "bertini2/parallel/thread_pool.hpp"
 
#include "bertini2/parallel/mpi_include.hpp"
 
#include <chrono>
#include <functional>
#include <limits>
#include <string>
#include <thread>
 
namespace bertini {
namespace parallel {
 

template<typename TaskT, typename ResultT>
void RunWorkerLoop(
    MPI_Comm comm,
    std::function<void(TaskT const&)> track_fn,
    std::function<ResultT(TaskT const&)> pack_fn)
{
    // Announce capacity: a serial worker can have exactly one task in flight.
    int capacity = 1;
    MPI_Send(&capacity, 1, MPI_INT, 0, TAG_CAPACITY, comm);
 
    while (true)
    {
        TaskT task;
        mpi_recv_serialized(comm, 0, TAG_WORK_ITEM, task);
 
        if (detail::is_sentinel(task))
            break;
 
        track_fn(task);
 
        ResultT result = pack_fn(task);
        mpi_send_serialized(comm, 0, TAG_RESULT, result);
    }
}
 

template<typename TaskT, typename ResultT, typename StateFactory, typename TrackFn>
void RunWorkerLoopThreaded(
    MPI_Comm comm,
    StateFactory state_factory,
    TrackFn track_fn,
    int n_threads)
{
    int capacity = n_threads;
    MPI_Send(&capacity, 1, MPI_INT, 0, TAG_CAPACITY, comm);
 
    WorkerThreadPool<TaskT, ResultT, StateFactory, TrackFn> pool(n_threads, state_factory, track_fn);
 
    int  in_flight    = 0;      // tasks submitted to the pool but not yet collected
    bool got_sentinel = false;
 
    while (!got_sentinel || in_flight > 0)
    {
        bool did_work = false;
 
        // Forward all completed results to the manager.
        while (auto opt = pool.try_collect())
        {
            mpi_send_serialized(comm, 0, TAG_RESULT, *opt);
            --in_flight;
            did_work = true;
        }
 
        // Non-blocking check for an incoming task (or the sentinel).
        if (!got_sentinel)
        {
            int flag = 0;
            MPI_Status status;
            MPI_Iprobe(0, TAG_WORK_ITEM, comm, &flag, &status);
            if (flag)
            {
                TaskT task;
                mpi_recv_serialized(comm, 0, TAG_WORK_ITEM, task);
 
                if (detail::is_sentinel(task))
                    got_sentinel = true;
                else
                {
                    pool.submit(std::move(task));
                    ++in_flight;
                }
                did_work = true;
            }
        }
 
        // Idle: nothing arrived and nothing finished.  Tracking a path takes
        // milliseconds to seconds, so a short sleep costs nothing measurable.
        if (!did_work)
            std::this_thread::sleep_for(std::chrono::microseconds(200));
    }
 
    pool.shutdown();
}
 
 
} // namespace parallel
} // namespace bertini
 
#endif // BERTINI2_HAVE_MPI