communicator.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 communicator_h
#define communicator_h
 
#include <cstdint>
#include "oofemenv.h"
#include "processcomm.h"
#include "commbufftype.h"
#include "communicatormode.h"
#include "error.h"
 
namespace oofem {
class EngngModel;

class OOFEM_EXPORT CommunicatorBuff
{
protected:
    std::vector<ProcessCommunicatorBuff> processCommBuffs;
 
public:
    CommunicatorBuff(int s, CommBuffType t = CBT_static);

    ProcessCommunicatorBuff *
    giveProcessCommunicatorBuff(int i) {
        return &this->processCommBuffs[i];
    }
};
 

class OOFEM_EXPORT Communicator
{
protected:
    int rank;
    std::vector<ProcessCommunicator> processComms;
    EngngModel *engngModel;
    CommunicatorMode mode;
 
public:
    Communicator(EngngModel * emodel, CommunicatorBuff * buff, int rank, int size, CommunicatorMode mode = CommMode_Static);
    virtual ~Communicator() {}

    ProcessCommunicator *
    giveProcessCommunicator(int i) {
        return &this->processComms[i];
    }

    template< class T > int packAllData( T *ptr, int ( T :: *packFunc )( ProcessCommunicator & ) );
    //template <class T> int packAllData (T* ptr, FloatArray* src, int (T::*packFunc) (FloatArray*, ProcessCommunicator&));
    template< class T, class P > int packAllData( T *ptr, P *src, int ( T :: *packFunc )( P *, ProcessCommunicator & ) );
    template< class T > int unpackAllData( T *ptr, int ( T :: *unpackFunc )( ProcessCommunicator & ) );
    //template <class T> int unpackAllData (T* ptr, FloatArray* dest, int (T::*unpackFunc) (FloatArray*, ProcessCommunicator&));
    template< class T, class P > int unpackAllData( T *ptr, P *src, int ( T :: *unpackFunc )( P *, ProcessCommunicator & ) );
    int initExchange(int tag);
    int initSend(int tag);
    int initReceive(int tag);
    int finishExchange();

    void clearBuffers();
    virtual void setUpCommunicationMaps(EngngModel *pm) { }

    std :: string errorInfo(const char *func) const;
};
 
template< class T > int
Communicator :: packAllData( T *ptr, int ( T :: *packFunc )( ProcessCommunicator & ) )
{
    int result = 1;
 
    for ( auto &pc : processComms ) {
        result &= pc.packData(ptr, packFunc);
    }
 
    return result;
}
 
#if 0
template <class T> int
Communicator :: packAllData (T* ptr, FloatArray* src, int (T::*packFunc) (FloatArray*, ProcessCommunicator&))
{
    int result = 1;
 
    for ( auto &pc : processComms ) result &= pc.packData (ptr, src, packFunc);
    return result;
}
#endif
 
template< class T, class P > int
Communicator :: packAllData( T *ptr, P *src, int ( T :: *packFunc )( P *, ProcessCommunicator & ) )
{
    int result = 1;
 
    for ( auto &pc : this->processComms ) {
        result &= pc.packData(ptr, src, packFunc);
    }
 
    return result;
}
 
template< class T > int
Communicator :: unpackAllData( T *ptr, int ( T :: *unpackFunc )( ProcessCommunicator & ) )
{
    int num_recv = 0, result = 1, size = processComms.size();
    IntArray recvFlag(size);
    //MPI_Status status;
 
    for  ( int i = 0; i < size; i++ ) {
        // receive if receive map is not empty or mode is dynamic
        if ( processComms[i].giveToRecvMap().giveSize() || this->mode == CommMode_Dynamic ) {
            recvFlag[i] = 1;
            num_recv++;
        }
    }
 
    while ( num_recv-- ) {
        // wait for any completion
        while ( 1 ) {
            bool received = false;
            for  ( int i = 0; i < size; i++ ) {
                if ( recvFlag[i] ) {
                    //if ( processComms[i].giveRecvBuff()->testCompletion() ) {
                    if ( processComms[i].receiveCompleted() ) {
 #ifdef __VERBOSE_PARALLEL
                        OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
                                        rank, "Communicator :: unpackAllData", i);
 #endif
 
                        recvFlag[i] = 0;
                        result &= processComms[i].unpackData(ptr, unpackFunc);
                        received = true;
                        break;
                    }
                }
            }
 
            if ( received ) {
                break;
            }
        }
    }
 
