Cleanup flush logic

[paolobarbolini]

This cleans-up the flushing logic from #1060.

This change makes it so flush is interrupted (doesn't get polled anymore) when new writes are in progress. The reason for this is that if a write returns Poll::Pending either the downstream write buffers are full already or tokio decided to force a yield to the executor. In either case there's no advantage in explicitly trying to flush.

Before:

• Successfully write first command
• No more commands are available
• Start flushing. Operation is Poll::Pending
• Get a second command and start writing it, getting Poll::Pending in the middle of writing it
• Call flush again. Poll::Pending is almost always guaranteed here.

After:

• Successfully write first command
• No more commands are available
• Start flushing. Operation is Poll::Pending
• Get a second command and start writing it, getting Poll::Pending in the middle of writing it
• The task yields
• The task gets woken up
• Continue writing the second command until Poll::Ready
• No more commands are queued for writing - resume flushing

---

I'm not sure about the nats::publish_throughput/32 benchmark result but it looks very flacky.

Benchmarks

nats::publish_throughput/32
                        time:   [138.21 ms 155.37 ms 176.86 ms]
                        thrpt:  [86.275 MiB/s 98.209 MiB/s 110.41 MiB/s]
                 change:
                        time:   [+3.0216% +21.250% +43.149%] (p = 0.05 < 0.05)
                        thrpt:  [-30.143% -17.526% -2.9330%]
                        Performance has regressed.
nats::publish_throughput/1024
                        time:   [215.63 ms 221.05 ms 226.61 ms]
                        thrpt:  [2.1042 GiB/s 2.1571 GiB/s 2.2114 GiB/s]
                 change:
                        time:   [-11.579% -7.5573% -2.8469%] (p = 0.01 < 0.05)
                        thrpt:  [+2.9303% +8.1751% +13.095%]
                        Performance has improved.
Benchmarking nats::publish_throughput/8192: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 14.0s.
nats::publish_throughput/8192
                        time:   [1.2862 s 1.3323 s 1.3820 s]
                        thrpt:  [2.7604 GiB/s 2.8633 GiB/s 2.9659 GiB/s]
                 change:
                        time:   [-11.125% -6.2950% -0.9346%] (p = 0.04 < 0.05)
                        thrpt:  [+0.9435% +6.7179% +12.517%]
                        Change within noise threshold.

nats::publish_amount/32 time:   [131.28 ms 155.98 ms 178.76 ms]
                        thrpt:  [2.7970 Melem/s 3.2056 Melem/s 3.8088 Melem/s]
                 change:
                        time:   [-14.054% +8.6289% +36.031%] (p = 0.50 > 0.05)
                        thrpt:  [-26.487% -7.9434% +16.352%]
                        No change in performance detected.
nats::publish_amount/1024
                        time:   [218.84 ms 227.07 ms 236.57 ms]
                        thrpt:  [2.1136 Melem/s 2.2019 Melem/s 2.2848 Melem/s]
                 change:
                        time:   [-13.606% -8.3031% -2.3712%] (p = 0.02 < 0.05)
                        thrpt:  [+2.4288% +9.0549% +15.748%]
                        Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) high mild
Benchmarking nats::publish_amount/8192: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 12.5s.
nats::publish_amount/8192
                        time:   [1.2770 s 1.3469 s 1.4188 s]
                        thrpt:  [352.41 Kelem/s 371.22 Kelem/s 391.56 Kelem/s]
                 change:
                        time:   [-11.572% -5.5703% +0.4424%] (p = 0.11 > 0.05)
                        thrpt:  [-0.4405% +5.8989% +13.087%]
                        No change in performance detected.

Benchmarking nats::subscribe_amount/32: Warming up for 3.0000 sthread 'tokio-runtime-worker' panicked at async-nats/benches/core_nats.rs:104:38:
called `Result::unwrap()` on an `Err` value: PublishError(SendError { .. })
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
Benchmarking nats::subscribe_amount/32: Collecting 10 samples in estimated 6.3417 s (20 iterations)thread 'tokio-runtime-worker' panicked at async-nats/benches/core_nats.rs:104:38:
called `Result::unwrap()` on an `Err` value: PublishError(SendError { .. })
nats::subscribe_amount/32
                        time:   [268.83 ms 295.58 ms 320.56 ms]
                        thrpt:  [1.5598 Melem/s 1.6916 Melem/s 1.8599 Melem/s]
                 change:
                        time:   [-13.631% +0.1028% +17.601%] (p = 0.99 > 0.05)
                        thrpt:  [-14.967% -0.1027% +15.783%]
                        No change in performance detected.
