BoxLayoutDataI.H

Go to the documentation of this file.

/*   _______              __
    / ___/ /  ___  __ _  / /  ___
   / /__/ _ \/ _ \/  V \/ _ \/ _ \
   \___/_//_/\___/_/_/_/_.__/\___/
*/
// CHOMBO Copyright (c) 2000-2004, The Regents of the University of
// California, through Lawrence Berkeley National Laboratory (subject to
// receipt of any required approvals from U.S. Dept. of Energy).  All
// rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
// (1) Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// (2) Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
// (3) Neither the name of Lawrence Berkeley National Laboratory, U.S.
// Dept. of Energy nor the names of its contributors may be used to endorse
// or promote products derived from this software without specific prior
// written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
// TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
// PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
// OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// You are under no obligation whatsoever to provide any bug fixes,
// patches, or upgrades to the features, functionality or performance of
// the source code ("Enhancements") to anyone; however, if you choose to
// make your Enhancements available either publicly, or directly to
// Lawrence Berkeley National Laboratory, without imposing a separate
// written license agreement for such Enhancements, then you hereby grant
// the following license: a non-exclusive, royalty-free perpetual license
// to install, use, modify, prepare derivative works, incorporate into
// other computer software, distribute, and sublicense such Enhancements or
// derivative works thereof, in binary and source code form.
//
// TRADEMARKS. Product and company names mentioned herein may be the
// trademarks of their respective owners.  Any rights not expressly granted
// herein are reserved.
//

#ifndef _BOXLAYOUTDATAI_H_
#define _BOXLAYOUTDATAI_H_

#include <cstdlib>
#include <algorithm>
#include <limits.h>

#include "parstream.H"
#include "memtrack.H"

using std::sort;

template<class T>
void BoxLayoutData<T>::makeItSo(const Interval&   a_srcComps,
                            const BoxLayoutData<T>& a_src,
                            BoxLayoutData<T>& a_dest,
                            const Interval&   a_destComps,
                            const Copier&     a_copier,
                            const LDOperator<T>& a_op) const
{
  // The following five functions are nullOps in uniprocessor mode

  // Instead of doing this here, do it an end of makeItSo (ndk)
  //completePendingSends(); // wait for sends from possible previous operation

  // new evil logic to determine under what conditions we can just
  // re-use our messaging pattern and buffers from the last call to
  // makeItSo.  pretty elaborate, I know.  bvs
#ifdef CH_MPI
  static Copier* lastCopier=NULL;

#ifndef NDEBUG
  //verifyCommunications();
#endif

  if(T::preAllocatable() == 2 || !a_copier.bufferAllocated() ||
     (m_fromMe.size() + m_toMe.size() == 0) ||lastCopier != &a_copier){
    allocateBuffers(a_src,  a_srcComps,
                    a_dest, a_destComps,
                    a_copier,
                    a_op);  //monkey with buffers, set up 'fromMe' and 'toMe' queues
    a_copier.setBufferAllocated(true);
  }
  lastCopier = (Copier*)(&a_copier);

#endif

  writeSendDataFromMeIntoBuffers(a_src, a_srcComps, a_op);

  // If there is nothing to recv/send, don't go into these functions
  // and allocate memory that will not be freed later.  (ndk)
  // The #ifdef CH_MPI is for the m_toMe and m_fromMe
#ifdef CH_MPI
  this->numReceives = m_toMe.size();
  if (this->numReceives > 0) {
    postReceivesToMe(); // all non-blocking
  }

  this->numSends = m_fromMe.size();
  if (this->numSends > 0) {
    postSendsFromMe();  // all non-blocking
  }
#endif

  //  computation that could occur during communication should really
  //  go here somehow.  while all the non-blocking sending and receiving is
  //  going on.
  //
  //  my thought is to return from this function at this point an object
  //  that encapsulates the argument list above.
  //  a "ChomboMessaging" object.
  //  The user can keep a reference
  //  to this object and do computations.  When they reach the limit of what
  //  they can compute without this communication completing, they call the
  //  "finalize()" function of their ChomboMessaging object and the rest of this
  //  code below gets executed.
  //  a real question though is: is there really enough computation to do while
  //  messaging is going on to justify the effort, and what machines really have
  //  good asynchronous messaging to make the work worthwhile.
  //
  //  the other approach is to more finely decompose the overlapping of
  //  messaging and computation by using the ChomboMessaging object in the
  //  DataIterator construction.  The DataIterator returns T objects as they
  //  are completed from messaging.  This preserves almost all of the Chombo
  //  code as is but would be mucho tricky to actually implement and might only
  //  gain little.  This would not be a thing to try unitl Chombo is
  //  heavily instrumented for performance measuring.  in this design, unpackRecievesToMe()
  //  would become a complicated process interwoven with a DataIterator.

  //  postReceivesToMe();

  // perform local copy
  for(CopyIterator it(a_copier, CopyIterator::LOCAL); it.ok(); ++it)
    {
      const MotionItem& item = it();
      a_op.op(a_dest[item.toIndex], item.fromRegion,
              a_destComps,
              item.toRegion,
              a_src[item.fromIndex],
              a_srcComps);
    }

  // Uncomment and Move this out of unpackReceivesToMe()  (ndk)
  completePendingSends(); // wait for sends from possible previous operation

  unpackReceivesToMe(a_dest, a_destComps, a_op); // nullOp in uniprocessor mode

}

