NAMD
Communicate.C
Go to the documentation of this file.
1 
7 #include <string.h>
8 #include <stdlib.h>
9 #include "Communicate.h"
10 #include "MStream.h"
11 #include "charm++.h"
12 
13 CkpvStaticDeclare(CmmTable, CsmMessages);
14 CkpvStaticDeclare(int, CsmAcks);
15 
16 static void CsmHandler(void *msg)
17 {
18  if ( CmiMyRank() ) NAMD_bug("Communicate CsmHandler on non-rank-zero pe");
19  // get start of user message
20  int *m = (int *) ((char *)msg+CmiMsgHeaderSizeBytes);
21  // sending node & tag act as tags
22  CmmPut(CkpvAccess(CsmMessages), 2, m, msg);
23 }
24 
25 static void CsmAckHandler(void *msg)
26 {
27  if ( CmiMyRank() ) NAMD_bug("Communicate CsmAckHandler on non-rank-zero pe");
28  CmiFree(msg);
29  CkpvAccess(CsmAcks) += 1;
30 }
31 
33 {
34  CkpvInitialize(CmmTable, CsmMessages);
35  CsmHandlerIndex = CmiRegisterHandler((CmiHandler) CsmHandler);
36  CsmAckHandlerIndex = CmiRegisterHandler((CmiHandler) CsmAckHandler);
37  CkpvAccess(CsmMessages) = CmmNew();
38 
39  int parent_node = 0;
40  int self = CkMyNode();
41  int range_begin = 0;
42  int range_end = CkNumNodes();
43  while ( self != range_begin ) {
44  parent_node = range_begin;
45  ++range_begin;
46  int split = range_begin + ( range_end - range_begin ) / 2;
47  if ( self < split ) { range_end = split; }
48  else { range_begin = split; }
49  }
50  int send_near = self + 1;
51  int send_far = send_near + ( range_end - send_near ) / 2;
52 
53  parent = CkNodeFirst(parent_node);
54  nchildren = 0;
55  if ( send_far < range_end ) children[nchildren++] = CkNodeFirst(send_far);
56  if ( send_near < send_far ) children[nchildren++] = CkNodeFirst(send_near);
57 
58  CkpvInitialize(int, CsmAcks);
59  CkpvAccess(CsmAcks) = nchildren;
60 
61  ackmsg = (char *) CmiAlloc(CmiMsgHeaderSizeBytes);
62  CmiSetHandler(ackmsg, CsmAckHandlerIndex);
63 }
64 
65 
67 {
68  CmiFree(ackmsg);
69 }
70 
72 {
73  MIStream *st = new MIStream(this, PE, tag);
74  return st;
75 }
76 
77 MOStream *Communicate::newOutputStream(int PE, int tag, unsigned int bufSize)
78 {
79  MOStream *st = new MOStream(this, PE, tag, bufSize);
80  return st;
81 }
82 
83 void *Communicate::getMessage(int PE, int tag)
84 {
85  if ( CmiMyRank() ) NAMD_bug("Communicate::getMessage called on non-rank-zero Pe\n");
86 
87  int itag[2], rtag[2];
88  void *msg;
89 
90  itag[0] = (PE==(-1)) ? (CmmWildCard) : PE;
91  itag[1] = (tag==(-1)) ? (CmmWildCard) : tag;
92  while((msg=CmmGet(CkpvAccess(CsmMessages),2,itag,rtag))==0) {
93  CmiDeliverMsgs(0);
94  // CmiDeliverSpecificMsg(CsmHandlerIndex);
95  }
96 
97  CmiSyncSend(parent, CmiMsgHeaderSizeBytes, ackmsg);
98 
99  while ( CkpvAccess(CsmAcks) < nchildren ) {
100  CmiDeliverMsgs(0);
101  // CmiDeliverSpecificMsg(CsmAckHandlerIndex);
102  }
103  CkpvAccess(CsmAcks) = 0;
104 
105  int size = SIZEFIELD(msg);
106  for ( int i = 0; i < nchildren; ++i ) {
107  CmiSyncSend(children[i],size,(char*)msg);
108  }
109 
110  return msg;
111 }
112 
113 void Communicate::sendMessage(int PE, void *msg, int size)
114 {
115  if ( CmiMyPe() ) NAMD_bug("Communicate::sendMessage not from Pe 0");
116 
117  while ( CkpvAccess(CsmAcks) < nchildren ) {
118  CmiDeliverMsgs(0);
119  // CmiDeliverSpecificMsg(CsmAckHandlerIndex);
120  }
121  CkpvAccess(CsmAcks) = 0;
122 
123  CmiSetHandler(msg, CsmHandlerIndex);
124  switch(PE) {
125  case ALL:
126  NAMD_bug("Unexpected Communicate::sendMessage(ALL,...)");
127  //CmiSyncBroadcastAll(size, (char *)msg);
128  break;
129  case ALLBUTME:
130  //CmiSyncBroadcast(size, (char *)msg);
131  for ( int i = 0; i < nchildren; ++i ) {
132  CmiSyncSend(children[i],size,(char*)msg);
133  }
134  break;
135  default:
136  NAMD_bug("Unexpected Communicate::sendMessage(PEL,...)");
137  //CmiSyncSend(PE, size, (char *)msg);
138  break;
139  }
140 }
void * getMessage(int PE, int tag)
Definition: Communicate.C:83
#define ALLBUTME
Definition: Communicate.h:14
MOStream * newOutputStream(int pe, int tag, unsigned int bufsize)
Definition: Communicate.C:77
#define ALL
Definition: Communicate.h:13
void NAMD_bug(const char *err_msg)
Definition: common.C:123
static void CsmHandler(void *msg)
Definition: Communicate.C:16
std::vector< std::string > split(const std::string &text, std::string delimiter)
Definition: MoleculeQM.C:73
Communicate(void)
Definition: Communicate.C:32
static void CsmAckHandler(void *msg)
Definition: Communicate.C:25
CkpvStaticDeclare(int, exitSchedHndlr)
void sendMessage(int PE, void *msg, int size)
Definition: Communicate.C:113
MIStream * newInputStream(int pe, int tag)
Definition: Communicate.C:71