DataExchanger.C File Reference

#include <stdio.h>
#include "DataExchanger.h"
#include "ProcessorPrivate.h"
#include "common.h"
#include "Node.h"
#include "CollectionMaster.h"
#include "Output.h"
#include "ScriptTcl.h"
#include "qd.h"
#include "DataExchanger.def.h"

Go to the source code of this file.

Functions
	CpvDeclare (int, breakScheduler)
	CpvDeclare (int, inEval)
void	packSend (int dst, int dstPart, const char *data, int size, int handler, int code)
void	sendReplicaDcdInit (int dstPart, ReplicaDcdInitMsg *msg, int msgsize)
void	sendReplicaDcdData (int dstPart, ReplicaDcdDataMsg *msg, int msgsize)
void	sendReplicaDcdAck (int dstPart, ReplicaDcdAckMsg *msg)
void	recvReplicaDcdInit (ReplicaDcdInitMsg *msg)
void	recvReplicaDcdData (ReplicaDcdDataMsg *msg)
void	recvReplicaDcdAck (ReplicaDcdAckMsg *msg)
void	recvData (DataMessage *dmsg)
void	recvAck (DataMessage *dmsg)
void	recvBcast (DataMessage *dmsg)
void	recvRed (DataMessage *dmsg)
void	recvEvalCommand (DataMessage *dmsg)
void	recvEvalResult (DataMessage *dmsg)
void	replica_send (const char *sndbuf, int sendcount, int destPart, int destPE)
void	replica_recv (DataMessage **precvMsg, int srcPart, int srcPE)
void	replica_sendRecv (const char *sndbuf, int sendcount, int destPart, int destPE, DataMessage **precvMsg, int srcPart, int srcPE)
void	replica_eval (const char *cmdbuf, int targPart, int targPE, DataMessage **precvMsg)
void	replica_barrier ()
void	replica_bcast (char *buf, int count, int root)
void	replica_min_double (double *dat, int count)
void	initializeReplicaConverseHandlers ()

Variables
static int	recvRedCalledEarly

Function Documentation

CpvDeclare	(	int	,
		breakScheduler
	)

CpvDeclare	(	int	,
		inEval
	)

void initializeReplicaConverseHandlers

(

)

Definition at line 252 of file DataExchanger.C.

References recvAck(), recvBcast(), recvData(), recvEvalCommand(), recvEvalResult(), recvRed(), recvReplicaDcdAck(), recvReplicaDcdData(), and recvReplicaDcdInit().

Referenced by ProcessorPrivateInit().

                                         {
  CkpvInitialize(int, recv_data_idx);
  CkpvInitialize(int, recv_ack_idx);
  CkpvInitialize(int, recv_bcast_idx);
  CkpvInitialize(int, recv_red_idx);
  CkpvInitialize(int, recv_eval_command_idx);
  CkpvInitialize(int, recv_eval_result_idx);
  CkpvInitialize(int, recv_replica_dcd_init_idx);
  CkpvInitialize(int, recv_replica_dcd_data_idx);
  CkpvInitialize(int, recv_replica_dcd_ack_idx);

  CkpvAccess(recv_data_idx) = CmiRegisterHandler((CmiHandler)recvData);                   
  CkpvAccess(recv_ack_idx) = CmiRegisterHandler((CmiHandler)recvAck);                     
  CkpvAccess(recv_red_idx) = CmiRegisterHandler((CmiHandler)recvRed);                     
  CkpvAccess(recv_bcast_idx) = CmiRegisterHandler((CmiHandler)recvBcast);                 
  CkpvAccess(recv_eval_command_idx) = CmiRegisterHandler((CmiHandler)recvEvalCommand);    
  CkpvAccess(recv_eval_result_idx) = CmiRegisterHandler((CmiHandler)recvEvalResult);      
  CkpvAccess(recv_replica_dcd_init_idx) = CmiRegisterHandler((CmiHandler)recvReplicaDcdInit);
  CkpvAccess(recv_replica_dcd_data_idx) = CmiRegisterHandler((CmiHandler)recvReplicaDcdData);
  CkpvAccess(recv_replica_dcd_ack_idx) = CmiRegisterHandler((CmiHandler)recvReplicaDcdAck);

#if CMK_IMMEDIATE_MSG && CMK_SMP && ! ( CMK_MULTICORE || CMK_SMP_NO_COMMTHD )
  int sleep_commthd_idx = CmiRegisterHandler((CmiHandler)recvSleepCommthdMsg);
  if ( CkMyPe() == 0 && CkNumNodes() == 1 ) {
    CkPrintf("Charm++ communication thread will sleep due to single-process run.\n");
    SleepCommthdMsg *msg = (SleepCommthdMsg *)malloc(sizeof(SleepCommthdMsg));
    CmiSetHandler(msg, sleep_commthd_idx);
    CmiBecomeImmediate(msg);
    CmiPushImmediateMsg(msg);
  }
#endif
}

