SynchronousCollectives Class Reference

#include <SynchronousCollectives.h>

Inheritance diagram for SynchronousCollectives:

Public Member Functions
	SynchronousCollectives ()
	~SynchronousCollectives ()
void	initAllScope ()
void	initMasterScope (const int isMasterPe, const int isMasterDevice, const int numDevices, const int deviceIndex, const std::vector< int > &masterPeList)
template<typename T >
std::vector< T >	allReduce (std::vector< T > &data, CkReduction::reducerType type, const SynchronousCollectiveScope scope)
template<typename T >
std::vector< T >	allGather (const T &data, const SynchronousCollectiveScope scope)
template<typename T >
std::vector< T >	alltoallv (const std::vector< T > &data, const SynchronousCollectiveScope scope)
template<typename T >
T	broadcast (const T &data, const bool isRoot, const SynchronousCollectiveScope scope)
void	waitAndAwaken ()
void	barrier (const SynchronousCollectiveScope scope)
void	forceBarrierAll ()
void	setupMulticastSection (SynchronousCollectivesMulticastMsg *msg)
void	handleReductionAll (CkReductionMsg *msg)
void	handleReductionMaster (CkReductionMsg *msg)
void	broadcastReductionResult (int n, char *data)
template<typename T >
void	recvIndexData (const int index, const T &data, const SynchronousCollectiveScope scope, const unsigned int key)
template<typename T >
void	recvBroadcast (const T &data, const unsigned int key)
void	wait ()
void	recvBarrierAll (const int PE)
void	recvBarrierMasterPe (const int deviceIndex)

Static Public Member Functions
static SynchronousCollectives *	Object ()
static SynchronousCollectives *	ObjectOnPe (const int pe)

Detailed Description

Definition at line 60 of file SynchronousCollectives.h.

Constructor & Destructor Documentation

SynchronousCollectives()

SynchronousCollectives::SynchronousCollectives

(

void

)

Definition at line 26 of file SynchronousCollectives.C.

References NAMD_bug().

{
  if (CkpvAccess(SynchronousCollectives_instance) == NULL) {
    CkpvAccess(SynchronousCollectives_instance) = this;
  } else {
    NAMD_bug("SynchronousCollectives instanced twice on same processor!");
  }
}

~SynchronousCollectives()

SynchronousCollectives::~SynchronousCollectives

(

void

)

Definition at line 35 of file SynchronousCollectives.C.

{ }

Member Function Documentation

allGather()

template<typename T >

std::vector< T > SynchronousCollectives::allGather	(	const T &	data,
		const SynchronousCollectiveScope	scope
	)

Definition at line 260 of file SynchronousCollectives.C.

References all, NAMD_bug(), NAMD_EVENT_START, NAMD_EVENT_STOP, and single.

Referenced by GlobalGPUMgr::initialize().

                                                                                                    {
  if (scope == SynchronousCollectiveScope::single) {
    NAMD_bug("SynchronousCollectives::allgather does not support single scope");
  }
  NAMD_EVENT_START(1, NamdProfileEvent::CHARM_ALLGATHER);

  std::vector<T> out;
  if (scope == SynchronousCollectiveScope::all || isMasterPe_) {
    const unsigned int key = getKey(scope);
    sendAllGather<T>(data, scope, key);
    suspendAndCheck(scope);

    out = retrieveTemp<std::vector<T>>(key);
  }
  NAMD_EVENT_STOP(1, NamdProfileEvent::CHARM_ALLGATHER);
  return out;
}

allReduce()

template<typename T >

std::vector< T > SynchronousCollectives::allReduce	(	std::vector< T > &	data,
		CkReduction::reducerType	type,
		const SynchronousCollectiveScope	scope
	)

Definition at line 187 of file SynchronousCollectives.C.

