RFC: run_loop: yield while waiting for in-flight tasks

The run loop drains queued work after finish() has been requested and also
waits for __task_count_ to reach zero. If the queue becomes empty while the
count is still non-zero, the current implementation busy-spins by repeatedly
calling __execute_all().

This can starve another thread that still needs to make progress in order to
decrement __task_count_. One concrete sequence is finish() incrementing the
count by two, setting __finishing_, and queueing the noop task. The worker can
then execute the noop, leaving the count at one, and immediately enter the
tight drain loop. The caller still needs to execute its final fetch_sub() to
bring the count to zero, but an unfair scheduler may keep running the spinning
worker instead.

This was reproducible under Valgrind's default scheduler as an intermittent
hang during run_loop shutdown. Using Valgrind's fair scheduler avoided the
hang, which pointed at a forward-progress issue rather than missing work.
Yielding when there is no queued work but tasks are still in flight allows the
thread responsible for completing those tasks to run.

With this change, the reproducer completed 700 consecutive runs under
Valgrind without --fair-sched=yes.

Co-Authored-By: GPT 5.6 Sol
Signed-off-by: Alexander Hansen <alexander.hansen@9elements.com>

Author: pointbazaar

include/stdexec/__detail/__run_loop.hpp

@@ -69,8 +69,21 @@ namespace STDEXEC
         }
         // drain the queue, taking care to execute any tasks that get added while
         // executing the remaining tasks (also wait for other tasks that might still be in flight):
-        while (__execute_all() || __task_count_.load(__std::memory_order_acquire) > 0)
-          ;
+        while (true)
+        {
+          if (__execute_all())
+          {
+            continue;
+          }
+
+          if (__task_count_.load(__std::memory_order_acquire) == 0)
+          {
+            break;
+          }
+
+          // Another thread still has work in flight. Let it make progress.
+          std::this_thread::yield();
+        }
       }
 
       STDEXEC_ATTRIBUTE(host, device)