[Bug] CONSUMED ring head can remain pinned when `advance_lock` contention drops the reclaim advance

### Platform

a2a3 (Ascend 910B/C hardware)

### Runtime Variant

tensormap_and_ringbuffer

### Description


A successful `COMPLETED -> CONSUMED` transition can lose the `last_task_alive`
advance when another scheduler thread holds `RingSchedState::advance_lock`.

The old scheduler path performs the consume transition and then only tries to
advance the ring head:

```text
check_and_handle_consumed(slot)
  fanout_refcount == fanout_count
  task_state: COMPLETED -> CONSUMED
  try_lock(advance_lock)
    success: advance_ring_pointers()
    failure: return
```

The failure path assumes that the current `advance_lock` owner will also scan the
task that just became `CONSUMED`. That assumption is not guaranteed.

A valid interleaving is:

```text
1. last_task_alive points at task H.
2. Thread A holds advance_lock and is advancing the same ring.
3. Thread A has already scanned the prefix and stopped before H because H was
   still COMPLETED at that moment.
4. Thread B changes H from COMPLETED to CONSUMED.
5. Thread B fails the old try-lock because Thread A still holds advance_lock.
6. Thread B returns without publishing another advance.
7. Thread A releases advance_lock, but it does not rescan H after Thread B's
   CONSUMED store.
```

The final state is inconsistent with the reclaim contract:

```text
H.task_state == CONSUMED
ring.fc.last_task_alive <= H
```

`PTO2_TASK_CONSUMED` means the task output is fully consumed and its buffers can
be released. Keeping a contiguous `CONSUMED` head in the published active range
is therefore not in-order reclaim semantics; it is a lost watermark advance.

This can pin FIFO task/heap reclaim even though the descriptor at the head is
already reclaimable. If no later transition happens to advance the same ring,
the orchestrator can keep observing stale `last_task_alive` and remain under
artificial heap/task-window backpressure.


### Steps to Reproduce

```markdown
1. Apply the C++ unit-test patch from **Complete UT patch** below to upstream
   `hw-native-sys/simpler` `origin/main`.

The reproducer was checked against upstream at:


f293be3bb2df638d9245e9f90991722aabc7a4ee Add: export Worker from the simpler package root (#1535)


2. Build the C++ unit tests:


cmake -B tests/ut/cpp/build -S tests/ut/cpp
cmake --build tests/ut/cpp/build --parallel 4


3. Run the a2a3 scheduler-state unit tests:


./tests/ut/cpp/build/test_scheduler_state \
  --gtest_filter=SchedulerStateTest.ConsumedHeadAdvancesAfterContendedAdvanceLock:SchedulerStateTest.ContendedConsumedHeadAdvanceStress


4. Run the a5 scheduler-state unit tests:


./tests/ut/cpp/build/test_a5_scheduler_state \
  --gtest_filter=SchedulerStateTest.ConsumedHeadAdvancesAfterContendedAdvanceLock:SchedulerStateTest.ContendedConsumedHeadAdvanceStress


Against the current upstream try-lock implementation, both architectures fail.
The deterministic test leaves `ring.fc.last_task_alive` at the consumed head.
The stress test enters a real contended advance window and observes a stale
watermark after the lock owner releases `advance_lock`.
```

### Expected Behavior


After a task successfully transitions to `PTO2_TASK_CONSUMED`, the scheduler
must guarantee a ring-advance pass that observes that transition after any
current `advance_lock` owner releases the lock.

The scheduler must not publish a stale active watermark over a contiguous
`CONSUMED` prefix:

```text
if ring slot H is at last_task_alive
and H.task_state == PTO2_TASK_CONSUMED
then a later ring advance must publish last_task_alive > H
```

This expected behavior does not change in-order reclaim semantics. The scheduler
still only advances across `CONSUMED` tasks and still must not skip a live task
with pending consumers.


### Actual Behavior


The old try-lock path can return immediately after a successful consume
transition:

```cpp
int32_t expected_lock = 0;
if (ring_sched_states[ring_id].advance_lock.compare_exchange_strong(
        expected_lock, 1, std::memory_order_acquire, std::memory_order_relaxed
    )) {
    ring_sched_states[ring_id].advance_ring_pointers();
    ring_sched_states[ring_id].advance_lock.store(0, std::memory_order_release);
}
```