void packSend	(	int	dst,
		int	dstPart,
		const char *	data,
		int	size,
		int	handler,
		int	code
	)

Definition at line 34 of file DataExchanger.C.

References DataMessage::setMessage().

Referenced by replica_send().

                                                                                         {
    int msgsize = sizeof(DataMessage) + size;
    DataMessage *dmsg = (DataMessage *)CmiAlloc(msgsize);
    dmsg->setMessage(data,CkMyPe(),CmiMyPartition(),size,handler,code);
#if CMK_HAS_PARTITION
    CmiInterSyncSendAndFree(dst,dstPart,msgsize,(char*)dmsg);
#else
    CmiSyncSendAndFree(dst,msgsize,(char*)dmsg);
#endif
  }

void recvAck

(

DataMessage *

dmsg

)

Definition at line 99 of file DataExchanger.C.

References CPROXY_DE, and NAMD_bug().

Referenced by initializeReplicaConverseHandlers().

                                  {
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) NAMD_bug("BOCgroup::dataExchanger is zero in recvAck!");
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_ack();
    CmiFree(dmsg);
  }

void recvBcast

(

DataMessage *

dmsg

)

Definition at line 105 of file DataExchanger.C.

References CPROXY_DE, and NAMD_bug().

Referenced by initializeReplicaConverseHandlers().

                                    {
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) NAMD_bug("BOCgroup::dataExchanger is zero in recvBcast!");
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_bcast();
    CmiFree(dmsg);
  }

void recvData

(

DataMessage *

dmsg

)

Definition at line 93 of file DataExchanger.C.

References CPROXY_DE, and NAMD_bug().

Referenced by initializeReplicaConverseHandlers().

                                   {
    Pointer msg(dmsg);
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) NAMD_bug("BOCgroup::dataExchanger is zero in recvData!");
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_data(msg);
  }

void recvEvalCommand

(

DataMessage *

dmsg

)

Definition at line 121 of file DataExchanger.C.

References CPROXY_DE, and NAMD_bug().

Referenced by initializeReplicaConverseHandlers().

                                          {
    Pointer msg(dmsg);
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) NAMD_bug("BOCgroup::dataExchanger is zero in recvEvalCommand!");
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_eval_command(msg);
  }

void recvEvalResult

(

DataMessage *

dmsg

)

Definition at line 127 of file DataExchanger.C.

References CPROXY_DE, and NAMD_bug().

Referenced by initializeReplicaConverseHandlers().

                                         {
    Pointer msg(dmsg);
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) NAMD_bug("BOCgroup::dataExchanger is zero in recvEvalResult!");
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_eval_result(msg);
  }

void recvRed

(

DataMessage *

dmsg

)

Definition at line 111 of file DataExchanger.C.

References CPROXY_DE, and recvRedCalledEarly.

Referenced by initializeReplicaConverseHandlers().

                                  {
    if ( CkpvAccess(BOCclass_group).dataExchanger.isZero() ) {
      ++recvRedCalledEarly;
      CmiFree(dmsg);
      return;
    }
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_red();
    CmiFree(dmsg);
  }

void recvReplicaDcdAck

(

ReplicaDcdAckMsg *

msg

)

Definition at line 87 of file DataExchanger.C.

References CollectionMaster::Object(), and CollectionMaster::unblockPositions().

Referenced by initializeReplicaConverseHandlers().

                                                {
    CmiFree(msg);
    CollectionMaster::Object()->unblockPositions();
    CkpvAccess(_qd)->process();
  }

void recvReplicaDcdData

(

ReplicaDcdDataMsg *

msg

)

Definition at line 82 of file DataExchanger.C.

References Node::Object(), Node::output, and Output::recvReplicaDcdData().

Referenced by initializeReplicaConverseHandlers().

                                                  {
    Node::Object()->output->recvReplicaDcdData(msg);
    CmiFree(msg);
  }

void recvReplicaDcdInit

(

ReplicaDcdInitMsg *

msg

)