 #ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier started", rank)
 #endif
 
    MPI_Barrier(MPI_COMM_WORLD);
 
 #ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier finished", rank)
 #endif
 
    return result;
}
 
 
#if 0
template <class T> int
Communicator :: unpackAllData (T* ptr, FloatArray* dest, int (T::*unpackFunc) (FloatArray*, ProcessCommunicator&))
{
    int num_recv = 0, result = 1, size = processComms.size();
    IntArray recvFlag(size);
    //MPI_Status status;
 
    for  ( int i = 0; i < size; i++) {
        // receive if receive map is not empty or mode is dynamic
        if ( processComms[i].giveToRecvMap()->giveSize() || this->mode == CommMode_Dynamic) {
            recvFlag[i] = 1;
            num_recv ++;
        }
    }
 
    while (num_recv--) {
        // wait for any completion
        while (1) {
            bool received = false;
            for ( int i = 0; i < size; i++ ) {
                if ( recvFlag[i] ) {
                    //if ( processComms[i].giveRecvBuff()->testCompletion() ) {
                    if ( processComms[i].receiveCompleted() ) {
 
#ifdef __VERBOSE_PARALLEL
                        OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
                            rank,"Communicator :: unpackAllData", i);
#endif
 
                        recvFlag[i] = 0;
                        result &= processComms[i].unpackData (ptr, dest, unpackFunc);
                        received = true;
                        break;
                    }
                }
            }
            if (received) break;
        }
    }
 
#ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier started",rank)
#endif
 
    MPI_Barrier (MPI_COMM_WORLD);
 
#ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier finished",rank)
#endif
 
    return result;
}
#endif
 
template< class T, class P > int
Communicator :: unpackAllData( T *ptr, P *dest, int ( T :: *unpackFunc )( P *, ProcessCommunicator & ) )
{
    int num_recv = 0, result = 1, size = processComms.size();
    IntArray recvFlag(size);
    //MPI_Status status;
 
    for  ( int i = 0; i < size; i++ ) {
        // receive if receive map is not empty or mode is dynamic
        if ( processComms[i].giveToRecvMap().giveSize() || this->mode == CommMode_Dynamic ) {
            recvFlag[i] = 1;
            num_recv++;
        }
    }
 
    while ( num_recv-- ) {
        // wait for any completion
        while ( 1 ) {
            bool received = false;
            for  ( int i = 0; i < size; i++ ) {
                if ( recvFlag[i] ) {
                    //if ( processComms[i].giveRecvBuff()->testCompletion() ) {
                    if ( processComms[i].receiveCompleted() ) {
 #ifdef __VERBOSE_PARALLEL
                        OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
                                        rank, "Communicator :: unpackAllData", i);
 #endif
 
                        recvFlag[i] = 0;
                        result &= processComms[i].unpackData(ptr, dest, unpackFunc);
                        received = true;
                        break;
                    }
                }
            }
 
            if ( received ) {
                break;
            }
        }
    }
 
 #ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier started", rank)
 #endif
 
    MPI_Barrier(MPI_COMM_WORLD);
 
 #ifdef __VERBOSE_PARALLEL
    VERBOSEPARALLEL_PRINT("Communicator :: unpackAllData", "Synchronize barrier finished", rank)
 #endif
 
    return result;
}
} // end namespace oofem
#endif // communicator_h