Fix completion-hang deadlock so all clients report BENCHMARK_DONE

Summary:
The distributed run never finalized cleanly: warmup completed (`warmup_done=N`) but `benchmark_done` stalled below `num_clients` (e.g. 13/16). Because the admin server only broadcasts `ALL_DONE` once `benchmarkCompleteClients_ >= numExpectedClients_`, the missing clients block EVERY client forever at `waitForNotification(0)`, so the run is torn down with no clean client-side results (only server-aggregated metrics survive).

Root cause is a `google::log_mutex`/stdio deadlock, not a coroutine hang. `gstack` on a stuck client showed 48 of 189 threads blocked forever (`futex abstime=0x0`) on `google::log_mutex` inside `LogMessage::Flush()`; the lock holder was blocked in `__libc_write(fd=1)` to stdout, its buffer full of `Benchmark GET error (sample 1/1000): mc_res_tko` repeated. The stdout/stderr pipe to benchpress/automark was backpressured, so `write()` blocked while holding the stdio FILE lock and glog's `log_mutex`, wedging every logging thread including the coro worker event-base threads -> `mainScope.joinAsync()` never returns -> `BENCHMARK_DONE` is never sent. Under the `mc_res_tko` storm at multi-M QPS, even 1-in-1000 error sampling is thousands of synchronous prints/sec, which is enough to back the pipe up.

Fix: the four hot-path sampled error prints now fire only on the FIRST error per worker (`errors == 1` instead of `errors % 1000 == 1`) -> at most one line per worker for the whole run, flood-proof, while keeping the error-type diagnostic (still gated by `--verbose`). The error counts themselves are unchanged and still reported at the end.

This change also folds in the related straggler-drain robustness the completion path needs: per-request `co_await` now uses `folly::makePromiseContract` + `SemiFuture::within(10s)` (global-timekeeper based, abandons the underlying future so the await always completes even if the McRouter callback never fires), and the warmup/measurement worker scopes use `CancellableAsyncScope::cancelAndJoinAsync()` to cancel rather than wait on requests still outstanding at window-end.

This entire diff stack contributed to the final result.

Differential Revision: D110091576

Author: charles-typ

packages/ucache_bench/client/UcacheBenchClient.cpp

@@ -28,6 +28,8 @@
 #include <folly/coro/Sleep.h>
 #include <folly/coro/Timeout.h>
 #include <folly/fibers/FiberManagerMap.h>
+#include <folly/futures/Future.h>
+#include <folly/futures/Promise.h>
 #include <folly/io/async/ScopedEventBaseThread.h>
 #include <folly/portability/GFlags.h>
 #include <mcrouter/McrouterFiberContext.h>
@@ -993,8 +995,7 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
         request.value_ref() = *folly::IOBuf::copyBuffer(value);
         request.exptime_ref() = 3600;
 
