communicator.h

Go to the documentation of this file.

/*
 *
 *                 #####    #####   ######  ######  ###   ###
 *               ##   ##  ##   ##  ##      ##      ## ### ##
 *              ##   ##  ##   ##  ####    ####    ##  #  ##
 *             ##   ##  ##   ##  ##      ##      ##     ##
 *            ##   ##  ##   ##  ##      ##      ##     ##
 *            #####    #####   ##      ######  ##     ##
 *
 *
 *             OOFEM : Object Oriented Finite Element Code
 *
 *               Copyright (C) 1993 - 2013   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 "oofemcfg.h"
#include "processcomm.h"
#include "commbufftype.h"
#include "communicatormode.h"
#include "error.h"

namespace oofem {
class EngngModel;

class OOFEM_EXPORT CommunicatorBuff
{
protected:
    int size;
    ProcessCommunicatorBuff **processCommBuffs;

public:
    CommunicatorBuff(int s, CommBuffType t = CBT_static);
    ~CommunicatorBuff();

    ProcessCommunicatorBuff *
    giveProcessCommunicatorBuff(int i) {
        if ( i < size ) {
            return processCommBuffs [ i ];
        } else {
            return NULL;
        }
    }
};


class OOFEM_EXPORT Communicator
{
protected:
    int rank;
    int size;
    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) {
        if ( i < size ) {
            return processComms [ i ];
        } else {
            return NULL;
        }
    }

    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 i = size, result = 1;

    if ( size ) {
        for ( i = 0; i < size; i++ ) {
            result &= giveProcessCommunicator(i)->packData(ptr, packFunc);
        }
    }

    return result;
}

/*
 * template <class T> int
 * Communicator :: packAllData (T* ptr, FloatArray* src, int (T::*packFunc) (FloatArray*, ProcessCommunicator&))
 * {
 * int i = size, result = 1;
 *
 * if (size)
 * for (i=0; i< size; i++) result &= giveProcessCommunicator(i)->packData (ptr, src, packFunc);
 * return result;
 * }
 */
template< class T, class P > int
Communicator :: packAllData( T *ptr, P *src, int ( T :: *packFunc )( P *, ProcessCommunicator & ) )
{
    int i = size, result = 1;

    if ( size ) {
        for ( i = 0; i < size; i++ ) {
            result &= giveProcessCommunicator(i)->packData(ptr, src, packFunc);
        }
    }

    return result;
}

template< class T > int
Communicator :: unpackAllData( T *ptr, int ( T :: *unpackFunc )( ProcessCommunicator & ) )
{
    int i, received, num_recv = 0, result = 1;
    IntArray recvFlag(size);
    //MPI_Status status;

    for  ( i = 0; i < size; i++ ) {
        // receive if receive map is not empty or mode is dynamic
        if ( ( giveProcessCommunicator(i)->giveToRecvMap()->giveSize() ) ||
            ( this->mode == CommMode_Dynamic ) ) {
            recvFlag.at(i + 1) = 1;
            num_recv++;
        }
    }

    while ( num_recv-- ) {
        // wait for any completion
        while ( 1 ) {
            received = 0;
            for  ( i = 0; i < size; i++ ) {
                if ( recvFlag.at(i + 1) ) {
                    //if (giveProcessCommunicator(i)->giveRecvBuff()->testCompletion()) {
                    if ( giveProcessCommunicator(i)->receiveCompleted() ) {
 #ifdef __VERBOSE_PARALLEL
                        OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
                                        rank, "Communicator :: unpackAllData", i);
 #endif

                        recvFlag.at(i + 1) = 0;
                        result &= giveProcessCommunicator(i)->unpackData(ptr, unpackFunc);
                        received = 1;
                        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;
}


/*
 * template <class T> int
 * Communicator :: unpackAllData (T* ptr, FloatArray* dest, int (T::*unpackFunc) (FloatArray*, ProcessCommunicator&))
 * {
 * int i, received, num_recv = 0, result = 1;
 * IntArray recvFlag (size);
 * //MPI_Status status;
 *
 * for  (i=0; i<size; i++) {
 * // receive if receive map is not empty or mode is dynamic
 * if ((giveProcessCommunicator(i)->giveToRecvMap()->giveSize()) ||
 *     (this->mode == CommMode_Dynamic)) {
 *   recvFlag.at(i+1) = 1;
 *   num_recv ++;
 * }
 * }
 *
 * while (num_recv--) {
 *
 * // wait for any completion
 * while (1) {
 * received = 0;
 * for  (i=0; i<size; i++)  {
 *  if (recvFlag.at(i+1)) {
 *    //if (giveProcessCommunicator(i)->giveRecvBuff()->testCompletion()) {
 *    if (giveProcessCommunicator(i)->receiveCompleted()) {
 *
 *
 ****#ifdef __VERBOSE_PARALLEL
 *     OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
 *                     rank,"Communicator :: unpackAllData", i);
 ****#endif
 *
 *    recvFlag.at(i+1) = 0;
 *    result &= giveProcessCommunicator(i)->unpackData (ptr, dest, unpackFunc);
 *    received = 1;
 *    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;
 * }
 */

template< class T, class P > int
Communicator :: unpackAllData( T *ptr, P *dest, int ( T :: *unpackFunc )( P *, ProcessCommunicator & ) )
{
    int i, received, num_recv = 0, result = 1;
    IntArray recvFlag(size);
    //MPI_Status status;

    for  ( i = 0; i < size; i++ ) {
        // receive if receive map is not empty or mode is dynamic
        if ( ( giveProcessCommunicator(i)->giveToRecvMap()->giveSize() ) ||
            ( this->mode == CommMode_Dynamic ) ) {
            recvFlag.at(i + 1) = 1;
            num_recv++;
        }
    }

    while ( num_recv-- ) {
        // wait for any completion
        while ( 1 ) {
            received = 0;
            for  ( i = 0; i < size; i++ ) {
                if ( recvFlag.at(i + 1) ) {
                    //if (giveProcessCommunicator(i)->giveRecvBuff()->testCompletion()) {
                    if ( giveProcessCommunicator(i)->receiveCompleted() ) {
 #ifdef __VERBOSE_PARALLEL
                        OOFEM_LOG_DEBUG("[process rank %3d]: %-30s: Received data from partition %3d\n",
                                        rank, "Communicator :: unpackAllData", i);
 #endif

                        recvFlag.at(i + 1) = 0;
                        result &= giveProcessCommunicator(i)->unpackData(ptr, dest, unpackFunc);
                        received = 1;
                        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