References all, handleReductionAll(), handleReductionMaster(), NAMD_bug(), NAMD_EVENT_START, NAMD_EVENT_STOP, and single.

                                    {
  if (scope == SynchronousCollectiveScope::single) {
    NAMD_bug("SynchronousCollectives::allreduce does not support single scope");
  }
  NAMD_EVENT_START(1, NamdProfileEvent::CHARM_ALLREDUCE);

  std::vector<T> out;
  reductionTemp_ = std::vector<T>(data.size());
  reductionPtr_ = (void*) std::any_cast<std::vector<T>&>(reductionTemp_).data();

  setThread(CthSelf());
  if (scope == SynchronousCollectiveScope::all) {
    CkCallback cb(CkReductionTarget(SynchronousCollectives, handleReductionAll),
      thisProxy[thisIndex]);

    contribute(data.size() * sizeof(T), data.data(), type, cb);
    suspendAndCheck(SynchronousCollectiveScope::single);
  } else if (isMasterPe_) {
    CkCallback cb(CkReductionTarget(SynchronousCollectives, handleReductionMaster),
      thisProxy[thisIndex]);

    CProxy_CkMulticastMgr mcastProxy = CkpvAccess(BOCclass_group).multicastMgr;
    CkMulticastMgr *mcastPtr = CProxy_CkMulticastMgr(mcastProxy).ckLocalBranch();
    mcastPtr->contribute(data.size() * sizeof(T), data.data(), type, reductionCookie_, cb);
    suspendAndCheck(SynchronousCollectiveScope::single);
  }
  out = std::move(std::any_cast<std::vector<T>&>(reductionTemp_));
  NAMD_EVENT_STOP(1, NamdProfileEvent::CHARM_ALLREDUCE);
  return out;
}

alltoallv()

template<typename T >

std::vector< T > SynchronousCollectives::alltoallv	(	const std::vector< T > &	data,
		const SynchronousCollectiveScope	scope
	)

Definition at line 296 of file SynchronousCollectives.C.

References all, NAMD_bug(), NAMD_EVENT_START, NAMD_EVENT_STOP, and single.

                                                                                                               {
  if (scope == SynchronousCollectiveScope::single) {
    NAMD_bug("SynchronousCollectives::alltoallv does not support single scope");
  }
  NAMD_EVENT_START(1, NamdProfileEvent::CHARM_ALLTOALL);

  std::vector<T> out;

  if (scope == SynchronousCollectiveScope::all || isMasterPe_) {
    const unsigned int key = getKey(scope);
    sendAlltoallv<T>(data, scope, key);
    suspendAndCheck(scope);
    out = retrieveTemp<std::vector<T>>(key);
  }

  NAMD_EVENT_STOP(1, NamdProfileEvent::CHARM_ALLTOALL);
  return out;
}

barrier()

void SynchronousCollectives::barrier

(

const SynchronousCollectiveScope

scope

)

Definition at line 145 of file SynchronousCollectives.C.

References all, forceBarrierAll(), master, NAMD_bug(), NAMD_EVENT_START, NAMD_EVENT_STOP, and single.

                                                                           {
  if (scope == SynchronousCollectiveScope::single) {
    NAMD_bug("SynchronousCollectives::barrier does not support single scope");
  }

  NAMD_EVENT_START(1, NamdProfileEvent::CHARM_BARRIER);
  if (scope == SynchronousCollectiveScope::all) {
    if (CkNumNodes() == 1) {
      CmiNodeBarrier();
    } else if (currentBarrierMasterPe_.size() == 0) {
      // If expectedBarrierMasterPe is not set, then we need to 
      // default back to a true all synchronization
      forceBarrierAll();
    } else {
      if (isMasterPe_) {
        barrier(SynchronousCollectiveScope::master);
      }
      CmiNodeBarrier();
    }
  } else if (isMasterPe_) {
    masterPes_.recvBarrierMasterPe(deviceIndex_);
    suspendAndCheck(SynchronousCollectiveScope::master);
  }
  NAMD_EVENT_STOP(1, NamdProfileEvent::CHARM_BARRIER);
}

broadcast()

template<typename T >

T SynchronousCollectives::broadcast	(	const T &	data,
		const bool	isRoot,
		const SynchronousCollectiveScope	scope
	)

Definition at line 332 of file SynchronousCollectives.C.

