BroadcastObject.h

Go to the documentation of this file.

/*
   Coordinates broadcast of a data type from a Controller/Seq
   to all other Controller/Sequencer type objects (they must
   run in a thread!)
*/

#ifndef _BCASTOBJ_H
#define _BCASTOBJ_H

#include "BroadcastMgr.h"
#include "BroadcastClient.h"
#include "LdbCoordinator.h"
#include "Node.h"
#include "SimParameters.h"
#include "common.h"

#include <map>
#include <any>


/*
 * @brief Broadcast object for intra-node GPU-resident broadcasts
 *
 * There will exist one of these in each logical node to facilitate shared memory
 * broadcasts between PEs within that logic node
 *
 * The data object stores the actual data associated with a given ID, and the current
 * objects stores the retrieval count for a given ID,tag pair. Once the retrieval count
 * has been reached, the entry for that ID in data is cleared
 */
class NodeBroadcast {
public:
  CmiNodeLock lock;
  std::map<int, std::any> data;
  std::map<int, std::map<int, int>> current;
};

template <class T> class SimpleBroadcastObject : public BroadcastClient {

  public:

    const LDObjHandle *ldObjPtr;

    bool useShared = false;
    int id = -1;
    NodeBroadcast *nodeBroadcast;

    /*
     * @brief Creates a new SimpleBroadcastObject with given id
     *
     * NodeBroadcast is used by the shared memory backend and the same pointer should be given to all
     * objects on all Pes. 
     *
     * Even if we are using the shared memory backend, some of the broadcast objects must still use 
     * the charm++ backend since they are used for barriers, like scriptBarrier. The shared memory backend must
     * be explicitly enabled using the useSharedIfPossible parameter
     *
     */
    SimpleBroadcastObject(int id_in, const LDObjHandle *h = 0, 
      NodeBroadcast* nodeBroadcastIn = 0, const bool useSharedIfPossible = false) 
      : BroadcastClient(id_in), ldObjPtr(h), nodeBroadcast(nodeBroadcastIn), id(id_in) {
      if ( sizeof(T) > BCASTMSGSIZE ) {
        NAMD_bug("SimpleBroadcastObject instantiated on class larger than BCASTMSGSIZE");
      }

      if (nodeBroadcast != 0) {
        useShared = Node::Object()->simParameters->GPUresidentSingleProcessMode && useSharedIfPossible;

        if (useShared) {
          // Add a map for this broadcast object if needed
          CmiLock(nodeBroadcast->lock);
          if (nodeBroadcast->data.find(id) == nodeBroadcast->data.end()) {
            nodeBroadcast->data[id] = new std::map<int, T>();
          }
          CmiUnlock(nodeBroadcast->lock);
        }
      }
    }
    ~SimpleBroadcastObject() { }

    /*
     * @brief Attempts to retrieve a previously published value for a given tag and id
     *
     * This function has slightly different behavior between the shared memory and charm++ backends. The Charm++
     * backend will suspend and wait if it attempts to retrieve an unpublished tag, while the shared memory 
     * backend will throw an error. The shared memory backend also should be given the expected number of 
     * retrievals so the data can be removed 
     *
     */
    T get(int tag, const int expected = -1) {
      T tmp;
      if (useShared) {
        CmiLock(nodeBroadcast->lock);
        if (nodeBroadcast->data.find(id) == nodeBroadcast->data.end()) {
          NAMD_bug("SimpleBroadcastObject: Did not find id");
        }
        std::map<int, T>* data = std::any_cast<std::map<int, T>*>(nodeBroadcast->data[id]);
        if (data->find(tag) == data->end()) {
          NAMD_bug("SimpleBroadcastObject: Did not find tag");
        } 
        tmp = (*data)[tag];
        nodeBroadcast->current[id][tag]++;
        if (nodeBroadcast->current[id][tag] == expected) {
          data->erase(tag);
          nodeBroadcast->current[id].erase(tag);
        }
        CmiUnlock(nodeBroadcast->lock);
      } else {
        if ( ldObjPtr ) LdbCoordinator::Object()->pauseWork(*ldObjPtr);
        while ( BroadcastMgr::Object()->getbuf(*this, tag, (void*)(&tmp)) < 0 ) {
          suspendFor(tag);
        }
        if ( ldObjPtr ) LdbCoordinator::Object()->startWork(*ldObjPtr);
      }
      return tmp;
    }
  int getSize() {
    return BroadcastMgr::Object()->boidSize();
  }
    int getBcastSize() {
    return BroadcastMgr::Object()->boidBroadcastSize();
  }
      int getTaggedSize() {
    return BroadcastMgr::Object()->boidTaggedMsgSize();
  }
    void publish(int tag,const T &t ) {
      if (useShared) {
        CmiLock(nodeBroadcast->lock);
        std::map<int, T>* data = std::any_cast<std::map<int, T>*>(nodeBroadcast->data[id]);
        data->insert({tag, t});
        nodeBroadcast->current[id][tag] = 0;
        CmiUnlock(nodeBroadcast->lock);
      } else {
        BroadcastMgr::Object()->send(*this, tag, (void*)(&t), sizeof(T));
      }
    }
};

#endif