feat(ccu): add direct signal wait backend

src/comm/CMakeLists.txt

@@ -145,6 +145,8 @@ set(TILEXR_SOURCE_FILE tilexr_comm.cpp
         ccu/tilexr_ccu_resource_allocator.cpp
         ccu/tilexr_ccu_runtime.h
         ccu/tilexr_ccu_runtime.cpp
+        ccu/tilexr_ccu_signal_wait_program.h
+        ccu/tilexr_ccu_signal_wait_program.cpp
         ccu/tilexr_ccu_runtime_session.h
         ccu/tilexr_ccu_runtime_session.cpp
         ccu/tilexr_ccu_executor.h

src/comm/ccu/tilexr_ccu_backend.cpp

@@ -20,6 +20,11 @@ class TileXRCcuBackend::Impl {
     bool Supports(const TileXRCcuCollectiveRequest &request) const;
     int PrepareCollective(const TileXRCcuCollectiveRequest &request, TileXRCcuCollectivePlan *plan);
     int SubmitCollective(const TileXRCcuCollectivePlan &plan, aclrtStream stream);
+    int PrepareSignalWait(const TileXRCcuSignalWaitRequest &request, TileXRCcuSignalWaitPlan *plan);
+    int SubmitSignalWait(
+        const TileXRCcuSignalWaitPlan &plan,
+        aclrtStream stream,
+        TileXRCcuDirectSubmitReport *report);
 #ifdef TILEXR_CCU_TESTING
     bool RuntimeInitializedForTest() const;
 #endif
@@ -105,6 +110,38 @@ int TileXRCcuBackend::Impl::SubmitCollective(const TileXRCcuCollectivePlan &plan
     return executor_->Submit(*runtimeSession_, plan, stream);
 }
 
+int TileXRCcuBackend::Impl::PrepareSignalWait(
+    const TileXRCcuSignalWaitRequest &request,
+    TileXRCcuSignalWaitPlan *plan)
+{
+    if (plan == nullptr) {
+        return TILEXR_ERROR_PARA_CHECK_FAIL;
+    }
+    *plan = TileXRCcuSignalWaitPlan {};
+    if (runtimeSession_ == nullptr || planner_ == nullptr) {
+        return TILEXR_ERROR_INTERNAL;
+    }
+    TileXRCcuDirectInstallReport report;
+    return planner_->PrepareSignalWait(*runtimeSession_, request, plan, &report);
+}
+
+int TileXRCcuBackend::Impl::SubmitSignalWait(
+    const TileXRCcuSignalWaitPlan &plan,
+    aclrtStream stream,
+    TileXRCcuDirectSubmitReport *report)
+{
+    if (report != nullptr) {
+        *report = TileXRCcuDirectSubmitReport {};
+    }
+    if (!plan.ready || plan.submitTasks.empty()) {
+        if (report != nullptr) {
+            report->message = "direct CCU signal/wait plan is not ready";
+        }
+        return TILEXR_ERROR_NOT_INITIALIZED;
+    }
+    return TileXRCcuSubmitPreparedTasks(plan.submitTasks, stream, nullptr, nullptr, report);
+}
+
 #ifdef TILEXR_CCU_TESTING
 bool TileXRCcuBackend::Impl::RuntimeInitializedForTest() const
 {
@@ -165,6 +202,27 @@ int TileXRCcuBackend::SubmitCollective(const TileXRCcuCollectivePlan &plan, aclr
     return impl_->SubmitCollective(plan, stream);
 }
 
+int TileXRCcuBackend::PrepareSignalWait(
+    const TileXRCcuSignalWaitRequest &request,
+    TileXRCcuSignalWaitPlan *plan)
+{
+    if (impl_ == nullptr) {
+        return TILEXR_ERROR_INTERNAL;
+    }
+    return impl_->PrepareSignalWait(request, plan);
+}
+
+int TileXRCcuBackend::SubmitSignalWait(
+    const TileXRCcuSignalWaitPlan &plan,
+    aclrtStream stream,
+    TileXRCcuDirectSubmitReport *report)
+{
+    if (impl_ == nullptr) {
+        return TILEXR_ERROR_INTERNAL;
+    }
+    return impl_->SubmitSignalWait(plan, stream, report);
+}
+
 #ifdef TILEXR_CCU_TESTING
 bool TileXRCcuBackend::RuntimeInitializedForTest() const
 {

src/comm/ccu/tilexr_ccu_backend.h

@@ -8,7 +8,9 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <vector>
 
+#include "ccu/tilexr_ccu_direct_orchestrator.h"
 #include "acl/acl_base.h"
 #include "tilexr_types.h"
 
@@ -42,6 +44,38 @@ struct TileXRCcuCollectivePlan {
     bool ready = false;
 };
 
+enum class TileXRCcuSignalWaitRole {
+    Signal = 0,
+    Wait = 1,
+    SignalAndWait = 2,
+};
+
+struct TileXRCcuSignalWaitRequest {
+    int peerRank = -1;
+    TileXRCcuSignalWaitRole role = TileXRCcuSignalWaitRole::Signal;
+    uint32_t syncInstructionCount = 0;
+    uint16_t missionStartId = 0;
+    uint16_t instructionStartId = 0;
+    uint16_t missionInstructionStartId = 0;
+    uint16_t xnStartId = 0;
+    uint16_t remoteXnStartId = 0;
+    uint16_t remoteXnCount = 0;
+    uint16_t ckeStartId = 0;
+    uint16_t channelStartId = 0;
+    uint16_t localWaitCkeStartId = 0;
+    uint16_t localWaitCkeCount = 0;
+    uint16_t remoteNotifyCkeStartId = 0;
+    uint16_t remoteNotifyCkeCount = 0;
+    uint16_t timeout = 0;
+    std::string provider;
+};
+
+struct TileXRCcuSignalWaitPlan {
+    bool ready = false;
+    TileXRCcuDirectInstallAttempt attempt;
+    std::vector<TileXRCcuTask> submitTasks;
+};
+
 class TileXRCcuBackend {
 public:
     TileXRCcuBackend();
@@ -56,6 +90,11 @@ class TileXRCcuBackend {
     bool Supports(const TileXRCcuCollectiveRequest &request) const;
     int PrepareCollective(const TileXRCcuCollectiveRequest &request, TileXRCcuCollectivePlan *plan);
     int SubmitCollective(const TileXRCcuCollectivePlan &plan, aclrtStream stream);
+    int PrepareSignalWait(const TileXRCcuSignalWaitRequest &request, TileXRCcuSignalWaitPlan *plan);
+    int SubmitSignalWait(
+        const TileXRCcuSignalWaitPlan &plan,
+        aclrtStream stream,
+        TileXRCcuDirectSubmitReport *report);
 #ifdef TILEXR_CCU_TESTING
     bool RuntimeInitializedForTest() const;
 #endif

src/comm/ccu/tilexr_ccu_barrier_program.cpp

@@ -221,7 +221,7 @@ int TileXRCcuBuildBarrierProgram(
             TileXRCcuSyncCkeSpec post;
             post.remoteCke = spec.remoteNotifyCke;
             post.localCke = spec.sourceCke;
-            post.localCkeMask = spec.remoteNotifyMask;
+            post.localCkeMask = spec.sourceCkeMask;
             post.channelId = spec.channelId;
             if (TileXRCcuEncodeSyncCke(post, &instr) != TILEXR_SUCCESS) {
                 return Fail(program, report, "failed to encode CCU barrier SyncCKE post instruction");