dyncombuff.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 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, int 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);
    int iRecv(MPI_Comm communicator, int source, int tag, int count = 0);
    virtual int testCompletion();
    virtual int waitCompletion();
 #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, int size, bool dynamic = 0);
    DynamicCommunicationBuffer(MPI_Comm comm, bool dynamic = 0);
    virtual ~DynamicCommunicationBuffer();

    virtual int resize(int newSize) { return 1; }
    virtual void init();

    virtual void initForPacking();
    virtual void initForUnpacking();

    virtual int write(const int *src, int n)
    { return __write(src, n, MPI_INT); }
    virtual int write(const long *src, int n)
    { return __write(src, n, MPI_LONG); }
    virtual int write(const unsigned long *src, int n)
    { return __write(src, n, MPI_UNSIGNED_LONG); }
    virtual int write(const double *src, int n)
    { return __write(src, n, MPI_DOUBLE); }
    virtual int write(const char *src, int n)
    { return __write(src, n, MPI_CHAR); }

    virtual int read(int *dest, int n)
    { return __read(dest, n, MPI_INT); }
    virtual int read(long *dest, int n)
    { return __read(dest, n, MPI_LONG); }
    virtual int read(unsigned long *dest, int n)
    { return __read(dest, n, MPI_UNSIGNED_LONG); }
    virtual int read(double *dest, int n)
    { return __read(dest, n, MPI_DOUBLE); }
    virtual int read(char *dest, int n)
    { return __read(dest, n, MPI_CHAR); }


    virtual int iSend(int dest, int tag);
    virtual int iRecv(int source, int tag, int count = 0);
    virtual int bcast(int root);

    int testCompletion();
    virtual int waitCompletion();


    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