If the lock owner has already scanned past the old state, no thread is guaranteed
to publish the new `CONSUMED` prefix. The deterministic test forces exactly this
case:

```text
advance_lock is held
head transitions to CONSUMED
old try-lock fails and returns
advance_lock is released later
last_task_alive remains pinned at the consumed head
```

A host-side C++ unit-test run against the current upstream implementation
produces the following a2a3 result:

```text
SchedulerStateTest.ConsumedHeadAdvancesAfterContendedAdvanceLock
  ring.fc.last_task_alive.load(std::memory_order_acquire) == 0
  expected 1

SchedulerStateTest.ContendedConsumedHeadAdvanceStress
  stale_watermark == 1
  expected 0

a2a3_rc=1
```

The a5 result:

```text
SchedulerStateTest.ConsumedHeadAdvancesAfterContendedAdvanceLock
  ring.fc.last_task_alive.load(std::memory_order_acquire) == 0
  expected 1

SchedulerStateTest.ContendedConsumedHeadAdvanceStress
  stale_watermark == 1
  expected 0

a5_rc=1
```


### Git Commit ID

f293be3bb2df638d9245e9f90991722aabc7a4ee

### CANN Version

9.0.0

### Driver Version

26.0.rc1

### Host Platform

Linux (aarch64)

### Additional Context


This issue only concerns the state after the head task has already become
`PTO2_TASK_CONSUMED`. There is no request to reclaim a live task, skip an
unconsumed head, or introduce out-of-order heap freeing. The only requirement is
to publish the already-safe in-order reclaim advance after lock contention
clears.

A possible upstream rejection reason would be:

```text
If try-lock fails, another thread is advancing the ring and will scan this task.
```

That is not a valid invariant. It only holds if the lock owner's scan happens
after the `CONSUMED` store. If the owner already scanned the head while it was
still `COMPLETED`, the old try-lock loser can be the only thread that knows a
new consumed head exists. Returning without a guaranteed follow-up scan is a lost
wakeup.

The deterministic UT proves the minimal contract violation by manually holding
`advance_lock` until the head has transitioned to `CONSUMED`. The stress UT
proves a more realistic interleaving: another scheduler thread enters
`advance_ring_pointers()` through a follower task, starts resetting an already
consumed prefix, and holds `advance_lock`; the blocked head becomes `CONSUMED`
after that owner's scan has already stopped before it. In both cases, the final
published `last_task_alive` remains stale.

This is not just a one-cycle delay. In this runtime, ring reclamation is
event-triggered by successful consume transitions; there is no background
reclaim thread, no `advance_pending` flag, no lock-release callback, and no
periodic scheduler-side scan that rechecks a stale `CONSUMED` head. The only
code path that publishes a newer shared-memory watermark is
`advance_ring_pointers() -> sync_to_sm()`, and the old implementation calls that
path only when the consumer thread wins the try-lock.

Allocator/orchestrator backpressure then observes only the published
`ring.fc.last_task_alive`, not the raw level state of every slot:

- `PTO2TaskAllocator::alloc()` waits for `last_task_alive`, uses it to update
  the heap tail, and reports heap/task-ring deadlock if the watermark does not
  advance.
- `PTO2FaninPool::ensure_space()` and `PTO2DepListPool::ensure_space()` reclaim
  from the same shared `last_task_alive` and diagnose a stuck watermark when it
  does not move.
- TensorMap cleanup is also driven from the orchestrator's read of
  `ring.fc.last_task_alive`.

Therefore the missed try-lock can be the last event that could have published
the reclaimable head. A later task on the same ring may happen to become
`CONSUMED` and successfully run `advance_ring_pointers()`, which would scan from
the stale head and repair the watermark. But that is only opportunistic. If the
stale head is already causing task-window, heap, fanin/dep-pool, or TensorMap
backpressure, the orchestrator may stop submitting the later task that would
accidentally trigger another advance. In that case `H.task_state == CONSUMED`
remains true, but `ring.fc.last_task_alive` can stay pinned at `H` until a
deadlock backstop fires or an unrelated later event perturbs the ring.

### Suggested fix

The correctness requirement is that a successful `COMPLETED -> CONSUMED`
transition must guarantee that some ring-advance pass observes that transition
after any current `advance_lock` owner releases the lock.