References all, NAMD_bug(), NAMD_EVENT_START, NAMD_EVENT_STOP, and single.

                                                                                                            {
  if (scope == SynchronousCollectiveScope::single) {
    NAMD_bug("SynchronousCollectives::broadcast does not support single scope");
  }
  NAMD_EVENT_START(1, NamdProfileEvent::CHARM_BROADCAST);

  T out = data;  // If we are not participating in the broadcast, just return the input
  if (scope == SynchronousCollectiveScope::all || isMasterPe_) {
    const unsigned int key = getKey(scope);
    if (isRoot) {
      sendBroadcast(data, scope, key);
    }
    suspendAndCheck(SynchronousCollectiveScope::single);

    out = retrieveTemp<T>(key);
  }
  NAMD_EVENT_STOP(1, NamdProfileEvent::CHARM_BROADCAST);
  return out;
}

broadcastReductionResult()

void SynchronousCollectives::broadcastReductionResult	(	int	n,
		char *	data
	)

Definition at line 181 of file SynchronousCollectives.C.

References single.

                                                                       {
  std::memcpy(reductionPtr_, (void*) data, n);
  incrementCount(SynchronousCollectiveScope::single, 0);
}

forceBarrierAll()

void SynchronousCollectives::forceBarrierAll

(

)

Definition at line 140 of file SynchronousCollectives.C.

References all.

Referenced by barrier(), GlobalGPUMgr::initialize(), initializeGPUResident(), and initMasterScope().

                                             {
  allPes_.recvBarrierAll(CkMyPe());
  suspendAndCheck(SynchronousCollectiveScope::all);
}

handleReductionAll()

void SynchronousCollectives::handleReductionAll

(

CkReductionMsg *

msg

)

Definition at line 171 of file SynchronousCollectives.C.

Referenced by allReduce().

                                                                   {
  allPes_.broadcastReductionResult(msg->getSize(), (char*) msg->getData()); 
  // TODO: Should this message be deleted here
}

handleReductionMaster()

void SynchronousCollectives::handleReductionMaster

(

CkReductionMsg *

msg

)

Definition at line 176 of file SynchronousCollectives.C.

Referenced by allReduce().

                                                                      {
  masterPes_.broadcastReductionResult(msg->getSize(), (char*) msg->getData()); 
  // TODO: Should this message be deleted here
}

initAllScope()

void SynchronousCollectives::initAllScope

(

)

Definition at line 37 of file SynchronousCollectives.C.

Referenced by Node::startup().

                                          {
  allPes_ = CkpvAccess(BOCclass_group).synchronousCollectives;
  currentBarrierAll_ = std::vector<int>(CkNumPes(), 0);
  currentBarrierSingle_ = std::vector<int>(1, 0);
}

initMasterScope()

void SynchronousCollectives::initMasterScope	(	const int	isMasterPe,
		const int	isMasterDevice,
		const int	numDevices,
		const int	deviceIndex,
		const std::vector< int > &	masterPeList
	)

Definition at line 43 of file SynchronousCollectives.C.

References forceBarrierAll(), masterPeList, and single.

Referenced by initializeGPUResident().

                                                                                   {

  isMasterPe_ = isMasterPe;
  numDevices_ = numDevices;
  deviceIndex_ = deviceIndex;
  masterPeList_ = masterPeList;

  currentBarrierMasterPe_ = std::vector<int>(numDevices_, 0);
  forceBarrierAll();  // Make sure all PEs have set expectedBarrierMasterPe

  masterPes_ = CProxySection_SynchronousCollectives(allPes_.ckGetGroupID(), 
    masterPeList_.data(), masterPeList_.size());
  masterPesMulticast_ = CProxySection_SynchronousCollectives(allPes_.ckGetGroupID(), 
    masterPeList_.data(), masterPeList_.size());

  //
  // For section broadcasts, we must use the multi-cast library; however, it requires explicitly
  // defined messages, so the multi-cast section will not be used for the non-reduction sections
  //
  CProxy_CkMulticastMgr mcastProxy = CkpvAccess(BOCclass_group).multicastMgr;
  CkMulticastMgr *mcastPtr = CProxy_CkMulticastMgr(mcastProxy).ckLocalBranch();
  masterPesMulticast_.ckSectionDelegate(mcastPtr);

  if (isMasterDevice && isMasterPe_) {
    SynchronousCollectivesMulticastMsg *msg = new SynchronousCollectivesMulticastMsg();
    setThread(CthSelf());
    masterPesMulticast_.setupMulticastSection(msg);
    suspendAndCheck(SynchronousCollectiveScope::single);
  } else if (isMasterPe_) {
    suspendAndCheck(SynchronousCollectiveScope::single);
  }

  forceBarrierAll();  // Make sure all PEs have set expectedBarrierMasterPe
}

