From c95320b088fcd44f16e311d57a6c55d35141b7c4 Mon Sep 17 00:00:00 2001 From: Kur0x Date: Tue, 28 Jul 2026 11:46:30 +0800 Subject: [PATCH 1/3] feat(ccu): generalize alltoall mesh ranks --- src/comm/ccu/tilexr_ccu_alltoall_program.cpp | 127 ++++++++++----- src/comm/ccu/tilexr_ccu_alltoall_program.h | 4 +- .../ccu/tilexr_ccu_collective_planner.cpp | 81 ++++++---- .../ccu/tilexr_ccu_direct_orchestrator.cpp | 148 ++++++++++++------ src/comm/ccu/tilexr_ccu_direct_orchestrator.h | 2 +- src/comm/ccu/tilexr_ccu_direct_runtime.cpp | 133 +++++++++------- src/comm/ccu/tilexr_ccu_direct_runtime.h | 4 + .../tilexr_ccu_lower_layer_plan_builder.cpp | 5 +- .../tilexr_ccu_ra_custom_channel_provider.cpp | 21 ++- .../ccu/tilexr_ccu_resource_allocator.cpp | 19 ++- src/comm/ccu/tilexr_ccu_resource_allocator.h | 2 + src/comm/ccu/tilexr_ccu_topology.cpp | 36 ++++- src/comm/ccu/tilexr_ccu_topology.h | 1 + tests/ccu/ccu_tilexr_direct_smoke_probe.cpp | 126 ++++++--------- tests/ccu/run_tilexr_ccu_direct_smoke.sh | 46 +++++- tests/ccu/test_tilexr_ccu_alltoall_program.py | 68 +++++++- tests/ccu/test_tilexr_ccu_backend_boundary.py | 5 +- .../test_tilexr_ccu_direct_orchestrator.py | 95 +++++++++-- .../ccu/test_tilexr_ccu_direct_smoke_probe.py | 21 ++- .../test_tilexr_ccu_direct_smoke_runner.py | 24 ++- ...est_tilexr_ccu_lower_layer_plan_builder.py | 52 ++++-- ...est_tilexr_ccu_ra_custom_channel_loader.py | 21 +++ ...t_tilexr_ccu_ra_custom_channel_provider.py | 62 ++++++++ .../ccu/test_tilexr_ccu_resource_allocator.py | 17 +- 24 files changed, 812 insertions(+), 308 deletions(-) diff --git a/src/comm/ccu/tilexr_ccu_alltoall_program.cpp b/src/comm/ccu/tilexr_ccu_alltoall_program.cpp index 7e32cf0a..e54981f0 100644 --- a/src/comm/ccu/tilexr_ccu_alltoall_program.cpp +++ b/src/comm/ccu/tilexr_ccu_alltoall_program.cpp @@ -19,6 +19,32 @@ constexpr uint16_t TILEXR_CCU_TRACE_TRANS_LOC_MEM_TO_RMT_MEM_HEADER = 0x1009U; constexpr uint16_t TILEXR_CCU_TRACE_TRANS_LOC_MEM_TO_LOC_MEM_HEADER = 0x100aU; constexpr uint16_t TILEXR_CCU_TRACE_TRANS_LOC_MEM_TO_LOC_MS_HEADER = 0x1000U; constexpr uint16_t TILEXR_CCU_TRACE_TRANS_LOC_MS_TO_LOC_MEM_HEADER = 0x1002U; +constexpr uint32_t TILEXR_CCU_ALLTOALL_MAX_RANK_SIZE = 64U; +constexpr uint32_t TILEXR_CCU_CKE_MASK_BITS = 16U; + +uint32_t CompletionCkeCount(size_t peerCount) +{ + return static_cast((peerCount + TILEXR_CCU_CKE_MASK_BITS - 1U) / TILEXR_CCU_CKE_MASK_BITS); +} + +uint16_t CompletionMaskForGroup(size_t peerCount, uint32_t group) +{ + const size_t begin = static_cast(group) * TILEXR_CCU_CKE_MASK_BITS; + const size_t remaining = peerCount > begin ? peerCount - begin : 0U; + const uint32_t bits = static_cast(std::min(remaining, TILEXR_CCU_CKE_MASK_BITS)); + return bits == TILEXR_CCU_CKE_MASK_BITS ? 0xffffU : static_cast((1U << bits) - 1U); +} + +size_t MeshPreSyncInstructionCount(size_t peerCount) +{ + return 3U + peerCount * 3U; +} + +size_t MeshCopyInstructionCountPerBlock(size_t peerCount) +{ + return peerCount * 6U + 9U + CompletionCkeCount(peerCount); +} + constexpr uint16_t TILEXR_CCU_TRACE_SYNC_CKE_HEADER = 0x100bU; constexpr uint16_t TILEXR_CCU_TRACE_SYNC_XN_HEADER = 0x100dU; constexpr uint16_t TILEXR_CCU_ALLTOALL_SOURCE_CKE_INIT_MASK = 0xffffU; @@ -486,8 +512,9 @@ int ValidateMeshSpec( if (program == nullptr) { return Fail(program, report, "missing output direct CCU alltoall mesh program"); } - if (spec.rankSize != 4U || spec.localRank >= spec.rankSize || spec.peers.size() != 3U) { - return Fail(program, report, "direct CCU alltoall mesh requires four ranks and three peers"); + if (spec.rankSize < 2U || spec.rankSize > TILEXR_CCU_ALLTOALL_MAX_RANK_SIZE || + spec.localRank >= spec.rankSize || spec.peers.size() != spec.rankSize - 1U) { + return Fail(program, report, "direct CCU alltoall mesh requires 2..64 ranks and rankSize-1 peers"); } if (spec.localSendAddr == 0 || spec.localRecvAddr == 0 || spec.localSendToken == 0 || spec.localRecvToken == 0 || spec.chunkBytes == 0 || @@ -496,16 +523,28 @@ int ValidateMeshSpec( } if (spec.selfSourceGsa == 0 || spec.selfDestinationGsa == 0 || spec.selfSourceXn == 0 || spec.selfDestinationXn == 0 || spec.selfLengthXn == 0 || - spec.selfCompletionCke == 0 || spec.remoteCompletionCke == 0) { + spec.selfCompletionCke == 0 || + spec.remoteCompletionCkes.size() != CompletionCkeCount(spec.peers.size())) { return Fail(program, report, "missing direct CCU alltoall mesh self-copy resource"); } - bool peerRanks[4] = {}; + std::vector peerRanks(spec.rankSize, false); std::set channelIds; + std::set completionCkes; const auto& sharedRoute = spec.peers.front().route; - if (spec.remoteCompletionCke == sharedRoute.sourceCke) { - return Fail(program, report, "alltoall mesh completion CKE overlaps source CKE"); + if (spec.selfSourceXn != sharedRoute.localXn || + spec.selfDestinationXn != sharedRoute.preSyncLocalTokenXn || + spec.selfLengthXn != sharedRoute.lengthXn) { + return Fail(program, report, "alltoall mesh self copy must share source, destination, and length XNs"); + } + for (uint16_t completionCke : spec.remoteCompletionCkes) { + if (completionCke == 0 || completionCke == sharedRoute.sourceCke || + !completionCkes.insert(completionCke).second) { + return Fail(program, report, + "alltoall mesh completion CKE overlaps source CKE or duplicates another completion CKE"); + } } - for (const auto& peer : spec.peers) { + for (size_t ordinal = 0; ordinal < spec.peers.size(); ++ordinal) { + const auto& peer = spec.peers[ordinal]; if (peer.peerRank >= spec.rankSize || peer.peerRank == spec.localRank || peerRanks[peer.peerRank]) { return Fail(program, report, "invalid direct CCU alltoall mesh peer rank"); } @@ -522,7 +561,7 @@ int ValidateMeshSpec( peer.route.preSyncChannelId != peer.route.copyChannelId || peer.route.preSyncTokenChannelId != peer.route.copyChannelId || peer.route.postSyncChannelId != peer.route.copyChannelId || - peer.route.copyCompletionCke != spec.remoteCompletionCke || + peer.route.copyCompletionCke != spec.remoteCompletionCkes[ordinal / TILEXR_CCU_CKE_MASK_BITS] || peer.route.ckeMask != TILEXR_CCU_ALLTOALL_POST_SYNC_MASK || !channelIds.insert(peer.route.copyChannelId).second) { return Fail(program, report, "duplicate direct CCU alltoall mesh peer resource"); @@ -769,8 +808,11 @@ int ValidateMeshProgramBindings( }); const uint32_t blocksPerChunk = static_cast( spec.chunkBytes / TILEXR_CCU_ALLTOALL_BLOCK_BYTES); - const size_t expectedSize = 12U + static_cast(blocksPerChunk) * 28U + 6U + 1U; - if (peers.size() != 3U || blocksPerChunk == 0 || program.size() != expectedSize) { + const size_t preSyncInstructions = MeshPreSyncInstructionCount(peers.size()); + const size_t copyInstructionsPerBlock = MeshCopyInstructionCountPerBlock(peers.size()); + const size_t expectedSize = preSyncInstructions + + static_cast(blocksPerChunk) * copyInstructionsPerBlock + peers.size() * 2U + 1U; + if (peers.size() != spec.rankSize - 1U || blocksPerChunk == 0 || program.size() != expectedSize) { return FailBindingValidation(report, "unexpected mesh program shape"); } @@ -804,7 +846,7 @@ int ValidateMeshProgramBindings( PreSyncTokenMask(route))) { return FailBindingValidation(report, "token SyncXn does not match its peer route"); } - const size_t wait = 9U + ordinal; + const size_t wait = 3U + peers.size() * 2U + ordinal; if (!MatchesWait( program[wait], TILEXR_CCU_TRACE_SET_CKE_HEADER, @@ -814,7 +856,7 @@ int ValidateMeshProgramBindings( } } - size_t instruction = 12U; + size_t instruction = preSyncInstructions; for (uint32_t block = 0; block < blocksPerChunk; ++block) { for (size_t ordinal = 0; ordinal < peers.size(); ++ordinal) { const auto& route = peers[ordinal].route; @@ -827,8 +869,8 @@ int ValidateMeshProgramBindings( route.localXn, route.lengthXn, route.copyChannelId, - spec.remoteCompletionCke, - static_cast(1U << ordinal))) { + route.copyCompletionCke, + static_cast(1U << (ordinal % TILEXR_CCU_CKE_MASK_BITS)))) { return FailBindingValidation(report, "remote copy does not match its peer route"); } instruction += 6U; @@ -866,14 +908,16 @@ int ValidateMeshProgramBindings( return FailBindingValidation(report, "self copy does not match its local route"); } instruction += 9U; - if (!MatchesWait( - program[instruction], - TILEXR_CCU_TRACE_CLEAR_CKE_HEADER, - spec.remoteCompletionCke, - 0x7U)) { - return FailBindingValidation(report, "combined remote copy wait does not match the mesh completion CKE"); + for (uint32_t group = 0; group < spec.remoteCompletionCkes.size(); ++group) { + if (!MatchesWait( + program[instruction], + TILEXR_CCU_TRACE_CLEAR_CKE_HEADER, + spec.remoteCompletionCkes[group], + CompletionMaskForGroup(peers.size(), group))) { + return FailBindingValidation(report, "grouped remote copy wait does not match the mesh completion CKE"); + } + ++instruction; } - ++instruction; } for (const auto& peer : peers) { const auto& route = peer.route; @@ -1003,7 +1047,10 @@ int TileXRCcuBuildAllToAllMeshProgram( }); const uint64_t bytesPerBlock = TILEXR_CCU_ALLTOALL_BLOCK_BYTES; const uint32_t blocksPerChunk = static_cast(spec.chunkBytes / bytesPerBlock); - program->reserve(12U + blocksPerChunk * 28U + 6U + 1U); + const size_t preSyncInstructions = MeshPreSyncInstructionCount(peers.size()); + const size_t copyInstructionsPerBlock = MeshCopyInstructionCountPerBlock(peers.size()); + program->reserve(preSyncInstructions + + static_cast(blocksPerChunk) * copyInstructionsPerBlock + peers.size() * 2U + 1U); ret = AppendMeshPeerPosts(peers, program, report); if (ret != TILEXR_SUCCESS) { @@ -1033,8 +1080,8 @@ int TileXRCcuBuildAllToAllMeshProgram( route, static_cast(block) * bytesPerBlock, bytesPerBlock, - spec.remoteCompletionCke, - static_cast(1U << ordinal), + peer.route.copyCompletionCke, + static_cast(1U << (ordinal % TILEXR_CCU_CKE_MASK_BITS)), program, report); if (ret != TILEXR_SUCCESS) { @@ -1050,15 +1097,17 @@ int TileXRCcuBuildAllToAllMeshProgram( if (ret != TILEXR_SUCCESS) { return ret; } - ret = AppendNotifyWait( - spec.remoteCompletionCke, - 0x7U, - "mesh Copy", - true, - program, - report); - if (ret != TILEXR_SUCCESS) { - return ret; + for (uint32_t group = 0; group < spec.remoteCompletionCkes.size(); ++group) { + ret = AppendNotifyWait( + spec.remoteCompletionCkes[group], + CompletionMaskForGroup(peers.size(), group), + "mesh Copy", + true, + program, + report); + if (ret != TILEXR_SUCCESS) { + return ret; + } } } @@ -1093,16 +1142,16 @@ int TileXRCcuBuildAllToAllMeshProgram( } if (report != nullptr) { - report->preSyncInstructionCount = 12U; + report->preSyncInstructionCount = static_cast(preSyncInstructions); report->blockCount = blocksPerChunk; report->bytesPerBlock = static_cast(bytesPerBlock); - report->copyInstructionCount = blocksPerChunk * 28U; - report->postSyncInstructionCount = 6U; + report->copyInstructionCount = static_cast(blocksPerChunk * copyInstructionsPerBlock); + report->postSyncInstructionCount = static_cast(peers.size() * 2U); report->finishInstructionCount = 1U; report->totalInstructionCount = static_cast(program->size()); - report->peerCount = 3U; - report->syncResourceCount = 3U; - report->remoteBlockCount = 3U * blocksPerChunk; + report->peerCount = static_cast(peers.size()); + report->syncResourceCount = static_cast(peers.size()); + report->remoteBlockCount = static_cast(peers.size()) * blocksPerChunk; report->selfBlockCount = blocksPerChunk; report->message = "ok"; } diff --git a/src/comm/ccu/tilexr_ccu_alltoall_program.h b/src/comm/ccu/tilexr_ccu_alltoall_program.h index 93783ba2..d02ab822 100644 --- a/src/comm/ccu/tilexr_ccu_alltoall_program.h +++ b/src/comm/ccu/tilexr_ccu_alltoall_program.h @@ -81,7 +81,7 @@ struct TileXRCcuAllToAllMeshPeerSpec { }; struct TileXRCcuAllToAllMeshProgramSpec { - uint32_t rankSize = 4; + uint32_t rankSize = 2; uint32_t localRank = 0; uint64_t localSendAddr = 0; uint64_t localSendToken = 0; @@ -95,7 +95,7 @@ struct TileXRCcuAllToAllMeshProgramSpec { uint16_t selfLengthXn = 0; uint16_t selfChannelId = 0; uint16_t selfCompletionCke = 0; - uint16_t remoteCompletionCke = 0; + std::vector remoteCompletionCkes; std::vector peers; }; diff --git a/src/comm/ccu/tilexr_ccu_collective_planner.cpp b/src/comm/ccu/tilexr_ccu_collective_planner.cpp index ad537e2e..c160012d 100644 --- a/src/comm/ccu/tilexr_ccu_collective_planner.cpp +++ b/src/comm/ccu/tilexr_ccu_collective_planner.cpp @@ -26,13 +26,23 @@ namespace TileXR { #ifdef TILEXR_CCU_TESTING constexpr uint32_t TILEXR_CCU_DIRECT_MEMORY_COPY_INSTRUCTION_COUNT = 7U; -constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT = - 7U + 64U * 7U; constexpr uint32_t TILEXR_CCU_DIRECT_SYNC_XN_PING_INSTRUCTION_COUNT = 2U; #endif constexpr uint32_t TILEXR_CCU_DIRECT_SIGNAL_INSTRUCTION_COUNT = 5U; constexpr uint32_t TILEXR_CCU_DIRECT_WAIT_INSTRUCTION_COUNT = 5U; constexpr uint32_t TILEXR_CCU_DIRECT_SIGNAL_WAIT_INSTRUCTION_COUNT = 6U; +constexpr uint32_t TILEXR_CCU_CHANNEL_XN_STRIDE = 8U; + +uint32_t DirectAllToAll2RankInstructionCapacity(uint64_t bytes) +{ + if (bytes == 0 || bytes % TILEXR_CCU_ALLTOALL_BLOCK_BYTES != 0) { + return 0; + } + const uint64_t blocks = bytes / TILEXR_CCU_ALLTOALL_BLOCK_BYTES; + const uint64_t instructions = 7ULL + blocks * 7ULL; + return instructions > std::numeric_limits::max() ? + 0U : static_cast(instructions); +} uint8_t SelectDirectCcuInstallDieId() { @@ -48,18 +58,15 @@ uint8_t SelectDirectCcuInstallDieId() return static_cast(parsed); } -uint32_t SelectDirectCcuPeerLocalXnOffset(size_t peerLocalIndex, uint32_t syncIndex, size_t peerRouteCount) -{ - if (peerRouteCount == 0) { - return 0; - } - return static_cast(peerLocalIndex) + - static_cast(syncIndex / peerRouteCount) * static_cast(peerRouteCount); -} - -uint32_t SelectDirectCcuChannelBoundRemoteXnOffset(size_t peerLocalIndex, uint32_t syncIndex, size_t peerRouteCount) +uint32_t SelectDirectCcuChannelBoundRemoteXnOffset( + size_t peerLocalIndex, + size_t routeWithinPeer, + size_t routesPerPeer, + bool channelStrided) { - return SelectDirectCcuPeerLocalXnOffset(peerLocalIndex, syncIndex, peerRouteCount); + const size_t stride = channelStrided ? TILEXR_CCU_CHANNEL_XN_STRIDE : routesPerPeer; + return static_cast(peerLocalIndex * stride) + + static_cast(routeWithinPeer); } uint16_t DirectCcuRemoteXnProofSpan(uint16_t syncRouteCount) @@ -73,12 +80,11 @@ uint16_t DirectCcuRemoteXnProofSpan(uint16_t syncRouteCount) uint16_t SelectDirectCcuChannelBoundRemoteXnId( uint16_t remoteXnStartId, size_t peerLocalIndex, - uint32_t syncIndex, - size_t peerRouteCount) + size_t routeWithinPeer) { return static_cast( static_cast(remoteXnStartId) + - SelectDirectCcuChannelBoundRemoteXnOffset(peerLocalIndex, syncIndex, peerRouteCount)); + SelectDirectCcuChannelBoundRemoteXnOffset(peerLocalIndex, routeWithinPeer, 1U, true)); } TileXRCcuSignalWaitProgramRole ToDirectSignalWaitProgramRole(TileXRCcuSignalWaitRole role) @@ -553,7 +559,7 @@ int TileXRCcuCollectivePlanner::ExchangeDirectCcuRemoteNotifyCke( return TILEXR_ERROR_PARA_CHECK_FAIL; } const size_t peerRouteCount = static_cast(rankSize - 1); - const size_t syncRouteCount = allocation.remoteXn.num; + const size_t syncRouteCount = allocation.channels.num; size_t routedPeerCount = peerRouteCount; int selectedDiagnosticPeer = -1; #ifdef TILEXR_CCU_TESTING @@ -570,16 +576,28 @@ int TileXRCcuCollectivePlanner::ExchangeDirectCcuRemoteNotifyCke( } #endif const size_t routesPerPeer = syncRouteCount / routedPeerCount; + const bool channelStridedRemoteXn = allocation.channels.num != 0U && + static_cast(allocation.remoteXn.num) >= + static_cast(allocation.channels.num) * TILEXR_CCU_CHANNEL_XN_STRIDE; if (allocation.localXn.num == 0 || allocation.localWaitCke.num == 0 || allocation.remoteNotifyCke.num == 0 || allocation.remoteXn.num < routedPeerCount || - allocation.localWaitCke.num < allocation.remoteXn.num || - allocation.remoteNotifyCke.num < allocation.remoteXn.num || + allocation.localWaitCke.num < syncRouteCount || + allocation.remoteNotifyCke.num < syncRouteCount || allocation.channels.num == 0 || routesPerPeer == 0 || syncRouteCount % routedPeerCount != 0 || + routesPerPeer > TILEXR_CCU_CHANNEL_XN_STRIDE || remoteCcuBuffers->size() != peerRouteCount) { if (report != nullptr) { - report->message = "invalid direct CCU peer XN/CKE exchange shape"; + report->message = "invalid direct CCU peer XN/CKE exchange shape" + " peerRoutes=" + std::to_string(peerRouteCount) + + " routedPeers=" + std::to_string(routedPeerCount) + + " syncRoutes=" + std::to_string(syncRouteCount) + + " localXn=" + std::to_string(allocation.localXn.num) + + " localWaitCke=" + std::to_string(allocation.localWaitCke.num) + + " remoteNotifyCke=" + std::to_string(allocation.remoteNotifyCke.num) + + " channels=" + std::to_string(allocation.channels.num) + + " peerBuffers=" + std::to_string(remoteCcuBuffers->size()); } return TILEXR_ERROR_PARA_CHECK_FAIL; } @@ -623,7 +641,7 @@ int TileXRCcuCollectivePlanner::ExchangeDirectCcuRemoteNotifyCke( peerRanks.push_back(selectedDiagnosticPeer); } for (int peer = 0; peer < rankSize; ++peer) { - if (peer != rank && peer != selectedDiagnosticPeer && routedPeerCount > 1U) { + if (peer != rank && peer != selectedDiagnosticPeer && peerRanks.size() < routedPeerCount) { peerRanks.push_back(peer); } } @@ -658,7 +676,7 @@ int TileXRCcuCollectivePlanner::ExchangeDirectCcuRemoteNotifyCke( remoteCcuBuffers->assign(syncRouteCount, TileXRCcuRemoteCcuBufferInfo{}); size_t routeIndex = 0; - for (uint32_t syncIndex = 0; syncIndex < allocation.remoteXn.num; ++syncIndex) { + for (uint32_t syncIndex = 0; syncIndex < syncRouteCount; ++syncIndex) { const size_t peerBufferIndex = syncIndex / routesPerPeer; const size_t routeWithinPeer = syncIndex % routesPerPeer; const int peer = peerRanks[peerBufferIndex]; @@ -668,7 +686,11 @@ int TileXRCcuCollectivePlanner::ExchangeDirectCcuRemoteNotifyCke( const uint32_t peerLocalResourceOffset = static_cast( peerLocalIndex * routesPerPeer + routeWithinPeer); const uint32_t peerLocalXnOffset = peerLocalResourceOffset; - const uint32_t selectedRemoteXnOffset = peerLocalResourceOffset; + const uint32_t selectedRemoteXnOffset = SelectDirectCcuChannelBoundRemoteXnOffset( + peerLocalIndex, + routeWithinPeer, + routesPerPeer, + channelStridedRemoteXn); const uint32_t peerLocalWaitCkeOffset = peerLocalResourceOffset; if (peerResources.localXnCount == 0 || peerResources.remoteXnCount == 0 || @@ -1412,7 +1434,7 @@ int TileXRCcuCollectivePlanner::PrepareDirectCcuAllToAll2RankInstallAttempt( next.syncResourceCount = 3; next.syncInstructionCount = std::max( next.syncInstructionCount, - TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT); + DirectAllToAll2RankInstructionCapacity(bytes)); next.bindingsPerSyncResource = next.bindingsPerSyncResource == 0 ? 1 : next.bindingsPerSyncResource; if (next.provider.empty()) { next.provider = "tilexr-comm-direct-ccu-alltoall"; @@ -1448,15 +1470,16 @@ int TileXRCcuCollectivePlanner::PrepareDirectCcuAllToAllMeshInstallAttempt( } const int rank = session.Rank(); const int rankSize = session.RankSize(); - if (rankSize != 4 || rank < 0 || rank >= rankSize || localSourceAddr == 0 || localDestinationAddr == 0 || - chunkBytes == 0 || chunkBytes > std::numeric_limits::max() / 4ULL) { + if (rankSize < 2 || rankSize > 64 || rank < 0 || rank >= rankSize || + localSourceAddr == 0 || localDestinationAddr == 0 || chunkBytes == 0 || + chunkBytes > std::numeric_limits::max() / static_cast(rankSize)) { if (report != nullptr) { *report = TileXRCcuDirectInstallReport {}; report->message = "invalid direct CCU alltoall mesh endpoint"; } return TILEXR_ERROR_PARA_CHECK_FAIL; } - const uint64_t bufferBytes = chunkBytes * 4ULL; + const uint64_t bufferBytes = chunkBytes * static_cast(rankSize); const std::string unavailableMessage = session.DirectCcuRuntimeUnavailableMessage(); if (!unavailableMessage.empty()) { if (report != nullptr) { @@ -1581,8 +1604,8 @@ int TileXRCcuCollectivePlanner::PrepareDirectCcuAllToAllMeshInstallAttempt( next.prepareLowerLayerPlan = &TileXRCcuCollectivePlanner::PrepareDirectCcuLowerLayerPlanCallback; next.lowerLayerPlanUserData = &callbackContext; next.sqeArgCount = 0; - next.syncResourceCount = 3U; - next.bindingsPerSyncResource = next.bindingsPerSyncResource == 0 ? 1 : next.bindingsPerSyncResource; + next.syncResourceCount = static_cast(rankSize - 1); + next.bindingsPerSyncResource = 1U; if (next.provider.empty()) { next.provider = "tilexr-comm-direct-ccu-alltoall-mesh"; } diff --git a/src/comm/ccu/tilexr_ccu_direct_orchestrator.cpp b/src/comm/ccu/tilexr_ccu_direct_orchestrator.cpp index 1df19674..953482ed 100644 --- a/src/comm/ccu/tilexr_ccu_direct_orchestrator.cpp +++ b/src/comm/ccu/tilexr_ccu_direct_orchestrator.cpp @@ -38,14 +38,37 @@ constexpr uint32_t TILEXR_CCU_DIRECT_MEMORY_COPY_INSTRUCTION_COUNT = 7U; constexpr uint32_t TILEXR_CCU_DIRECT_MEMORY_COPY_LOCAL_XN_COUNT = 3U; constexpr uint32_t TILEXR_CCU_DIRECT_MEMORY_COPY_LOCAL_GSA_COUNT = 2U; constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_SYNC_RESOURCE_COUNT = 3U; -constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT = - 7U + 64U * 7U; -constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT = 3U; +constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_MAX_RANK_SIZE = 64U; +constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_CKE_MASK_BITS = 16U; +constexpr uint32_t TILEXR_CCU_DIRECT_ALLTOALL_XN_BINDINGS_PER_CHANNEL = 3U; + +uint32_t DirectAllToAllMeshPeerCount(uint32_t rankSize) +{ + return rankSize >= 2U && rankSize <= TILEXR_CCU_DIRECT_ALLTOALL_MAX_RANK_SIZE ? rankSize - 1U : 0U; +} + +uint32_t DirectAllToAllMeshCompletionCkeCount(uint32_t rankSize) +{ + const uint32_t peers = DirectAllToAllMeshPeerCount(rankSize); + return peers == 0 ? 0U : (peers + TILEXR_CCU_DIRECT_ALLTOALL_CKE_MASK_BITS - 1U) / + TILEXR_CCU_DIRECT_ALLTOALL_CKE_MASK_BITS; +} constexpr uint32_t TILEXR_CCU_DIRECT_SIGNAL_INSTRUCTION_COUNT = 5U; constexpr uint32_t TILEXR_CCU_DIRECT_WAIT_INSTRUCTION_COUNT = 5U; constexpr uint32_t TILEXR_CCU_DIRECT_SIGNAL_WAIT_INSTRUCTION_COUNT = 6U; constexpr uint32_t TILEXR_CCU_DIRECT_SYNC_XN_PING_INSTRUCTION_COUNT = 2U; +uint32_t DirectAllToAll2RankInstructionCapacity(uint64_t bytes) +{ + if (bytes == 0 || bytes % TILEXR_CCU_ALLTOALL_BLOCK_BYTES != 0) { + return 0; + } + const uint64_t blocks = bytes / TILEXR_CCU_ALLTOALL_BLOCK_BYTES; + const uint64_t instructions = 7ULL + blocks * 7ULL; + return instructions > std::numeric_limits::max() ? + 0U : static_cast(instructions); +} + uint32_t SyncXnPingAllocationInstructionCount(uint32_t syncResourceCount) { if (syncResourceCount > std::numeric_limits::max() / 2U) { @@ -54,18 +77,24 @@ uint32_t SyncXnPingAllocationInstructionCount(uint32_t syncResourceCount) return syncResourceCount * 2U; } -uint32_t DirectAllToAllMeshInstructionCount(uint64_t chunkBytes) +uint32_t DirectAllToAllMeshInstructionCount(uint32_t rankSize, uint64_t chunkBytes) { - if (chunkBytes == 0 || chunkBytes % TILEXR_CCU_ALLTOALL_BLOCK_BYTES != 0) { + const uint64_t peers = DirectAllToAllMeshPeerCount(rankSize); + const uint64_t completionCkes = DirectAllToAllMeshCompletionCkeCount(rankSize); + if (peers == 0 || chunkBytes == 0 || chunkBytes % TILEXR_CCU_ALLTOALL_BLOCK_BYTES != 0) { return 0; } const uint64_t blocks = chunkBytes / TILEXR_CCU_ALLTOALL_BLOCK_BYTES; - const uint64_t instructions = 11ULL + blocks * 28ULL + 7ULL + 1ULL; + const uint64_t preSync = 3ULL + peers * 3ULL; + const uint64_t perBlock = peers * 6ULL + 9ULL + completionCkes; + const uint64_t postSync = peers * 2ULL; + const uint64_t instructions = preSync + blocks * perBlock + postSync + 1ULL; return instructions > std::numeric_limits::max() ? 0U : static_cast(instructions); } bool DirectAllToAllMeshCapacityFits( const TileXRCcuResourceSpec& resources, + uint32_t rankSize, uint32_t instructionCount, std::string* message) { @@ -89,20 +118,23 @@ bool DirectAllToAllMeshCapacityFits( resources.ckeCount : resources.localWaitCkeCount; const uint32_t remoteNotifyCkeCount = resources.remoteNotifyCkeCount == 0 ? resources.ckeCount : resources.remoteNotifyCkeCount; - return require("mission", 1U, resources.missionCount) && + const uint32_t peers = DirectAllToAllMeshPeerCount(rankSize); + const uint32_t completionCkes = DirectAllToAllMeshCompletionCkeCount(rankSize); + const uint32_t localXns = std::max(peers, TILEXR_CCU_DIRECT_ALLTOALL_XN_BINDINGS_PER_CHANNEL); + const uint32_t remoteXns = TILEXR_CCU_DIRECT_ALLTOALL_XN_BINDINGS_PER_CHANNEL; + return peers != 0U && completionCkes != 0U && require("mission", 1U, resources.missionCount) && require("instruction", repositoryPrefix + instructionCount, resources.instructionCount) && require("GSA", TILEXR_CCU_DIRECT_MEMORY_COPY_LOCAL_GSA_COUNT, resources.gsaCount) && - require("local XN", TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT, resources.xnCount) && - require("remote XN", TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT, + require("local XN", localXns, resources.xnCount) && + require("remote XN", remoteXns, resources.remoteXnCount == 0 ? - (resources.xnCount > TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT ? - resources.xnCount - TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT : 0U) : + (resources.xnCount > localXns ? resources.xnCount - localXns : 0U) : resources.remoteXnCount) && - require("local CKE", TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT + 2U, + require("local CKE", peers + 1U + completionCkes, localWaitCkeCount) && - require("remote CKE", TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT, + require("remote CKE", peers, remoteNotifyCkeCount) && - require("channel", TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT, resources.channelCount); + require("channel", peers, resources.channelCount); } void ResetReport(TileXRCcuDirectInstallReport* report) @@ -842,10 +874,12 @@ int ConfigureDirectMemoryCopyResources( int ConfigureDirectAllToAll2RankResources( const TileXRCcuDirectInstallOptions& options, + const TileXRCcuDirectAllToAll2RankSpec& alltoall, TileXRCcuDirectInstallAttempt* attempt, TileXRCcuDirectInstallReport* report) { - if (attempt == nullptr || + const uint32_t instructionCapacity = DirectAllToAll2RankInstructionCapacity(alltoall.bytes); + if (attempt == nullptr || instructionCapacity == 0 || attempt->plan.syncResources.size() != TILEXR_CCU_DIRECT_ALLTOALL_SYNC_RESOURCE_COUNT || attempt->plan.taskWindows.size() != 1) { if (report != nullptr) { @@ -882,7 +916,7 @@ int ConfigureDirectAllToAll2RankResources( attempt->plan.taskWindows[0].instCnt = static_cast(std::max( attempt->plan.taskWindows[0].instCnt, - TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT)); + instructionCapacity)); return TILEXR_SUCCESS; } @@ -892,12 +926,14 @@ int ConfigureDirectAllToAllMeshResources( TileXRCcuDirectInstallAttempt* attempt, TileXRCcuDirectInstallReport* report) { - const uint32_t instructionCount = DirectAllToAllMeshInstructionCount(alltoall.chunkBytes); + const uint32_t peerCount = DirectAllToAllMeshPeerCount(alltoall.rankSize); + const uint32_t completionCkeCount = DirectAllToAllMeshCompletionCkeCount(alltoall.rankSize); + const uint32_t instructionCount = DirectAllToAllMeshInstructionCount(alltoall.rankSize, alltoall.chunkBytes); if (attempt == nullptr || instructionCount == 0 || - attempt->plan.syncResources.size() != TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || + attempt->plan.syncResources.size() != peerCount || attempt->plan.taskWindows.size() != 1U) { if (report != nullptr) { - report->message = "alltoall mesh direct CCU plan requires three peer sync resources and one task"; + report->message = "alltoall mesh direct CCU plan requires rankSize-1 peer resources and one task"; } return TILEXR_ERROR_PARA_CHECK_FAIL; } @@ -907,12 +943,10 @@ int ConfigureDirectAllToAllMeshResources( } return TILEXR_ERROR_PARA_CHECK_FAIL; } - if (attempt->allocation.localXn.num < TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || - attempt->allocation.remoteXn.num < TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || - attempt->allocation.localWaitCke.num < TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || - attempt->allocation.remoteNotifyCke.num < TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || - attempt->allocation.channels.num < TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || - attempt->allocation.sourceCke.num < 2U) { + if (attempt->allocation.localXn.num < peerCount || attempt->allocation.remoteXn.num < peerCount || + attempt->allocation.localWaitCke.num < peerCount || attempt->allocation.remoteNotifyCke.num < peerCount || + attempt->allocation.channels.num < peerCount || + attempt->allocation.sourceCke.num < 1U + completionCkeCount) { if (report != nullptr) { report->message = "alltoall mesh direct CCU allocation is missing XN/CKE/channel resources"; } @@ -1201,8 +1235,7 @@ int ValidateDirectAllToAllMeshRouteResources( const TileXRCcuProducerPlan& plan, TileXRCcuDirectInstallReport* report) { - if (mesh.peers.size() != 3U || - plan.syncResources.size() != TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT) { + if (mesh.peers.size() != mesh.rankSize - 1U || plan.syncResources.size() != mesh.peers.size()) { if (report != nullptr) { report->message = "alltoall mesh route binding validation has an invalid shape"; } @@ -1222,7 +1255,8 @@ int ValidateDirectAllToAllMeshRouteResources( route.preSyncRemoteNotifyCke != resource.notifyCke || route.preSyncRemoteTokenNotifyCke != resource.notifyCke || route.postSyncRemoteNotifyCke != resource.notifyCke || - route.copyCompletionCke != mesh.remoteCompletionCke) { + route.copyCompletionCke != + mesh.remoteCompletionCkes[ordinal / TILEXR_CCU_DIRECT_ALLTOALL_CKE_MASK_BITS]) { if (report != nullptr) { std::ostringstream stream; stream << "alltoall mesh route binding mismatch peerRank=" << mesh.peers[ordinal].peerRank @@ -1241,7 +1275,7 @@ int BuildDirectAllToAllMeshLaunchPackage( TileXRCcuDirectInstallReport* report) { if (attempt == nullptr || - attempt->plan.syncResources.size() != TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT || + attempt->plan.syncResources.size() != DirectAllToAllMeshPeerCount(alltoall.rankSize) || attempt->plan.kernelLocalGsa.num < TILEXR_CCU_DIRECT_MEMORY_COPY_LOCAL_GSA_COUNT) { if (report != nullptr) { report->message = "missing direct CCU alltoall mesh producer resources"; @@ -1267,7 +1301,11 @@ int BuildDirectAllToAllMeshLaunchPackage( const uint16_t localXnStart = attempt->allocation.localXn.startId; const uint16_t remoteXnStart = attempt->allocation.remoteXn.startId; - mesh.remoteCompletionCke = static_cast(attempt->allocation.sourceCke.startId + 1U); + const uint32_t completionCkeCount = DirectAllToAllMeshCompletionCkeCount(alltoall.rankSize); + for (uint32_t group = 0; group < completionCkeCount; ++group) { + mesh.remoteCompletionCkes.push_back( + static_cast(attempt->allocation.sourceCke.startId + 1U + group)); + } for (uint32_t ordinal = 0; ordinal < peers.size(); ++ordinal) { const TileXRCcuSyncResource& resource = attempt->plan.syncResources[ordinal]; TileXRCcuAllToAllMeshPeerSpec peer; @@ -1295,7 +1333,8 @@ int BuildDirectAllToAllMeshLaunchPackage( route.preSyncTokenChannelId = resource.channelId; route.copyChannelId = resource.channelId; route.postSyncChannelId = resource.channelId; - route.copyCompletionCke = mesh.remoteCompletionCke; + route.copyCompletionCke = + mesh.remoteCompletionCkes[ordinal / TILEXR_CCU_DIRECT_ALLTOALL_CKE_MASK_BITS]; route.preSyncLocalWaitCke = resource.localWaitCke; route.preSyncRemoteNotifyCke = resource.notifyCke; route.preSyncTokenLocalWaitCke = resource.localWaitCke; @@ -1819,7 +1858,8 @@ int RunDirectInstallAttemptImpl( std::string capacityMessage; if (!DirectAllToAllMeshCapacityFits( attempt->resourceSpec, - DirectAllToAllMeshInstructionCount(alltoallMesh->chunkBytes), + alltoallMesh->rankSize, + DirectAllToAllMeshInstructionCount(alltoallMesh->rankSize, alltoallMesh->chunkBytes), &capacityMessage)) { return Fail(attempt, report, capacityMessage); } @@ -1829,7 +1869,7 @@ int RunDirectInstallAttemptImpl( signalWait != nullptr || syncXnPing != nullptr; attempt->resourceRequest.sqeArgCount = customProgram ? 0U : options.sqeArgCount; attempt->resourceRequest.syncResourceCount = - alltoallMesh != nullptr ? TILEXR_CCU_DIRECT_ALLTOALL_MESH_SYNC_RESOURCE_COUNT : + alltoallMesh != nullptr ? DirectAllToAllMeshPeerCount(alltoallMesh->rankSize) : alltoall != nullptr ? TILEXR_CCU_DIRECT_ALLTOALL_SYNC_RESOURCE_COUNT : syncXnPing != nullptr ? options.syncResourceCount : customProgram ? 1U : options.syncResourceCount; @@ -1837,10 +1877,12 @@ int RunDirectInstallAttemptImpl( memoryCopy != nullptr ? std::max(options.syncInstructionCount, TILEXR_CCU_DIRECT_MEMORY_COPY_INSTRUCTION_COUNT) : alltoall != nullptr ? - std::max(options.syncInstructionCount, TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT) : + std::max( + options.syncInstructionCount, + DirectAllToAll2RankInstructionCapacity(alltoall->bytes)) : alltoallMesh != nullptr ? std::max(options.syncInstructionCount, - DirectAllToAllMeshInstructionCount(alltoallMesh->chunkBytes)) : + DirectAllToAllMeshInstructionCount(alltoallMesh->rankSize, alltoallMesh->chunkBytes)) : signalWait != nullptr ? std::max(options.syncInstructionCount, SignalWaitInstructionCount(*signalWait)) : syncXnPing != nullptr ? @@ -1848,8 +1890,16 @@ int RunDirectInstallAttemptImpl( options.syncInstructionCount, SyncXnPingAllocationInstructionCount(options.syncResourceCount)) : options.syncInstructionCount; - attempt->resourceRequest.bindingsPerSyncResource = options.bindingsPerSyncResource; - attempt->resourceRequest.sourceCkeCount = alltoallMesh != nullptr ? 2U : 1U; + attempt->resourceRequest.bindingsPerSyncResource = alltoallMesh != nullptr ? + 1U : options.bindingsPerSyncResource; + attempt->resourceRequest.minimumLocalXnCount = + alltoallMesh != nullptr ? + TILEXR_CCU_DIRECT_ALLTOALL_XN_BINDINGS_PER_CHANNEL : 0U; + attempt->resourceRequest.minimumRemoteXnCount = + alltoallMesh != nullptr ? + TILEXR_CCU_DIRECT_ALLTOALL_XN_BINDINGS_PER_CHANNEL : 0U; + attempt->resourceRequest.sourceCkeCount = alltoallMesh != nullptr ? + 1U + DirectAllToAllMeshCompletionCkeCount(alltoallMesh->rankSize) : 1U; attempt->resourceRequest.barrierMode = alltoallMesh != nullptr ? TileXRCcuBarrierMode::SyncCke : alltoall != nullptr ? TileXRCcuBarrierMode::SyncXn : @@ -1882,7 +1932,7 @@ int RunDirectInstallAttemptImpl( report->message); } } else if (alltoall != nullptr) { - ret = ConfigureDirectAllToAll2RankResources(options, attempt, report); + ret = ConfigureDirectAllToAll2RankResources(options, *alltoall, attempt, report); if (ret != TILEXR_SUCCESS) { return Fail( attempt, @@ -2075,18 +2125,22 @@ int TileXRCcuRunDirectAllToAllMeshInstallAttempt( TileXRCcuDirectInstallAttempt* attempt, TileXRCcuDirectInstallReport* report) { - bool peerRanks[4] = {}; - bool valid = alltoall.rankSize == 4U && alltoall.localRank < alltoall.rankSize && + bool valid = alltoall.rankSize >= 2U && alltoall.rankSize <= TILEXR_CCU_DIRECT_ALLTOALL_MAX_RANK_SIZE && + alltoall.localRank < alltoall.rankSize && alltoall.localSendAddr != 0 && alltoall.localSendToken != 0 && alltoall.localRecvAddr != 0 && alltoall.localRecvToken != 0 && - DirectAllToAllMeshInstructionCount(alltoall.chunkBytes) != 0 && alltoall.peers.size() == 3U; - for (const auto& peer : alltoall.peers) { - if (peer.peerRank >= alltoall.rankSize || peer.peerRank == alltoall.localRank || - peerRanks[peer.peerRank] || peer.remoteRecvAddr == 0 || peer.remoteRecvToken == 0) { - valid = false; - break; + DirectAllToAllMeshInstructionCount(alltoall.rankSize, alltoall.chunkBytes) != 0 && + alltoall.peers.size() == alltoall.rankSize - 1U; + if (valid) { + std::vector peerRanks(alltoall.rankSize, false); + for (const auto& peer : alltoall.peers) { + if (peer.peerRank >= alltoall.rankSize || peer.peerRank == alltoall.localRank || + peerRanks[peer.peerRank] || peer.remoteRecvAddr == 0 || peer.remoteRecvToken == 0) { + valid = false; + break; + } + peerRanks[peer.peerRank] = true; } - peerRanks[peer.peerRank] = true; } if (!valid) { ResetReport(report); diff --git a/src/comm/ccu/tilexr_ccu_direct_orchestrator.h b/src/comm/ccu/tilexr_ccu_direct_orchestrator.h index a98b945f..e5a32cfb 100644 --- a/src/comm/ccu/tilexr_ccu_direct_orchestrator.h +++ b/src/comm/ccu/tilexr_ccu_direct_orchestrator.h @@ -92,7 +92,7 @@ struct TileXRCcuDirectAllToAllMeshPeerSpec { }; struct TileXRCcuDirectAllToAllMeshSpec { - uint32_t rankSize = 4; + uint32_t rankSize = 2; uint32_t localRank = 0; uint64_t localSendAddr = 0; uint64_t localSendToken = 0; diff --git a/src/comm/ccu/tilexr_ccu_direct_runtime.cpp b/src/comm/ccu/tilexr_ccu_direct_runtime.cpp index 1183e2a2..be8eafb9 100644 --- a/src/comm/ccu/tilexr_ccu_direct_runtime.cpp +++ b/src/comm/ccu/tilexr_ccu_direct_runtime.cpp @@ -26,6 +26,10 @@ constexpr uint32_t TILEXR_CCU_DIRECT_CCUM_SQE_BYTES = 64; constexpr uint32_t TILEXR_CCU_DIRECT_SQ_EBB_WORDS = 4; constexpr uint32_t TILEXR_CCU_DIRECT_LOOP_JETTY_ID = 1024; constexpr uint32_t TILEXR_CCU_DIRECT_LOOP_JETTY_CTX_ID = 0; +constexpr uint32_t TILEXR_CCU_HCOMM_INNER_FE_JETTY_NUM = 23; +constexpr uint32_t TILEXR_CCU_HCOMM_OUTER_FE_START_JETTY_CTX_ID = 92; +constexpr uint32_t TILEXR_CCU_HCOMM_OUTER_FE_JETTY_NUM = 36; +constexpr uint32_t TILEXR_CCU_HCOMM_MAX_INNER_FE_ID = 7; constexpr uint64_t TILEXR_CCU_V1_WQE_BASIC_BLOCK_OFFSET = TILEXR_CCU_V1_CCUM_OFFSET + 0x800000ULL; constexpr uint64_t TILEXR_CCU_DIRECT_SQ_BUFFER_BYTES = 256ULL * 1024ULL; constexpr uint32_t TILEXR_CCU_DIRECT_CCU_POLL_CQ_DEPTH = 64; @@ -33,6 +37,7 @@ constexpr uint32_t TILEXR_CCU_DIRECT_ENDPOINT_ROUTE_ASYNC_MAX_POLLS = 1000; constexpr uint32_t TILEXR_CCU_DIRECT_ENDPOINT_ROUTE_ASYNC_SLEEP_US = 1000; constexpr uint32_t TILEXR_CCU_DIRECT_ENDPOINT_ROUTE_MAX_ATTEMPTS = 8; constexpr uint8_t TILEXR_CCU_DIRECT_ENDPOINT_ERR_TIMEOUT = 16; +constexpr uint8_t TILEXR_CCU_DIRECT_CTP_ENDPOINT_ERR_TIMEOUT = 8; constexpr int TILEXR_CCU_DIRECT_MAX_RANK_SIZE = 128; constexpr int TILEXR_CCU_HCCP_JFC_MODE_CCU_POLL = 2; constexpr int TILEXR_CCU_HCCP_ASYNC_EAGAIN = 128301; @@ -458,10 +463,30 @@ uint32_t SelectEndpointRouteSqBytes(uint32_t sqDepth) return sqDepth * TILEXR_CCU_DIRECT_SQ_EBB_WORDS * TILEXR_CCU_DIRECT_CCUM_SQE_BYTES; } -uint64_t SelectEndpointRouteSqVa(const TileXRCcuLocalResourceWindowInfo& localResourceWindow) +bool SelectEndpointRouteJettyCtxId(uint32_t pfeId, uint32_t peerOrdinal, uint16_t* jettyCtxId) +{ + if (jettyCtxId == nullptr) { + return false; + } + const uint32_t start = pfeId > TILEXR_CCU_HCOMM_MAX_INNER_FE_ID ? + TILEXR_CCU_HCOMM_OUTER_FE_START_JETTY_CTX_ID : + pfeId * TILEXR_CCU_HCOMM_INNER_FE_JETTY_NUM; + const uint32_t count = pfeId > TILEXR_CCU_HCOMM_MAX_INNER_FE_ID ? + TILEXR_CCU_HCOMM_OUTER_FE_JETTY_NUM : + TILEXR_CCU_HCOMM_INNER_FE_JETTY_NUM; + if (peerOrdinal >= count || start + peerOrdinal >= 128U) { + return false; + } + *jettyCtxId = static_cast(start + peerOrdinal); + return true; +} + +uint64_t SelectEndpointRouteSqVa( + const TileXRCcuLocalResourceWindowInfo& localResourceWindow, + uint16_t jettyCtxId) { return localResourceWindow.addr + TILEXR_CCU_V1_WQE_BASIC_BLOCK_OFFSET + - static_cast(TILEXR_CCU_DIRECT_LOOP_JETTY_CTX_ID) * TILEXR_CCU_DIRECT_SQ_BUFFER_BYTES; + static_cast(jettyCtxId) * TILEXR_CCU_DIRECT_SQ_BUFFER_BYTES; } int WaitRaCtxAsyncRequest(TileXRCcuHccpLoader& loader, void* reqHandle) @@ -1371,7 +1396,9 @@ int TileXRCcuDirectRuntime::CollectLocalEndpointRouteWithRaCtxOnce( ReleaseLocalEndpointRoute(); const uint32_t sqDepth = SelectEndpointRouteSqDepth(); - const uint64_t sqVa = SelectEndpointRouteSqVa(localResourceWindow_); + const uint64_t sqVa = SelectEndpointRouteSqVa( + localResourceWindow_, + TILEXR_CCU_DIRECT_LOOP_JETTY_CTX_ID); const uint32_t sqBytes = SelectEndpointRouteSqBytes(sqDepth); if (TraceEndpointRoute()) { std::cerr << "TileXRDirectCcuTrace endpointRoute begin" @@ -1648,6 +1675,7 @@ int TileXRCcuDirectRuntime::CreatePeerEndpointState( uint32_t peerDevicePhyId, const std::array& localEid, const std::array& peerEid, + uint32_t tpType, uint32_t peerOrdinal, TileXRCcuPeerEndpointState* state) { @@ -1658,6 +1686,10 @@ int TileXRCcuDirectRuntime::CreatePeerEndpointState( *state = TileXRCcuPeerEndpointState {}; state->peerRank = peerRank; state->peerDevicePhyId = peerDevicePhyId; + state->tpType = tpType; + if (tpType != TILEXR_CCU_HCCP_TP_TYPE_RTP && tpType != TILEXR_CCU_HCCP_TP_TYPE_CTP) { + return TILEXR_ERROR_PARA_CHECK_FAIL; + } TileXRCcuRaInfo raInfo {}; raInfo.mode = TILEXR_CCU_NETWORK_OFFLINE; @@ -1709,47 +1741,34 @@ int TileXRCcuDirectRuntime::CreatePeerEndpointState( TileXRCcuRaInfo randomInfo {}; randomInfo.mode = TILEXR_CCU_NETWORK_OFFLINE; randomInfo.phyId = devicePhyId_; - ret = loader_.RaGetSecRandom(&randomInfo, &state->resourceWindow.tokenValue); + ret = loader_.RaGetSecRandom(&randomInfo, &state->jettyTokenValue); if (ret != 0) { ReleasePeerEndpointState(state); return TILEXR_ERROR_MKIRT; } - - const uint64_t alignedAddr = AlignResourceWindowAddr(localResourceWindow_.addr); - TileXRCcuHccpMrRegInfo mr {}; - mr.in.mem.addr = alignedAddr; - mr.in.mem.size = localResourceWindow_.bytes + (localResourceWindow_.addr - alignedAddr); - mr.in.ub.flags.value = 0; - mr.in.ub.flags.bs.tokenPolicy = TILEXR_CCU_HCCP_TOKEN_POLICY_PLAIN_TEXT; - mr.in.ub.flags.bs.tokenIdValid = 1; - mr.in.ub.flags.bs.access = TILEXR_CCU_HCCP_MEM_SEG_ACCESS_DEFAULT; - mr.in.ub.flags.bs.nonPin = 1; - mr.in.ub.tokenValue = state->resourceWindow.tokenValue; - mr.in.ub.tokenIdHandle = state->resourceWindow.tokenIdHandle; - ret = loader_.RaCtxLmemRegister( - state->resourceWindow.raCtxHandle, - &mr, - &state->resourceWindow.lmemHandle); - if (ret != 0 || state->resourceWindow.lmemHandle == nullptr) { - ReleasePeerEndpointState(state); - return TILEXR_ERROR_MKIRT; - } - const uint32_t rawTokenId = mr.out.ub.tokenId != 0 ? mr.out.ub.tokenId : allocatedToken.tokenId; state->resourceWindow.addr = localResourceWindow_.addr; state->resourceWindow.bytes = localResourceWindow_.bytes; - state->resourceWindow.rawTokenId = rawTokenId; - state->resourceWindow.tokenId = rawTokenId >> TILEXR_CCU_URMA_TOKEN_ID_RIGHT_SHIFT; - state->resourceWindow.targetSegHandle = mr.out.ub.targetSegHandle; + state->resourceWindow.rawTokenId = localResourceWindow_.rawTokenId; + state->resourceWindow.tokenId = localResourceWindow_.tokenId; + state->resourceWindow.tokenValue = localResourceWindow_.tokenValue; + state->resourceWindow.targetSegHandle = localResourceWindow_.targetSegHandle; state->resourceWindow.eid = localEid; state->resourceWindow.eidIndex = state->eidInfo.eidIndex; state->resourceWindow.funcId = state->eidInfo.funcId; state->resourceWindow.funcIdValid = true; - state->resourceWindow.raCtxRegistered = true; + state->resourceWindow.raCtxRegistered = false; + + uint16_t jettyCtxId = 0; + if (!SelectEndpointRouteJettyCtxId(state->eidInfo.funcId, peerOrdinal, &jettyCtxId)) { + ReleasePeerEndpointState(state); + return TILEXR_ERROR_PARA_CHECK_FAIL; + } ret = SelectTpRouteForPeer( state->resourceWindow.raCtxHandle, localEid, peerEid, + state->tpType, &state->localTpHandle, &state->mappedJettyPriority); if (ret != TILEXR_SUCCESS) { @@ -1779,19 +1798,19 @@ int TileXRCcuDirectRuntime::CreatePeerEndpointState( qpAttr.rqDepth = TILEXR_CCU_HCCP_RQ_DEPTH_DEFAULT; qpAttr.transportMode = TILEXR_CCU_HCCP_TRANSPORT_MODE_RM; qpAttr.ub.mode = static_cast(TILEXR_CCU_HCCP_JETTY_MODE_CCU); - qpAttr.ub.jettyId = static_cast(TILEXR_CCU_DIRECT_LOOP_JETTY_ID + peerOrdinal); + qpAttr.ub.jettyId = static_cast(TILEXR_CCU_DIRECT_LOOP_JETTY_ID + jettyCtxId); qpAttr.ub.tokenIdHandle = state->resourceWindow.tokenIdHandle; - qpAttr.ub.tokenValue = state->resourceWindow.tokenValue; + qpAttr.ub.tokenValue = state->jettyTokenValue; qpAttr.ub.flag.value = 0; qpAttr.ub.flag.bs.shareJfr = 1; qpAttr.ub.jfsFlag.bs.errorSuspend = 1; qpAttr.ub.priority = state->mappedJettyPriority; qpAttr.ub.rnrRetry = TILEXR_CCU_HCCP_RNR_RETRY_DEFAULT; - qpAttr.ub.errTimeout = TILEXR_CCU_DIRECT_ENDPOINT_ERR_TIMEOUT; + qpAttr.ub.errTimeout = state->tpType == TILEXR_CCU_HCCP_TP_TYPE_CTP ? + TILEXR_CCU_DIRECT_CTP_ENDPOINT_ERR_TIMEOUT : TILEXR_CCU_DIRECT_ENDPOINT_ERR_TIMEOUT; qpAttr.ub.extMode.cstmFlag.value = 0; qpAttr.ub.extMode.cstmFlag.bs.sqCstm = 1; - qpAttr.ub.extMode.sq.buffVa = SelectEndpointRouteSqVa(localResourceWindow_) + - static_cast(peerOrdinal) * TILEXR_CCU_DIRECT_SQ_BUFFER_BYTES; + qpAttr.ub.extMode.sq.buffVa = SelectEndpointRouteSqVa(localResourceWindow_, jettyCtxId); qpAttr.ub.extMode.sq.buffSize = SelectEndpointRouteSqBytes(sqDepth); qpAttr.ub.extMode.sqebbNum = sqDepth; ret = loader_.RaCtxQpCreate( @@ -1814,6 +1833,9 @@ int TileXRCcuDirectRuntime::CreatePeerEndpointState( << " funcId=" << state->eidInfo.funcId << " tpHandle=0x" << std::hex << state->localTpHandle << std::dec << " priority=" << static_cast(state->mappedJettyPriority) + << " tpType=" << state->tpType + << " jettyCtxId=" << jettyCtxId + << " sqVa=0x" << std::hex << qpAttr.ub.extMode.sq.buffVa << std::dec << " qpId=" << state->qpInfo.ub.id << " psn=" << state->psn << std::endl; @@ -1885,6 +1907,7 @@ int TileXRCcuDirectRuntime::PreparePeerEndpointRoutes(TileXRCcuDirectRuntimeRepo peerDevicePhyIds[ordinal], topologyRoutes[ordinal].localEid, peerEid, + topologyRoutes[ordinal].tpType, ordinal, &state); if (ret != TILEXR_SUCCESS) { @@ -1901,7 +1924,7 @@ int TileXRCcuDirectRuntime::PreparePeerEndpointRoutes(TileXRCcuDirectRuntimeRepo offer.resourceTokenId = state.resourceWindow.tokenId; offer.resourceRawTokenId = state.resourceWindow.rawTokenId; offer.resourceTokenValue = state.resourceWindow.tokenValue; - offer.jettyTokenValue = state.resourceWindow.tokenValue; + offer.jettyTokenValue = state.jettyTokenValue; offer.eid = state.resourceWindow.eid; offer.qpKey = state.qpInfo.key; if (offer.qpKey.size == 0 || offer.qpKey.size > TILEXR_CCU_HCCP_QP_KEY_BYTES) { @@ -1973,7 +1996,7 @@ int TileXRCcuDirectRuntime::PreparePeerEndpointRoutes(TileXRCcuDirectRuntimeRepo importInfo.in.ub.expImportCfg.peerTpHandle = peerTpHandle; importInfo.in.ub.expImportCfg.txPsn = state.psn; importInfo.in.ub.expImportCfg.rxPsn = peerOffer.psn; - importInfo.in.ub.tpType = TILEXR_CCU_HCCP_TP_TYPE_RTP; + importInfo.in.ub.tpType = state.tpType; ret = loader_.RaCtxQpImport( state.resourceWindow.raCtxHandle, &importInfo, @@ -1987,7 +2010,7 @@ int TileXRCcuDirectRuntime::PreparePeerEndpointRoutes(TileXRCcuDirectRuntimeRepo state.route.doorbellVa = state.qpInfo.ub.dbAddr; state.route.doorbellTokenId = state.qpInfo.ub.dbTokenId >> TILEXR_CCU_URMA_TOKEN_ID_RIGHT_SHIFT; - state.route.doorbellTokenValue = state.resourceWindow.tokenValue; + state.route.doorbellTokenValue = state.jettyTokenValue; state.route.sqDepth = SelectEndpointRouteSqDepth(); state.route.startJettyId = static_cast(state.qpInfo.ub.id); state.route.remoteCcuVa = peerOffer.resourceAddr; @@ -2041,6 +2064,7 @@ int TileXRCcuDirectRuntime::SelectTpRouteForPeer( void* ctxHandle, const std::array& localEid, const std::array& peerEid, + uint32_t tpType, uint64_t* tpHandle, uint8_t* mappedJettyPriority) { @@ -2054,7 +2078,8 @@ int TileXRCcuDirectRuntime::SelectTpRouteForPeer( *mappedJettyPriority = 0; TileXRCcuHccpGetTpCfg tpCfg {}; - tpCfg.flag.bs.rtp = 1; + tpCfg.flag.bs.rtp = tpType == TILEXR_CCU_HCCP_TP_TYPE_RTP ? 1 : 0; + tpCfg.flag.bs.ctp = tpType == TILEXR_CCU_HCCP_TP_TYPE_CTP ? 1 : 0; tpCfg.transMode = TILEXR_CCU_HCCP_TRANSPORT_MODE_RM; std::copy(localEid.begin(), localEid.end(), tpCfg.localEid.raw); std::copy(peerEid.begin(), peerEid.end(), tpCfg.peerEid.raw); @@ -2094,21 +2119,23 @@ int TileXRCcuDirectRuntime::SelectTpRouteForPeer( return TILEXR_ERROR_NOT_FOUND; } - TileXRCcuHccpTpAttr setAttr {}; - setAttr.sl = mappedSl; - reqHandle = nullptr; - ret = loader_.RaSetTpAttrAsync( - ctxHandle, - tpInfos[tpIndex].tpHandle, - TILEXR_CCU_TP_ATTR_BITMAP_SL, - &setAttr, - &reqHandle); - if (ret != 0 || reqHandle == nullptr) { - return TILEXR_ERROR_MKIRT; - } - ret = WaitRaCtxAsyncRequest(loader_, reqHandle); - if (ret != TILEXR_SUCCESS) { - return ret; + if (tpType == TILEXR_CCU_HCCP_TP_TYPE_RTP) { + TileXRCcuHccpTpAttr setAttr {}; + setAttr.sl = mappedSl; + reqHandle = nullptr; + ret = loader_.RaSetTpAttrAsync( + ctxHandle, + tpInfos[tpIndex].tpHandle, + TILEXR_CCU_TP_ATTR_BITMAP_SL, + &setAttr, + &reqHandle); + if (ret != 0 || reqHandle == nullptr) { + return TILEXR_ERROR_MKIRT; + } + ret = WaitRaCtxAsyncRequest(loader_, reqHandle); + if (ret != TILEXR_SUCCESS) { + return ret; + } } if (TraceEndpointRoute()) { diff --git a/src/comm/ccu/tilexr_ccu_direct_runtime.h b/src/comm/ccu/tilexr_ccu_direct_runtime.h index f664a89e..d016c740 100644 --- a/src/comm/ccu/tilexr_ccu_direct_runtime.h +++ b/src/comm/ccu/tilexr_ccu_direct_runtime.h @@ -137,6 +137,8 @@ struct TileXRCcuPeerEndpointState { void* remoteQpHandle = nullptr; TileXRCcuHccpQpCreateInfo qpInfo {}; TileXRCcuLowerLayerTransportRoute route; + uint32_t jettyTokenValue = 0; + uint32_t tpType = TILEXR_CCU_HCCP_TP_TYPE_RTP; uint32_t psn = 0; uint64_t localTpHandle = 0; uint8_t mappedJettyPriority = 0; @@ -176,12 +178,14 @@ class TileXRCcuDirectRuntime { uint32_t peerDevicePhyId, const std::array& localEid, const std::array& peerEid, + uint32_t tpType, uint32_t peerOrdinal, TileXRCcuPeerEndpointState* state); int SelectTpRouteForPeer( void* ctxHandle, const std::array& localEid, const std::array& peerEid, + uint32_t tpType, uint64_t* tpHandle, uint8_t* mappedJettyPriority); int QueryTpHandleForPeer( diff --git a/src/comm/ccu/tilexr_ccu_lower_layer_plan_builder.cpp b/src/comm/ccu/tilexr_ccu_lower_layer_plan_builder.cpp index 39515270..03b8dc8d 100644 --- a/src/comm/ccu/tilexr_ccu_lower_layer_plan_builder.cpp +++ b/src/comm/ccu/tilexr_ccu_lower_layer_plan_builder.cpp @@ -500,10 +500,7 @@ int TileXRCcuBuildLowerLayerTransportTemplate( remoteNotifyCke.num == 0 || allocation.remoteXn.num == 0) { return Fail(nullptr, report, "missing lower-layer CCU allocated resources"); } - if (remoteCcuBuffers.empty() || remoteCcuBuffers.size() != allocation.remoteXn.num) { - return Fail(nullptr, report, "remote CCU buffer template count does not match remote XN allocation"); - } - if (allocation.channels.num < remoteCcuBuffers.size()) { + if (remoteCcuBuffers.empty() || remoteCcuBuffers.size() != allocation.channels.num) { return Fail(nullptr, report, "channel allocation count does not match lower-layer route count"); } if (remoteCcuBuffers.size() > std::numeric_limits::max()) { diff --git a/src/comm/ccu/tilexr_ccu_ra_custom_channel_provider.cpp b/src/comm/ccu/tilexr_ccu_ra_custom_channel_provider.cpp index f364290d..b160ca62 100644 --- a/src/comm/ccu/tilexr_ccu_ra_custom_channel_provider.cpp +++ b/src/comm/ccu/tilexr_ccu_ra_custom_channel_provider.cpp @@ -5,9 +5,17 @@ #include "ccu/tilexr_ccu_ra_custom_channel_provider.h" +#include +#include +#include + namespace TileXR { namespace { +constexpr int TILEXR_CCU_ROCE_EAGAIN = 128101; +constexpr uint32_t TILEXR_CCU_RA_EAGAIN_RETRY_COUNT = 100; +constexpr uint32_t TILEXR_CCU_RA_EAGAIN_RETRY_INTERVAL_MS = 100; + void ResetReport(TileXRCcuRaCustomChannelProviderReport* report) { if (report != nullptr) { @@ -103,10 +111,15 @@ int TileXRCcuRaCustomChannelProvider::AdapterCallback( TileXRCcuRaInfo info {}; info.mode = TILEXR_CCU_NETWORK_OFFLINE; info.phyId = devicePhyId; - return provider->raCustomChannel_( - info, - in, - out); + for (uint32_t retry = 0; retry <= TILEXR_CCU_RA_EAGAIN_RETRY_COUNT; ++retry) { + std::memset(out, 0, sizeof(*out)); + const int ret = provider->raCustomChannel_(info, in, out); + if (ret != TILEXR_CCU_ROCE_EAGAIN || retry == TILEXR_CCU_RA_EAGAIN_RETRY_COUNT) { + return ret; + } + std::this_thread::sleep_for(std::chrono::milliseconds(TILEXR_CCU_RA_EAGAIN_RETRY_INTERVAL_MS)); + } + return TILEXR_CCU_ROCE_EAGAIN; } } // namespace TileXR diff --git a/src/comm/ccu/tilexr_ccu_resource_allocator.cpp b/src/comm/ccu/tilexr_ccu_resource_allocator.cpp index 5b4840f7..a797a149 100644 --- a/src/comm/ccu/tilexr_ccu_resource_allocator.cpp +++ b/src/comm/ccu/tilexr_ccu_resource_allocator.cpp @@ -14,6 +14,7 @@ namespace { constexpr const char* TILEXR_CCU_HCOMM_DERIVED_PROVIDER = "tilexr-hcomm-derived-resource-allocator"; constexpr uint32_t TILEXR_CCU_HCOMM_TASK1_PRELUDE_INSTRUCTION_COUNT = 5U; constexpr uint32_t TILEXR_CCU_HCOMM_TASK1_PRELUDE_RESERVED_XN_COUNT = 1U; +constexpr uint32_t TILEXR_CCU_CHANNEL_XN_STRIDE = 8U; void ResetReport(TileXRCcuResourceAllocatorReport* report) { @@ -212,8 +213,19 @@ int TileXRCcuResourceAllocator::Allocate( } const uint32_t localSqeXnCount = RequiredSqeLoadXnCount(request.sqeArgCount, hcommStyleTask1Prelude); - const uint32_t localXnCount = std::max(localSqeXnCount, request.syncResourceCount); - const uint32_t remoteXnCount = request.syncResourceCount; + const uint32_t localXnCount = std::max( + std::max( + std::max(localSqeXnCount, request.syncResourceCount), + request.bindingsPerSyncResource), + request.minimumLocalXnCount); + const uint32_t remoteXnStride = request.bindingsPerSyncResource > 1U ? + TILEXR_CCU_CHANNEL_XN_STRIDE : 1U; + if (request.syncResourceCount > std::numeric_limits::max() / remoteXnStride) { + return Fail(report, "remote XN resource count overflows"); + } + const uint32_t remoteXnCount = std::max( + request.syncResourceCount * remoteXnStride, + request.minimumRemoteXnCount); const uint32_t localGsaCount = hcommStyleTask1Prelude && spec_.gsaCount != 0 ? 1U : 0U; const uint32_t totalXnCount = localXnCount + remoteXnCount; const uint32_t localWaitCkeCount = request.syncResourceCount; @@ -301,7 +313,8 @@ int TileXRCcuResourceAllocator::Allocate( TileXRCcuSyncResource resource; resource.dieId = spec_.dieId; resource.localXn = static_cast(result.localXn.startId + i); - resource.remoteXn = static_cast(static_cast(result.remoteXn.startId) + i); + resource.remoteXn = static_cast( + static_cast(result.remoteXn.startId) + i * remoteXnStride); resource.notifyCke = static_cast(static_cast(result.remoteNotifyCke.startId) + i); resource.channelId = static_cast(static_cast(result.channels.startId) + i); resource.bindingCount = CheckedU16(request.bindingsPerSyncResource); diff --git a/src/comm/ccu/tilexr_ccu_resource_allocator.h b/src/comm/ccu/tilexr_ccu_resource_allocator.h index 959e712d..67e7e040 100644 --- a/src/comm/ccu/tilexr_ccu_resource_allocator.h +++ b/src/comm/ccu/tilexr_ccu_resource_allocator.h @@ -43,6 +43,8 @@ struct TileXRCcuResourceRequest { uint32_t syncResourceCount = 0; uint32_t syncInstructionCount = 0; uint32_t bindingsPerSyncResource = 1; + uint32_t minimumLocalXnCount = 0; + uint32_t minimumRemoteXnCount = 0; uint32_t sourceCkeCount = 1; TileXRCcuBarrierMode barrierMode = TileXRCcuBarrierMode::SyncXn; }; diff --git a/src/comm/ccu/tilexr_ccu_topology.cpp b/src/comm/ccu/tilexr_ccu_topology.cpp index fc20242a..c03e6ee2 100644 --- a/src/comm/ccu/tilexr_ccu_topology.cpp +++ b/src/comm/ccu/tilexr_ccu_topology.cpp @@ -17,6 +17,9 @@ namespace TileXR { namespace { +constexpr const char* TILEXR_CCU_DIRECT_FORCE_TP_TYPE_ENV = + "TILEXR_CCU_DIRECT_FORCE_TP_TYPE"; + struct RootInfo { std::string topoPath; std::unordered_map deviceToLocalId; @@ -29,6 +32,7 @@ struct TopoEdge { uint32_t localB = 0; std::vector localAPorts; std::vector localBPorts; + bool supportsCtp = false; }; std::string ReadTextFile(const std::string& path) @@ -195,6 +199,8 @@ std::vector ParseTopoInfo(const std::string& path) } edge.localAPorts = JsonStringArrayField(edgeObject, "local_a_ports"); edge.localBPorts = JsonStringArrayField(edgeObject, "local_b_ports"); + const auto protocols = JsonStringArrayField(edgeObject, "protocols"); + edge.supportsCtp = std::find(protocols.begin(), protocols.end(), "UB_CTP") != protocols.end(); if (!edge.localAPorts.empty() && !edge.localBPorts.empty()) { edges.push_back(edge); } @@ -206,24 +212,43 @@ bool ResolveLocalPort( const std::vector& edges, uint32_t localId, uint32_t peerLocalId, - std::string* localPort) + std::string* localPort, + bool* supportsCtp) { - if (localPort == nullptr) { + if (localPort == nullptr || supportsCtp == nullptr) { return false; } for (const auto& edge : edges) { if (edge.localA == localId && edge.localB == peerLocalId) { *localPort = edge.localAPorts.front(); + *supportsCtp = edge.supportsCtp; return true; } if (edge.localB == localId && edge.localA == peerLocalId) { *localPort = edge.localBPorts.front(); + *supportsCtp = edge.supportsCtp; return true; } } return false; } +int ForcedTpType() +{ + const char* value = std::getenv(TILEXR_CCU_DIRECT_FORCE_TP_TYPE_ENV); + if (value == nullptr) { + return -1; + } + const std::string text(value); + if (text == "rtp" || text == "RTP" || text == "0") { + return static_cast(TILEXR_CCU_HCCP_TP_TYPE_RTP); + } + if (text == "ctp" || text == "CTP" || text == "1") { + return static_cast(TILEXR_CCU_HCCP_TP_TYPE_CTP); + } + return -1; +} + } // namespace int TileXRCcuResolvePeerEidRoutes( @@ -262,8 +287,9 @@ int TileXRCcuResolvePeerEidRoutes( for (const uint32_t peerDevicePhyId : peerDevicePhyIds) { const auto peerIdIt = root.deviceToLocalId.find(peerDevicePhyId); std::string localPort; + bool supportsCtp = false; if (peerIdIt == root.deviceToLocalId.end() || - !ResolveLocalPort(edges, localIdIt->second, peerIdIt->second, &localPort)) { + !ResolveLocalPort(edges, localIdIt->second, peerIdIt->second, &localPort, &supportsCtp)) { if (message != nullptr) { *message = "HCCL topology has no device-pair edge"; } @@ -282,6 +308,10 @@ int TileXRCcuResolvePeerEidRoutes( route.peerDevicePhyId = peerDevicePhyId; route.localEid = eidIt->second; route.localPort = localPort; + const int forcedTpType = ForcedTpType(); + route.tpType = forcedTpType >= 0 ? + static_cast(forcedTpType) : + (supportsCtp ? TILEXR_CCU_HCCP_TP_TYPE_CTP : TILEXR_CCU_HCCP_TP_TYPE_RTP); routes->push_back(route); } if (message != nullptr) { diff --git a/src/comm/ccu/tilexr_ccu_topology.h b/src/comm/ccu/tilexr_ccu_topology.h index cbb68d96..de8ac29f 100644 --- a/src/comm/ccu/tilexr_ccu_topology.h +++ b/src/comm/ccu/tilexr_ccu_topology.h @@ -20,6 +20,7 @@ struct TileXRCcuPeerEidRoute { uint32_t peerDevicePhyId = 0; std::array localEid {}; std::string localPort; + uint32_t tpType = TILEXR_CCU_HCCP_TP_TYPE_RTP; }; int TileXRCcuResolvePeerEidRoutes( diff --git a/tests/ccu/ccu_tilexr_direct_smoke_probe.cpp b/tests/ccu/ccu_tilexr_direct_smoke_probe.cpp index 288b04c1..a486fb03 100644 --- a/tests/ccu/ccu_tilexr_direct_smoke_probe.cpp +++ b/tests/ccu/ccu_tilexr_direct_smoke_probe.cpp @@ -606,17 +606,6 @@ uint64_t BuildAllToAllLoopMarker(int rank, int loopIndex) static_cast(loopIndex & 0xffff); } -std::vector BuildAllToAllLoopPattern(int rank, int loopIndex, size_t bytes) -{ - std::vector pattern(bytes); - for (size_t i = 0; i < bytes; ++i) { - pattern[i] = static_cast( - (static_cast(rank + 1) * 17U + - static_cast(loopIndex + 1) * 29U + i * 13U) & 0xffU); - } - return pattern; -} - uint8_t BuildAllToAllMeshByte( uint32_t sourceRank, uint32_t targetRank, @@ -630,7 +619,7 @@ uint8_t BuildAllToAllMeshByte( int InitAllToAllMeshState(int rank, int rankSize, AllToAllState* state) { - if (state == nullptr || rank < 0 || rank >= rankSize || rankSize != 4) { + if (state == nullptr || rankSize < 2 || rankSize > 64 || rank < 0 || rank >= rankSize) { return TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; } state->chunkBytes = AllToAllBytesFromEnv(); @@ -666,7 +655,8 @@ int InitAllToAllMeshState(int rank, int rankSize, AllToAllState* state) int ResetAllToAllMeshStateForLoop(int rank, int loopIndex, AllToAllState* state) { if (state == nullptr || state->source.ptr == nullptr || state->destination.ptr == nullptr || - state->rankSize != 4 || state->bytes != static_cast(state->rankSize) * state->chunkBytes || + state->rankSize < 2 || state->rankSize > 64 || + state->bytes != static_cast(state->rankSize) * state->chunkBytes || loopIndex < 0) { return TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; } @@ -716,7 +706,10 @@ int InitAllToAllState(int rank, int peer, AllToAllState* state) return TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; } state->bytes = AllToAllBytesFromEnv(); - if (state->bytes != 2U * 1024U * 1024U || AllToAllMemSlicePerLoopFromEnv() != 8) { + const bool supportedBytes = state->bytes == 2U * 1024U * 1024U || + state->bytes == 8U * 1024U * 1024U || + state->bytes == 16U * 1024U * 1024U; + if (!supportedBytes || AllToAllMemSlicePerLoopFromEnv() != 8) { state->initRet = TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; return state->initRet; } @@ -755,46 +748,6 @@ int InitAllToAllState(int rank, int peer, AllToAllState* state) return ret; } -int ResetAllToAllStateForLoop(int rank, int peer, int loopIndex, AllToAllState* state) -{ - if (state == nullptr || state->source.ptr == nullptr || state->destination.ptr == nullptr || - state->bytes == 0 || loopIndex < 0) { - return TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; - } - state->expected = BuildAllToAllLoopPattern(peer, loopIndex, state->bytes); - state->observed.assign(state->bytes, 0); - state->readRet = ACL_SUCCESS; - state->mismatchCount = 0; - state->firstMismatchOffset = 0; - state->lastMismatchOffset = 0; - state->firstMismatchObserved = 0; - state->firstMismatchExpected = 0; - state->mismatchedBlockCount = 0; - state->firstMismatchedBlock = 0; - state->lastMismatchedBlock = 0; - state->passed = false; - - const std::vector source = BuildAllToAllLoopPattern(rank, loopIndex, state->bytes); - const std::vector destination( - state->bytes, - static_cast(0xa5U ^ static_cast(loopIndex))); - int ret = aclrtMemcpy( - state->source.ptr, - state->bytes, - source.data(), - source.size(), - ACL_MEMCPY_HOST_TO_DEVICE); - if (ret != ACL_SUCCESS) { - return ret; - } - return aclrtMemcpy( - state->destination.ptr, - state->bytes, - destination.data(), - destination.size(), - ACL_MEMCPY_HOST_TO_DEVICE); -} - int CheckAllToAllState(AllToAllState* state) { if (state == nullptr || state->destination.ptr == nullptr || state->bytes == 0) { @@ -1468,7 +1421,9 @@ void PrintCcuResourceState( uint8_t dieId, const TileXRDirectCcuPrepareOptions& options, const char* label, - uint32_t resourceCount = 3U) + uint32_t resourceCount = 3U, + uint32_t extraCkeStartId = 0U, + uint32_t extraCkeCount = 0U) { if (context == nullptr || label == nullptr || resourceCount == 0) { return; @@ -1486,6 +1441,7 @@ void PrintCcuResourceState( std::vector remoteXn(resourceCount, 0); std::vector localWaitCke(resourceCount, 0); std::vector remoteNotifyCke(resourceCount, 0); + std::vector extraCke(extraCkeCount, 0); const uint32_t localXnStartId = options.xnStartId; const uint32_t remoteXnStartId = options.remoteXnStartId; const uint32_t localWaitCkeStartId = options.localWaitCkeStartId; @@ -1498,6 +1454,8 @@ void PrintCcuResourceState( dieId, localWaitCkeStartId, localWaitCke.data(), resourceCount, &report); const int remoteCkeRet = adapter.ReadCkeRange( dieId, remoteNotifyCkeStartId, remoteNotifyCke.data(), resourceCount, &report); + const int extraCkeRet = extraCkeCount == 0U ? TileXR::TILEXR_SUCCESS : + adapter.ReadCkeRange(dieId, extraCkeStartId, extraCke.data(), extraCkeCount, &report); const auto values = [](const std::vector& data) { std::ostringstream out; @@ -1524,6 +1482,9 @@ void PrintCcuResourceState( << " remoteNotifyCkeStartId=" << remoteNotifyCkeStartId << " remoteCkeRet=" << remoteCkeRet << " remoteCke=" << values(remoteNotifyCke) + << " extraCkeStartId=" << extraCkeStartId + << " extraCkeRet=" << extraCkeRet + << " extraCke=" << values(extraCke) << std::endl; } @@ -2091,9 +2052,9 @@ int RunAllToAllMeshLongMissionSmokeForRank( if (context == nullptr) { return TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; } - if (rankSize != 4) { + if (rankSize < 2 || rankSize > 64) { std::cout << "tilexr_ccu_alltoall skipped rankSize=" << rankSize - << " reason=\"direct CCU alltoall mesh requires four ranks\"" << std::endl; + << " reason=\"direct CCU alltoall mesh requires 2..64 ranks\"" << std::endl; return 0; } @@ -2104,10 +2065,17 @@ int RunAllToAllMeshLongMissionSmokeForRank( alltoall.initRet = TileXR::TILEXR_ERROR_PARA_CHECK_FAIL; } TileXRDirectCcuPrepareOptions options = MakePrepareOptions(rank, rankSize, device); - options.syncResourceCount = 3U; + options.syncResourceCount = static_cast(rankSize - 1); options.sqeArgCount = TILEXR_DIRECT_CCU_SQE_ARGS_LEN; if (std::getenv("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT") == nullptr) { - options.syncInstructionCount = 131U; + const uint32_t peerCount = static_cast(rankSize - 1); + const uint32_t completionCkeCount = (peerCount + 15U) / 16U; + const uint64_t blockCount = alltoall.chunkBytes / TileXR::TILEXR_CCU_ALLTOALL_BLOCK_BYTES; + const uint64_t copyPerBlock = peerCount * 6ULL + 9ULL + completionCkeCount; + options.syncInstructionCount = static_cast( + 3ULL + peerCount * 3ULL + + blockCount * copyPerBlock + + peerCount * 2ULL + 1ULL); } if (options.gsaStartId == 0) { options.gsaStartId = 1; @@ -2118,7 +2086,7 @@ int RunAllToAllMeshLongMissionSmokeForRank( << " chunkBytes=" << alltoall.chunkBytes << " bytes=" << alltoall.bytes << " loopCount=" << loopCount - << " resourceCount=3" + << " resourceCount=" << (rankSize - 1) << " mesh=1" << " longMission=1" << std::endl; @@ -2146,7 +2114,8 @@ int RunAllToAllMeshLongMissionSmokeForRank( if (prepareRet != TileXR::TILEXR_SUCCESS) { finalRet = 6; } else if (attempt.submitTasks.size() != 1U || attempt.submitTasks.front().argSize != - TILEXR_DIRECT_CCU_SQE_ARGS_LEN || attempt.plan.syncResources.size() != 3U) { + TILEXR_DIRECT_CCU_SQE_ARGS_LEN || + attempt.plan.syncResources.size() != static_cast(rankSize - 1)) { std::cerr << "tilexr_ccu_alltoall invalidMeshPreparedTask" << " rank=" << rank << " taskCount=" << attempt.submitTasks.size() @@ -2216,7 +2185,7 @@ int RunAllToAllMeshLongMissionSmokeForRank( << " rank=" << rank << " loopIndex=" << loopIndex << " ret=" << finalRet - << " resourceCount=3" + << " resourceCount=" << (rankSize - 1) << " selfCopyCompletionCke=" << attempt.plan.syncResources[0].localWaitCke << std::endl; PrintMissionContext(context, attempt.submitTasks.front(), "tilexr_ccu_alltoall"); @@ -2225,7 +2194,9 @@ int RunAllToAllMeshLongMissionSmokeForRank( attempt.submitTasks.front().dieId, options, "tilexr_ccu_alltoall", - 3U); + static_cast(rankSize - 1), + attempt.allocation.sourceCke.startId, + attempt.allocation.sourceCke.num); break; } std::cout << "tilexr_ccu_alltoall stableResources=1" @@ -2274,8 +2245,9 @@ int RunAllToAllLongMissionSmokeForRank(DirectCcuSmokeContext* context, int rank, TileXRDirectCcuPrepareOptions options = MakePrepareOptions(rank, rankSize, device); options.syncResourceCount = 3; options.sqeArgCount = TILEXR_DIRECT_CCU_SQE_ARGS_LEN; + const size_t blockCount = alltoall.bytes / TileXR::TILEXR_CCU_ALLTOALL_BLOCK_BYTES; if (std::getenv("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT") == nullptr) { - options.syncInstructionCount = 5 + 64 * 7; + options.syncInstructionCount = static_cast(5U + blockCount * 7U); } if (options.gsaStartId == 0) { options.gsaStartId = 1; @@ -2287,7 +2259,7 @@ int RunAllToAllLongMissionSmokeForRank(DirectCcuSmokeContext* context, int rank, << " bytes=" << alltoall.bytes << " loopCount=" << loopCount << " memSlicePerLoop=" << AllToAllMemSlicePerLoopFromEnv() - << " blockCount=64" + << " blockCount=" << blockCount << " longMission=1" << " preSync=1" << " postSync=0" @@ -2335,19 +2307,18 @@ int RunAllToAllLongMissionSmokeForRank(DirectCcuSmokeContext* context, int rank, finalRet = 7; } else { bool skipStreamDestroy = false; + int lastLoopIndex = -1; const int syncTimeoutMs = std::max(1, EnvInt("TILEXR_CCU_DIRECT_SUBMIT_TIMEOUT", 6000)); for (int loopIndex = 0; loopIndex < loopCount; ++loopIndex) { - const int resetRet = ResetAllToAllStateForLoop(rank, peer, loopIndex, &alltoall); + lastLoopIndex = loopIndex; const uint64_t localLoopMarker = BuildAllToAllLoopMarker(rank, loopIndex); attempt.submitTasks.front().args[0] = localLoopMarker; const bool collectiveSubmitReady = WaitForCollectiveSubmitReadiness( rank, rankSize, - resetRet == ACL_SUCCESS && installReport.submitReady, + installReport.submitReady, loopIndex); - if (resetRet != ACL_SUCCESS) { - finalRet = 14; - } else if (!collectiveSubmitReady) { + if (!collectiveSubmitReady) { finalRet = 13; } @@ -2403,13 +2374,16 @@ int RunAllToAllLongMissionSmokeForRank(DirectCcuSmokeContext* context, int rank, finalRet = 15; } } - if (finalRet == 0 && CheckAllToAllState(&alltoall) != ACL_SUCCESS) { - finalRet = 14; - } if (!WaitForCollectiveSubmitDone(rank, rankSize, finalRet, loopIndex) && finalRet == 0) { finalRet = 13; } - PrintAllToAllResult(rank, loopIndex, finalRet, alltoall); + std::cout << "tilexr_ccu_alltoall loopResult" + << " passed=" << (finalRet == 0 ? 1 : 0) + << " rank=" << rank + << " loopIndex=" << loopIndex + << " ret=" << finalRet + << " dataCheckDeferred=1" + << std::endl; if (finalRet != 0) { std::cerr << "tilexr_ccu_alltoall loopFailure" << " rank=" << rank @@ -2424,6 +2398,10 @@ int RunAllToAllLongMissionSmokeForRank(DirectCcuSmokeContext* context, int rank, break; } } + if (finalRet == 0 && CheckAllToAllState(&alltoall) != ACL_SUCCESS) { + finalRet = 14; + } + PrintAllToAllResult(rank, lastLoopIndex, finalRet, alltoall); if (skipStreamDestroy) { std::cout << "tilexr_ccu_alltoall skipDestroyStream=1" << " rank=" << rank diff --git a/tests/ccu/run_tilexr_ccu_direct_smoke.sh b/tests/ccu/run_tilexr_ccu_direct_smoke.sh index 806594ec..1ba2d8ab 100644 --- a/tests/ccu/run_tilexr_ccu_direct_smoke.sh +++ b/tests/ccu/run_tilexr_ccu_direct_smoke.sh @@ -234,23 +234,40 @@ apply_alltoall_defaults() export TILEXR_CCU_PROBE_INSTRUCTION_START="${TILEXR_CCU_PROBE_INSTRUCTION_START:-475}" export TILEXR_CCU_PROBE_MISSION_INSTRUCTION_START="${TILEXR_CCU_PROBE_MISSION_INSTRUCTION_START:-489}" if alltoall_mesh_mode_enabled; then + mesh_rank_size="$(parse_int "${TILEXR_CCU_RANK_SIZE:-${TILEXR_CCU_PROBE_RANK_SIZE:-2}}" 2)" + mesh_peer_count=$((mesh_rank_size - 1)) + mesh_completion_cke_count=$(((mesh_peer_count + 15) / 16)) + mesh_chunk_bytes="$(parse_int "${TILEXR_CCU_ALLTOALL_BYTES:-131072}" 131072)" + mesh_block_count=$((mesh_chunk_bytes / 32768)) + mesh_pre_sync_count=$((3 + mesh_peer_count * 3)) + mesh_copy_per_block=$((mesh_peer_count * 6 + 9 + mesh_completion_cke_count)) + if [ "${mesh_peer_count}" -gt 16 ]; then + mesh_remote_xn_count=${mesh_peer_count} + else + mesh_remote_xn_count=16 + fi + mesh_instruction_count=$((mesh_pre_sync_count + mesh_block_count * mesh_copy_per_block + mesh_peer_count * 2 + 1)) + mesh_local_cke_count=$((mesh_peer_count + 1 + mesh_completion_cke_count)) export TILEXR_CCU_DIRECT_LOWER_LAYER_PFE_OFFSET_SOURCE="${TILEXR_CCU_DIRECT_LOWER_LAYER_PFE_OFFSET_SOURCE:-hcomm_die}" export TILEXR_CCU_DIRECT_LOWER_LAYER_PFE_PARTITION="${TILEXR_CCU_DIRECT_LOWER_LAYER_PFE_PARTITION:-hcomm}" export TILEXR_CCU_PROBE_SQE_ARG_COUNT="${TILEXR_CCU_PROBE_SQE_ARG_COUNT:-0}" - export TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT="${TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT:-3}" - export TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT="${TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-131}" + export TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT="${TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT:-${mesh_peer_count}}" + export TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT="${TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-${mesh_instruction_count}}" export TILEXR_CCU_PROBE_XN_START="${TILEXR_CCU_PROBE_XN_START:-1961}" export TILEXR_CCU_PROBE_REMOTE_XN_START="${TILEXR_CCU_PROBE_REMOTE_XN_START:-2361}" - export TILEXR_CCU_PROBE_REMOTE_XN_COUNT="${TILEXR_CCU_PROBE_REMOTE_XN_COUNT:-16}" + export TILEXR_CCU_PROBE_REMOTE_XN_COUNT="${TILEXR_CCU_PROBE_REMOTE_XN_COUNT:-${mesh_remote_xn_count}}" export TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_START="${TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_START:-332}" - export TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_COUNT="${TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_COUNT:-16}" + export TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_COUNT="${TILEXR_CCU_PROBE_LOCAL_WAIT_CKE_COUNT:-${mesh_local_cke_count}}" export TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_START="${TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_START:-364}" - export TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_COUNT="${TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_COUNT:-16}" + export TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_COUNT="${TILEXR_CCU_PROBE_REMOTE_NOTIFY_CKE_COUNT:-${mesh_peer_count}}" export TILEXR_CCU_DIRECT_RESOURCE_WINDOW_EID_INDEX="${TILEXR_CCU_DIRECT_RESOURCE_WINDOW_EID_INDEX:-3}" elif [ "${TILEXR_CCU_DIRECT_SMOKE_ALLTOALL_LONG_MISSION:-0}" = "1" ]; then + long_mission_bytes="$(parse_int "${TILEXR_CCU_ALLTOALL_BYTES:-2097152}" 2097152)" + long_mission_block_count=$((long_mission_bytes / 32768)) + long_mission_instruction_count=$((7 + long_mission_block_count * 7)) export TILEXR_CCU_PROBE_SQE_ARG_COUNT="${TILEXR_CCU_PROBE_SQE_ARG_COUNT:-13}" export TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT="${TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT:-3}" - export TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT="${TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-453}" + export TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT="${TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-${long_mission_instruction_count}}" else export TILEXR_CCU_PROBE_SQE_ARG_COUNT="${TILEXR_CCU_PROBE_SQE_ARG_COUNT:-0}" export TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT="${TILEXR_CCU_PROBE_SYNC_RESOURCE_COUNT:-1}" @@ -288,6 +305,10 @@ if [ "${rank_size}" -lt 1 ]; then echo "ERROR: rank size must be positive: ${rank_size}" >&2 exit 2 fi +if alltoall_mesh_mode_enabled && { [ "${rank_size}" -lt 2 ] || [ "${rank_size}" -gt 64 ]; }; then + echo "ERROR: direct CCU alltoall mesh rank size must be in [2,64]: ${rank_size}" >&2 + exit 2 +fi devices="${TILEXR_CCU_SMOKE_DEVICES:-${TILEXR_TEST_DEVICES:-0,1}}" IFS=',' read -r -a device_list <<< "${devices}" if [ "${#device_list[@]}" -ne "${rank_size}" ]; then @@ -949,9 +970,20 @@ if [ "${TILEXR_CCU_DIRECT_SMOKE_SUBMIT:-0}" = "1" ]; then done fi -if alltoall_mode_enabled; then +if alltoall_mode_enabled && [ "${TILEXR_CCU_DIRECT_SMOKE_SUBMIT:-0}" = "1" ]; then loop_count="$(parse_int "${TILEXR_CCU_ALLTOALL_LOOP_COUNT:-1}" 1)" expected_results=$((rank_size * loop_count)) + if [ "${TILEXR_CCU_DIRECT_SMOKE_ALLTOALL_LONG_MISSION:-0}" = "1" ] && + [ "${TILEXR_CCU_DIRECT_SMOKE_ALLTOALL_MESH:-0}" != "1" ]; then + expected_loop_results=$((rank_size * loop_count)) + actual_loop_results="$(grep -h -c "tilexr_ccu_alltoall loopResult passed=1" "${rank_logs[@]}" | awk '{ total += $1 } END { print total + 0 }')" + echo "tilexr_ccu_direct_smoke_runner alltoallLoopCounts expectedResults=${expected_loop_results} actualResults=${actual_loop_results}" + if [ "${actual_loop_results}" -ne "${expected_loop_results}" ]; then + echo "ERROR: direct CCU alltoall loop result count mismatch expected=${expected_loop_results} actual=${actual_loop_results}" >&2 + exit 9 + fi + expected_results="${rank_size}" + fi actual_results="$(grep -h -c "tilexr_ccu_alltoall result passed=1" "${rank_logs[@]}" | awk '{ total += $1 } END { print total + 0 }')" echo "tilexr_ccu_direct_smoke_runner alltoallCounts expectedResults=${expected_results} actualResults=${actual_results}" if [ "${actual_results}" -ne "${expected_results}" ]; then diff --git a/tests/ccu/test_tilexr_ccu_alltoall_program.py b/tests/ccu/test_tilexr_ccu_alltoall_program.py index 7aeac4fe..1027808b 100644 --- a/tests/ccu/test_tilexr_ccu_alltoall_program.py +++ b/tests/ccu/test_tilexr_ccu_alltoall_program.py @@ -600,12 +600,12 @@ def test_four_rank_mesh_posts_all_peers_then_copies_remote_and_self_chunks(self) spec.chunkBytes = 2ULL * 1024ULL * 1024ULL; spec.selfSourceGsa = 0x180; spec.selfDestinationGsa = 0x181; - spec.selfSourceXn = 0x280; - spec.selfDestinationXn = 0x281; - spec.selfLengthXn = 0x282; + spec.selfSourceXn = 0x200; + spec.selfDestinationXn = 0x201; + spec.selfLengthXn = 0x202; spec.selfChannelId = 0; spec.selfCompletionCke = 0x480; - spec.remoteCompletionCke = 0x491; + spec.remoteCompletionCkes = {0x491}; spec.peers = {Peer(2, 3, 2), Peer(2, 0, 0), Peer(2, 1, 1)}; std::vector program; @@ -693,9 +693,9 @@ def test_four_rank_mesh_posts_all_peers_then_copies_remote_and_self_chunks(self) return 7; } auto overlappingCke = spec; - overlappingCke.remoteCompletionCke = overlappingCke.peers[0].route.sourceCke; + overlappingCke.remoteCompletionCkes[0] = overlappingCke.peers[0].route.sourceCke; for (auto& peer : overlappingCke.peers) { - peer.route.copyCompletionCke = overlappingCke.remoteCompletionCke; + peer.route.copyCompletionCke = overlappingCke.remoteCompletionCkes[0]; } if (TileXRCcuBuildAllToAllMeshProgram(overlappingCke, &program, &report) != TILEXR_ERROR_PARA_CHECK_FAIL || @@ -733,6 +733,62 @@ def test_four_rank_mesh_posts_all_peers_then_copies_remote_and_self_chunks(self) return 9; } } + auto spec8 = spec; + spec8.rankSize = 8; + spec8.localRank = 3; + spec8.peers.clear(); + spec8.remoteCompletionCkes = {0x491}; + uint16_t ordinal8 = 0; + for (uint32_t peerRank = 0; peerRank < spec8.rankSize; ++peerRank) { + if (peerRank != spec8.localRank) { + spec8.peers.push_back(Peer(spec8.localRank, peerRank, ordinal8++)); + } + } + if (TileXRCcuBuildAllToAllMeshProgram(spec8, &program, &report) != TILEXR_SUCCESS || + program.size() != 3367 || report.peerCount != 7 || report.syncResourceCount != 7 || + report.remoteBlockCount != 448 || report.selfBlockCount != 64) { + std::cerr << "unexpected 8-rank 2MB mesh: " << report.message + << " instructions=" << program.size() << "\n"; + return 10; + } + + auto spec2 = spec; + spec2.rankSize = 2; + spec2.localRank = 0; + spec2.peers = {Peer(0, 1, 0)}; + spec2.remoteCompletionCkes = {0x491}; + if (TileXRCcuBuildAllToAllMeshProgram(spec2, &program, &report) != TILEXR_SUCCESS || + program.size() != 1033 || report.peerCount != 1 || report.syncResourceCount != 1 || + report.remoteBlockCount != 64 || report.selfBlockCount != 64) { + std::cerr << "unexpected 2-rank 2MB full mesh: " << report.message + << " instructions=" << program.size() << "\n"; + return 11; + } + + auto spec64 = spec; + spec64.rankSize = 64; + spec64.localRank = 17; + spec64.chunkBytes = 128ULL * 1024ULL; + spec64.peers.clear(); + spec64.remoteCompletionCkes = {0x491, 0x492, 0x493, 0x494}; + uint16_t ordinal64 = 0; + for (uint32_t peerRank = 0; peerRank < spec64.rankSize; ++peerRank) { + if (peerRank == spec64.localRank) { + continue; + } + auto peer = Peer(spec64.localRank, peerRank, ordinal64); + peer.route.bytes = spec64.chunkBytes; + peer.route.copyCompletionCke = spec64.remoteCompletionCkes[ordinal64 / 16U]; + spec64.peers.push_back(peer); + ++ordinal64; + } + if (TileXRCcuBuildAllToAllMeshProgram(spec64, &program, &report) != TILEXR_SUCCESS || + program.size() != 1883 || report.peerCount != 63 || report.syncResourceCount != 63 || + report.remoteBlockCount != 252 || report.selfBlockCount != 4) { + std::cerr << "unexpected 64-rank 128KB mesh: " << report.message + << " instructions=" << program.size() << "\n"; + return 12; + } return 0; } ''' diff --git a/tests/ccu/test_tilexr_ccu_backend_boundary.py b/tests/ccu/test_tilexr_ccu_backend_boundary.py index 6622c390..689af500 100644 --- a/tests/ccu/test_tilexr_ccu_backend_boundary.py +++ b/tests/ccu/test_tilexr_ccu_backend_boundary.py @@ -235,6 +235,7 @@ def test_alltoall_overrides_only_copy_route_memory_not_sync_routes(self): ] self.assertIn("SetDirectCcuRemoteRouteMemoryOverrideForSyncRoute(", prepare_alltoall) + self.assertIn("peerRanks.size() < routedPeerCount", planner) self.assertIn("0U", prepare_alltoall) self.assertIn("uint32_t routeIndex = 0", override_apply) self.assertIn("override.syncRouteIndex != routeIndex", override_apply) @@ -251,7 +252,9 @@ def test_four_rank_mesh_gathers_imports_and_maps_three_routes_per_peer(self): planner.index("int TileXRCcuCollectivePlanner::PrepareDirectCcuAllToAllMeshInstallAttempt"): planner.index("int TileXRCcuCollectivePlanner::PrepareDirectCcuSyncXnPingInstallAttempt") ] - self.assertIn("rankSize != 4", mesh_body) + self.assertIn("rankSize < 2", mesh_body) + self.assertIn("rankSize > 64", mesh_body) + self.assertIn("rankSize - 1", mesh_body) self.assertEqual(1, mesh_body.count("session.AllGather(")) self.assertIn("endpoint.rank != peerRank", mesh_body) self.assertIn("session.ImportRemoteMemoryBuffer", mesh_body) diff --git a/tests/ccu/test_tilexr_ccu_direct_orchestrator.py b/tests/ccu/test_tilexr_ccu_direct_orchestrator.py index f30e01bf..62359813 100644 --- a/tests/ccu/test_tilexr_ccu_direct_orchestrator.py +++ b/tests/ccu/test_tilexr_ccu_direct_orchestrator.py @@ -506,8 +506,8 @@ def test_direct_install_attempt_becomes_submit_ready_with_remote_xn_peer_exchang lowerLayer.xnClears.push_back({1, 1961, 14}); lowerLayer.ckeClears.push_back({1, 332, 3}); lowerLayer.remoteXnBindings.push_back({1, 2, 1961, 1975, 332, 0, true, 0, true, true, true}); - lowerLayer.remoteXnBindings.push_back({1, 3, 1962, 1976, 333, 0, true, 0, true, true, true}); - lowerLayer.remoteXnBindings.push_back({1, 4, 1963, 1977, 334, 0, true, 0, true, true, true}); + lowerLayer.remoteXnBindings.push_back({1, 3, 1962, 1983, 333, 0, true, 0, true, true, true}); + lowerLayer.remoteXnBindings.push_back({1, 4, 1963, 1991, 334, 0, true, 0, true, true, true}); return lowerLayer; } @@ -872,7 +872,7 @@ def test_direct_install_attempt_can_prepare_lower_layer_plan_after_allocation(se { auto* state = static_cast(userData); ++state->callCount; - state->syncResourceCount = allocation.remoteXn.num; + state->syncResourceCount = allocation.channels.num; if (plan == nullptr || report == nullptr) { return TILEXR_ERROR_PARA_CHECK_FAIL; } @@ -912,12 +912,12 @@ def test_direct_install_attempt_can_prepare_lower_layer_plan_after_allocation(se } plan->xnClears.push_back({1, allocation.localXn.startId, allocation.localXn.num}); plan->ckeClears.push_back({1, allocation.notifyCke.startId, allocation.notifyCke.num}); - for (uint32_t i = 0; i < allocation.remoteXn.num; ++i) { + for (uint32_t i = 0; i < allocation.channels.num; ++i) { plan->remoteXnBindings.push_back({ 1, static_cast(allocation.channels.startId + i), static_cast(allocation.localXn.startId + i), - static_cast(allocation.remoteXn.startId + i), + static_cast(allocation.remoteXn.startId + i * 8U), static_cast(allocation.notifyCke.startId + i), i, true, @@ -973,7 +973,7 @@ def test_direct_install_attempt_can_prepare_lower_layer_plan_after_allocation(se options.xnStartId = 1961; options.gsaStartId = 510; options.remoteXnStartId = 2361; - options.remoteXnCount = 8; + options.remoteXnCount = 24; options.ckeStartId = 332; options.remoteNotifyCkeStartId = 364; options.remoteNotifyCkeCount = 8; @@ -1133,12 +1133,12 @@ def test_direct_install_attempt_passes_split_cke_ranges_to_lower_layer_callback( } plan->xnClears.push_back({1, allocation.localXn.startId, allocation.localXn.num}); plan->ckeClears.push_back({1, allocation.localWaitCke.startId, allocation.localWaitCke.num}); - for (uint32_t i = 0; i < allocation.remoteXn.num; ++i) { + for (uint32_t i = 0; i < allocation.channels.num; ++i) { plan->remoteXnBindings.push_back({ 1, allocation.channels.startId + i, static_cast(allocation.localXn.startId + i), - static_cast(allocation.remoteXn.startId + i), + static_cast(allocation.remoteXn.startId + i * 8U), static_cast(allocation.remoteNotifyCke.startId + i), i, true, @@ -1756,8 +1756,8 @@ def test_direct_alltoall_uses_three_sync_resources_and_distinct_phases(self): ] self.assertIn("TILEXR_CCU_DIRECT_ALLTOALL_SYNC_RESOURCE_COUNT = 3U", source) - self.assertIn("TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT =\n 7U + 64U * 7U", source) - self.assertIn("TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT =\n 7U + 64U * 7U", planner) + self.assertIn("DirectAllToAll2RankInstructionCapacity", source) + self.assertIn("DirectAllToAll2RankInstructionCapacity", planner) self.assertIn("alltoall != nullptr ? TILEXR_CCU_DIRECT_ALLTOALL_SYNC_RESOURCE_COUNT", source) self.assertIn("alltoall != nullptr ? TileXRCcuBarrierMode::SyncXn", source) self.assertIn("const TileXRCcuSyncResource& copyResource = attempt->plan.syncResources[0]", source) @@ -1896,12 +1896,83 @@ def test_direct_four_rank_mesh_builds_one_three_channel_launch_package(self): attempt.package.program.sync.size() != 1811 || attempt.plan.taskWindows[0].instCnt != 1811 || attempt.plan.kernelLocalGsa.num != 2 || attempt.allocation.sourceCke.num != 2 || + attempt.allocation.remoteXn.num != 3 || + attempt.plan.syncResources[1].remoteXn - + attempt.plan.syncResources[0].remoteXn != 1 || + attempt.plan.syncResources[2].remoteXn - + attempt.plan.syncResources[1].remoteXn != 1 || attempt.plan.barrierMode != TileXRCcuBarrierMode::SyncCke) { std::cerr << "unexpected mesh package: " << report.message << " resources=" << attempt.plan.syncResources.size() << " instructions=" << attempt.package.program.sync.size() << "\n"; return 1; } + auto mesh8 = mesh; + mesh8.rankSize = 8; + mesh8.peers.clear(); + for (uint32_t peerRank = 0; peerRank < mesh8.rankSize; ++peerRank) { + if (peerRank == mesh8.localRank) { + continue; + } + TileXRCcuDirectAllToAllMeshPeerSpec peer; + peer.peerRank = peerRank; + peer.remoteRecvAddr = 0x40000000ULL + peerRank * 0x1000000ULL; + peer.remoteRecvToken = TileXRCcuPackMemoryToken(30 + peerRank, 40 + peerRank, true); + mesh8.peers.push_back(peer); + } + TileXRCcuDirectInstallAttempt attempt8; + TileXRCcuDirectInstallReport report8; + const int ret8 = TileXRCcuRunDirectAllToAllMeshInstallAttempt( + options, mesh8, &attempt8, &report8); + (void)ret8; + if (!report8.pipelineBuilt || attempt8.plan.syncResources.size() != 7 || + attempt8.package.program.sync.size() != 3367 || + attempt8.plan.taskWindows[0].instCnt != 3367 || + attempt8.allocation.remoteXn.num != 7 || + attempt8.plan.syncResources[6].remoteXn - + attempt8.plan.syncResources[5].remoteXn != 1 || + attempt8.allocation.sourceCke.num != 2) { + std::cerr << "unexpected 8-rank mesh package: " << report8.message + << " resources=" << attempt8.plan.syncResources.size() + << " instructions=" << attempt8.package.program.sync.size() << "\n"; + return 3; + } + auto mesh2 = mesh; + mesh2.rankSize = 2; + mesh2.localRank = 0; + mesh2.peers.clear(); + TileXRCcuDirectAllToAllMeshPeerSpec peer2; + peer2.peerRank = 1; + peer2.remoteRecvAddr = 0x50000000ULL; + peer2.remoteRecvToken = TileXRCcuPackMemoryToken(50, 60, true); + mesh2.peers.push_back(peer2); + TileXRCcuDirectInstallAttempt attempt2; + TileXRCcuDirectInstallReport report2; + const int ret2 = TileXRCcuRunDirectAllToAllMeshInstallAttempt( + options, mesh2, &attempt2, &report2); + (void)ret2; + if (!report2.pipelineBuilt || attempt2.plan.syncResources.size() != 1 || + attempt2.package.program.sync.size() != 1033 || + attempt2.plan.taskWindows[0].instCnt != 1033 || + attempt2.allocation.localXn.num != 3 || + attempt2.allocation.remoteXn.num != 3 || + attempt2.allocation.sourceCke.num != 2) { + std::cerr << "unexpected 2-rank full mesh package: " << report2.message + << " resources=" << attempt2.plan.syncResources.size() + << " instructions=" << attempt2.package.program.sync.size() << "\n"; + return 4; + } + auto invalidMesh = mesh; + invalidMesh.rankSize = 0xffffffffU; + invalidMesh.peers.clear(); + TileXRCcuDirectInstallAttempt invalidAttempt; + TileXRCcuDirectInstallReport invalidReport; + if (TileXRCcuRunDirectAllToAllMeshInstallAttempt( + options, invalidMesh, &invalidAttempt, &invalidReport) != + TILEXR_ERROR_PARA_CHECK_FAIL || invalidReport.pipelineBuilt) { + std::cerr << "unexpected invalid rank-size result: " << invalidReport.message << "\n"; + return 5; + } basic.caps.cap0 = (7U << 24) | (11U << 16) | 1599U; TileXRCcuDirectInstallAttempt smallAttempt; TileXRCcuDirectInstallReport smallReport; @@ -1959,7 +2030,7 @@ def test_collective_planner_has_private_alltoall_prepare_path(self): self.assertIn("alltoall.remoteRecvAddr = remoteImportRequest.addr", source) self.assertIn("alltoall.remoteRecvToken", source) self.assertNotIn("alltoall.remoteRecvAddr = peerEndpoint.destinationAddr", source) - self.assertIn("TILEXR_CCU_DIRECT_ALLTOALL_INSTRUCTION_COUNT", source) + self.assertIn("DirectAllToAll2RankInstructionCapacity(bytes)", source) self.assertIn("tilexr-comm-direct-ccu-alltoall", source) self.assertIn("TileXRCcuRunDirectAllToAll2RankInstallAttempt", source) @@ -1990,7 +2061,7 @@ def test_direct_sync_xn_ping_uses_one_mission_route_and_full_4p_transport_resour self.assertIn("TILEXR_CCU_DIRECT_SMOKE_SYNC_XN_PING_NOTIFY_MASK", planner) self.assertIn("TILEXR_CCU_DIRECT_SMOKE_SYNC_XN_PING_WAIT_MASK", planner) self.assertIn("RegisterCcuResourceRmaBuffer", planner) - self.assertIn("next.syncResourceCount = rankSize == 4 ? 3U : 1U", planner) + self.assertIn("next.syncResourceCount = static_cast(rankSize - 1)", planner) self.assertIn("attempt->plan.syncResources.empty()", source) self.assertIn("syncXnPing != nullptr ? options.syncResourceCount", source) diff --git a/tests/ccu/test_tilexr_ccu_direct_smoke_probe.py b/tests/ccu/test_tilexr_ccu_direct_smoke_probe.py index ae8ad68f..0e5a8b06 100644 --- a/tests/ccu/test_tilexr_ccu_direct_smoke_probe.py +++ b/tests/ccu/test_tilexr_ccu_direct_smoke_probe.py @@ -660,6 +660,7 @@ def test_alltoall_smoke_mode_is_opt_in_and_validates_peer_pattern(self): self.assertIn("TILEXR_CCU_DIRECT_SMOKE_ALLTOALL_SINGLE_ROUTE_BIDIRECTIONAL", source) self.assertIn('kAllToAllBytesEnv = "TILEXR_CCU_ALLTOALL_BYTES"', source) self.assertIn('kAllToAllMemSlicePerLoopEnv = "TILEXR_CCU_ALLTOALL_MEM_SLICE_PER_LOOP"', source) + self.assertIn("state->bytes == 16U * 1024U * 1024U", source) self.assertIn("struct AllToAllState", source) self.assertIn("AllToAllSmokeEnabled", source) self.assertIn("AllToAllLongMissionEnabled", source) @@ -693,8 +694,8 @@ def test_alltoall_long_mission_reuses_prepare_with_loop_specific_state(self): self.assertIn("std::strtol", loop_count_body) self.assertIn("parsed < 1 || parsed > 1024", loop_count_body) self.assertIn("BuildAllToAllLoopMarker", source) - self.assertIn("BuildAllToAllLoopPattern", source) - self.assertIn("ResetAllToAllStateForLoop", source) + self.assertNotIn("BuildAllToAllLoopPattern", source) + self.assertNotIn("ResetAllToAllStateForLoop", source) self.assertIn("ReadAndValidatePeerLoopMarker", source) self.assertIn("for (int loopIndex = 0; loopIndex < loopCount; ++loopIndex)", body) self.assertIn("attempt.submitTasks.front().args[0] =", body) @@ -703,6 +704,13 @@ def test_alltoall_long_mission_reuses_prepare_with_loop_specific_state(self): self.assertIn("WaitForCollectiveSubmitDone(rank, rankSize, finalRet, loopIndex)", body) self.assertIn("adapter.ReadXnRange", source) self.assertIn("peerLoopMarker", source) + loop_body = body[body.index("for (int loopIndex = 0; loopIndex < loopCount; ++loopIndex)"):] + self.assertNotIn( + "CheckAllToAllState(&alltoall)", + loop_body[:loop_body.index("lastLoopIndex = loopIndex")], + ) + self.assertIn("CheckAllToAllState(&alltoall)", body) + self.assertIn("dataCheckDeferred=1", body) self.assertIn("attempt.plan.syncResources[0].remoteXn", body) self.assertNotIn("attempt.plan.syncResources[0].localXn,", body) self.assertIn("loopIndex=", body) @@ -738,7 +746,9 @@ def test_four_rank_mesh_reuses_one_prepare_and_validates_full_matrix_each_loop(s source.index("int RunAllToAllLongMissionSmokeForRank") ] loop = "for (int loopIndex = 0; loopIndex < loopCount; ++loopIndex)" - self.assertIn("rankSize != 4", body) + self.assertIn("rankSize < 2", body) + self.assertIn("rankSize > 64", body) + self.assertIn("rankSize - 1", body) self.assertIn("PrepareDirectCcuAllToAllMeshInstallAttempt", body) self.assertIn("aclrtCreateStream", body) self.assertIn(loop, body) @@ -751,7 +761,7 @@ def test_four_rank_mesh_reuses_one_prepare_and_validates_full_matrix_each_loop(s self.assertNotIn("ReadAndValidatePeerLoopMarker", body) self.assertIn("CheckAllToAllState(&alltoall)", body) self.assertIn("PrintCcuResourceState", body) - self.assertIn("resourceCount=3", body) + self.assertIn('" resourceCount=" << (rankSize - 1)', body) pattern = source[ source.index("uint8_t BuildAllToAllMeshByte"): @@ -836,8 +846,9 @@ def test_smoke_runner_forwards_alltoall_env(self): self.assertIn("TILEXR_CCU_ALLTOALL_BYTES", runner) self.assertIn("TILEXR_CCU_ALLTOALL_MEM_SLICE_PER_LOOP", runner) self.assertIn("TILEXR_CCU_ALLTOALL_LOOP_COUNT", runner) + self.assertIn("tilexr_ccu_direct_smoke_runner alltoallLoopCounts", runner) self.assertIn('if [ "${TILEXR_CCU_DIRECT_SMOKE_ALLTOALL_LONG_MISSION:-0}" = "1" ]; then', runner) - self.assertIn("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-453", runner) + self.assertIn("long_mission_instruction_count=$((7 + long_mission_block_count * 7))", runner) self.assertNotIn("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-451", runner) self.assertNotIn("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-452", runner) self.assertNotIn("TILEXR_CCU_PROBE_SYNC_INSTRUCTION_COUNT:-458", runner) diff --git a/tests/ccu/test_tilexr_ccu_direct_smoke_runner.py b/tests/ccu/test_tilexr_ccu_direct_smoke_runner.py index 99278d67..e5c83c3c 100644 --- a/tests/ccu/test_tilexr_ccu_direct_smoke_runner.py +++ b/tests/ccu/test_tilexr_ccu_direct_smoke_runner.py @@ -17,7 +17,9 @@ class TileXRCcuDirectSmokeRunnerTest(unittest.TestCase): - def run_fake_mesh_runner(self, devices="4,5,6,7", rank_size="4", loop_count="10"): + def run_fake_mesh_runner( + self, devices="4,5,6,7", rank_size="4", loop_count="10", submit=True + ): temp_dir = tempfile.TemporaryDirectory() temp_path = Path(temp_dir.name) fake_bin = temp_path / "bin" @@ -35,6 +37,7 @@ def run_fake_mesh_runner(self, devices="4,5,6,7", rank_size="4", loop_count="10" "rank_size=${TILEXR_CCU_PROBE_RANK_SIZE}\n" "loops=${TILEXR_CCU_ALLTOALL_LOOP_COUNT}\n" "echo \"tilexr_ccu_alltoall prepare ret=0 installSucceeded=1 submitReady=1\"\n" + "if [ \"${TILEXR_CCU_DIRECT_SMOKE_SUBMIT:-0}\" = 1 ]; then\n" "for ((loop=0; loop(allocation.remoteXn.startId + 8U); remote1.remoteNotifyCke = 0x361; remoteCcuBuffers.push_back(remote1); @@ -1236,7 +1244,7 @@ def test_builds_transport_template_from_basic_info_and_resource_allocation(self) if (snapshot.routes.size() != 2 || snapshot.routes[0].channelId != allocation.channels.startId || snapshot.routes[1].channelId != allocation.channels.startId + 1U || snapshot.routes[0].remoteXnId != allocation.remoteXn.startId || - snapshot.routes[1].remoteXnId != allocation.remoteXn.startId + 1U || + snapshot.routes[1].remoteXnId != allocation.remoteXn.startId + 8U || snapshot.routes[0].remoteNotifyCke != 0x360 || snapshot.routes[1].remoteNotifyCke != 0x361 || snapshot.routes[0].wqeBasicBlockStartId != 0 || @@ -2237,7 +2245,7 @@ def test_plan_builder_surface_is_wired_into_tilexr_comm_without_udma_boundary(se self.assertIn("TileXRCcuBuildPfeCtx", source) self.assertIn("TileXRCcuBuildLocalJettyCtx", source) self.assertIn("TileXRCcuBuildChannelCtxV1", source) - self.assertIn("allocation.channels.num < remoteCcuBuffers.size()", source) + self.assertIn("remoteCcuBuffers.size() != allocation.channels.num", source) self.assertIn("channel allocation count does not match lower-layer route count", source) self.assertNotIn("TILEXR_CCU_DIRECT_SYNC_RESOURCE_MAP", source) self.assertNotIn("UseHcommTraceSyncResourceMap", source) @@ -2534,6 +2542,12 @@ def test_remote_xn_exchange_uses_peer_channel_bound_remote_xn_operand(self): def test_lower_layer_clears_the_complete_allocated_remote_xn_range(self): source = BUILDER_SOURCE.read_text(encoding="utf-8") + self.assertIn( + "remoteCcuBuffers.size() != allocation.channels.num", + source) + self.assertNotIn( + "remoteCcuBuffers.size() != allocation.remoteXn.num", + source) self.assertIn("result.remoteXnStartId = allocation.remoteXn.startId", source) self.assertIn("result.remoteXnCount = allocation.remoteXn.num", source) self.assertIn("snapshot.remoteXnStartId", source) @@ -2570,20 +2584,26 @@ def test_peer_xn_exchange_expands_one_peer_window_to_multiple_sync_routes(self): compact_body = " ".join(exchange_body.split()) self.assertIn("const size_t peerRouteCount = static_cast(rankSize - 1)", compact_body) - self.assertIn("const size_t syncRouteCount = allocation.remoteXn.num", compact_body) + self.assertIn("const size_t syncRouteCount = allocation.channels.num", compact_body) self.assertIn("allocation.remoteXn.num < routedPeerCount", compact_body) + self.assertIn("allocation.localWaitCke.num < syncRouteCount", compact_body) + self.assertIn("allocation.remoteNotifyCke.num < syncRouteCount", compact_body) self.assertNotIn("allocation.remoteXn.num != static_cast(rankSize - 1)", compact_body) self.assertIn("peerCcuBuffersByRank", compact_body) self.assertIn("peerCcuBuffer.peerRank", compact_body) self.assertIn("invalid direct CCU peer buffer rank mapping", compact_body) self.assertIn("incomplete direct CCU peer buffer rank mapping", compact_body) self.assertIn("remoteCcuBuffers->assign(syncRouteCount, TileXRCcuRemoteCcuBufferInfo{})", compact_body) - self.assertIn("for (uint32_t syncIndex = 0; syncIndex < allocation.remoteXn.num; ++syncIndex)", compact_body) + self.assertIn("for (uint32_t syncIndex = 0; syncIndex < syncRouteCount; ++syncIndex)", compact_body) self.assertIn("const size_t peerBufferIndex = syncIndex / routesPerPeer", compact_body) self.assertIn( "(*remoteCcuBuffers)[routeIndex] = *peerCcuBuffersByRank[static_cast(peer)]", compact_body) self.assertIn("peerLocalIndex * routesPerPeer + routeWithinPeer", compact_body) + self.assertIn( + "SelectDirectCcuChannelBoundRemoteXnOffset( peerLocalIndex, routeWithinPeer, " + "routesPerPeer, channelStridedRemoteXn)", + compact_body) self.assertIn("peerResources.remoteXnStartId", compact_body) self.assertIn("DirectCcuRemoteXnProofSpan(allocation.remoteXn.num)", compact_body) @@ -2611,9 +2631,17 @@ def test_direct_ccu_runtime_imports_peer_endpoint_route_before_export(self): def test_peer_endpoints_keep_per_peer_resource_and_jetty_tokens(self): source = DIRECT_RUNTIME_SOURCE.read_text(encoding="utf-8") + self.assertIn("SelectEndpointRouteJettyCtxId(state->eidInfo.funcId, peerOrdinal", source) + self.assertIn( + "TILEXR_CCU_DIRECT_LOOP_JETTY_ID + jettyCtxId", + source) + self.assertIn("SelectEndpointRouteSqVa(localResourceWindow_, jettyCtxId)", source) + self.assertIn("state->resourceWindow.tokenId = localResourceWindow_.tokenId", source) + self.assertIn("state->resourceWindow.tokenValue = localResourceWindow_.tokenValue", source) + self.assertNotIn("mr.in.ub.tokenValue = state->resourceWindow.tokenValue", source) self.assertIn("offer.resourceTokenId = state.resourceWindow.tokenId", source) self.assertIn("offer.resourceTokenValue = state.resourceWindow.tokenValue", source) - self.assertIn("offer.jettyTokenValue = state.resourceWindow.tokenValue", source) + self.assertIn("offer.jettyTokenValue = state.jettyTokenValue", source) self.assertIn("importInfo.in.ub.tokenValue = peerOffer.jettyTokenValue", source) self.assertIn("state.route.memoryTokenValue = peerOffer.resourceTokenValue", source) @@ -2890,8 +2918,14 @@ def test_direct_runtime_source_supports_selecting_ra_ctx_resource_window_eid(sel self.assertIn("loopEidCandidate", source) self.assertIn("TraceRaCtxEidInfos", source) self.assertIn("ctxAttr.ub.eidIndex = candidate.eidIndex", source) - self.assertEqual(2, source.count( + self.assertEqual(1, source.count( "qpAttr.ub.errTimeout = TILEXR_CCU_DIRECT_ENDPOINT_ERR_TIMEOUT")) + self.assertIn( + "qpAttr.ub.errTimeout = state->tpType == TILEXR_CCU_HCCP_TP_TYPE_CTP ?", + source) + self.assertIn( + "TILEXR_CCU_DIRECT_CTP_ENDPOINT_ERR_TIMEOUT : TILEXR_CCU_DIRECT_ENDPOINT_ERR_TIMEOUT", + source) def test_direct_ccu_runtime_can_select_ra_ctx_resource_window_eid_by_env(self): code = textwrap.dedent( diff --git a/tests/ccu/test_tilexr_ccu_ra_custom_channel_loader.py b/tests/ccu/test_tilexr_ccu_ra_custom_channel_loader.py index 56cc7fe2..1f04d59c 100644 --- a/tests/ccu/test_tilexr_ccu_ra_custom_channel_loader.py +++ b/tests/ccu/test_tilexr_ccu_ra_custom_channel_loader.py @@ -1305,6 +1305,27 @@ def test_direct_runtime_selects_tp_sl_before_creating_peer_qp(self): self.assertIn("qpAttr.ub.priority = state->mappedJettyPriority", create_body) self.assertNotIn("qpAttr.ub.priority = 2", create_body) + def test_direct_runtime_maps_ctp_jetty_priority_without_setting_tp_sl(self): + source = DIRECT_RUNTIME_SOURCE.read_text(encoding="utf-8") + select_start = source.index("int TileXRCcuDirectRuntime::SelectTpRouteForPeer(") + select_body = source[ + select_start: + source.index("int TileXRCcuDirectRuntime::QueryTpHandleForPeer(", select_start) + ] + compact_body = " ".join(select_body.split()) + + self.assertNotIn( + "if (tpType == TILEXR_CCU_HCCP_TP_TYPE_CTP) {", + select_body, + ) + self.assertIn("RaGetTpAttrAsync(", select_body) + self.assertIn("MapQosToTpAndSl(", select_body) + self.assertIn( + "if (tpType == TILEXR_CCU_HCCP_TP_TYPE_RTP) { TileXRCcuHccpTpAttr setAttr", + compact_body, + ) + self.assertIn("*mappedJettyPriority = mappedSl", select_body) + def test_peer_endpoint_route_uses_the_driver_returned_jetty_id(self): source = DIRECT_RUNTIME_SOURCE.read_text(encoding="utf-8") prepare_body = source[ diff --git a/tests/ccu/test_tilexr_ccu_ra_custom_channel_provider.py b/tests/ccu/test_tilexr_ccu_ra_custom_channel_provider.py index 6052edf4..5b74a0cb 100644 --- a/tests/ccu/test_tilexr_ccu_ra_custom_channel_provider.py +++ b/tests/ccu/test_tilexr_ccu_ra_custom_channel_provider.py @@ -227,6 +227,68 @@ def test_provider_accepts_opaque_ra_custom_channel_c_abi_shape(self): self.assertEqual("", result.stderr) self.assertEqual(0, result.returncode, result.stdout + result.stderr) + def test_provider_retries_transient_roce_eagain(self): + code = textwrap.dedent( + r''' + #include "ccu/tilexr_ccu_ra_custom_channel_provider.h" + + #include + + using namespace TileXR; + + int g_calls = 0; + + int FakeRaCustomChannel( + TileXRCcuRaInfo, + TileXRCcuCustomChannelIn*, + TileXRCcuCustomChannelOut* out) + { + ++g_calls; + if (g_calls < 3) { + out->opRet = 99; + return 128101; + } + out->opRet = 0; + out->data.dataInfo.dataArray[0].baseinfo.msId = 0x55; + out->data.dataInfo.dataArray[0].baseinfo.missionKey = 0xabcdef01U; + out->data.dataInfo.dataArray[0].baseinfo.resourceAddr = 0x500000000ULL; + out->data.dataInfo.dataArray[0].baseinfo.caps.cap0 = (1U << 24) | (2U << 16) | 31U; + out->data.dataInfo.dataArray[0].baseinfo.caps.cap1 = (15U << 16) | 7U; + out->data.dataInfo.dataArray[0].baseinfo.caps.cap2 = (3U << 16) | 5U; + out->data.dataInfo.dataArray[0].baseinfo.caps.cap3 = (9U << 16) | 1U; + return 0; + } + + int main() + { + TileXRCcuRaCustomChannelProvider provider; + TileXRCcuRaCustomChannelProviderReport providerReport; + if (provider.Init(3, FakeRaCustomChannel, &providerReport) != TILEXR_SUCCESS) { + return 1; + } + TileXRCcuDriverAdapter adapter; + TileXRCcuDriverAdapterReport adapterReport; + if (provider.CreateAdapter(&adapter, &adapterReport) != TILEXR_SUCCESS) { + return 2; + } + TileXRCcuBasicInfo basic; + if (adapter.GetBasicInfo(0, &basic, &adapterReport) != TILEXR_SUCCESS) { + std::cerr << adapterReport.message << "\n"; + return 3; + } + if (g_calls != 3 || basic.msId != 0x55 || basic.missionKey != 0xabcdef01U) { + return 4; + } + return 0; + } + ''' + ) + + result = self.compile_and_run(code) + + self.assertEqual("", result.stderr) + self.assertEqual(0, result.returncode, result.stdout + result.stderr) + def test_provider_is_wired_and_keeps_hcomm_runtime_out_of_ccu_surface(self): cmake = COMM_CMAKE.read_text(encoding="utf-8") header = PROVIDER_HEADER.read_text(encoding="utf-8") diff --git a/tests/ccu/test_tilexr_ccu_resource_allocator.py b/tests/ccu/test_tilexr_ccu_resource_allocator.py index b9f2ca28..f7d60de2 100644 --- a/tests/ccu/test_tilexr_ccu_resource_allocator.py +++ b/tests/ccu/test_tilexr_ccu_resource_allocator.py @@ -143,7 +143,7 @@ def test_allocator_builds_complete_tilexr_owned_producer_plan(self): std::cerr << "first sync resource mismatch\n"; return 7; } - if (plan.syncResources[2].localXn != 1963 || plan.syncResources[2].remoteXn != 1977 || + if (plan.syncResources[2].localXn != 1963 || plan.syncResources[2].remoteXn != 1991 || plan.syncResources[2].notifyCke != 334 || plan.syncResources[2].channelId != 4) { std::cerr << "last sync resource mismatch\n"; return 8; @@ -163,6 +163,7 @@ def test_allocator_builds_complete_tilexr_owned_producer_plan(self): } if (allocation.receiptId == 0 || allocation.packageProvider != "tilexr-hcomm-derived-resource-allocator" || allocation.localXn.startId != 1961 || allocation.remoteXn.startId != 1975 || + allocation.remoteXn.num != 24 || allocation.localGsa.startId != 510 || allocation.localGsa.num != 1 || allocation.notifyCke.startId != 332 || allocation.channels.startId != 2 || allocation.channels.num != 3) { @@ -171,7 +172,7 @@ def test_allocator_builds_complete_tilexr_owned_producer_plan(self): } if (report.missionAllocated != 1 || report.localXnAllocated != 14 || report.localGsaAllocated != 1 || - report.remoteXnAllocated != 3 || report.notifyCkeAllocated != 3 || + report.remoteXnAllocated != 24 || report.notifyCkeAllocated != 3 || report.channelBindingsAllocated != 9 || report.repositoryAllocated != 156 || report.message != "ok") { std::cerr << "report mismatch\n"; @@ -391,6 +392,8 @@ def test_allocator_builds_pure_barrier_plan_without_sqe_load_task(self): request.syncResourceCount = 1; request.syncInstructionCount = 2; request.bindingsPerSyncResource = 1; + request.minimumLocalXnCount = 3; + request.minimumRemoteXnCount = 3; TileXRCcuResourceAllocator allocator; if (allocator.Init(spec) != TILEXR_SUCCESS) { @@ -416,15 +419,15 @@ def test_allocator_builds_pure_barrier_plan_without_sqe_load_task(self): std::cerr << "pure barrier sync task mismatch\n"; return 4; } - if (plan.kernelLocalXn.startId != 1 || plan.kernelLocalXn.num != 1 || - allocation.localXn.startId != 1 || allocation.localXn.num != 1 || - allocation.remoteXn.startId != 2 || allocation.remoteXn.num != 1 || + if (plan.kernelLocalXn.startId != 1 || plan.kernelLocalXn.num != 3 || + allocation.localXn.startId != 1 || allocation.localXn.num != 3 || + allocation.remoteXn.startId != 4 || allocation.remoteXn.num != 3 || allocation.repository.startId != 1 || allocation.repository.num != 2) { std::cerr << "pure barrier allocation mismatch\n"; return 5; } - if (report.localXnAllocated != 1 || - report.remoteXnAllocated != 1 || + if (report.localXnAllocated != 3 || + report.remoteXnAllocated != 3 || report.repositoryAllocated != 2) { std::cerr << "pure barrier report mismatch\n"; return 6; From 5a61a413d403e90790e3159ba68ba743be58d792 Mon Sep 17 00:00:00 2001 From: Kur0x Date: Tue, 28 Jul 2026 12:26:31 +0800 Subject: [PATCH 2/3] ci: restore main-only PR gate --- .github/workflows/pr-ci.yml | 4 +--- tests/ci/test_workflows.rb | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index f6489f24..ae8ab0d7 100644 --- a/.github/workflows/pr-ci.yml +++ b/.github/workflows/pr-ci.yml @@ -2,9 +2,7 @@ name: PR CI "on": pull_request: - branches: - - main - - "codex/ccu-stack-*" + branches: [main] types: [opened, reopened, synchronize, ready_for_review, converted_to_draft, closed] permissions: diff --git a/tests/ci/test_workflows.rb b/tests/ci/test_workflows.rb index 4a0e099e..69690d28 100644 --- a/tests/ci/test_workflows.rb +++ b/tests/ci/test_workflows.rb @@ -45,8 +45,8 @@ def named_step(workflow, job_name, step_name) assert_equal(["pull_request"], pr.fetch("on").keys, "PR workflow must use only pull_request") pull_request = pr.fetch("on").fetch("pull_request") -assert_equal(["main", "codex/ccu-stack-*"], pull_request.fetch("branches"), - "PR workflow must target main and the temporary CCU stack") +assert_equal(["main"], pull_request.fetch("branches"), + "PR workflow must target main") assert_equal(%w[opened reopened synchronize ready_for_review converted_to_draft closed], pull_request.fetch("types"), "PR event types differ") assert_equal({"contents" => "read"}, pr.fetch("permissions"), From 6db8f1216fd56561f575293d00423b08228955dd Mon Sep 17 00:00:00 2001 From: Kur0x Date: Tue, 28 Jul 2026 12:31:51 +0800 Subject: [PATCH 3/3] ci: keep validation through final stack layer --- .github/workflows/pr-ci.yml | 4 +++- tests/ci/test_workflows.rb | 4 ++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index ae8ab0d7..f6489f24 100644 --- a/.github/workflows/pr-ci.yml +++ b/.github/workflows/pr-ci.yml @@ -2,7 +2,9 @@ name: PR CI "on": pull_request: - branches: [main] + branches: + - main + - "codex/ccu-stack-*" types: [opened, reopened, synchronize, ready_for_review, converted_to_draft, closed] permissions: diff --git a/tests/ci/test_workflows.rb b/tests/ci/test_workflows.rb index 69690d28..4a0e099e 100644 --- a/tests/ci/test_workflows.rb +++ b/tests/ci/test_workflows.rb @@ -45,8 +45,8 @@ def named_step(workflow, job_name, step_name) assert_equal(["pull_request"], pr.fetch("on").keys, "PR workflow must use only pull_request") pull_request = pr.fetch("on").fetch("pull_request") -assert_equal(["main"], pull_request.fetch("branches"), - "PR workflow must target main") +assert_equal(["main", "codex/ccu-stack-*"], pull_request.fetch("branches"), + "PR workflow must target main and the temporary CCU stack") assert_equal(%w[opened reopened synchronize ready_for_review converted_to_draft closed], pull_request.fetch("types"), "PR event types differ") assert_equal({"contents" => "read"}, pr.fetch("permissions"),