The minimal implementation is to make the thread that performed the successful
consume transition wait for `advance_lock` and then run `advance_ring_pointers()`
once. This preserves the existing in-order scan. It does not release any task
that is not already `CONSUMED`.

An equivalent handoff implementation would also be valid, for example a
per-ring `advance_pending` flag that the lock owner drains before unlocking.
The important part is that a failed try-lock cannot silently drop the only
follow-up scan for a newly consumed head.

<details>
<summary>Show scheduler fix shape</summary>

```diff
diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h
index ec79a388..612479e9 100644
--- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h
+++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h
@@ -526,6 +526,39 @@ struct PTO2SchedulerState {
     // Inline hot-path methods
     // =========================================================================
 
+    void advance_ring_after_consumed(int32_t ring_id) {
+        auto &ring_sched = ring_sched_states[ring_id];
+        int32_t expected_lock = 0;
+        while (!ring_sched.advance_lock.compare_exchange_weak(
+            expected_lock, 1, std::memory_order_acquire, std::memory_order_relaxed
+        )) {
+            expected_lock = 0;
+            SPIN_WAIT_HINT();
+        }
+        ring_sched.advance_ring_pointers();
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+    }
+
+#if SIMPLER_ORCH_PROFILING || SIMPLER_SCHED_PROFILING
+    void advance_ring_after_consumed(int32_t ring_id, uint64_t &atomic_count) {
+        auto &ring_sched = ring_sched_states[ring_id];
+        uint64_t ops = 0;
+        int32_t expected_lock = 0;
+        while (!ring_sched.advance_lock.compare_exchange_weak(
+            expected_lock, 1, std::memory_order_acquire, std::memory_order_relaxed
+        )) {
+            ops++;
+            expected_lock = 0;
+            SPIN_WAIT_HINT();
+        }
+        ops++;
+        ring_sched.advance_ring_pointers();
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+        ops++;
+        atomic_count += ops;
+    }
+#endif
+
     void check_and_handle_consumed(PTO2TaskSlotState &slot_state) {
         bool became_consumed = false;
         slot_state.lock_fanout();
@@ -581,17 +614,8 @@ struct PTO2SchedulerState {
 #endif
 
         int32_t ring_id = slot_state.ring_id;
-        int32_t expected_lock = 0;
-        if (ring_sched_states[ring_id].advance_lock.compare_exchange_strong(
-                expected_lock, 1, std::memory_order_acquire, std::memory_order_relaxed
-            )) {
-            ring_sched_states[ring_id].advance_ring_pointers();
-            ring_sched_states[ring_id].advance_lock.store(0, std::memory_order_release);
-        }
+        advance_ring_after_consumed(ring_id);
     }
 
 #if SIMPLER_ORCH_PROFILING || SIMPLER_SCHED_PROFILING
@@ -621,18 +645,8 @@ struct PTO2SchedulerState {
 
         int32_t ring_id = slot_state.ring_id;
-        int32_t expected_lock = 0;
-        if (ring_sched_states[ring_id].advance_lock.compare_exchange_strong(
-                expected_lock, 1, std::memory_order_acquire, std::memory_order_relaxed
-            )) {
-            ring_sched_states[ring_id].advance_ring_pointers();
-            ring_sched_states[ring_id].advance_lock.store(0, std::memory_order_release);
-            atomic_count += 2;
-        } else {
-            atomic_count += 1;
-        }
+        advance_ring_after_consumed(ring_id, atomic_count);
     }
 #endif
```

The same fix shape applies to the a5 scheduler path.

</details>

### Complete UT patch

The following test-only patch adds coverage to both
`tests/ut/cpp/a2a3/test_scheduler_state.cpp` and
`tests/ut/cpp/a5/test_scheduler_state.cpp`:

- `SchedulerStateTest.ConsumedHeadAdvancesAfterContendedAdvanceLock`
- `SchedulerStateTest.ContendedConsumedHeadAdvanceStress`

<details>
<summary>Full test-only diff</summary>