nats::subscribe_amount/1024
                        time:   [343.50 ms 360.97 ms 374.73 ms]
                        thrpt:  [1.3343 Melem/s 1.3851 Melem/s 1.4556 Melem/s]
                 change:
                        time:   [-10.144% -4.3353% +1.2316%] (p = 0.19 > 0.05)
                        thrpt:  [-1.2166% +4.5318% +11.289%]
                        No change in performance detected.
Found 2 outliers among 10 measurements (20.00%)
  1 (10.00%) low severe
  1 (10.00%) low mild
Benchmarking nats::subscribe_amount/8192: Warming up for 3.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 16.1s.
nats::subscribe_amount/8192
                        time:   [1.6068 s 1.6303 s 1.6545 s]
                        thrpt:  [302.21 Kelem/s 306.70 Kelem/s 311.17 Kelem/s]
                 change:
                        time:   [-2.0916% +0.5039% +3.3302%] (p = 0.72 > 0.05)
                        thrpt:  [-3.2229% -0.5013% +2.1363%]
                        No change in performance detected.

Benchmarking nats::request_amount/32: Warming up for 3.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 8.0s.
nats::request_amount/32 time:   [784.59 ms 798.66 ms 811.78 ms]
                        thrpt:  [12.319 Kelem/s 12.521 Kelem/s 12.745 Kelem/s]
                 change:
                        time:   [-5.3628% -2.0794% +1.9515%] (p = 0.32 > 0.05)
                        thrpt:  [-1.9142% +2.1236% +5.6667%]
                        No change in performance detected.
Benchmarking nats::request_amount/1024: Warming up for 3.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 8.1s.
nats::request_amount/1024
                        time:   [750.99 ms 798.83 ms 833.83 ms]
                        thrpt:  [11.993 Kelem/s 12.518 Kelem/s 13.316 Kelem/s]
                 change:
                        time:   [-8.5641% -2.2194% +2.5287%] (p = 0.51 > 0.05)
                        thrpt:  [-2.4663% +2.2698% +9.3662%]
                        No change in performance detected.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) low severe
Benchmarking nats::request_amount/8192: Warming up for 3.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 8.4s.
nats::request_amount/8192
                        time:   [837.38 ms 852.38 ms 866.79 ms]
                        thrpt:  [11.537 Kelem/s 11.732 Kelem/s 11.942 Kelem/s]
                 change:
                        time:   [-1.4246% +1.5684% +4.5172%] (p = 0.32 > 0.05)
                        thrpt:  [-4.3220% -1.5442% +1.4452%]
                        No change in performance detected.

Benchmarking jetstream::sync_publish_throughput/32: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 40.5s.
jetstream::sync_publish_throughput/32
                        time:   [3.8569 s 3.9371 s 4.0148 s]
                        thrpt:  [389.19 KiB/s 396.87 KiB/s 405.12 KiB/s]
                 change:
                        time:   [-0.9903% +1.7186% +4.3620%] (p = 0.26 > 0.05)
                        thrpt:  [-4.1797% -1.6896% +1.0002%]
                        No change in performance detected.
Benchmarking jetstream::sync_publish_throughput/1024: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 39.5s.
jetstream::sync_publish_throughput/1024
                        time:   [4.0830 s 4.1402 s 4.1925 s]
                        thrpt:  [11.647 MiB/s 11.794 MiB/s 11.959 MiB/s]
                 change:
                        time:   [-2.1806% -0.7481% +0.7124%] (p = 0.36 > 0.05)
                        thrpt:  [-0.7073% +0.7538% +2.2292%]
                        No change in performance detected.
Benchmarking jetstream::sync_publish_throughput/8192: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 50.1s.
jetstream::sync_publish_throughput/8192
                        time:   [4.9634 s 5.0036 s 5.0456 s]
                        thrpt:  [77.419 MiB/s 78.069 MiB/s 78.701 MiB/s]
                 change:
                        time:   [-1.0301% +0.0967% +1.2997%] (p = 0.88 > 0.05)
                        thrpt:  [-1.2831% -0.0966% +1.0408%]
                        No change in performance detected.

