|
|
@@ -117,18 +117,45 @@ func main() {
|
|
|
// Status ticker.
|
|
|
ticker := time.NewTicker(5 * time.Second)
|
|
|
defer ticker.Stop()
|
|
|
+
|
|
|
+ // done is closed when every worker has returned (e.g. when
|
|
|
+ // --duration expires in runWorker via prodCtx.Done()). Without
|
|
|
+ // this, main only exits on SIGTERM and the loadgen hangs past
|
|
|
+ // --duration even though the workers themselves shut down cleanly
|
|
|
+ // (producer closes ackCh, consumer's `for range ackCh` exits,
|
|
|
+ // runWorker returns). Symptom: smoke step 3 times out at 60s
|
|
|
+ // with "backpressure loadgen did not exit within 60s".
|
|
|
+ done := make(chan struct{})
|
|
|
+ go func() {
|
|
|
+ wg.Wait()
|
|
|
+ close(done)
|
|
|
+ }()
|
|
|
+
|
|
|
for {
|
|
|
select {
|
|
|
case <-ctx.Done():
|
|
|
- wg.Wait()
|
|
|
+ // Signal received: wait for in-flight workers to drain
|
|
|
+ // via the done channel, then print and return.
|
|
|
+ <-done
|
|
|
var tsent, tfailed, tdupes, trlHits uint64
|
|
|
- for i := range results {
|
|
|
- tsent += results[i].sent
|
|
|
- tfailed += results[i].failed
|
|
|
- tdupes += results[i].dupes
|
|
|
- trlHits += results[i].rlHits
|
|
|
- }
|
|
|
- printFinal(logger, tsent, tfailed, tdupes, trlHits, &rttHist)
|
|
|
+ for i := range results {
|
|
|
+ tsent += results[i].sent
|
|
|
+ tfailed += results[i].failed
|
|
|
+ tdupes += results[i].dupes
|
|
|
+ trlHits += results[i].rlHits
|
|
|
+ }
|
|
|
+ printFinal(logger, tsent, tfailed, tdupes, trlHits, &rttHist)
|
|
|
+ return
|
|
|
+ case <-done:
|
|
|
+ // All workers finished naturally (e.g. --duration expired).
|
|
|
+ var tsent, tfailed, tdupes, trlHits uint64
|
|
|
+ for i := range results {
|
|
|
+ tsent += results[i].sent
|
|
|
+ tfailed += results[i].failed
|
|
|
+ tdupes += results[i].dupes
|
|
|
+ trlHits += results[i].rlHits
|
|
|
+ }
|
|
|
+ printFinal(logger, tsent, tfailed, tdupes, trlHits, &rttHist)
|
|
|
return
|
|
|
case <-ticker.C:
|
|
|
// Aggregate per-worker counters.
|