Definition at line 77 of file DataExchanger.C.

References Node::Object(), Node::output, and Output::recvReplicaDcdInit().

Referenced by initializeReplicaConverseHandlers().

                                                  {
    Node::Object()->output->recvReplicaDcdInit(msg);
    CmiFree(msg);
  }

void replica_barrier

(

)

Definition at line 168 of file DataExchanger.C.

References CPROXY_DE, and recvRedCalledEarly.

Referenced by colvarproxy_namd::replica_comm_barrier(), and ScriptTcl::run().

                         {
    for ( ; recvRedCalledEarly > 0; --recvRedCalledEarly ) CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv_red();
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].barrier();
    CpvAccess(breakScheduler) = 0;
    while(!CpvAccess(breakScheduler)) CsdSchedulePoll();
  }

void replica_bcast	(	char *	buf,
		int	count,
		int	root
	)

Definition at line 175 of file DataExchanger.C.

References DataMessage::data, NAMD_bug(), replica_recv(), replica_send(), and DataMessage::size.

Referenced by replica_min_double().

                                                     {
    if ( root != 0 ) NAMD_bug("replica_bcast root must be zero");
    int rank = CmiMyPartition();
    int size = CmiNumPartitions();
    int i;
    for ( i=1; i<size; i*=2 );
    for ( i/=2; i>0; i/=2 ) {
      if ( rank & (i - 1) ) continue;
      if ( rank & i ) {
        int src = rank - i;
        //CkPrintf("rank %d recv from %d\n", rank, src);
        DataMessage *recvMsg = NULL;
        replica_recv(&recvMsg, src, CkMyPe());
        if ( recvMsg == NULL ) NAMD_bug("recvMsg == NULL in replica_bcast");
        if ( recvMsg->size != count ) NAMD_bug("size != count in replica_bcast");
        memcpy(buf, recvMsg->data, count);
        CmiFree(recvMsg);
      } else {
        int dst = rank + i;
        if ( dst < size ) {
          //CkPrintf("rank %d send to %d\n", rank, dst);
          replica_send(buf, count, dst, CkMyPe());
        }
      }
    }
  }

void replica_eval	(	const char *	cmdbuf,
		int	targPart,
		int	targPE,
		DataMessage **	precvMsg
	)

Definition at line 159 of file DataExchanger.C.

References CPROXY_DE.

                                                                                          {
    ConstPointer sendPointer(cmdbuf);
    Pointer recvPointer((char *) precvMsg);
    int sendcount = strlen(cmdbuf) + 1;
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].eval(sendPointer,sendcount,targPart,targPE,recvPointer);
    CpvAccess(breakScheduler) = 0;
    while(!CpvAccess(breakScheduler)) CsdSchedulePoll();
  }

void replica_min_double	(	double *	dat,
		int	count
	)

Definition at line 202 of file DataExchanger.C.

References DataMessage::data, NAMD_bug(), replica_bcast(), replica_recv(), replica_send(), and DataMessage::size.

Referenced by PatchMap::makePatches().

                                                  {
    int rank = CmiMyPartition();
    int size = CmiNumPartitions();
    for ( int i=1; i<size; i*=2 ) {
      if ( rank & i ) {
        int dst = rank - i;
        //CkPrintf("rank %d send to %d\n", rank, dst);
        replica_send((char*)dat, count * sizeof(double), dst, CkMyPe());
      } else {
        int src = rank + i;
        if ( src < size ) {
          //CkPrintf("rank %d recv from %d\n", rank, src);
          DataMessage *recvMsg = NULL;
          replica_recv(&recvMsg, src, CkMyPe());
          if ( recvMsg == NULL ) NAMD_bug("recvMsg == NULL in replica_bcast");
          if ( recvMsg->size != count * sizeof(double) ) NAMD_bug("size != count in replica_min_double");
          double *rdat = new double[count];
          memcpy(rdat, recvMsg->data, count * sizeof(double));
          CmiFree(recvMsg);
          for ( int j=0; j<count; ++j ) {
            if ( rdat[j] < dat[j] ) dat[j] = rdat[j];
          }
          delete [] rdat;
        }
      }
      if ( rank & (2 * i - 1) ) break;
    }
    replica_bcast((char*)dat, count * sizeof(double), 0);
  }

void replica_recv	(	DataMessage **	precvMsg,
		int	srcPart,
		int	srcPE
	)

Definition at line 144 of file DataExchanger.C.

References CPROXY_DE.

