feat: zerocopy for hypercube

Author: Sh0g0-1758

sim

@@ -71,12 +71,13 @@ void AllGather::local_buff_done(CkDataMsg *m) {
 }
 
 void AllGather::startGather() {
+  for (int i = 0; i < k; i++) {
+    store[k * thisIndex + i] = data[i];
+  }
+  CkNcpyBuffer src(data, k*sizeof(long int), dum_dum, CK_BUFFER_UNREG);
+  
   switch (type) {
   case allGatherType::ALL_GATHER_DEFAULT: {
-    for (int i = 0; i < k; i++) {
-      store[k * thisIndex + i] = data[i];
-    }
-    CkNcpyBuffer src(data, k*sizeof(long int), dum_dum, CK_BUFFER_UNREG);
 #ifdef TIMESTAMP
     thisProxy[(thisIndex + 1) % n].recvDefault(
         thisIndex, src, (timeStamp + alpha + beta * k * 8));
@@ -87,17 +88,11 @@ void AllGather::startGather() {
   } break;
   case allGatherType::ALL_GATHER_HYPERCUBE: {
     hyperCubeIndx.push_back(thisIndex);
-    for (int i = 0; i < k; i++) {
-      hyperCubeStore.push_back(data[i]);
-    }
+    hyperCubeStore.push_back(src);
     thisProxy(thisIndex).Hypercube();
   } break;
   case allGatherType::ALL_GATHER_FLOODING: {
-    for (int i = 0; i < k; i++) {
-      store[k * thisIndex + i] = data[i];
-    }
     recvFloodMsg[thisIndex] = true;
-    CkNcpyBuffer src(data, k*sizeof(long int), dum_dum, CK_BUFFER_UNREG);
     for (int i = 0; i < n; i++) {
       if (graph[thisIndex][i] == 1) {
 #ifdef TIMESTAMP

src/allGather/allGather.cc

@@ -15,28 +15,27 @@ module allGather {
         serial { HypercubeToSend = thisIndex ^ ((int)pow(2, iter)); }
         if (HypercubeToSend < n) {
           serial {
-            long int data[hyperCubeStore.size()];
-            for(int i = 0; i < hyperCubeStore.size(); i++) {
-              data[i] = hyperCubeStore[i];
-            }
+            CkNcpyBuffer data[hyperCubeStore.size()];
             int dataIndx[hyperCubeIndx.size()];
-            for(int i = 0; i < hyperCubeIndx.size(); i++) {
+            int numBuffers = hyperCubeStore.size();
+            for(int i = 0; i < numBuffers; i++) {
+              data[i] = hyperCubeStore[i];
               dataIndx[i] = hyperCubeIndx[i];
             }
 #ifdef TIMESTAMP
-            thisProxy(HypercubeToSend).recvHypercube(iter, data, hyperCubeStore.size(), dataIndx, hyperCubeIndx.size(), (timeStamp + alpha + beta * hyperCubeStore.size() * 8));
+            thisProxy(HypercubeToSend).recvHypercube(iter, data, dataIndx, numBuffers, (timeStamp + alpha + beta * hyperCubeStore.size() * 8));
             timeStamp += alpha;
 #else
-            thisProxy(HypercubeToSend).recvHypercube(iter, data, hyperCubeStore.size(), dataIndx, hyperCubeIndx.size(), 0.0);
+            thisProxy(HypercubeToSend).recvHypercube(iter, data, dataIndx, numBuffers, 0.0);
 #endif
           }
-          when recvHypercube[iter](int ref, long int data[size], int size, int dataIndx[indxSize], int indxSize, double recvTime) {
+          when recvHypercube[iter](int ref, CkNcpyBuffer data[size], int dataIndx[size], int size, double recvTime) {
             serial {
               for(int m = 0; m < size; m++) {
                 hyperCubeStore.emplace_back(data[m]);
-              }
-              for(int m = 0; m < indxSize; m++) {
                 hyperCubeIndx.emplace_back(dataIndx[m]);
+                CkNcpyBuffer dst(store + dataIndx[m] * k, k * sizeof(long int), zero_copy_callback, CK_BUFFER_UNREG);
+                dst.get(data[m]);
               }
 #ifdef TIMESTAMP
               timeStamp = std::max(recvTime, timeStamp);
@@ -45,17 +44,8 @@ module allGather {
           }
         }
       }
-      serial {
-        for(int m = 0; m < hyperCubeIndx.size(); m++) {
-          int currIndx = hyperCubeIndx[m];
-          for(int j = 0; j < k; j++) {
-            store[k * currIndx + j] = hyperCubeStore[m * k + j];
-          }
-        }
-        lib_done_callback.send(msg);
-      }
     };
-    entry void recvHypercube(int ref, long int data[size], int size, int dataIndx[indxSize], int indxSize, double recvTime);
+    entry void recvHypercube(int ref, CkNcpyBuffer data[size], int dataIndx[size], int size, double recvTime);
     entry void Flood(int sender, CkNcpyBuffer data, double recvTime);
   };
 };

src/allGather/allGather.ci

@@ -34,7 +34,7 @@ private:
   std::map<int, bool> recvFloodMsg{};
   int randCounter{};
   std::vector<int> hyperCubeIndx{};
-  std::vector<long int> hyperCubeStore{};
+  std::vector<CkNcpyBuffer> hyperCubeStore{};
   allGatherMsg *msg = new allGatherMsg;
   long int* data;
   CkCallback zero_copy_callback;