src/Message/GlobalComm.cpp

Go to the documentation of this file.

// -*- C++ -*-
/***************************************************************************
 *
 * The IPPL Framework
 * 
 * This program was prepared by PSI. 
 * All rights in the program are reserved by PSI.
 * Neither PSI nor the author(s)
 * makes any warranty, express or implied, or assumes any liability or
 * responsibility for the use of this software
 *
 * Visit http://www.acl.lanl.gov/POOMS for more details
 *
 ***************************************************************************/

// -*- C++ -*-
/***************************************************************************
 *
 * The IPPL Framework
 * 
 *
 * Visit http://people.web.psi.ch/adelmann/ for more details
 *
 ***************************************************************************/

// include files
#include "Message/GlobalComm.h"
#include "Message/Communicate.h"
#include "Message/Message.h"
#include "Message/Tags.h"
#include "Utility/IpplInfo.h"
#include "Utility/IpplStats.h"
#include "Profile/Profiler.h"

#ifdef IPPL_STDSTL
// Standard STL names this file "algorithm"
#include <algorithm>
using namespace std;
#else
#include <algo.h>
#endif // IPPL_STDSTL


// Reduce equally-sized arrays across the machine, by sending to node
// 0 and broadcasting back the result.  The arguments are two begin,end
// iterators for the source of the data, an iterator pointing to
// where the summed data should go, and an operation to perform in
// the reduction ... this last should be from PETE, e.g., OpAdd, etc.
// template classes found in either ../Expression/Applicative.h or
// ../Expression/TypeComputations.h. The are simple classes  such as
// OpAnd(), OpMax(), OpAdd(), OpMultipply(), etc....
// Return success.
// The final argument indicates whether the LOCAL NODE should have it's
// values included in the reduction (by default, this is true).  If this
// pointer to the boolean array is null, all the values will be included.
// NOTE: The input iterators must iterate over simple data objects,
// which do not require their own special getMessage/putMessage.  If you
// need to reduce a complex quantity, use the scalar version of reduce.
template <class InputIterator, class OutputIterator, class ReduceOp>
bool reduce(Communicate& comm, InputIterator s1, InputIterator s2,
            OutputIterator t1, const ReduceOp& op, bool *IncludeVal) {

  TAU_TYPE_STRING(profileString, "bool (Communicate, " + CT(s1) + ", " 
    + CT(s2) + " )");

  TAU_PROFILE("reduce()", profileString, TAU_MESSAGE);

  // Inform dbgmsg("reduce-vector", INFORM_ALL_NODES);

  // determine destination node and tags
  int parent = 0;
  int sendtag = comm.next_tag(COMM_REDUCE_SEND_TAG, COMM_REDUCE_CYCLE);
  int rectag  = comm.next_tag(COMM_REDUCE_RECV_TAG, COMM_REDUCE_CYCLE);

  // determine how many elements we have to reduce
  unsigned int elements = 0;
  for (InputIterator tmps1 = s1; tmps1 != s2; ++tmps1, ++elements);
  if (elements == 0) {
    ERRORMSG("reduce: nothing to reduce." << endl);
  }

  // create flags, if they are not provided
  bool *useFlags = IncludeVal;
  if (useFlags == 0 && elements > 0) {
    useFlags = new bool[elements];
    for (unsigned int u=0; u < elements; useFlags[u++] = true);
  }

  if ( comm.myNode() != parent ) {
    // send the source data to node 0 if we are not node 0
    Message *msg = new Message;
    // dbgmsg << "sending message with " << elements << " elements to node ";
    // dbgmsg << parent << " with tag " << sendtag << endl;
    ::putMessage(*msg, elements);
    if (elements > 0) {
      ::putMessage(*msg, s1, s2);
      ::putMessage(*msg, useFlags, useFlags + elements);
    }
    if ( ! comm.send(msg, parent, sendtag) ) {
      Ippl::abort("reduce: cannot send reduce buffers.");
    }

    // then we get the results back
    msg = comm.receive_block(parent, rectag);
    // dbgmsg << "received message with size = " << msg->size();
    // dbgmsg << " from node " << parent << " with tag " << rectag << endl;
    if ( ! msg || msg->size() < 1 )
      Ippl::abort("reduce: cannot receive reduce results.");
    getMessage(*msg, *t1);
    delete msg;

  } else {
    // first copy the source into the target; this is like receiving
    // from ourselves
    InputIterator  tmp1  = s1;
    OutputIterator t2    = t1;
    bool*          copyf = useFlags;
    for ( ; tmp1 != s2; ++tmp1, ++t2, ++copyf)
      if (*copyf)
        *t2 = *tmp1;

    // the parent receives all the messages and then broadcasts the
    // reduced result
    int notReceived = comm.getNodes() - 1;
    while (notReceived > 0) {
      // receive message
      int fromnode = COMM_ANY_NODE;
      Message *recmsg = comm.receive_block(fromnode, sendtag);
      // dbgmsg << "received message with size = " << recmsg->size();
      // dbgmsg << " from node " << fromnode << " with tag "<<sendtag<<endl;
      if ( ! recmsg || recmsg->size() < 1 )
        Ippl::abort("reduce: cannot receive reduce buffers.");

      // get data from message
      int recelems;
      ::getMessage(*recmsg, recelems);
      if (recelems != elements)
        Ippl::abort("reduce: mismatched element count in vector reduction.");
      if (elements > 0) {
        InputIterator reci = (InputIterator)(recmsg->item(0).data());
        bool *recflag = (bool *)(recmsg->item(1).data());

        // the target buffer must have size >= size of the source, so
        // we can iterate over the source iterator
        unsigned int u;
        for (u=0, t2=t1; u < elements; ++t2, ++reci, ++u) {
          if (recflag[u]) {
            if (useFlags[u]) {
              PETE_apply(op, *t2, *reci);
            } else {
              *t2 = *reci;
              useFlags[u] = true;
            }
          }
        }
      }

      // finished with this node's data
      delete recmsg;
      notReceived--;
    }

    // Finally, broadcast the results out.  t2 should now point to the
    // end of the target buffer.
    if (comm.getNodes() > 1) {
      Message *sendmsg = new Message();
      putMessage(*sendmsg, t1, t2);
      // dbgmsg << "sending message with size " << sendmsg->size();
      // dbgmsg << " to all nodes with tag " << rectag << endl;
      if (comm.broadcast_others(sendmsg, rectag) != (comm.getNodes() - 1))
        Ippl::abort("reduce: cannot send reduce results.");
    }
  }

  // we're done
  if (useFlags != 0 && useFlags != IncludeVal)
    delete [] useFlags;

  //INCIPPLSTAT(incReductions);
  return true;
}