Referenced by replica_bcast(), colvarproxy_namd::replica_comm_recv(), and replica_min_double().

                                                                    {
    Pointer recvPointer((char *) precvMsg);
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].recv(recvPointer,srcPart,srcPE);
    CpvAccess(breakScheduler) = 0;
    while(!CpvAccess(breakScheduler)) CsdSchedulePoll();
  }

void replica_send	(	const char *	sndbuf,
		int	sendcount,
		int	destPart,
		int	destPE
	)

Definition at line 133 of file DataExchanger.C.

References CPROXY_DE, and packSend().

Referenced by replica_bcast(), colvarproxy_namd::replica_comm_send(), and replica_min_double().

                                                                                 {
    if ( CpvAccess(inEval) ) {
      packSend(destPE,destPart,sndbuf,sendcount,CkpvAccess(recv_data_idx),1);
      return;
    }
    ConstPointer sendPointer(sndbuf);
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].send(sendPointer,sendcount,destPart,destPE); 
    CpvAccess(breakScheduler) = 0;
    while(!CpvAccess(breakScheduler)) CsdSchedulePoll();
  }

void replica_sendRecv	(	const char *	sndbuf,
		int	sendcount,
		int	destPart,
		int	destPE,
		DataMessage **	precvMsg,
		int	srcPart,
		int	srcPE
	)

Definition at line 151 of file DataExchanger.C.

References CPROXY_DE.

                                                                                                                                      {
    ConstPointer sendPointer(sndbuf);
    Pointer recvPointer((char *) precvMsg);
    CPROXY_DE(CkpvAccess(BOCclass_group).dataExchanger)[CkMyPe()].sendRecv(sendPointer,sendcount,destPart,destPE,recvPointer,srcPart,srcPE);
    CpvAccess(breakScheduler) = 0;
    while(!CpvAccess(breakScheduler)) CsdSchedulePoll();
  }

void sendReplicaDcdAck	(	int	dstPart,
		ReplicaDcdAckMsg *	msg
	)

Definition at line 67 of file DataExchanger.C.

References ReplicaDcdAckMsg::core.

Referenced by Output::recvReplicaDcdData(), and Output::recvReplicaDcdInit().

                                                             {
    CmiSetHandler(msg->core, CkpvAccess(recv_replica_dcd_ack_idx));
    int msgsize = sizeof(ReplicaDcdAckMsg);
#if CMK_HAS_PARTITION
    CmiInterSyncSendAndFree(0,dstPart,msgsize,(char*)msg);
#else
    CmiSyncSendAndFree(0,msgsize,(char*)msg);
#endif
  }

void sendReplicaDcdData	(	int	dstPart,
		ReplicaDcdDataMsg *	msg,
		int	msgsize
	)

Definition at line 56 of file DataExchanger.C.

References CollectionMaster::blockPositions(), ReplicaDcdDataMsg::core, and CollectionMaster::Object().

                                                                            {
    CmiSetHandler(msg->core, CkpvAccess(recv_replica_dcd_data_idx));
#if CMK_HAS_PARTITION
    CmiInterSyncSendAndFree(0,dstPart,msgsize,(char*)msg);
#else
    CmiSyncSendAndFree(0,msgsize,(char*)msg);
#endif
    CollectionMaster::Object()->blockPositions();  // ensure ordering
    CkpvAccess(_qd)->create();  // ensure completion
  }

void sendReplicaDcdInit	(	int	dstPart,
		ReplicaDcdInitMsg *	msg,
		int	msgsize
	)

Definition at line 45 of file DataExchanger.C.

References CollectionMaster::blockPositions(), ReplicaDcdInitMsg::core, and CollectionMaster::Object().

Referenced by Output::replicaDcdInit().

                                                                            {
    CmiSetHandler(msg->core, CkpvAccess(recv_replica_dcd_init_idx));
#if CMK_HAS_PARTITION
    CmiInterSyncSendAndFree(0,dstPart,msgsize,(char*)msg);
#else
    CmiSyncSendAndFree(0,msgsize,(char*)msg);
#endif
    CollectionMaster::Object()->blockPositions();  // ensure ordering
    CkpvAccess(_qd)->create();  // ensure completion
  }

Variable Documentation

int recvRedCalledEarly

static

Copyright (c) 1995, 1996, 1997, 1998, 1999, 2000 by The Board of Trustees of the University of Illinois. All rights reserved.

Definition at line 27 of file DataExchanger.C.

Referenced by recvRed(), and replica_barrier().