AsyncSendBuffers.h

Go to the documentation of this file.

//
// Class AsyncSendBuffer and AsyncSendBuffers
//
// 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/>.
//
#include <algorithm>
#include <memory>
#include <sstream>
#include <string>
#include <vector>
 
#include "mpi.h"
 
class AsyncSendBuffer {
 
public:
    AsyncSendBuffer(std::ostringstream& os) {
        this->size_req   = new MPI_Request();
        this->buffer_req = new MPI_Request();
        this->buf_size   = os.str().length();
        buffer           = new char[buf_size];
        std::memcpy(buffer, os.str().c_str(), buf_size);
    }
 
    ~AsyncSendBuffer() {
        delete   size_req;
        delete   buffer_req;
        delete[] buffer;
    }
 
    bool hasCompleted() {
        int bufferflag = 0;
        MPI_Test(this->buffer_req, &bufferflag, MPI_STATUS_IGNORE);
        if(bufferflag) {
            int sizeflag = 0;
            MPI_Test(this->buffer_req, &sizeflag, MPI_STATUS_IGNORE);
            if(sizeflag) {
                return true;
            }
        }
        return false;
    }
 
    void send(int recv_rank, int size_tag, int data_tag, MPI_Comm comm) {
        MPI_Isend(&buf_size, 1, MPI_LONG, recv_rank, size_tag, comm, size_req);
        MPI_Isend(buffer, buf_size, MPI_CHAR, recv_rank, data_tag, comm, buffer_req);
    }
 
private:
    // can't use smart pointers because MPI will hold last valid reference to
    // pointer
    MPI_Request *size_req;
    MPI_Request *buffer_req;
    char        *buffer;
 
    std::size_t buf_size;
};
 
 
class AsyncSendBuffers {
 
public:
    AsyncSendBuffers() {}
 
    void insert(std::shared_ptr<AsyncSendBuffer> buf) {
        collection_.push_back(buf);
    }
 
        void cleanup() {
        collection_.erase(
            std::remove_if(
                collection_.begin(),
                collection_.end(),
                [](const std::shared_ptr<AsyncSendBuffer>& buf) {
                    return buf->hasCompleted();
                }),
            collection_.end());
    }
 
    std::size_t size() {
        return collection_.size();
    }
 
private:
    std::vector< std::shared_ptr<AsyncSendBuffer> > collection_;
};