// same as above, but this uses the default Communicate object
template <class InputIterator, class OutputIterator, class ReduceOp>
bool reduce(InputIterator s1, InputIterator s2,
            OutputIterator t1, const ReduceOp& op, bool *IncludeVal) {

  TAU_TYPE_STRING(profileString, "bool (" + CT(s1) + ", " + CT(s2) + ", " 
    + CT(t1) + ", " + CT(op) + ")");
  TAU_PROFILE("reduce()", profileString, TAU_MESSAGE);

  return reduce(*Ippl::Comm, s1, s2, t1, op, IncludeVal);
}


// masked scalar versions of reduce ... instead of iterators, these versions
// expect a single quantity to reduce and a location to place the result.
// The final argument indicates whether the LOCAL NODE should have it's
// value included in the reduction (by default, this is true).
// Return success of operation.
template <class T, class ReduceOp>
bool reduce_masked(Communicate& comm, T& input, T& output,
                   const ReduceOp& op, bool IncludeVal) {

  TAU_TYPE_STRING(profileString, "bool (Communicate, " + CT(input) + ", " 
                  + CT(output) + " )");
  TAU_PROFILE("reduce_masked()", profileString, TAU_MESSAGE);

  // Inform dbgmsg("reduce_masked", INFORM_ALL_NODES);

  // determine destination node and tags
  int parent = 0;
  int sendtag = comm.next_tag(COMM_REDUCE_SEND_TAG, COMM_REDUCE_CYCLE);
  int rectag  = comm.next_tag(COMM_REDUCE_RECV_TAG, COMM_REDUCE_CYCLE);

  if (comm.myNode() != parent) {
    // send the source data to node 0 if we are not node 0
    Message *msg = new Message;
    // dbgmsg << "sending message, includeflag=" << IncludeVal << ", to node ";
    // dbgmsg << parent << " with tag " << sendtag << endl;
    ::putMessage(*msg, IncludeVal);
    if (IncludeVal)
      ::putMessage(*msg, input);
    if ( ! comm.send(msg, parent, sendtag) ) {
      Ippl::abort("reduce: cannot send reduce scalar.");
    }

    // then we get the results back
    msg = comm.receive_block(parent, rectag);
    // dbgmsg << "received message with size = " << msg->size();
    // dbgmsg << " from node " << parent << " with tag " << rectag << endl;
    if ( ! msg || msg->size() < 1 )
      Ippl::abort("reduce: cannot receive reduce results.");
    getMessage(*msg, output);
    delete msg;

  } else {
    // first copy the source into the target; this is like receiving
    // from ourselves
    if (IncludeVal)
      output = input;

    // if there are several nodes, we must get the other results
    if (comm.getNodes() > 1) {

      // the parent receives all the messages and then broadcasts the
      // reduced result
      int notReceived = comm.getNodes() - 1;

      // create a temporary array to store values from other nodes
      T *recval = new T[notReceived];
      bool *recflag = new bool[notReceived];

      // get all messages
      while (notReceived > 0) {
        // receive message
        int fromnode = COMM_ANY_NODE;
        Message *recmsg = comm.receive_block(fromnode, sendtag);
        if ( ! recmsg || recmsg->size() < 1 )
          Ippl::abort("reduce: cannot receive reduce buffers.");

        // get flag indicating if the message has any data; if it does,
        // get it and store it
	::getMessage(*recmsg, recflag[fromnode - 1]);
        if (recflag[fromnode - 1])
	  ::getMessage(*recmsg, recval[fromnode - 1]);

        // finished with this node's data
        delete recmsg;
        notReceived--;
      }

      // now loop through the received values and do the reduction
      for (int n=1; n < comm.getNodes(); ++n) {
        if (recflag[n-1]) {
          if (IncludeVal) {
            PETE_apply(op, output, recval[n-1]);
          } else {
            output = recval[n-1];
            IncludeVal = true;
          }
        }
      }

      // done with the temporary storage
      delete [] recflag;
      delete [] recval;
    }

    // Finally, broadcast the results out.  t2 should now point to the
    // end of the target buffer.
    if (comm.getNodes() > 1) {
      Message *sendmsg = new Message();
      ::putMessage(*sendmsg, output);
      // dbgmsg << "sending message with size " << sendmsg->size();
      // dbgmsg << " to all nodes with tag " << rectag << endl;
      if (comm.broadcast_others(sendmsg, rectag) != (comm.getNodes() - 1))
        Ippl::abort("reduce: cannot send reduce results.");
    }

    // we're done ... but do a check to see that we reduced SOMETHING
    if (!IncludeVal) {
      WARNMSG("reduce: there was nothing to reduce, since the masks ");
      WARNMSG("were all false." << endl);
    }
  }

  //INCIPPLSTAT(incReductions);
  return true;
}