Benchmarking jetstream sync publish messages amount/32: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 39.0s.
jetstream sync publish messages amount/32
                        time:   [3.7760 s 3.8772 s 3.9614 s]
                        thrpt:  [12.622 Kelem/s 12.896 Kelem/s 13.242 Kelem/s]
                 change:
                        time:   [-3.3061% -0.4687% +1.9495%] (p = 0.76 > 0.05)
                        thrpt:  [-1.9123% +0.4709% +3.4192%]
                        No change in performance detected.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) low mild
Benchmarking jetstream sync publish messages amount/1024: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 29.8s.
jetstream sync publish messages amount/1024
                        time:   [2.9294 s 3.2153 s 3.6307 s]
                        thrpt:  [13.772 Kelem/s 15.551 Kelem/s 17.069 Kelem/s]
                 change:
                        time:   [-28.580% -22.132% -12.599%] (p = 0.00 < 0.05)
                        thrpt:  [+14.416% +28.423% +40.018%]
                        Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) high severe
Benchmarking jetstream sync publish messages amount/8192: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 66.4s.
jetstream sync publish messages amount/8192
                        time:   [4.2100 s 4.5687 s 4.8608 s]
                        thrpt:  [10.286 Kelem/s 10.944 Kelem/s 11.876 Kelem/s]
                 change:
                        time:   [-15.303% -8.5802% -3.1430%] (p = 0.01 < 0.05)
                        thrpt:  [+3.2450% +9.3856% +18.069%]
                        Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) low mild

jetstream async publish throughput/32
                        time:   [145.63 ms 172.13 ms 199.90 ms]
                        thrpt:  [7.6331 MiB/s 8.8648 MiB/s 10.478 MiB/s]
                 change:
                        time:   [-22.841% -2.5488% +24.431%] (p = 0.84 > 0.05)
                        thrpt:  [-19.634% +2.6155% +29.602%]
                        No change in performance detected.
Benchmarking jetstream async publish throughput/1024: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 9.6s or enable flat sampling.
jetstream async publish throughput/1024
                        time:   [145.32 ms 173.53 ms 220.91 ms]
                        thrpt:  [221.03 MiB/s 281.38 MiB/s 336.01 MiB/s]
                 change:
                        time:   [-24.285% -1.2861% +30.390%] (p = 0.92 > 0.05)
                        thrpt:  [-23.307% +1.3029% +32.074%]
                        No change in performance detected.
jetstream async publish throughput/8192
                        time:   [304.40 ms 389.03 ms 486.66 ms]
                        thrpt:  [802.67 MiB/s 1004.1 MiB/s 1.2532 GiB/s]
                 change:
                        time:   [-44.722% -24.640% +2.6866%] (p = 0.11 > 0.05)
                        thrpt:  [-2.6163% +32.697% +80.904%]
                        No change in performance detected.

jetstream::async_publish_messages_amount/32
                        time:   [101.84 ms 105.74 ms 110.11 ms]
                        thrpt:  [454.08 Kelem/s 472.86 Kelem/s 490.97 Kelem/s]
                 change:
                        time:   [-46.471% -32.583% -10.058%] (p = 0.03 < 0.05)
                        thrpt:  [+11.183% +48.331% +86.815%]
                        Performance has improved.
Benchmarking jetstream::async_publish_messages_amount/1024: Warming up for 1.0000 s
Warning: Unable to complete 10 samples in 5.0s. You may wish to increase target time to 7.3s or enable flat sampling.
jetstream::async_publish_messages_amount/1024
                        time:   [153.90 ms 161.24 ms 170.96 ms]
                        thrpt:  [292.47 Kelem/s 310.10 Kelem/s 324.88 Kelem/s]
                 change:
                        time:   [-42.377% -37.288% -31.687%] (p = 0.00 < 0.05)
                        thrpt:  [+46.386% +59.459% +73.542%]
                        Performance has improved.
jetstream::async_publish_messages_amount/8192
                        time:   [315.64 ms 364.90 ms 431.02 ms]
                        thrpt:  [116.00 Kelem/s 137.02 Kelem/s 158.41 Kelem/s]
                 change:
                        time:   [-42.287% -30.916% -16.313%] (p = 0.00 < 0.05)
                        thrpt:  [+19.492% +44.752% +73.272%]
                        Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
  1 (10.00%) high severe

[Jarema]

Thanks!
Will take a look.

[Jarema]

LGTM!