RFC: run loop finish forward progress

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)

test/run_loop_finish_repro.cpp

@@ -0,0 +1,40 @@
+#include <stdexec/execution.hpp>
+#include <iostream>
+
+#include <thread>
+
+void f(){
+	stdexec::run_loop loop;
+
+	std::thread worker([&loop] {
+		loop.run();
+	});
+
+	// Ensure run() is actively servicing the loop before initiating
+	// shutdown from this thread.
+	if (!stdexec::sync_wait(
+		stdexec::schedule(loop.get_scheduler()) |  stdexec::then([] (){})))
+	{
+		loop.finish();
+		worker.join();
+	}
+
+	loop.finish();
+	worker.join();
+
+	std::cout << ".";
+}
+
+int main()
+{
+	// Repeat the normal run/finish lifecycle to make the forward-progress
+	// failure easy to reproduce under an unfair scheduler such as Valgrind's
+	// default scheduler.
+	for (int i = 0; i < 1000; ++i)
+	{
+		f();
+	}
+	std::cout << std::endl;
+
+	return 0;
+}