Avoid calling destroy event inside host func

.github/workflows/test.yml

@@ -53,12 +53,9 @@ jobs:
           if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
             from_ref="origin/$GITHUB_HEAD_REF"
             to_ref="origin/$GITHUB_BASE_REF"
-          else
-            from_ref="HEAD^"
-            to_ref="HEAD"
+            echo "From reference: $from_ref; To reference: $to_ref"
+            pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
           fi
-          echo "From reference: $from_ref; To reference: $to_ref"
-          pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
         continue-on-error: false
 
       - name: Check the current working directory

.github/workflows/torch-api-test.yml

@@ -50,12 +50,9 @@ jobs:
           if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
             from_ref="origin/$GITHUB_HEAD_REF"
             to_ref="origin/$GITHUB_BASE_REF"
-          else
-            from_ref="HEAD^"
-            to_ref="HEAD"
+            echo "From reference: $from_ref; To reference: $to_ref"
+            pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
           fi
-          echo "From reference: $from_ref; To reference: $to_ref"
-          pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
         continue-on-error: false
 
       - name: Run `make` to build the project

.github/workflows/unit-test.yml

@@ -53,12 +53,9 @@ jobs:
           if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
             from_ref="origin/$GITHUB_HEAD_REF"
             to_ref="origin/$GITHUB_BASE_REF"
-          else
-            from_ref="HEAD^"
-            to_ref="HEAD"
+            echo "From reference: $from_ref; To reference: $to_ref"
+            pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
           fi
-          echo "From reference: $from_ref; To reference: $to_ref"
-          pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
         continue-on-error: false
 
       - name: Build Google Test

flagcx/core/group.cc