-        auto [promise, future] =
-            folly::coro::makePromiseContract<UcbSetReply>();
+        auto [promise, future] = folly::makePromiseContract<UcbSetReply>();
 
         clientPtr->send(
             request,
@@ -1004,8 +1005,7 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
             });
 
         try {
-          co_await folly::coro::timeout(
-              std::move(future), std::chrono::seconds(10));
+          co_await std::move(future).within(std::chrono::seconds(10));
           scanOps++;
         } catch (const std::exception&) {
           // Request timed out or failed (e.g. send() returned false and the
@@ -1106,7 +1106,9 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
   auto warmupWorker =
       [&](memcache::mcrouter::CarbonRouterClient<UcacheBenchRouterInfo>*
               clientPtr) -> folly::coro::Task<void> {
-    folly::coro::AsyncScope scope;
+    // CancellableAsyncScope so stragglers at warmup-end are cancelled, not
+    // waited on forever (same hang as the measurement worker — see below).
+    folly::coro::CancellableAsyncScope scope;
     auto exe = co_await folly::coro::co_current_executor;
 
     std::atomic<uint64_t> inflight{0};
@@ -1140,7 +1142,7 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
       }
 
       // Same pattern as production McrouterAdapter::coro()
-      auto [promise, future] = folly::coro::makePromiseContract<UcbSetReply>();
+      auto [promise, future] = folly::makePromiseContract<UcbSetReply>();
 
       clientPtr->send(
           request,
@@ -1151,8 +1153,7 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
 
       UcbSetReply result;
       try {
-        result = co_await folly::coro::timeout(
-            std::move(future), std::chrono::seconds(10));
+        result = co_await std::move(future).within(std::chrono::seconds(10));
       } catch (const std::exception&) {
         // Request never completed (send() rejected without callback, or stuck
         // pre-send during the connection storm). Count as error and return so
@@ -1168,11 +1169,15 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
         localSuccesses++;
       } else {
         localErrors++;
-        // Rate-limit verbose error output to avoid flooding stdout pipe
-        // which can deadlock when run through benchpress/automark
-        if (FLAGS_verbose && (localErrors.load() % 1000 == 1)) {
+        // Print only the FIRST error per worker. Any periodic sampling (even
+        // 1/1000) floods stdout under a TKO storm at multi-M QPS; the
+        // synchronous write() blocks on the backpressured benchpress/automark
+        // pipe while holding the stdio FILE lock and glog's log_mutex, wedging
+        // every worker thread -> joinAsync never returns -> BENCHMARK_DONE is
+        // never sent.
+        if (FLAGS_verbose && localErrors.load() == 1) {
           printf(
-              "Warmup SET error (sample 1/1000): %s\n",
+              "Warmup SET error (first sample): %s\n",
               carbon::resultToString(*result.result_ref()));
         }
       }
@@ -1223,7 +1228,8 @@ UcacheBenchClient::WarmupResults UcacheBenchClient::warmup() {
       co_await folly::futures::sleep(std::chrono::milliseconds(1));
     }
 
-    co_await scope.joinAsync();
+    // Warmup window over: cancel stragglers and join (deterministic drain).
+    co_await scope.cancelAndJoinAsync();
 
     // Final update - add any remaining counts not yet synced
     uint64_t finalSuccesses = localSuccesses.load();
@@ -1627,7 +1633,14 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
       [&](size_t workerId,
           memcache::mcrouter::CarbonRouterClient<UcacheBenchRouterInfo>*
               clientPtr) -> folly::coro::Task<void> {
-    folly::coro::AsyncScope scope;
+    // CancellableAsyncScope (not AsyncScope) so that when the measurement
+    // window ends we CANCEL any still-outstanding requests instead of waiting
+    // for them indefinitely. A small number of in-flight requests at endTime
+    // can have responses that never arrive (and the per-request timeout's timer
+    // may sit on a momentarily-saturated proxy event base), which otherwise
+    // hangs joinAsync forever -> the client never sends BENCHMARK_DONE -> the
+    // whole run stalls. Cancelling at window-end is the correct semantics.
+    folly::coro::CancellableAsyncScope scope;
     auto exe = co_await folly::coro::co_current_executor;
 
     // Send one GET request - matches production McrouterAdapter::coro() pattern
@@ -1650,7 +1663,7 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
       }
 
       // Same pattern as production McrouterAdapter::coro()
-      auto [promise, future] = folly::coro::makePromiseContract<UcbGetReply>();
+      auto [promise, future] = folly::makePromiseContract<UcbGetReply>();
 
       clientPtr->send(
           request,
@@ -1661,8 +1674,7 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
 
       UcbGetReply result;
       try {
-        result = co_await folly::coro::timeout(
-            std::move(future), std::chrono::seconds(10));
+        result = co_await std::move(future).within(std::chrono::seconds(10));
       } catch (const std::exception&) {
         // Request never completed; count as a GET error and return so the
         // measurement worker's scope.joinAsync() can't hang.
@@ -1711,7 +1723,7 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
         }
 
         auto [setPromise, setFuture] =
-            folly::coro::makePromiseContract<UcbSetReply>();
+            folly::makePromiseContract<UcbSetReply>();
 
         clientPtr->send(
             setRequest,
@@ -1722,8 +1734,8 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
 
         UcbSetReply setResult;
         try {
-          setResult = co_await folly::coro::timeout(
-              std::move(setFuture), std::chrono::seconds(10));
+          setResult =
+              co_await std::move(setFuture).within(std::chrono::seconds(10));
         } catch (const std::exception&) {
           workerSetOps[workerId]->fetch_add(1);
           workerSetErrors[workerId]->fetch_add(1);
@@ -1735,18 +1747,17 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
           workerSetSuccesses[workerId]->fetch_add(1);
         } else {
           workerSetErrors[workerId]->fetch_add(1);
-          if (FLAGS_verbose &&
-              (workerSetErrors[workerId]->load() % 1000 == 1)) {
+          if (FLAGS_verbose && workerSetErrors[workerId]->load() == 1) {
             printf(
-                "Benchmark SET error (on GET miss, sample 1/1000): %s\n",
+                "Benchmark SET error (on GET miss, first sample): %s\n",
                 carbon::resultToString(*setResult.result_ref()));
           }
         }
       } else {
         workerGetErrors[workerId]->fetch_add(1);
-        if (FLAGS_verbose && (workerGetErrors[workerId]->load() % 1000 == 1)) {
+        if (FLAGS_verbose && workerGetErrors[workerId]->load() == 1) {
           printf(
-              "Benchmark GET error (sample 1/1000): %s\n",
+              "Benchmark GET error (first sample): %s\n",
               carbon::resultToString(*result.result_ref()));
         }
       }
@@ -1777,7 +1788,7 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
       }
 
       // Same pattern as production McrouterAdapter::coro()
-      auto [promise, future] = folly::coro::makePromiseContract<UcbSetReply>();
+      auto [promise, future] = folly::makePromiseContract<UcbSetReply>();
 
       clientPtr->send(
           request,
@@ -1788,8 +1799,7 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
 
       UcbSetReply result;
       try {
-        result = co_await folly::coro::timeout(
-            std::move(future), std::chrono::seconds(10));
+        result = co_await std::move(future).within(std::chrono::seconds(10));
       } catch (const std::exception&) {
         workerTotalOps[workerId]->fetch_add(1);
         workerSetOps[workerId]->fetch_add(1);
@@ -1814,9 +1824,9 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
         workerSetSuccesses[workerId]->fetch_add(1);
       } else {
         workerSetErrors[workerId]->fetch_add(1);
-        if (FLAGS_verbose && (workerSetErrors[workerId]->load() % 1000 == 1)) {
+        if (FLAGS_verbose && workerSetErrors[workerId]->load() == 1) {
           printf(
-              "Benchmark SET error (sample 1/1000): %s\n",
+              "Benchmark SET error (first sample): %s\n",
               carbon::resultToString(*result.result_ref()));
         }
       }
@@ -1867,7 +1877,9 @@ UcacheBenchClient::BenchmarkResults UcacheBenchClient::runBenchmark() {
       co_await folly::futures::sleep(std::chrono::milliseconds(1));
     }
 
-    co_await scope.joinAsync();
+    // Measurement window is over: cancel any still-outstanding requests and
+    // join. This drains deterministically instead of hanging on stragglers.
+    co_await scope.cancelAndJoinAsync();
 
     // No need to sync to global counters - they'll be summed at the end
     co_return;