#ifndef CH_MPI
// uniprocessor version of all these nullop functions.
template<class T>
void BoxLayoutData<T>::completePendingSends() const
{}

template<class T>
void BoxLayoutData<T>::allocateBuffers(const BoxLayoutData<T>& a_src,
                                   const Interval& a_srcComps,
                                   const BoxLayoutData<T>& a_dest,
                                   const Interval& a_destComps,
                                   const Copier&   a_copier,
                                   const LDOperator<T>& a_op
                                   ) const
{}

template<class T>
void BoxLayoutData<T>::writeSendDataFromMeIntoBuffers(const BoxLayoutData<T>& a_src,
                                                  const Interval&     a_srcComps,
                                                  const LDOperator<T>& a_op) const
{}

template<class T>
void BoxLayoutData<T>::postSendsFromMe() const
{}

template<class T>
void BoxLayoutData<T>::postReceivesToMe() const
{}

template<class T>
void BoxLayoutData<T>::unpackReceivesToMe(BoxLayoutData<T>& a_dest,
                                      const Interval&   a_destComps,
                                      const LDOperator<T>& a_op) const
{}

template<class T>
void BoxLayoutData<T>::unpackReceivesToMe_append(LayoutData<Vector<RefCountedPtr<T> > >& a_dest,
                                                 const Interval&   a_destComps,
                                                 int ncomp,
                                                 const DataFactory<T>& factory,
                                                 const LDOperator<T>& a_op) const
{}

#else

// MPI versions of the above codes.

template<class T>
void BoxLayoutData<T>::completePendingSends() const
{
  if(this->numSends > 0){
    int result = MPI_Waitall(this->numSends, m_sendRequests, m_sendStatus);
    if(result != MPI_SUCCESS)
      {
        //hell if I know what to do about failed messaging here
      }

    delete[] m_sendRequests;
    delete[] m_sendStatus;
  }
  this->numSends = 0;
}


