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>


class NodeBroadcast {
public:
  CmiNodeLock lock;
  // Map from [id][tag] -> data
  //
  // std::any will be cast to type std::map<int, T> where T is the data type being broadcast.
  std::map<int, std::any> data;
  // Map from [id][tag] -> retrieval count
  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;

    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() { }

    T get(int tag, const int expected = -1) {
      T tmp;
      if (useShared) {
        CmiLock(nodeBroadcast->lock);
        // Attempt to retrieve data from map
        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];
        // Check if the data has been retrieved the expected number of times, if so, delete it.
        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);
        // Insert new data into map.
        std::map<int, T>* data = std::any_cast<std::map<int, T>*>(nodeBroadcast->data[id]);
        data->insert({tag, t});
        // Reset retrieval counter to 0.
        nodeBroadcast->current[id][tag] = 0;
        CmiUnlock(nodeBroadcast->lock);
      } else {
        BroadcastMgr::Object()->send(*this, tag, (void*)(&t), sizeof(T));
      }
    }
};

#endif