Poller.h

Go to the documentation of this file.

#ifndef __POLLER_H__
#define __POLLER_H__

#include <iostream>

#include "Util/MPIHelper.h"

class Poller {

public:

    Poller(MPI_Comm comm, double delay = 0.1)
        : comm_m(comm)
        , is_running_(false)
        , poll_delay_(delay)
    {
        last_polled_ = MPI_Wtime();
    }

    virtual ~Poller()
    {}

protected:
    MPI_Comm comm_m;

    bool is_running_;

    double last_polled_;
    double poll_delay_;

    virtual bool onMessage(MPI_Status status, size_t recv_value) = 0;
    virtual void onStop() = 0;

    virtual void setupPoll() = 0;
    virtual void prePoll()  = 0;
    virtual void postPoll() = 0;

    virtual void run() {
        MPI_Status status;
        MPI_Request req;
        size_t recv_value = 0;
        int flag = 0;

        setupPoll();

        MPI_Irecv(&recv_value, 1, MPI_UNSIGNED_LONG, MPI_ANY_SOURCE, MPI_ANY_TAG,
                  comm_m, &req);

        is_running_ = true;

        while(true) {

            // regulate the amount of MPI_Test calls (expensive)
            double tnow = MPI_Wtime();
            if(tnow - last_polled_ > poll_delay_)
                last_polled_ = tnow;
            else
                continue;

            prePoll();

            if(req != MPI_REQUEST_NULL) {
                MPI_Test(&req, &flag, &status);
                if(flag) {
                    if(status.MPI_TAG == MPI_STOP_TAG) {
                        is_running_ = false;
                        onStop();
                        return;
                    } else {
                        if(onMessage(status, recv_value))
                            MPI_Irecv(&recv_value, 1, MPI_UNSIGNED_LONG, MPI_ANY_SOURCE,
                                      MPI_ANY_TAG, comm_m, &req);
                        else
                            break;
                    }
                }
            }

            postPoll();
        }
    }
};

#endif