```diff
 tests/ut/cpp/a2a3/test_scheduler_state.cpp | 129 +++++++++++++++++++++++++++++
 tests/ut/cpp/a5/test_scheduler_state.cpp   | 129 +++++++++++++++++++++++++++++
 2 files changed, 258 insertions(+)

diff --git a/tests/ut/cpp/a2a3/test_scheduler_state.cpp b/tests/ut/cpp/a2a3/test_scheduler_state.cpp
index e00ce44d..b2af4974 100644
--- a/tests/ut/cpp/a2a3/test_scheduler_state.cpp
+++ b/tests/ut/cpp/a2a3/test_scheduler_state.cpp
@@ -17,7 +17,9 @@
 #include <gtest/gtest.h>
 
 #include <atomic>
+#include <chrono>
 #include <cstring>
+#include <thread>
 
 #include "utils/device_arena.h"
 #include "scheduler/scheduler_types.h"
@@ -99,6 +101,48 @@ protected:
         memset(&slot_pl, 0, sizeof(slot_pl));
         slot.payload = &slot_pl;
     }
+
+    void init_ring_slot(
+        PTO2SharedMemoryRingHeader &ring, int32_t local_id, PTO2TaskState state, uint8_t ring_id,
+        uint32_t fanout_count = 1, uint32_t fanout_refcount = 1
+    ) {
+        PTO2TaskSlotState &slot = ring.get_slot_state_by_task_id(local_id);
+        memset(&slot, 0, sizeof(slot));
+        slot.task_state.store(state, std::memory_order_relaxed);
+        slot.fanin_count = 0;
+        slot.fanin_refcount.store(0, std::memory_order_relaxed);
+        slot.fanout_count = fanout_count;
+        slot.fanout_refcount.store(fanout_refcount, std::memory_order_relaxed);
+        slot.fanout_lock.store(0, std::memory_order_relaxed);
+        slot.fanout_head = nullptr;
+        slot.ring_id = ring_id;
+        slot.active_mask = ActiveMask(PTO2_SUBTASK_MASK_AIC);
+        slot.completed_subtasks.store(0, std::memory_order_relaxed);
+        slot.total_required_subtasks = 1;
+        slot.logical_block_num = 1;
+        slot.lifecycle_flags.store(PTO2_COMPLETION_DONE, std::memory_order_relaxed);
+        PTO2TaskPayload &payload = ring.get_payload_by_task_id(local_id);
+        PTO2TaskDescriptor &task = ring.get_task_by_task_id(local_id);
+        memset(&payload, 0, sizeof(payload));
+        memset(&task, 0, sizeof(task));
+        slot.payload = &payload;
+        slot.task = &task;
+    }
+
+    void setup_ring_for_reclaim_race(int32_t ring_id, int32_t current_task_index, int32_t blocked_task_id) {
+        PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+        PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+
+        ring.fc.current_task_index.store(current_task_index, std::memory_order_release);
+        ring.fc.last_task_alive.store(0, std::memory_order_release);
+        ring_sched.last_task_alive = 0;
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+
+        for (int32_t task_id = 0; task_id < current_task_index; task_id++) {
+            PTO2TaskState state = task_id < blocked_task_id ? PTO2_TASK_CONSUMED : PTO2_TASK_COMPLETED;
+            init_ring_slot(ring, task_id, state, static_cast<uint8_t>(ring_id));
+        }
+    }
 };
 
 // =============================================================================
@@ -123,6 +167,91 @@ TEST_F(SchedulerStateTest, ConsumedTransition) {
     EXPECT_EQ(slot.task_state.load(), PTO2_TASK_CONSUMED);
 }
 
+TEST_F(SchedulerStateTest, ConsumedHeadAdvancesAfterContendedAdvanceLock) {
+    constexpr int32_t ring_id = PTO2_MAX_RING_DEPTH - 1;
+    constexpr int32_t head_task_id = 0;
+    setup_ring_for_reclaim_race(ring_id, /*current_task_index=*/1, head_task_id);
+
+    PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+    PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+    PTO2TaskSlotState &head = ring.get_slot_state_by_task_id(head_task_id);
+
+    ring_sched.advance_lock.store(1, std::memory_order_release);
+    std::thread unlocker([&]() {
+        while (head.task_state.load(std::memory_order_acquire) != PTO2_TASK_CONSUMED) {
+            std::this_thread::yield();
+        }
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+    });
+
+    sched.check_and_handle_consumed(head);
+    unlocker.join();
+
+    EXPECT_EQ(head.task_state.load(std::memory_order_acquire), PTO2_TASK_CONSUMED);
+    EXPECT_EQ(ring.fc.last_task_alive.load(std::memory_order_acquire), 1)
+        << "a CONSUMED ring head must not remain pinned after advance_lock contention clears";
+}
+
+TEST_F(SchedulerStateTest, ContendedConsumedHeadAdvanceStress) {
+    constexpr int32_t ring_id = PTO2_MAX_RING_DEPTH - 1;
+    constexpr int32_t blocked_task_id = PTO2_TASK_WINDOW_SIZE - 384;
+    constexpr int32_t follower_task_id = blocked_task_id + 1;
+    constexpr int32_t current_task_index = follower_task_id + 1;
+    constexpr int32_t attempts = 64;
+
+    int observed_race_window = 0;
+    int stale_watermark = 0;
+
+    for (int32_t attempt = 0; attempt < attempts && stale_watermark == 0; attempt++) {
+        setup_ring_for_reclaim_race(ring_id, current_task_index, blocked_task_id);
+
+        PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+        PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+        PTO2TaskSlotState &first_prefix = ring.get_slot_state_by_task_id(0);
+        PTO2TaskSlotState &blocked_head = ring.get_slot_state_by_task_id(blocked_task_id);
+        PTO2TaskSlotState &follower = ring.get_slot_state_by_task_id(follower_task_id);
+        std::atomic<bool> start{false};
+
+        std::thread owner([&]() {
+            while (!start.load(std::memory_order_acquire)) {
+                std::this_thread::yield();
+            }
+            sched.check_and_handle_consumed(follower);
+        });
+
+        start.store(true, std::memory_order_release);
+
+        auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+        bool in_reset_window = false;
+        while (std::chrono::steady_clock::now() < deadline) {
+            bool owner_holds_lock = ring_sched.advance_lock.load(std::memory_order_acquire) != 0;
+            bool owner_started_reset =
+                first_prefix.lifecycle_flags.load(std::memory_order_acquire) == PTO2_LIFECYCLE_FLAGS_NONE;
+            if (owner_holds_lock && owner_started_reset) {
+                in_reset_window = true;
+                break;
+            }
+        }
+
+        if (in_reset_window) {
+            observed_race_window++;
+        }
+        sched.check_and_handle_consumed(blocked_head);
+        owner.join();
+
+        int32_t published_last_alive = ring.fc.last_task_alive.load(std::memory_order_acquire);
+        if (blocked_head.task_state.load(std::memory_order_acquire) == PTO2_TASK_CONSUMED &&
+            published_last_alive <= blocked_task_id) {
+            stale_watermark++;
+        }
+    }
+
+    EXPECT_GT(observed_race_window, 0) << "stress did not enter the contested advance window";
+    EXPECT_EQ(
+        stale_watermark, 0
+    ) << "a consumed ring head was left behind after another thread's advance pass released the lock";
+}
+
 TEST_F(SchedulerStateTest, ConsumedNotCompletedState) {
     alignas(64) PTO2TaskSlotState slot;
     init_slot(slot, PTO2_TASK_PENDING, 1, 1);
diff --git a/tests/ut/cpp/a5/test_scheduler_state.cpp b/tests/ut/cpp/a5/test_scheduler_state.cpp
index 5a11e65f..5accc77e 100644
--- a/tests/ut/cpp/a5/test_scheduler_state.cpp
+++ b/tests/ut/cpp/a5/test_scheduler_state.cpp
@@ -17,7 +17,9 @@
 #include <gtest/gtest.h>
 
 #include <atomic>
+#include <chrono>
 #include <cstring>
+#include <thread>
 
 #include "utils/device_arena.h"
 #include "scheduler/scheduler_types.h"
@@ -99,6 +101,48 @@ protected:
         memset(&slot_pl, 0, sizeof(slot_pl));
         slot.payload = &slot_pl;
     }
+
+    void init_ring_slot(
+        PTO2SharedMemoryRingHeader &ring, int32_t local_id, PTO2TaskState state, uint8_t ring_id,
+        uint32_t fanout_count = 1, uint32_t fanout_refcount = 1
+    ) {
+        PTO2TaskSlotState &slot = ring.get_slot_state_by_task_id(local_id);
+        memset(&slot, 0, sizeof(slot));
+        slot.task_state.store(state, std::memory_order_relaxed);
+        slot.fanin_count = 0;
+        slot.fanin_refcount.store(0, std::memory_order_relaxed);
+        slot.fanout_count = fanout_count;
+        slot.fanout_refcount.store(fanout_refcount, std::memory_order_relaxed);
+        slot.fanout_lock.store(0, std::memory_order_relaxed);
+        slot.fanout_head = nullptr;
+        slot.ring_id = ring_id;
+        slot.active_mask = ActiveMask(PTO2_SUBTASK_MASK_AIC);
+        slot.completed_subtasks.store(0, std::memory_order_relaxed);
+        slot.total_required_subtasks = 1;
+        slot.logical_block_num = 1;
+        slot.lifecycle_flags.store(PTO2_COMPLETION_DONE, std::memory_order_relaxed);
+        PTO2TaskPayload &payload = ring.get_payload_by_task_id(local_id);
+        PTO2TaskDescriptor &task = ring.get_task_by_task_id(local_id);
+        memset(&payload, 0, sizeof(payload));
+        memset(&task, 0, sizeof(task));
+        slot.payload = &payload;
+        slot.task = &task;
+    }
+
+    void setup_ring_for_reclaim_race(int32_t ring_id, int32_t current_task_index, int32_t blocked_task_id) {
+        PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+        PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+
+        ring.fc.current_task_index.store(current_task_index, std::memory_order_release);
+        ring.fc.last_task_alive.store(0, std::memory_order_release);
+        ring_sched.last_task_alive = 0;
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+
+        for (int32_t task_id = 0; task_id < current_task_index; task_id++) {
+            PTO2TaskState state = task_id < blocked_task_id ? PTO2_TASK_CONSUMED : PTO2_TASK_COMPLETED;
+            init_ring_slot(ring, task_id, state, static_cast<uint8_t>(ring_id));
+        }
+    }
 };
 
 // =============================================================================
@@ -123,6 +167,91 @@ TEST_F(SchedulerStateTest, ConsumedTransition) {
     EXPECT_EQ(slot.task_state.load(), PTO2_TASK_CONSUMED);
 }
 
+TEST_F(SchedulerStateTest, ConsumedHeadAdvancesAfterContendedAdvanceLock) {
+    constexpr int32_t ring_id = PTO2_MAX_RING_DEPTH - 1;
+    constexpr int32_t head_task_id = 0;
+    setup_ring_for_reclaim_race(ring_id, /*current_task_index=*/1, head_task_id);
+
+    PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+    PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+    PTO2TaskSlotState &head = ring.get_slot_state_by_task_id(head_task_id);
+
+    ring_sched.advance_lock.store(1, std::memory_order_release);
+    std::thread unlocker([&]() {
+        while (head.task_state.load(std::memory_order_acquire) != PTO2_TASK_CONSUMED) {
+            std::this_thread::yield();
+        }
+        ring_sched.advance_lock.store(0, std::memory_order_release);
+    });
+
+    sched.check_and_handle_consumed(head);
+    unlocker.join();
+
+    EXPECT_EQ(head.task_state.load(std::memory_order_acquire), PTO2_TASK_CONSUMED);
+    EXPECT_EQ(ring.fc.last_task_alive.load(std::memory_order_acquire), 1)
+        << "a CONSUMED ring head must not remain pinned after advance_lock contention clears";
+}
+
+TEST_F(SchedulerStateTest, ContendedConsumedHeadAdvanceStress) {
+    constexpr int32_t ring_id = PTO2_MAX_RING_DEPTH - 1;
+    constexpr int32_t blocked_task_id = PTO2_TASK_WINDOW_SIZE - 384;
+    constexpr int32_t follower_task_id = blocked_task_id + 1;
+    constexpr int32_t current_task_index = follower_task_id + 1;
+    constexpr int32_t attempts = 64;
+
+    int observed_race_window = 0;
+    int stale_watermark = 0;
+
+    for (int32_t attempt = 0; attempt < attempts && stale_watermark == 0; attempt++) {
+        setup_ring_for_reclaim_race(ring_id, current_task_index, blocked_task_id);
+
+        PTO2SharedMemoryRingHeader &ring = sm_handle->header->rings[ring_id];
+        PTO2SchedulerState::RingSchedState &ring_sched = sched.ring_sched_states[ring_id];
+        PTO2TaskSlotState &first_prefix = ring.get_slot_state_by_task_id(0);
+        PTO2TaskSlotState &blocked_head = ring.get_slot_state_by_task_id(blocked_task_id);
+        PTO2TaskSlotState &follower = ring.get_slot_state_by_task_id(follower_task_id);
+        std::atomic<bool> start{false};
+
+        std::thread owner([&]() {
+            while (!start.load(std::memory_order_acquire)) {
+                std::this_thread::yield();
+            }
+            sched.check_and_handle_consumed(follower);
+        });
+
+        start.store(true, std::memory_order_release);
+
+        auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+        bool in_reset_window = false;
+        while (std::chrono::steady_clock::now() < deadline) {
+            bool owner_holds_lock = ring_sched.advance_lock.load(std::memory_order_acquire) != 0;
+            bool owner_started_reset =
+                first_prefix.lifecycle_flags.load(std::memory_order_acquire) == PTO2_LIFECYCLE_FLAGS_NONE;
+            if (owner_holds_lock && owner_started_reset) {
+                in_reset_window = true;
+                break;
+            }
+        }
+
+        if (in_reset_window) {
+            observed_race_window++;
+        }
+        sched.check_and_handle_consumed(blocked_head);
+        owner.join();
+
+        int32_t published_last_alive = ring.fc.last_task_alive.load(std::memory_order_acquire);
+        if (blocked_head.task_state.load(std::memory_order_acquire) == PTO2_TASK_CONSUMED &&
+            published_last_alive <= blocked_task_id) {
+            stale_watermark++;
+        }
+    }
+
+    EXPECT_GT(observed_race_window, 0) << "stress did not enter the contested advance window";
+    EXPECT_EQ(
+        stale_watermark, 0
+    ) << "a consumed ring head was left behind after another thread's advance pass released the lock";
+}
+
 TEST_F(SchedulerStateTest, ConsumedNotCompletedState) {
     alignas(64) PTO2TaskSlotState slot;
     init_slot(slot, PTO2_TASK_PENDING, 1, 1);
```

