Add: L2 input window support to L3-L2 message queue#1236
Open
ccyywwen wants to merge 2 commits into
Open
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the L3-L2 Message Queue, enabling ordered message exchange between an L3 Host Orchestrator and a persistent L2 AICPU Orchestrator task, complete with documentation, Python/C++ implementations, examples, and unit tests. The review feedback focuses on enhancing C++ memory safety and resource management within the l3_l2_message_queue.h header. Specifically, the reviewer recommends explicitly deleting copy and move operations for InputQueue and L3L2QueueEndpoint to prevent double-free vulnerabilities, and safeguarding InputQueue::initialize against potential memory leaks from multiple invocations.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+267
to
+270
| explicit InputQueue(L3L2QueueEndpoint *parent) : | ||
| parent_(parent) {} | ||
|
|
||
| ~InputQueue() { delete[] active_entries_; } |
There was a problem hiding this comment.
To prevent accidental copying or moving of InputQueue which manages the raw pointer active_entries_ (allocated via new[]), we should explicitly delete its copy and move constructors and assignment operators. This avoids potential double-free vulnerabilities or memory leaks.
explicit InputQueue(L3L2QueueEndpoint *parent) :
parent_(parent) {}
InputQueue(const InputQueue &) = delete;
InputQueue &operator=(const InputQueue &) = delete;
InputQueue(InputQueue &&) = delete;
InputQueue &operator=(InputQueue &&) = delete;
~InputQueue() { delete[] active_entries_; }
Comment on lines
+636
to
+653
| L3L2QueueEndpoint( | ||
| const L3L2OrchRegionDesc &desc, const L3L2QueueArgs &args, const L3L2QueueEndpointConfig &config | ||
| ) : | ||
| endpoint_(desc), | ||
| config_(config), | ||
| input_queue_(this), | ||
| output_queue_(this) { | ||
| if (endpoint_.error().kind != L3L2EndpointErrorKind::NONE || | ||
| !l3_l2_queue_validate_region(desc, args, &layout_)) { | ||
| set_error(L3L2QueueErrorKind::BAD_DESCRIPTOR, "init", desc.region_id, "invalid queue descriptor"); | ||
| return; | ||
| } | ||
| if (config_.max_l2_input_inflight == 0 || config_.max_l2_input_inflight > layout_.depth) { | ||
| set_error(L3L2QueueErrorKind::BAD_ARGUMENT, "init", desc.region_id, "invalid endpoint config"); | ||
| return; | ||
| } | ||
| input_queue_.initialize(config_.max_l2_input_inflight); | ||
| } |
There was a problem hiding this comment.
To prevent accidental copying or moving of L3L2QueueEndpoint (which owns input_queue_ and manages its lifetime), we should explicitly delete its copy and move constructors and assignment operators. This ensures proper RAII and resource safety.
L3L2QueueEndpoint(
const L3L2OrchRegionDesc &desc, const L3L2QueueArgs &args, const L3L2QueueEndpointConfig &config
) :
endpoint_(desc),
config_(config),
input_queue_(this),
output_queue_(this) {
if (endpoint_.error().kind != L3L2EndpointErrorKind::NONE ||
!l3_l2_queue_validate_region(desc, args, &layout_)) {
set_error(L3L2QueueErrorKind::BAD_DESCRIPTOR, "init", desc.region_id, "invalid queue descriptor");
return;
}
if (config_.max_l2_input_inflight == 0 || config_.max_l2_input_inflight > layout_.depth) {
set_error(L3L2QueueErrorKind::BAD_ARGUMENT, "init", desc.region_id, "invalid endpoint config");
return;
}
input_queue_.initialize(config_.max_l2_input_inflight);
}
L3L2QueueEndpoint(const L3L2QueueEndpoint &) = delete;
L3L2QueueEndpoint &operator=(const L3L2QueueEndpoint &) = delete;
L3L2QueueEndpoint(L3L2QueueEndpoint &&) = delete;
L3L2QueueEndpoint &operator=(L3L2QueueEndpoint &&) = delete;
Comment on lines
+272
to
+274
| bool initialize(uint64_t max_l2_input_inflight) { | ||
| entry_capacity_ = max_l2_input_inflight + 1; | ||
| active_entries_ = new (std::nothrow) ActiveInputEntry[entry_capacity_]; |
There was a problem hiding this comment.
If initialize is called multiple times on the same InputQueue instance, the previously allocated active_entries_ array will be leaked. To prevent memory leaks, we should explicitly delete any existing allocation before allocating a new one.
bool initialize(uint64_t max_l2_input_inflight) {
if (active_entries_ != nullptr) {
delete[] active_entries_;
}
entry_capacity_ = max_l2_input_inflight + 1;
active_entries_ = new (std::nothrow) ActiveInputEntry[entry_capacity_];
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughAdds an ordered L3-L2 message queue: C++ ABI/endpoint header, Python queue wrapper ( ChangesL3-L2 Message Queue
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant L3Script as l3_l2_message_queue.py
participant L3Queue as L3L2Queue (Python)
participant Region as L3L2OrchRegion
participant L2Endpoint as L3L2QueueEndpoint (C++)
participant AIVKernel as kernel_queue_transform.cpp
L3Script->>L3Queue: create_l3_l2_queue(depth, arena bytes)
L3Queue->>Region: allocate descriptor rings + payload arenas
L3Script->>L3Queue: enqueue(input payload, mode)
L3Queue->>Region: write descriptor + payload, notify tail counter
L2Endpoint->>Region: peek(timeout) input queue
Region-->>L2Endpoint: DATA descriptor + payload
L2Endpoint->>AIVKernel: rt_submit_aiv_task(tile/scalar op)
AIVKernel-->>L2Endpoint: computed output tile
L2Endpoint->>Region: reserve + publish output descriptor
L3Script->>L3Queue: peek/read_into output
L3Queue->>Region: read payload, release, advance head
L3Script->>L3Queue: request_stop()
L3Queue->>Region: publish STOP descriptor
L2Endpoint->>Region: peek observes STOP, drains
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
src/common/platform/include/aicpu/l3_l2_message_queue.h (1)
38-43: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPin the descriptor ABI in the header.
The size/offset guarantees only live in the unit test right now. Adding
static_asserts besideL3L2QueueDescSlotwill make any ABI drift fail in every build, not just the test target.Suggested hardening
struct L3L2QueueDescSlot { uint64_t seq; uint64_t opcode; uint64_t payload_offset; uint64_t payload_nbytes; }; + +static_assert(sizeof(L3L2QueueDescSlot) == 32, "L3L2QueueDescSlot ABI size changed"); +static_assert(offsetof(L3L2QueueDescSlot, seq) == 0, "L3L2QueueDescSlot::seq offset changed"); +static_assert(offsetof(L3L2QueueDescSlot, opcode) == 8, "L3L2QueueDescSlot::opcode offset changed"); +static_assert(offsetof(L3L2QueueDescSlot, payload_offset) == 16, "L3L2QueueDescSlot::payload_offset changed"); +static_assert(offsetof(L3L2QueueDescSlot, payload_nbytes) == 24, "L3L2QueueDescSlot::payload_nbytes changed");🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/platform/include/aicpu/l3_l2_message_queue.h` around lines 38 - 43, The descriptor ABI for L3L2QueueDescSlot is only enforced in tests right now, so add compile-time static_assert checks next to the struct definition to pin its size and member offsets. Use the L3L2QueueDescSlot type in the header to assert the expected layout so any ABI drift fails in every build, not just the unit test target.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/l3-l2-message-queue.md`:
- Around line 169-172: Update the message-queue docs around
queue.input().try_peek() and queue.output().try_reserve() to remove timeout from
the false-return contract, since these methods are non-blocking and false should
only cover no progress, validation failure, or poison; keep the guidance to
check queue.error().kind for terminal errors and ensure the wording is
consistent with the try_* API semantics.
In
`@examples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/kernels/orchestration/l3_l2_message_queue_orch.cpp`:
- Around line 123-145: process_first_pair() leaves consumed ActiveRequest slots
populated after release_input(), which lets the next batch overwrite or reuse
stale handles. Clear the released entries in the active array after each
successful release and make the mode == 1 path verify that active[0] is free
before assigning a new input handle. Apply the same cleanup to the related batch
handling code in the other matching block so both paths reset
active[0]/active[1] consistently before continuing the receive/process/publish
loop.
In `@src/common/platform/include/aicpu/l3_l2_message_queue.h`:
- Around line 360-419: The input peek path in the queue handling logic accepts
STOP descriptors with payload bytes, which should be rejected by the STOP
contract. Update the `try_peek` flow in the `L3L2Queue` input handling to
validate `opcode == L3L2QueueOpcode::STOP` before any payload read, and poison
with `L3L2QueueErrorKind::INVALID_DESCRIPTOR` if `slot.payload_nbytes` is
nonzero. Ensure `payload_matches_head`, `endpoint_.payload_read`, and the
`active_entries_` / `input_payload_head_` bookkeeping are only reached for valid
zero-byte STOP descriptors.
---
Nitpick comments:
In `@src/common/platform/include/aicpu/l3_l2_message_queue.h`:
- Around line 38-43: The descriptor ABI for L3L2QueueDescSlot is only enforced
in tests right now, so add compile-time static_assert checks next to the struct
definition to pin its size and member offsets. Use the L3L2QueueDescSlot type in
the header to assert the expected layout so any ABI drift fails in every build,
not just the unit test target.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: e6cfe6c1-0173-4585-8104-6d1cd2e9d978
📒 Files selected for processing (12)
docs/l3-l2-message-queue.mddocs/l3-l2-orch-comm.mdexamples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/kernels/aiv/kernel_queue_transform.cppexamples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/kernels/orchestration/l3_l2_message_queue_orch.cppexamples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/l3_l2_message_queue.pyexamples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/test_l3_l2_message_queue.pypython/simpler/l3_l2_message_queue.pypython/simpler/orchestrator.pysrc/common/platform/include/aicpu/l3_l2_message_queue.htests/ut/cpp/CMakeLists.txttests/ut/cpp/common/test_l3_l2_message_queue.cpptests/ut/py/test_worker/test_l3_l2_message_queue.py
Comment on lines
+169
to
+172
| `queue.input().try_peek(&input)` and | ||
| `queue.output().try_reserve(nbytes, &reservation)` are non-blocking. A `false` | ||
| return can mean no progress, timeout, validation failure, or poison; check | ||
| `queue.error().kind` to distinguish ordinary no-progress from terminal error. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Remove timeout from the try_* contract.
try_peek() and try_reserve() are non-blocking, so false here cannot mean timeout. Keeping timeout in this list makes the API contract internally inconsistent for callers.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/l3-l2-message-queue.md` around lines 169 - 172, Update the message-queue
docs around queue.input().try_peek() and queue.output().try_reserve() to remove
timeout from the false-return contract, since these methods are non-blocking and
false should only cover no progress, validation failure, or poison; keep the
guidance to check queue.error().kind for terminal errors and ensure the wording
is consistent with the try_* API semantics.
Comment on lines
+123
to
+145
| bool process_first_pair(L3L2QueueEndpoint &queue, ActiveRequest *active, const L3L2QueueInputHandle &input) { | ||
| active[1].handle = input; | ||
| if (!parse_input_header(input, &active[1].header) || active[0].header.request_id == 0) { | ||
| rt_report_fatal(PTO2_ERROR_EXPLICIT_ORCH_FATAL, "invalid L3-L2 queue example request"); | ||
| return false; | ||
| } | ||
| if (!publish_aiv_output( | ||
| queue, active[1].handle, active[1].handle, active[1].header.request_id, 20, 0, ADD_SCALAR, 20.0F | ||
| )) { | ||
| return false; | ||
| } | ||
| if (!release_input(queue, active[1].handle)) { | ||
| return false; | ||
| } | ||
| if (!publish_aiv_output( | ||
| queue, active[0].handle, active[0].handle, active[0].header.request_id, 10, 0, ADD_SCALAR, 10.0F | ||
| ) || | ||
| !publish_aiv_output( | ||
| queue, active[0].handle, active[0].handle, active[0].header.request_id, 11, 0, ADD_SCALAR, 11.0F | ||
| )) { | ||
| return false; | ||
| } | ||
| return release_input(queue, active[0].handle); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Clear consumed active slots before the next batch.
process_first_pair() releases active[0]/active[1] but leaves both slots populated, and the mode == 1 path overwrites active[0] without checking whether a prior request is still pending. A second batch can therefore leak the overwritten handle or reuse a handle that was already released, which breaks the example’s persistent receive/process/publish loop.
Proposed fix
bool process_first_pair(L3L2QueueEndpoint &queue, ActiveRequest *active, const L3L2QueueInputHandle &input) {
active[1].handle = input;
if (!parse_input_header(input, &active[1].header) || active[0].header.request_id == 0) {
rt_report_fatal(PTO2_ERROR_EXPLICIT_ORCH_FATAL, "invalid L3-L2 queue example request");
return false;
@@
if (!release_input(queue, active[1].handle)) {
return false;
}
+ active[1] = {};
if (!publish_aiv_output(
queue, active[0].handle, active[0].handle, active[0].header.request_id, 10, 0, ADD_SCALAR, 10.0F
) ||
!publish_aiv_output(
queue, active[0].handle, active[0].handle, active[0].header.request_id, 11, 0, ADD_SCALAR, 11.0F
)) {
return false;
}
- return release_input(queue, active[0].handle);
+ if (!release_input(queue, active[0].handle)) {
+ return false;
+ }
+ active[0] = {};
+ return true;
}
@@
if (header.mode == 1) {
+ if (active[0].header.request_id != 0) {
+ rt_report_fatal(PTO2_ERROR_EXPLICIT_ORCH_FATAL, "L3-L2 queue example received mode=1 while a request is still pending");
+ return false;
+ }
active[0].handle = input;
active[0].header = header;
return true;
}Also applies to: 169-178
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@examples/a2a3/tensormap_and_ringbuffer/l3_l2_message_queue/kernels/orchestration/l3_l2_message_queue_orch.cpp`
around lines 123 - 145, process_first_pair() leaves consumed ActiveRequest slots
populated after release_input(), which lets the next batch overwrite or reuse
stale handles. Clear the released entries in the active array after each
successful release and make the mode == 1 path verify that active[0] is free
before assigning a new input handle. Apply the same cleanup to the related batch
handling code in the other matching block so both paths reset
active[0]/active[1] consistently before continuing the receive/process/publish
loop.
Comment on lines
+360
to
+419
| L3L2QueueOpcode opcode = static_cast<L3L2QueueOpcode>(slot.opcode); | ||
| if (!l3_l2_queue_valid_opcode(opcode)) { | ||
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "invalid input opcode"); | ||
| return false; | ||
| } | ||
| bool counts_against_window = opcode == L3L2QueueOpcode::DATA || opcode == L3L2QueueOpcode::ERROR; | ||
| if (counts_against_window && active_non_stop_count_ >= parent_->config_.max_l2_input_inflight) { | ||
| return false; | ||
| } | ||
| if (active_count_ >= entry_capacity_) { | ||
| parent_->poison(L3L2QueueErrorKind::OWNERSHIP, "input.try_peek", "input window state full"); | ||
| return false; | ||
| } | ||
|
|
||
| L3L2OrchPayloadView view{0, 0}; | ||
| if (slot.payload_nbytes == 0) { | ||
| if (slot.payload_offset != 0) { | ||
| parent_->poison( | ||
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | ||
| "zero-byte descriptor uses nonzero payload offset" | ||
| ); | ||
| return false; | ||
| } | ||
| } else if (!parent_->payload_in_arena( | ||
| slot.payload_offset, slot.payload_nbytes, parent_->layout_.input_arena_offset, | ||
| parent_->layout_.input_arena_bytes | ||
| )) { | ||
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "input payload out of arena"); | ||
| return false; | ||
| } else if (!parent_->payload_matches_head( | ||
| parent_->input_payload_head_, slot.payload_offset, slot.payload_nbytes, | ||
| parent_->layout_.input_arena_offset, parent_->layout_.input_arena_bytes, "input.try_peek" | ||
| )) { | ||
| return false; | ||
| } else if (!parent_->endpoint_.payload_read(slot.payload_offset, slot.payload_nbytes, &view)) { | ||
| parent_->poison( | ||
| L3L2QueueErrorKind::ENDPOINT_ERROR, "input.try_peek", parent_->endpoint_.error().message | ||
| ); | ||
| return false; | ||
| } | ||
|
|
||
| *out = L3L2QueueInputHandle{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view}; | ||
| active_entries_[active_count_] = | ||
| ActiveInputEntry{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view, false}; | ||
| active_count_ += 1; | ||
| if (counts_against_window) { | ||
| active_non_stop_count_ += 1; | ||
| } | ||
| input_acquire_ += 1; | ||
| if (opcode == L3L2QueueOpcode::STOP) { | ||
| stop_observed_ = true; | ||
| if (parent_->input_tail_ != input_acquire_) { | ||
| parent_->poison( | ||
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | ||
| "input descriptor published after STOP" | ||
| ); | ||
| return false; | ||
| } | ||
| } | ||
| return true; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Reject STOP descriptors that carry payload bytes.
The STOP contract is zero-byte, but this path currently accepts opcode == STOP with nonzero payload_nbytes, exposes a payload view, and then advances input_payload_head_ on release. That lets malformed shared state bypass INVALID_DESCRIPTOR poison and breaks the documented STOP ABI.
Suggested fix
L3L2QueueOpcode opcode = static_cast<L3L2QueueOpcode>(slot.opcode);
if (!l3_l2_queue_valid_opcode(opcode)) {
parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "invalid input opcode");
return false;
}
+ if (opcode == L3L2QueueOpcode::STOP && (slot.payload_offset != 0 || slot.payload_nbytes != 0)) {
+ parent_->poison(
+ L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "STOP descriptor must be zero-byte"
+ );
+ return false;
+ }
bool counts_against_window = opcode == L3L2QueueOpcode::DATA || opcode == L3L2QueueOpcode::ERROR;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| L3L2QueueOpcode opcode = static_cast<L3L2QueueOpcode>(slot.opcode); | |
| if (!l3_l2_queue_valid_opcode(opcode)) { | |
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "invalid input opcode"); | |
| return false; | |
| } | |
| bool counts_against_window = opcode == L3L2QueueOpcode::DATA || opcode == L3L2QueueOpcode::ERROR; | |
| if (counts_against_window && active_non_stop_count_ >= parent_->config_.max_l2_input_inflight) { | |
| return false; | |
| } | |
| if (active_count_ >= entry_capacity_) { | |
| parent_->poison(L3L2QueueErrorKind::OWNERSHIP, "input.try_peek", "input window state full"); | |
| return false; | |
| } | |
| L3L2OrchPayloadView view{0, 0}; | |
| if (slot.payload_nbytes == 0) { | |
| if (slot.payload_offset != 0) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | |
| "zero-byte descriptor uses nonzero payload offset" | |
| ); | |
| return false; | |
| } | |
| } else if (!parent_->payload_in_arena( | |
| slot.payload_offset, slot.payload_nbytes, parent_->layout_.input_arena_offset, | |
| parent_->layout_.input_arena_bytes | |
| )) { | |
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "input payload out of arena"); | |
| return false; | |
| } else if (!parent_->payload_matches_head( | |
| parent_->input_payload_head_, slot.payload_offset, slot.payload_nbytes, | |
| parent_->layout_.input_arena_offset, parent_->layout_.input_arena_bytes, "input.try_peek" | |
| )) { | |
| return false; | |
| } else if (!parent_->endpoint_.payload_read(slot.payload_offset, slot.payload_nbytes, &view)) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::ENDPOINT_ERROR, "input.try_peek", parent_->endpoint_.error().message | |
| ); | |
| return false; | |
| } | |
| *out = L3L2QueueInputHandle{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view}; | |
| active_entries_[active_count_] = | |
| ActiveInputEntry{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view, false}; | |
| active_count_ += 1; | |
| if (counts_against_window) { | |
| active_non_stop_count_ += 1; | |
| } | |
| input_acquire_ += 1; | |
| if (opcode == L3L2QueueOpcode::STOP) { | |
| stop_observed_ = true; | |
| if (parent_->input_tail_ != input_acquire_) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | |
| "input descriptor published after STOP" | |
| ); | |
| return false; | |
| } | |
| } | |
| return true; | |
| L3L2QueueOpcode opcode = static_cast<L3L2QueueOpcode>(slot.opcode); | |
| if (!l3_l2_queue_valid_opcode(opcode)) { | |
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "invalid input opcode"); | |
| return false; | |
| } | |
| if (opcode == L3L2QueueOpcode::STOP && (slot.payload_offset != 0 || slot.payload_nbytes != 0)) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "STOP descriptor must be zero-byte" | |
| ); | |
| return false; | |
| } | |
| bool counts_against_window = opcode == L3L2QueueOpcode::DATA || opcode == L3L2QueueOpcode::ERROR; | |
| if (counts_against_window && active_non_stop_count_ >= parent_->config_.max_l2_input_inflight) { | |
| return false; | |
| } | |
| if (active_count_ >= entry_capacity_) { | |
| parent_->poison(L3L2QueueErrorKind::OWNERSHIP, "input.try_peek", "input window state full"); | |
| return false; | |
| } | |
| L3L2OrchPayloadView view{0, 0}; | |
| if (slot.payload_nbytes == 0) { | |
| if (slot.payload_offset != 0) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | |
| "zero-byte descriptor uses nonzero payload offset" | |
| ); | |
| return false; | |
| } | |
| } else if (!parent_->payload_in_arena( | |
| slot.payload_offset, slot.payload_nbytes, parent_->layout_.input_arena_offset, | |
| parent_->layout_.input_arena_bytes | |
| )) { | |
| parent_->poison(L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", "input payload out of arena"); | |
| return false; | |
| } else if (!parent_->payload_matches_head( | |
| parent_->input_payload_head_, slot.payload_offset, slot.payload_nbytes, | |
| parent_->layout_.input_arena_offset, parent_->layout_.input_arena_bytes, "input.try_peek" | |
| )) { | |
| return false; | |
| } else if (!parent_->endpoint_.payload_read(slot.payload_offset, slot.payload_nbytes, &view)) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::ENDPOINT_ERROR, "input.try_peek", parent_->endpoint_.error().message | |
| ); | |
| return false; | |
| } | |
| *out = L3L2QueueInputHandle{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view}; | |
| active_entries_[active_count_] = | |
| ActiveInputEntry{slot.seq, opcode, slot.payload_offset, slot.payload_nbytes, view, false}; | |
| active_count_ += 1; | |
| if (counts_against_window) { | |
| active_non_stop_count_ += 1; | |
| } | |
| input_acquire_ += 1; | |
| if (opcode == L3L2QueueOpcode::STOP) { | |
| stop_observed_ = true; | |
| if (parent_->input_tail_ != input_acquire_) { | |
| parent_->poison( | |
| L3L2QueueErrorKind::INVALID_DESCRIPTOR, "input.try_peek", | |
| "input descriptor published after STOP" | |
| ); | |
| return false; | |
| } | |
| } | |
| return true; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/common/platform/include/aicpu/l3_l2_message_queue.h` around lines 360 -
419, The input peek path in the queue handling logic accepts STOP descriptors
with payload bytes, which should be rejected by the STOP contract. Update the
`try_peek` flow in the `L3L2Queue` input handling to validate `opcode ==
L3L2QueueOpcode::STOP` before any payload read, and poison with
`L3L2QueueErrorKind::INVALID_DESCRIPTOR` if `slot.payload_nbytes` is nonzero.
Ensure `payload_matches_head`, `endpoint_.payload_read`, and the
`active_entries_` / `input_payload_head_` bookkeeping are only reached for valid
zero-byte STOP descriptors.
ccyywwen
force-pushed
the
l3-l2-message-queue-input-window
branch
3 times, most recently
from
cd1b527 to
5904101
Compare
jvjhfhg
force-pushed
the
l3-l2-message-queue-input-window
branch
from
5904101 to
cf44edb
Compare
ccyywwen
force-pushed
the
l3-l2-message-queue-input-window
branch
2 times, most recently
from
0cfe6dc to
3e290c4
Compare
- Template L3L2QueueEndpoint on MaxInflight and remove runtime config - Move active input tracking to a fixed ring with STOP extra capacity - Unify L3-L2 helper namespaces, error metadata, cache, and time helpers - Link public AICPU helper sources into JIT orchestration builds - Update docs, examples, and C++ UT coverage
ccyywwen
force-pushed
the
l3-l2-message-queue-input-window
branch
from
a0b02b4 to
f5f6473
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
L3L2QueueEndpointConfig{.max_l2_input_inflight = N}.queue.input().drained().tensormap_and_ringbuffer/l3_l2_message_queueexample, using PTO-ISA AIV tile compute while L2 keeps a persistent receive/process/publish loop.