add counter reset and message free

Author: ritvikrao

include/converse.h

@@ -760,6 +760,8 @@ typedef struct DecrementToEnqueueMsg{
 
 DecrementToEnqueueMsg *CmiCreateDecrementToEnqueue(unsigned int initialCount, void *msg);
 void CmiDecrementCounter(DecrementToEnqueueMsg *dteMsg);
+void CmiResetCounter(unsigned int newCount, DecrementToEnqueueMsg *dteMsg);
+void CmiFreeDecrementToEnqueue(DecrementToEnqueueMsg *dteMsg);
 
 // error checking
 

src/convcore.cpp

@@ -700,14 +700,35 @@ void CmiDecrementCounter(DecrementToEnqueueMsg *dteMsg){
       destPE >= 0 && destPE < Cmi_npes,
       "CmiDecrementCounter: destPE out of range"
     );
-    // enqueue the message (node send)
-    CmiSyncSendAndFree(destPE, header->messageSize, dteMsg->msg);
-    // free the counter and dteMsg
-    free(dteMsg->counter);
-    free(dteMsg);
+    // enqueue the message without freeing
+    CmiSyncSend(destPE, header->messageSize, dteMsg->msg);
   }
 }
 
+void CmiResetCounter(unsigned int newCount, DecrementToEnqueueMsg *dteMsg){
+  if(dteMsg == nullptr){
+    CmiAbort("CmiResetCounter: dteMsg is nullptr\n");
+  }
+  if(dteMsg->counter == nullptr){
+    CmiAbort("CmiResetCounter: counter is nullptr\n");
+  }
+  if(newCount == 0){
+    CmiAbort("CmiResetCounter: newCount cannot be zero\n");
+  }
+  __atomic_store_n(dteMsg->counter, newCount, __ATOMIC_SEQ_CST);
+}
+
+void CmiFreeDecrementToEnqueue(DecrementToEnqueueMsg *dteMsg){
+  if(dteMsg == nullptr){
+    CmiAbort("CmiFreeDecrementToEnqueue: dteMsg is nullptr\n");
+  }
+  if(dteMsg->counter == nullptr){
+    CmiAbort("CmiFreeDecrementToEnqueue: counter is nullptr\n");
+  }
+  free(dteMsg->counter);
+  free(dteMsg);
+}
+
 void CmiNodeBarrier(void) {
   static Barrier nodeBarrier(CmiMyNodeSize());
   int64_t ticket = nodeBarrier.arrive();

tests/decrement_bcast/decrement_bcast.cpp

@@ -8,18 +8,19 @@
 #include "converse.h"
 #include <string.h>
 
-// Handler called when final message is delivered (counter reached zero).
-void exit_handler(void *msg) {
-  CmiPrintf("Exit handler: counter reached zero on PE %d\n", CmiMyPe());
-  CmiExit(0);
-}
-
 // Global pointer to the DecrementToEnqueueMsg created on PE 0. Other PEs do not
 // need direct access to its fields; PE 0 will own and use this pointer when
 // decrementing. Making it global simplifies the broadcast message.
 static DecrementToEnqueueMsg *g_dte = NULL;
 static int g_decInvH = -1;
 
+// Handler called when final message is delivered (counter reached zero).
+void exit_handler(void *msg) {
+  CmiPrintf("Exit handler: counter reached zero on PE %d\n", CmiMyPe());
+  CmiFreeDecrementToEnqueue(g_dte);
+  CmiExit(0);
+}
+
 // Handler that will be invoked on PE0 for each incoming decrement-invoker
 // message. It calls CmiDecrementCounter on the global DTE.
 void decrement_invoker(void *msg) {

tests/decrement_enqueue/decrement_enqueue.cpp

@@ -4,9 +4,12 @@
 #include "converse.h"
 #include <string.h>
 
+DecrementToEnqueueMsg *dte;
+
 // Handler invoked when the decrement counter reaches zero. Exits the program.
 void exit_handler(void *msg) {
-  // We don't need the message payload; just exit.
+  // free DecrementToEnqueueMsg
+  CmiFreeDecrementToEnqueue(dte);
   CmiPrintf("Exit handler called, exiting...\n");
   CmiExit(0);
 }
@@ -32,7 +35,7 @@ void test_start(int argc, char **argv) {
   hdr->messageSize = msgSize;
 
   // Create the decrement-to-enqueue helper with initial count 16.
-  DecrementToEnqueueMsg *dte = CmiCreateDecrementToEnqueue(16u, msg);
+  dte = CmiCreateDecrementToEnqueue(16u, msg);
 
   // Decrement 16 times; on the 16th call the message will be sent and the
   // registered handler will call CmiExit.