</details>

The deterministic test sets up one ring head at `COMPLETED` with
`fanout_refcount == fanout_count`, holds that ring's `advance_lock`, calls
`check_and_handle_consumed(head)`, and releases the lock only after observing the
head transition to `CONSUMED`. A correct implementation must publish
`last_task_alive == 1` after contention clears.

The stress test sets up a long consumed prefix, one blocked head, and one
follower. The follower thread enters the real `check_and_handle_consumed() ->
advance_ring_pointers()` path and holds `advance_lock` while resetting the
already-consumed prefix. The blocked head is then consumed while that owner is
past the old state. A correct implementation must not leave
`published_last_alive <= blocked_task_id` after the owner releases the lock.

## 中文简要说明

这个问题本质上是 `check_and_handle_consumed()` 在成功把任务从
`COMPLETED` 切到 `CONSUMED` 后，只做了一次 `advance_lock` try-lock。如果 try-lock
失败，代码假设当前 lock owner 会看到这个新 `CONSUMED` head 并推进
`last_task_alive`。这个假设不成立：lock owner 可能已经在 head 还是 `COMPLETED` 时扫过
并停下，随后另一个线程才把 head 写成 `CONSUMED`。

当前 upstream `f293be3b` 上，仅加入 UT 后，a2a3 和 a5 都稳定失败：

```text
deterministic: last_task_alive == 0, expected 1
stress: stale_watermark == 1, expected 0
```

这不是要求乱序回收，也不是要求跳过 live task；测试只要求已经连续
`CONSUMED` 的 ring head 在 `advance_lock` contention 清除后最终被发布为可回收。


Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[Bug] CONSUMED ring head can remain pinned when `advance_lock` contention drops the reclaim advance #1545

Platform

Runtime Variant

Description

Steps to Reproduce

Expected Behavior

Actual Behavior

Git Commit ID

CANN Version

Driver Version

Host Platform

Additional Context

Suggested fix

Complete UT patch

中文简要说明

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Uh oh!

[Bug] CONSUMED ring head can remain pinned when advance_lock contention drops the reclaim advance #1545

Description

Platform

Runtime Variant

Description

Steps to Reproduce

Expected Behavior

Actual Behavior

Git Commit ID

CANN Version

Driver Version

Host Platform

Additional Context

Suggested fix

Complete UT patch

中文简要说明

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

[Bug] CONSUMED ring head can remain pinned when `advance_lock` contention drops the reclaim advance #1545