@@ -111,7 +111,8 @@ static flagcxResult_t groupLaunch(struct flagcxAsyncJob *job_) {
   // and a stream to launch host or device func
   struct flagcxHostSemaphore *semaphore;
   FLAGCXCHECK(flagcxCalloc(&semaphore, 1));
-  semaphore->flag = 0;
+  semaphore->start = 0;
+  semaphore->end = 0;
   semaphore->counter = 0;
   flagcxStream_t launchStream = nullptr;
 
@@ -208,6 +209,7 @@ static flagcxResult_t groupLaunch(struct flagcxAsyncJob *job_) {
           } else {
             op->args.semaphore = semaphore;
             op->event = semaphore->getEvent();
+            op->eventId = semaphore->counter;
             FLAGCXCHECK(deviceAdaptor->eventRecord(op->event, op->stream));
             semaphore->counter++;
             if (semaphore->counter == 1) {
@@ -267,6 +269,7 @@ static flagcxResult_t groupLaunch(struct flagcxAsyncJob *job_) {
           } else {
             op->args.semaphore = semaphore;
             op->event = semaphore->getEvent();
+            op->eventId = semaphore->counter;
             FLAGCXCHECK(deviceAdaptor->eventRecord(op->event, op->stream));
             semaphore->counter++;
             if (semaphore->counter == 1) {

flagcx/core/launch_kernel.cc

@@ -50,8 +50,7 @@ void cpuAsyncLoadWithMaxSpinCount(void *args) {
 
 void cpuAsyncKernel(void *args) {
   flagcxHostSemaphore *semaphore = (flagcxHostSemaphore *)args;
-  semaphore->signalFlag();
+  semaphore->signalStart();
   semaphore->wait();
-  free(semaphore);
-  semaphore = nullptr;
+  semaphore->signalEnd();
 }

flagcx/core/launch_kernel.h

@@ -36,28 +36,33 @@ void cpuAsyncLoad(void *args);
 void cpuAsyncLoadWithMaxSpinCount(void *args);
 
 struct flagcxHostSemaphore {
-  int flag;    // if ready to be triggered
+  int start;   // started or not
+  int end;     // ended or not
   int counter; // total operations to wait for inside the group
   std::vector<flagcxEvent_t> events;
 
+  ~flagcxHostSemaphore() {
+    for (auto event : events) {
+      deviceAdaptor->eventDestroy(event);
+    }
+  }
   flagcxEvent_t getEvent() {
     events.push_back(nullptr);
     auto &event = events.back();
     deviceAdaptor->eventCreate(&event);
     return event;
   }
-  void signalFlag() { __atomic_store_n(&flag, 1, __ATOMIC_RELEASE); }
+  void signalStart() { __atomic_store_n(&start, 1, __ATOMIC_RELEASE); }
+  void signalEnd() { __atomic_store_n(&end, 1, __ATOMIC_RELEASE); }
   void signalCounter(int value) {
     __atomic_fetch_sub(&counter, value, __ATOMIC_RELEASE);
   }
-  int poll() { return __atomic_load_n(&flag, __ATOMIC_ACQUIRE); }
+  int pollStart() { return __atomic_load_n(&start, __ATOMIC_ACQUIRE); }
+  int pollEnd() { return __atomic_load_n(&end, __ATOMIC_ACQUIRE); }
   void wait() {
     while (__atomic_load_n(&counter, __ATOMIC_ACQUIRE) > 0) {
       sched_yield();
     }
-    for (auto event : events) {
-      deviceAdaptor->eventDestroy(event);
-    }
   }
 };
 

flagcx/core/net.cc

@@ -136,7 +136,7 @@ flagcxResult_t flagcxNetInit(struct flagcxHeteroComm *comm) {
 
 flagcxResult_t flagcxProxySend(sendNetResources *resources, void *data,
                                size_t size, flagcxProxyArgs *args) {
-  if (!args->semaphore->poll()) {
+  if (!args->semaphore->pollStart()) {
     return flagcxSuccess;
   }
   if (args->transmitted < args->chunkSteps) {
@@ -220,7 +220,7 @@ flagcxResult_t flagcxProxySend(sendNetResources *resources, void *data,
 
 flagcxResult_t flagcxProxyRecv(recvNetResources *resources, void *data,
                                size_t size, flagcxProxyArgs *args) {
-  if (!args->semaphore->poll()) {
+  if (!args->semaphore->pollStart()) {
     return flagcxSuccess;
   }
   if (args->copied < args->chunkSteps) {

flagcx/core/proxy.cc

@@ -197,8 +197,17 @@ static flagcxResult_t progressOps(struct flagcxProxyState *proxyState,
               }
             } else {
               if (op->args.done == 1) {
-                flagcxIntruQueueDelete(queue, op);
-                free(op);
+                // Let the last operation within the group release the semaphore
+                if (op->eventId >= (int)op->args.semaphore->events.size() - 1) {
+                  if (op->args.semaphore->pollEnd()) {
+                    free(op->args.semaphore);
+                    flagcxIntruQueueDelete(queue, op);
+                    free(op);
+                  }
+                } else {
+                  flagcxIntruQueueDelete(queue, op);
+                  free(op);
+                }
               }
             }
           }
@@ -221,8 +230,17 @@ static flagcxResult_t progressOps(struct flagcxProxyState *proxyState,
               }
             } else {
               if (op->args.done == 1) {
-                flagcxIntruQueueDelete(queue, op);
-                free(op);
+                // Let the last operation within the group release the semaphore
+                if (op->eventId >= (int)op->args.semaphore->events.size() - 1) {
+                  if (op->args.semaphore->pollEnd()) {
+                    free(op->args.semaphore);
+                    flagcxIntruQueueDelete(queue, op);
+                    free(op);
+                  }
+                } else {
+                  flagcxIntruQueueDelete(queue, op);
+                  free(op);
+                }
               }
             }
           }

flagcx/core/proxy.h

@@ -170,6 +170,8 @@ struct flagcxProxyOp {
   flagcxProxyArgs args;
   flagcxStream_t stream;
   flagcxEvent_t event; // used to record host/device func
+  int eventId; // The event ID associated with the semaphore in the underlying
+               // group
 };
 
 #define FLAGCX_MAX_NETDEVS 128