template<class T>
void BoxLayoutData<T>::allocateBuffers(const BoxLayoutData<T>& a_src,
                                   const Interval& a_srcComps,
                                   const BoxLayoutData<T>& a_dest,
                                   const Interval& a_destComps,
                                   const Copier&   a_copier,
                                   const LDOperator<T>& a_op) const
{
  m_fromMe.resize(0);
  m_toMe.resize(0);
  size_t sendBufferSize = 0;
  size_t recBufferSize  = 0;
  // two versions of code here.  one for preAllocatable T, one not.

  T dummy;
  for(CopyIterator it(a_copier, CopyIterator::FROM); it.ok(); ++it)
    {
      const MotionItem& item = it();
      bufEntry b;
      b.item = &item;
      b.size = a_op.size(a_src[item.fromIndex], item.fromRegion, a_srcComps);
      sendBufferSize+=b.size;
      b.procID = item.procID;
      m_fromMe.push_back(b);
    }
  sort(m_fromMe.begin(), m_fromMe.end());
  for(CopyIterator it(a_copier, CopyIterator::TO); it.ok(); ++it)
    {
      const MotionItem& item = it();
      bufEntry b;
      b.item = &item;
      if(T::preAllocatable() == 0)
        {
          b.size = a_op.size(dummy, item.toRegion, a_destComps);
          recBufferSize+=b.size;
        }
      else if (T::preAllocatable() == 1)
        {
          b.size = a_op.size(a_dest[item.toIndex], item.toRegion, a_destComps);
          recBufferSize+=b.size;
        }
      b.procID = item.procID;
      m_toMe.push_back(b);
    }
  sort(m_toMe.begin(), m_toMe.end());

  if(T::preAllocatable() == 2) // dynamic allocatable, need two pass
    {
      // in the non-preallocatable case, I need to message the
      // values for the m_toMe[*].size
      Vector<unsigned long> fdata;
      Vector<unsigned long> tdata;
      int count = 1;
      int scount = 1;
      if(m_toMe.size() > 0)
        {
          tdata.resize(m_toMe.size(), ULONG_MAX);
          m_receiveRequests = new MPI_Request[numProc()-1];
          m_receiveStatus   = new MPI_Status [numProc()-1];
          MPI_Request* Rptr = m_receiveRequests;

          int lastProc = m_toMe[0].procID;
          int messageSize = 1;
          unsigned long * dataPtr = &(tdata[0]);
          int i = 1;
          
          for(;i<m_toMe.size(); ++i)
            {
              bufEntry& b = m_toMe[i];
              if(b.procID == lastProc) 
                messageSize++;
              else
                {                     
                 
                  MPI_Irecv(dataPtr, messageSize, MPI_UNSIGNED_LONG, lastProc,
                            1, Chombo_MPI::comm, Rptr);
                  Rptr++;
              
                  lastProc = b.procID;
                  messageSize = 1;
                  dataPtr = &(tdata[i]);
                  count++;
                }
            }
          
          MPI_Irecv(dataPtr, messageSize, MPI_UNSIGNED_LONG, lastProc,
                    1, Chombo_MPI::comm, Rptr );

         
        }
      if(m_fromMe.size() > 0)
        {
          fdata.resize(m_fromMe.size());
          fdata[0]=m_fromMe[0].size;
          m_sendRequests = new MPI_Request[numProc()-1];
          m_sendStatus   = new MPI_Status [numProc()-1];
          MPI_Request* Rptr = m_sendRequests;

          int lastProc = m_fromMe[0].procID;
          int messageSize = 1;
          unsigned long * dataPtr = &(fdata[0]);
          int i = 1;
          for(;i<m_fromMe.size(); ++i)
            {
              fdata[i]    = m_fromMe[i].size;
              bufEntry& b = m_fromMe[i];
              if(b.procID == lastProc) 
                messageSize++;
              else
                {                     
                
                  MPI_Isend(dataPtr, messageSize, MPI_UNSIGNED_LONG, lastProc,
                            1, Chombo_MPI::comm, Rptr);
                 
                
                  Rptr++;
                  lastProc = b.procID;
                  messageSize = 1;
                  dataPtr = &(fdata[i]);
                  scount++;
                }
            }
   
          MPI_Isend(dataPtr, messageSize, MPI_UNSIGNED_LONG, lastProc,
                    1, Chombo_MPI::comm, Rptr);
     
          
        }

      if(m_toMe.size() > 0)
        {
          
          int result = MPI_Waitall(count, m_receiveRequests, m_receiveStatus);
          if(result != MPI_SUCCESS)
            {
              MayDay::Error("First pass of two-phase communication failed");
            }
          
          for(int i=0; i<m_toMe.size(); ++i) {
            CH_assert(tdata[i] != ULONG_MAX);
            m_toMe[i].size = tdata[i];
            recBufferSize+= tdata[i];
          }
          
          delete[] m_receiveRequests;
          delete[] m_receiveStatus;
        }
      if(m_fromMe.size() > 0)
        {
                   
          int result = MPI_Waitall(scount, m_sendRequests, m_sendStatus);
          if(result != MPI_SUCCESS)
            {
              MayDay::Error("First pass of two-phase communication failed");
            }
          delete[] m_sendRequests;
          delete[] m_sendStatus;
        }
    }

  // allocate send and receveive buffer space.

  if(sendBufferSize > m_sendcapacity)
    {
      free(m_sendbuffer);
      m_sendbuffer = malloc(sendBufferSize);
      if(m_sendbuffer == NULL)
        {
          MayDay::Error("Out of memory in BoxLayoutData::allocatebuffers");
        }
      m_sendcapacity = sendBufferSize;
    }

  if(recBufferSize > m_reccapacity)
    {
      free(m_recbuffer);
      m_recbuffer = malloc(recBufferSize);
      if(m_recbuffer == NULL)
        {
          MayDay::Error("Out of memory in BoxLayoutData::allocatebuffers");
        }
      m_reccapacity = recBufferSize;
    }

  /*
    pout()<<"\n";
    for(int i=0; i<m_fromMe.size(); i++)
    pout()<<m_fromMe[i].item->region<<"{"<<m_fromMe[i].procID<<"}"<<" ";
    pout() <<"::::";
    for(int i=0; i<m_toMe.size(); i++)
    pout()<<m_toMe[i].item->region<<"{"<<m_toMe[i].procID<<"}"<<" ";
    pout() << endl;
  */

  char* nextFree = (char*)m_sendbuffer;
  if(m_fromMe.size() > 0)
    {
      for(unsigned int i=0; i<m_fromMe.size(); ++i)
        {
          m_fromMe[i].bufPtr = nextFree;
          nextFree += m_fromMe[i].size;
        }
    }

  nextFree = (char*)m_recbuffer;
  if(m_toMe.size() > 0)
    {
      for(unsigned int i=0; i<m_toMe.size(); ++i)
        {
          m_toMe[i].bufPtr = nextFree;
          nextFree += m_toMe[i].size;
        }
    }

  // since fromMe and toMe are sorted based on procID, messages can now be grouped
  // together on a per-processor basis.

}