// same as above, but this uses the default Communicate object
template <class T, class ReduceOp>
bool reduce_masked(T& input, T& output, const ReduceOp& op,
                   bool IncludeVal) {

  TAU_TYPE_STRING(profileString, "bool (" + CT(input) + ", " 
    + CT(output) + " )");
  TAU_PROFILE("reduce()", profileString, TAU_MESSAGE);

  return reduce_masked(*Ippl::Comm, input, output, op, IncludeVal);
}


// Scatter the data in the given source container to all other nodes.
// The data is read using the first two begin,end iterators, and written
// to the location indicated by the third iterator.  The next two
// arrays are for target nodes and target indices for the data in the
// source array; they should be of the same length as the source array.
// the final argument is an STL predicate which is used to combine data
// when two or more items are scattered into the same location on the
// same node.
// Return success of operation.
template <class InputIterator, class RandomIterator, class ScatterOp>
bool scatter(Communicate& comm, InputIterator s1, InputIterator s2,
             RandomIterator t1, int *target_node,
             int *target_position, const ScatterOp& op) {

  TAU_TYPE_STRING(profileString, "bool (Communicate&, " + CT(s1) + ", " 
    + CT(s2) + ", " + CT(t1) + ", int *, int *, " + CT(op) + ")");
  TAU_PROFILE("scatter()", profileString, TAU_MESSAGE);
  
  int i;                        // loop variables
  int tag = comm.next_tag(COMM_REDUCE_SCATTER_TAG, COMM_REDUCE_CYCLE);

  // Create a number of send messages equal to TotalNodes
  // these messages will be packed with the data from the source
  // data and sent to the node indicated by target node array
  // some empty messages will be sent so the recieving node knows when
  // it has recieved all the messages
  Message* msg = new Message[comm.getNodes()];

  // Loop over each item of the source array and pack the send messages.
  // The message is packed in pairs, the first element of each pair is
  // an integer representing the array offset in the target. The second
  // element is the data to be placed in that offset.
  int *tn = target_node;
  int *tp = target_position;
  InputIterator si;
  for ( si = s1; si != s2 ; si++, tn++, tp++ ) {
    if ( *tn < 0 || *tn >= comm.getNodes() ) {
      ERRORMSG("scatter: bad scatter target " << *tn << endl);
      return false;
    }
    //    msg[*tn].put(*tp).put(*si);
    putMessage(msg[*tn], *tp);
    putMessage(msg[*tn], *si);
  }

  // Send out the messages.  We do not delete the messages here after the
  // send, however.
  for ( i = comm.getNodes() - 1; i >= 0; i-- ) {
    if ( ! comm.send(msg + i, i, tag, false) ) {
      ERRORMSG("scatter: cannot send scatter buffer " << i << endl);
      return false;
    }
  }

  // Receive the scatter messages back now.
  int notReceived = comm.getNodes();
  while (notReceived > 0) {
    int fromnode = COMM_ANY_NODE;
    Message *recmsg = comm.receive_block(fromnode, tag);
    if ( ! recmsg ) {
      ERRORMSG("scatter: cannot receive scatter message." << endl);
      return false;
    }

    // for each (pos, val) pair, get it and put results in target storage
    int pairs = recmsg->size() / 2;
    int datapos;
    InputIterator reci;
    for ( i = 0 ; i < pairs ; i++ ) {
      //      recmsg->get(datapos);
      getMessage(*recmsg, datapos);
      reci = (InputIterator)(recmsg->item(0).data());
      PETE_apply(op, t1[datapos], *reci);
      recmsg->get();    // cleans out the item without another copy
    }

    // Finished with this message.  Delete it if it is from another node; if
    // it is not, we sent it to ourselves and will delete it later.
    if ( fromnode != comm.myNode() )
      delete recmsg;
    notReceived--;
  }

  // at the end, delete the scatter messages, and return success
  delete [] msg;

  //INCIPPLSTAT(incScatters);
  return true;
}


// same as above, but this uses the default Communicate object
template <class InputIterator, class RandomIterator, class ScatterOp>
bool scatter(InputIterator s1, InputIterator s2,
             RandomIterator t1, int *target_node,
             int *target_position, const ScatterOp& op) {

  TAU_TYPE_STRING(profileString, "bool (" + CT(s1) + ", " 
    + CT(s2) + ", " + CT(t1) + ", " + "int *, int *, " + CT(op) + ")");
  TAU_PROFILE("scatter()", profileString, TAU_MESSAGE);

  return scatter(*Ippl::Comm, s1, s2, t1, target_node, target_position, op);
}


/***************************************************************************
 * $RCSfile: GlobalComm.cpp,v $   $Author: adelmann $
 * $Revision: 1.1.1.1 $   $Date: 2003/01/23 07:40:28 $
 * IPPL_VERSION_ID: $Id: GlobalComm.cpp,v 1.1.1.1 2003/01/23 07:40:28 adelmann Exp $ 
 ***************************************************************************/

---