Object()

static SynchronousCollectives* SynchronousCollectives::Object ( void  )

inlinestatic

Definition at line 63 of file SynchronousCollectives.h.

Referenced by GlobalGPUMgr::initialize(), initializeGPUResident(), Sequencer::Sequencer(), and Node::startup().

                                          { 
    return CkpvAccess(SynchronousCollectives_instance); 
  }

ObjectOnPe()

static SynchronousCollectives* SynchronousCollectives::ObjectOnPe ( const int  pe )

inlinestatic

Definition at line 66 of file SynchronousCollectives.h.

                                                          {
    return CkpvAccessOther(SynchronousCollectives_instance, CmiRankOf(pe));
  }

recvBarrierAll()

void SynchronousCollectives::recvBarrierAll

(

const int

PE

)

Definition at line 132 of file SynchronousCollectives.C.

References all.

                                                        {
  incrementCount(SynchronousCollectiveScope::all, PE);
}

recvBarrierMasterPe()

void SynchronousCollectives::recvBarrierMasterPe

(

const int

deviceIndex

)

Definition at line 136 of file SynchronousCollectives.C.

References master.

                                                                      {
  incrementCount(SynchronousCollectiveScope::master, deviceIndex);
}

recvBroadcast()

template<typename T >

void SynchronousCollectives::recvBroadcast	(	const T &	data,
		const unsigned int	key
	)

Definition at line 325 of file SynchronousCollectives.C.

References single.

                                                                                {
  // Since we are only expecting one message, the key should not exist in the map
  tempData_.insert({key, std::move(data)});
  incrementCount(SynchronousCollectiveScope::single, 0);
}

recvIndexData()

template<typename T >

void SynchronousCollectives::recvIndexData	(	const int	index,
		const T &	data,
		const SynchronousCollectiveScope	scope,
		const unsigned int	key
	)

Definition at line 235 of file SynchronousCollectives.C.

References all, and NAMD_die().

                                                                                                                                         {
  const int tempSize = (scope == SynchronousCollectiveScope::all) ? CkNumPes() : numDevices_;
  auto res = tempData_.try_emplace(key, std::in_place_type<std::vector<T>>, tempSize);

  std::vector<T>& tempVec = std::any_cast<std::vector<T>&>(res.first->second);
  if (index >= tempVec.size()) {
    NAMD_die("SynchronousCollectives::recvIndexData: temp array not large enough");
  }

  tempVec[index] = std::move(data);
  incrementCount(scope, index);
}

setupMulticastSection()

void SynchronousCollectives::setupMulticastSection

(

SynchronousCollectivesMulticastMsg *

msg

)

Definition at line 219 of file SynchronousCollectives.C.

References single.

                                                                                          {
  CkGetSectionInfo(reductionCookie_, msg);
  delete msg;
  incrementCount(SynchronousCollectiveScope::single, 0);
}

wait()

void SynchronousCollectives::wait

(

)

Definition at line 103 of file SynchronousCollectives.C.

Referenced by waitAndAwaken().

                                  {
  int finished = false;
  switch (waitPhase_) {
        case 0:
        break;
        case 1:
            finished = true;
        break;
  }
  
  waitPhase_++;
  if (!CkMyPe()) {
    if (!finished) {
      CkStartQD(CkCallback(CkIndex_SynchronousCollectives::wait(), thisgroup));
    } 
  }

  if (finished) {
    waitPhase_ = 0;
    CthAwaken(self_awaken_thread_);
  }
}

waitAndAwaken()

void SynchronousCollectives::waitAndAwaken

(

)

Definition at line 126 of file SynchronousCollectives.C.

References wait().

                                           {
  setThread(CthSelf());
  wait();
  CthSuspend();
}

---

The documentation for this class was generated from the following files:

• SynchronousCollectives.h
• SynchronousCollectives.C