template<class T>
void BoxLayoutData<T>::writeSendDataFromMeIntoBuffers(const BoxLayoutData<T>& a_src,
                                                  const Interval&     a_srcComps,
                                                  const LDOperator<T>& a_op) const
{
  for(unsigned int i=0; i<m_fromMe.size(); ++i)
    {
      const bufEntry& entry = m_fromMe[i];
      a_op.linearOut(a_src[entry.item->fromIndex], entry.bufPtr,
                     entry.item->fromRegion, a_srcComps);
    }
}

template<class T>
void BoxLayoutData<T>::postSendsFromMe() const
{

  // now we get the magic of message coalescence
  // fromMe has already been sorted in the allocateBuffers() step.

  this->numSends = m_fromMe.size();
  if(this->numSends > 1){
    for(unsigned int i=m_fromMe.size()-1; i>0; --i)
      {
        if(m_fromMe[i].procID == m_fromMe[i-1].procID)
          {
            this->numSends--;
            m_fromMe[i-1].size+=m_fromMe[i].size;
            m_fromMe[i].size = 0;
          }
      }
  }
  m_sendRequests = new MPI_Request[this->numSends];
  m_sendStatus = new MPI_Status[this->numSends];

  unsigned int next=0;
  for(int i=0; i<this->numSends; ++i)
    {
      const bufEntry& entry = m_fromMe[next];
      // cout<<procID()<< ": sending message of "<<entry.size;
      // cout<<" to proc "<<  entry.procID<<endl;
      MPI_Isend(entry.bufPtr, entry.size, MPI_BYTE, entry.procID,
                0, Chombo_MPI::comm, m_sendRequests+i);
      ++next;
      while(next < m_fromMe.size() && m_fromMe[next].size == 0) ++next;
    }
}

template<class T>
void BoxLayoutData<T>::postReceivesToMe() const
{
  this->numReceives = m_toMe.size();

  if(this->numReceives > 1){
    for(unsigned int i=m_toMe.size()-1; i>0; --i)
      {
        if(m_toMe[i].procID == m_toMe[i-1].procID)
          {
            this->numReceives--;
            m_toMe[i-1].size+=m_toMe[i].size;
            m_toMe[i].size = 0;
          }
      }
  }
  m_receiveRequests = new MPI_Request[this->numReceives];
  m_receiveStatus = new MPI_Status[this->numReceives];

  unsigned int next=0;
  for(int i=0; i<this->numReceives; ++i)
    {
      const bufEntry& entry = m_toMe[next];
      //cout<<procID()<< ": receiving message of "<<entry.size;
      //cout<<" from proc "<<  entry.procID<<endl;
      MPI_Irecv(entry.bufPtr, entry.size, MPI_BYTE, entry.procID,
                0, Chombo_MPI::comm, m_receiveRequests+i);
      ++next;
      while(next < m_toMe.size() && m_toMe[next].size == 0) ++next;
    }

}

