server_pack.h

Go to the documentation of this file.

#ifndef FPMAS_SERVER_PACK_H
#define FPMAS_SERVER_PACK_H
 
#include "fpmas/api/communication/communication.h"
#include "./api/client_server.h"
#include "fpmas/utils/log.h"
 
namespace fpmas { namespace synchro { namespace hard {
 
    class VoidServer : public api::Server {
        public:
            void setEpoch(api::Epoch) override {};
            api::Epoch getEpoch() const override {
                return {};
            };
            void handleIncomingRequests() override {};
    };
 
    class ServerPackBase : public api::Server {
        typedef api::Epoch Epoch;
 
        private:
        std::vector<fpmas::api::communication::Request> pending_requests;
 
        fpmas::api::communication::MpiCommunicator& comm;
        api::TerminationAlgorithm& termination;
        api::Server& mutex_server;
        api::Server& link_server;
        Epoch epoch;
 
        public:
        ServerPackBase(
                fpmas::api::communication::MpiCommunicator& comm,
                api::TerminationAlgorithm& termination,
                api::Server& mutex_server,
                api::Server& link_server)
            : comm(comm), termination(termination), mutex_server(mutex_server), link_server(link_server) {
                setEpoch(Epoch::EVEN);
            }
 
        virtual api::Server& mutexServer() {
            return mutex_server;
        }
 
        virtual api::Server& linkServer() {
            return link_server;
        }
 
        void handleIncomingRequests() override {
            mutex_server.handleIncomingRequests();
            link_server.handleIncomingRequests();
        }
 
        void setEpoch(Epoch epoch) override {
            this->epoch = epoch;
            mutex_server.setEpoch(epoch);
            link_server.setEpoch(epoch);
        }
 
        Epoch getEpoch() const override {
            return epoch;
        }
 
        void terminate() {
            FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "wait all...", "");
            // Applies the termination algorithm: keeps handling requests
            termination.terminate(*this);
 
            // Completes Isend responses performed while handling requests
            // since the last terminate() call.
            comm.waitAll(pending_requests);
            pending_requests.clear();
        }
 
        std::vector<fpmas::api::communication::Request>& pendingRequests() {
            return pending_requests;
        }
 
        void waitSendRequest(fpmas::api::communication::Request& req) {
            FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "wait for send...", "");
            bool sent = comm.test(req);
 
            while(!sent) {
                handleIncomingRequests();
                sent = comm.test(req);
            }
            FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "Request sent.", "");
        }
 
        template<typename T>
            void waitResponse(
                    fpmas::api::communication::TypedMpi<T>& mpi,
                    int source, api::Tag tag,
                    fpmas::api::communication::Status& status) {
                FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "wait for response...", "");
                bool response_available = mpi.Iprobe(source, getEpoch() | tag, status);
 
                while(!response_available) {
                    handleIncomingRequests();
                    response_available = mpi.Iprobe(source, getEpoch() | tag, status);
                }
                FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "Response available.", "");
            }
 
        void waitVoidResponse(
                fpmas::api::communication::MpiCommunicator& comm,
                int source, api::Tag tag,
                fpmas::api::communication::Status& status) {
            FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "wait for response...", "");
            bool response_available = comm.Iprobe(
                    fpmas::api::communication::MpiCommunicator::IGNORE_TYPE,
                    source, getEpoch() | tag, status);
 
            while(!response_available) {
                handleIncomingRequests();
                response_available = comm.Iprobe(
                        fpmas::api::communication::MpiCommunicator::IGNORE_TYPE,
                        source, getEpoch() | tag, status);
            }
            FPMAS_LOGV(comm.getRank(), "SERVER_PACK", "Response available.", "");
        }
 
    };
 
    
    template<typename MutexServer, typename LinkServer>
        class ServerPack : public ServerPackBase {
            private:
                MutexServer& mutex_server;
                LinkServer& link_server;
 
            public:
                ServerPack(
                        fpmas::api::communication::MpiCommunicator& comm,
                        api::TerminationAlgorithm& termination,
                        MutexServer& mutex_server, LinkServer& link_server) :
                    ServerPackBase(comm, termination, mutex_server, link_server),
                    mutex_server(mutex_server), link_server(link_server) {
                    }
 
                MutexServer& mutexServer() override {
                    return mutex_server;
                }
 
                LinkServer& linkServer() override {
                    return link_server;
                }
        };
}}}
#endif