dyncombuff.h

Go to the documentation of this file.

/*
 *
 *                 #####    #####   ######  ######  ###   ###
 *               ##   ##  ##   ##  ##      ##      ## ### ##
 *              ##   ##  ##   ##  ####    ####    ##  #  ##
 *             ##   ##  ##   ##  ##      ##      ##     ##
 *            ##   ##  ##   ##  ##      ##      ##     ##
 *            #####    #####   ##      ######  ##     ##
 *
 *
 *             OOFEM : Object Oriented Finite Element Code
 *
 *               Copyright (C) 1993 - 2025   Borek Patzak
 *
 *
 *
 *       Czech Technical University, Faculty of Civil Engineering,
 *   Department of Structural Mechanics, 166 29 Prague, Czech Republic
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This program 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
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 */
 
#ifndef dyncombuff_h
#define dyncombuff_h
 
#include "parallel.h"
#include "combuff.h"
#include <list>
 
namespace oofem {
class IntArray;
class FloatArray;
class FloatMatrix;
 
 #define __CommunicationPacket_DEFAULT_SIZE 40960

 
class OOFEM_EXPORT CommunicationPacket : public MPIBuffer
{
protected:
    int number;
    bool EOF_Flag;
 
public:
 
 #ifdef __USE_MPI
    CommunicationPacket(MPI_Comm comm, std::size_t size, int num);
    CommunicationPacket(MPI_Comm comm, int num);
 #endif
    virtual ~CommunicationPacket();

    virtual void init(MPI_Comm comm);

 #ifdef __USE_MPI
    int iSend(MPI_Comm communicator, int dest, int tag) override;
    int iRecv(MPI_Comm communicator, int source, int tag, std::size_t count = 0) override;
    virtual int testCompletion();
    int waitCompletion() override;
 #endif
 
    void setNumber(int _num) { this->number = _num; }
    void setEOFFlag() { this->EOF_Flag = true; }
    int getNumber() { return number; }
    bool hasEOFFlag() { return EOF_Flag; }

    int packHeader(MPI_Comm);
    int unpackHeader(MPI_Comm);
};
 
class OOFEM_EXPORT CommunicationPacketPool
{
private:
    std :: list< CommunicationPacket * >available_packets;
    std :: list< CommunicationPacket * >leased_packets;
 
    int allocatedPackets, leasedPackets, freePackets;
public:
    CommunicationPacketPool() : available_packets(), leased_packets() {
        allocatedPackets = leasedPackets = freePackets;
    }
    ~CommunicationPacketPool() {
        this->clear();
    }
 
    CommunicationPacket *popPacket(MPI_Comm);
    void pushPacket(CommunicationPacket *);
 
    void printInfo();
 
private:
    void clear();
};
 
 
class OOFEM_EXPORT DynamicCommunicationBuffer : public CommunicationBuffer
{
protected:
    std :: list< CommunicationPacket * >packet_list;
    std :: list< CommunicationPacket * > :: iterator recvIt;
    CommunicationPacket *active_packet;
    int active_tag, active_rank;
    int number_of_packets;

    enum DCB_Mode { DCB_null, DCB_send, DCB_receive } mode;
    static CommunicationPacketPool packetPool;
    bool completed;
public:
    DynamicCommunicationBuffer(MPI_Comm comm, std::size_t size, bool dynamic = 0);
    DynamicCommunicationBuffer(MPI_Comm comm, bool dynamic = 0);
    virtual ~DynamicCommunicationBuffer();
 
    int resize(std::size_t newSize) override { return 1; }
    void init() override;

    void initForPacking() override;
    void initForUnpacking() override;
 
    int write(const int *src, std::size_t n) override
    { return __write(src, n, MPI_INT); }
    int write(const long *src, std::size_t n) override
    { return __write(src, n, MPI_LONG); }
    int write(const unsigned long *src, std::size_t n) override
    { return __write(src, n, MPI_UNSIGNED_LONG); }
    int write(const double *src, std::size_t n) override
    { return __write(src, n, MPI_DOUBLE); }
    int write(const char *src, std::size_t n) override
    { return __write(src, n, MPI_CHAR); }
 
    int read(int *dest, std::size_t n) override
    { return __read(dest, n, MPI_INT); }
    int read(long *dest, std::size_t n) override
    { return __read(dest, n, MPI_LONG); }
    int read(unsigned long *dest, std::size_t n) override
    { return __read(dest, n, MPI_UNSIGNED_LONG); }
    int read(double *dest, std::size_t n) override
    { return __read(dest, n, MPI_DOUBLE); }
    int read(char *dest, std::size_t n) override
    { return __read(dest, n, MPI_CHAR); }
 
 
    int iSend(int dest, int tag) override;
    int iRecv(int source, int tag, std::size_t count = 0) override;
    int bcast(int root) override;

    int testCompletion() override;
    int waitCompletion() override;
 
 
    static void printInfo() { packetPool.printInfo(); }
 
protected:
    CommunicationPacket *allocateNewPacket(int);
    void freePacket(CommunicationPacket *);
 
    int receiveCompleted();
    int sendCompleted();
 
    void popNewRecvPacket();
    void pushNewRecvPacket(CommunicationPacket *);
 
    void clear();
    int giveFitSize(MPI_Datatype type, int availableSpace, int arrySize);

    template< class T > int __write(T *src, int n, MPI_Datatype type) {
        int _result = 1;
        int start_indx = 0, end_indx, _size;
        int remaining_size = n;
 
        do {
            _size = this->giveFitSize(type, active_packet->giveAvailableSpace(), remaining_size);
            end_indx = start_indx + _size;
 
            if ( _size ) {
                _result &= active_packet->packArray(communicator, src + start_indx, _size, type);
            }
 
            if ( end_indx >= n ) {
                break;
            }
 
            // active packet full, allocate a new one
            active_packet = this->allocateNewPacket(++number_of_packets);
            packet_list.push_back(active_packet);
            start_indx = end_indx;
            remaining_size -= _size;
        } while ( 1 );
 
        return _result;
    }

    template< class T > int __read(T *dest, int n, MPI_Datatype type) {
        int _result = 1;
        int start_indx = 0, end_indx, _size;
        int remaining_size = n;
 
        do {
            _size = this->giveFitSize(type, active_packet->giveAvailableSpace(), remaining_size);
            end_indx = start_indx + _size;
 
            if ( _size ) {
                _result &= active_packet->unpackArray(communicator, dest + start_indx, _size, type);
            }
 
            if ( end_indx >= n ) {
                break;
            }
 
            // active packet exhausted, pop a new one
            this->popNewRecvPacket();
            start_indx = end_indx;
            remaining_size -= _size;
        } while ( 1 );
 
        return _result;
    }
};
} // end namespace oofem
 
#endif // dyncombuff_h