template<class T>
void BoxLayoutData<T>::unpackReceivesToMe(BoxLayoutData<T>& a_dest,
                                      const Interval&   a_destComps,
                                      const LDOperator<T>& a_op) const
{

  if(this->numReceives > 0){
    int result = MPI_Waitall(this->numReceives, m_receiveRequests, m_receiveStatus);
    if(result != MPI_SUCCESS)
      {
        //hell if I know what to do about failed messaging here
      }

    for(unsigned int i=0; i<m_toMe.size(); ++i)
      {
        const bufEntry& entry = m_toMe[i];
        a_op.linearIn(a_dest[entry.item->toIndex], entry.bufPtr, entry.item->toRegion, a_destComps);
      }

    delete[] m_receiveRequests;
    delete[] m_receiveStatus;
  }
  this->numReceives = 0;
}

template<class T>
void BoxLayoutData<T>::unpackReceivesToMe_append(LayoutData<Vector<RefCountedPtr<T> > >& a_dest,
                                                 const Interval&   a_destComps,
                                                 int ncomp,
                                                 const DataFactory<T>& factory,

                                                 const LDOperator<T>& a_op) const
{

  if(this->numReceives > 0){
    int result = MPI_Waitall(this->numReceives, m_receiveRequests, m_receiveStatus);
    if(result != MPI_SUCCESS)
      {
        //hell if I know what to do about failed messaging here
      }

    for(unsigned int i=0; i<m_toMe.size(); ++i)
      {
        const bufEntry& entry = m_toMe[i];
        const MotionItem& item = *(entry.item);
        RefCountedPtr<T> newT = factory.create(item.toRegion, ncomp, item.toIndex);

        a_op.linearIn(*newT, entry.bufPtr, item.toRegion, a_destComps);
        a_dest[item.toIndex].push_back(newT);
      }

    delete[] m_receiveRequests;
    delete[] m_receiveStatus;
  }
  this->numReceives = 0;
}
#endif

template <class T>
void BoxLayoutData<T>::generalCopyTo(const BoxLayout& a_destGrids,
                                     LayoutData<Vector<RefCountedPtr<T> > >& a_dest,
                                     const Interval& a_srcComps,
                                     const ProblemDomain& a_domain,
                                     const DataFactory<T>& factory) const
{

 CH_assert(T::preAllocatable() == 0);
  LDOperator<T> a_op;

  a_dest.define(a_destGrids);
  Copier copier;
  copier.define(this->m_boxLayout, a_destGrids, a_domain, IntVect::Zero);

  int ncomp = a_srcComps.size();
  Interval destComps(0, ncomp-1);
  allocateBuffers(*this,  a_srcComps,
                  *this,  destComps,
                  copier, a_op);

  writeSendDataFromMeIntoBuffers(*this, a_srcComps, a_op);

  // If there is nothing to recv/send, don't go into these functions
  // and allocate memory that will not be freed later.  (ndk)
  // The #ifdef CH_MPI is for the m_toMe and m_fromMe
#ifdef CH_MPI
  this->numReceives = m_toMe.size();
  if (this->numReceives > 0) {
    postReceivesToMe(); // all non-blocking
  }

  this->numSends = m_fromMe.size();
  if (this->numSends > 0) {
    postSendsFromMe();  // all non-blocking
  }
#endif

    // perform local copy
  for(CopyIterator it(copier, CopyIterator::LOCAL); it.ok(); ++it)
    {
      const MotionItem& item = it();
      RefCountedPtr<T> newT = factory.create(item.toRegion, ncomp, item.toIndex);

      a_op.op(*newT, item.fromRegion,
              destComps,
              item.toRegion,
              this->operator[](item.fromIndex),
              a_srcComps);
      a_dest[item.toIndex].push_back(newT);
    }

  // Uncomment and Move this out of unpackReceivesToMe()  (ndk)
  completePendingSends(); // wait for sends from possible previous operation

  unpackReceivesToMe_append(a_dest, destComps, ncomp, factory, a_op); // nullOp in uniprocessor mode
}

#endif

---