refactor(async client): refactor background task #1145

niklasad1 · 2023-06-21T16:31:21Z

Split send and receive to separate tasks to support multiplexing reads and writes

Summary of what was wrong before:

A request from frontend came into this task
We try to send it off to backend, .awaiting that to complete
But! The backend is applying backpressure, oh no! so it won't accept any new messages until some messages are read from it to clear that queue.
So this send keeps .awaiting because reading happened in the same task, no reads will happen until the send completes, so essentially we are in a deadlock situation.

Benches

group                                           client-multiplex                        master
-----                                           ----------------                        ------
async/ws_batch_requests/fast_call/10            1.00     42.2±4.86µs 231.5 KElem/sec    1.30     54.6±0.85µs 178.7 KElem/sec
async/ws_batch_requests/fast_call/100           1.00   317.8±20.38µs 307.3 KElem/sec    1.12   356.3±19.86µs 274.1 KElem/sec
async/ws_batch_requests/fast_call/2             1.56     52.9±3.86µs 36.9 KElem/sec     1.00     33.8±5.25µs 57.7 KElem/sec
async/ws_batch_requests/fast_call/5             1.00     37.3±5.52µs 130.8 KElem/sec    1.08     40.5±5.66µs 120.7 KElem/sec
async/ws_batch_requests/fast_call/50            1.11    192.4±4.70µs 253.8 KElem/sec    1.00   173.3±33.70µs 281.7 KElem/sec
async/ws_concurrent_conn_calls/fast_call/128    1.05      2.1±0.05ms        ? ?/sec     1.00  1966.4±11.49µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/16     1.00  1091.3±89.12µs        ? ?/sec     1.06  1160.0±134.30µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/2      1.00  1092.4±112.38µs        ? ?/sec    1.07  1174.0±86.38µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/256    1.06      3.9±0.03ms        ? ?/sec     1.00      3.7±0.04ms        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/32     1.00  1204.2±135.04µs        ? ?/sec    1.00  1209.4±143.03µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/4      1.02  1241.3±71.76µs        ? ?/sec     1.00  1212.0±104.79µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/64     1.05  1217.5±15.00µs        ? ?/sec     1.00  1159.5±10.37µs        ? ?/sec
async/ws_concurrent_conn_calls/fast_call/8      1.01  1186.8±123.04µs        ? ?/sec    1.00  1178.7±73.72µs        ? ?/sec

Split send and receive to separate tasks to support multiplexing reads/writes

core/src/client/async_client/mod.rs

core/src/client/async_client/manager.rs

core/src/client/async_client/helpers.rs

core/src/client/async_client/mod.rs

jsdw

Code looks good to me overall, and the separate taks looks pretty clean!

If the ThreadSafeRequestManager didn't need to lock over await points then maybe a standard mutex would be a better choice (a safe way to do this would be to make the mutex an integral part of RequestManager so that each function only locks internall for as long as it needs and it's impossible to hold a lock over any await points or whatever).

I'd be interested to know whether this has any impact on benchmarks in either direction :)

core/src/client/async_client/helpers.rs

core/src/client/async_client/manager.rs

core/src/client/async_client/mod.rs

lexnv · 2023-06-27T12:51:04Z

core/src/client/async_client/mod.rs

+				}
+			}
+			_ = close_tx.closed() => {
+				break Ok(())


Could some tracing debug logs be useful on the break branches to know the reason of the send_task exit?

I added Client dropped logs when if either the send or receive task returns Ok(()) ok?

These channels should be closed if send/receive tasks are closed, if something terminates with an error that must occur before this send fails.

However, if it's bug somewhere that could be tricky detect by looking at the logs.

core/src/client/async_client/mod.rs

…ure-support

This reverts commit 49edaee.

jsdw · 2023-08-01T14:44:47Z

core/src/client/async_client/mod.rs

+
+	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+		if self.futs.is_empty() {
+			return Poll::Pending;


At this point, the waker is not handed anywhere and so if any other futures are added, they will be ignored.

I think that this is ok though because you just call next() on this in a loop, so it'll get polled at least once per loop anyway.

I tried using pure FutureUnordered and then the select loop will be busy that it returns Poll::Ready i.e None when it's empty and steal CPU cycles from more important stuff, that's the reason why I added it...

Yeah, if it's empty and futures are added those will read on the next iteration

Ah I see, so when it's empty it'll spam Poll::Ready(None) whenever polled; yeah that sucks. Your version looks good to me offhand in the current code now, though it's a little scary that if used wrong it'll just .await forever, so a small footgun to be wary of for future us! Perhaps it's impossible to misuse though given it needs &mut self, but I havent convinced myself 100% :D

I believe this should be fine for now, because we use the read_task to drive this:

read_task select (server_receiver, closed, pending_unsub) server_receiver -> if Ok(Some(m)) = handle_backend_messages pending_unsub.push( to_send_task.send(m) )

One case where this might be problematic is when:

server_receiver and closed futures are not ready

the pending_unsub is empty and thus will return Poll::Pending and I believe will never woke up for the current iteration loop because there's nothing that will wake up the cx.waker again of the future

then pending_unsub has some elements added to it

But sincepending_unsub having elements added is dependent on server_receiver being triggered first; it should be fine here as long as we don't save the current pending_unsub future that returned Pending and instead call next directly in the loop future::select(rx_or_closed, pending_unsubscribes.next())

Let me know if my reasoning is right 🙏

Indeed there might be dragons if we change something in the read_task (maybe a comment to not save pending_unsub.next() future is enough here).

We may be able to handle this by:

if futs.is_empty(), then we could save the cx

when adding a new future, if we have Some(cx) we could wake up the waker

preserve the same pending_unsub.next() between iterations similar to let rx_item = backend_event.next();

Not quite sure if worth going that way

yeah, you are correct.

did what you suggested it didn't require much code but instead I had to clone the Waker because Context is !Send

Save the Waker and not the context; that's sendable :)

Ah that's what you did; I misread the earlier comment and github was showing me old stuff, whoops!

core/src/client/async_client/mod.rs

jsdw · 2023-08-01T15:14:49Z

There's a lot to work through here, but offhand it looks good to me now; no more unbounded queues from what I could see, and so backpressure looks like it'll work with sending and reading, and the two queues shouldn't block progrss from eachother I think now, so nice one!

Couple of nits above but I'm happy to see this merge, and if it does well enough on burnins then it's an improvement from where we were regardless :)

This reverts commit d73aeeb.

lexnv · 2023-08-02T08:21:49Z

core/src/client/async_client/mod.rs

+	});
+
+	let closed = close_tx.closed();
+	let pending_unsubscribes = MaybePendingFutures::new();


I do wonder, is this only used for unsubscribes? Wouldn't the receiver: R also provide pings and plain messages?

No, this is really an edge-case if a subscription fails

i) a subscription call is answered but the client dropped the subscription (we want to unsubscribe then)
ii) if the subscription couldn't keep it with the server.

Thus, this really just is a matter to unsubscribe then so that no further message are received for that particular subscription...

lexnv · 2023-08-02T08:23:45Z

This looks good to me! I believe the backpressure would work fine, nice job here!
Happy to merge this

refactor(async client): refactor background task

0b66421

Split send and receive to separate tasks to support multiplexing reads/writes

niklasad1 requested a review from a team as a code owner June 21, 2023 16:31

niklasad1 added 2 commits June 22, 2023 11:07

fix tests

5220dac

dont rely on tokio::spawn handles anymore

94b438e

jsdw reviewed Jun 22, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

jsdw reviewed Jun 22, 2023

View reviewed changes

core/src/client/async_client/manager.rs Show resolved Hide resolved

jsdw reviewed Jun 22, 2023

View reviewed changes

core/src/client/async_client/helpers.rs Show resolved Hide resolved

jsdw reviewed Jun 22, 2023

View reviewed changes

core/src/client/async_client/mod.rs Show resolved Hide resolved

niklasad1 commented Jun 22, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

niklasad1 added 2 commits June 22, 2023 16:20

fix build

920dc2d

fix build again

049b711

jsdw approved these changes Jun 22, 2023

View reviewed changes

lexnv reviewed Jun 22, 2023

View reviewed changes

core/src/client/async_client/helpers.rs Show resolved Hide resolved

niklasad1 added 3 commits June 22, 2023 18:13

switch to std::sync::Mutex

60ec208

fix tests again

8478e14

bench stuff

bdffddf

niklasad1 commented Jun 26, 2023

View reviewed changes

core/src/client/async_client/manager.rs Show resolved Hide resolved

lexnv reviewed Jun 27, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

lexnv reviewed Jun 27, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

lexnv approved these changes Jun 27, 2023

View reviewed changes

jsdw reviewed Jun 30, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

don't block in read task

05bb6c8

niklasad1 force-pushed the na-client-backpressure-support branch from 0509338 to 05bb6c8 Compare July 7, 2023 13:17

niklasad1 added 6 commits July 7, 2023 16:06

fix build

194bc95

fix nits

2da8ab7

works now

9de33e4

revert bench code

ad91354

Merge remote-tracking branch 'origin/master' into na-client-backpress…

aa7fff1

…ure-support

replace unreachable with None

49edaee

niklasad1 added 2 commits August 1, 2023 13:15

Revert "replace unreachable with None"

1b0f255

This reverts commit 49edaee.

fix nits

dd3f5f7

jsdw reviewed Aug 1, 2023

View reviewed changes

core/src/client/async_client/mod.rs Outdated Show resolved Hide resolved

niklasad1 added 2 commits August 1, 2023 17:15

use dashmap instead of Arc<Mutex<RequestManager>

d73aeeb

Revert "use dashmap instead of Arc<Mutex<RequestManager>"

ef8161b

This reverts commit d73aeeb.

lexnv reviewed Aug 2, 2023

View reviewed changes

niklasad1 added 5 commits August 2, 2023 10:28

refactor select loops prio for closed futs

64dd8fb

grumbles: save waker to wake when new items are pushed

55e61d2

fix build

4fb0994

fix build again

bbfee35

fix some nits

e664782

niklasad1 merged commit 54f4dcd into master Aug 3, 2023

niklasad1 deleted the na-client-backpressure-support branch August 3, 2023 11:07

niklasad1 mentioned this pull request Aug 10, 2023

[async client]: event loop process heavy responses in separate task #955

Closed

niklasad1 mentioned this pull request Aug 18, 2023

Response handling on the client is very slow comparing to server #1189

Open

niklasad1 mentioned this pull request Dec 7, 2023

Spamming requests hits timeout after upgrading v0.16 to v0.17/v0.20 #1259

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(async client): refactor background task #1145

refactor(async client): refactor background task #1145

niklasad1 commented Jun 21, 2023 •

edited

Loading

jsdw left a comment

lexnv Jun 27, 2023

niklasad1 Jun 27, 2023 •

edited

Loading

jsdw Aug 1, 2023

niklasad1 Aug 1, 2023

niklasad1 Aug 1, 2023

jsdw Aug 1, 2023 •

edited

Loading

lexnv Aug 2, 2023

niklasad1 Aug 2, 2023

niklasad1 Aug 2, 2023

jsdw Aug 3, 2023

jsdw Aug 3, 2023 •

edited

Loading

jsdw commented Aug 1, 2023 •

edited

Loading

lexnv Aug 2, 2023

niklasad1 Aug 2, 2023 •

edited

Loading

lexnv commented Aug 2, 2023

refactor(async client): refactor background task #1145

refactor(async client): refactor background task #1145

Conversation

niklasad1 commented Jun 21, 2023 • edited Loading

Benches

jsdw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

niklasad1 Jun 27, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsdw Aug 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsdw Aug 3, 2023 • edited Loading

Choose a reason for hiding this comment

jsdw commented Aug 1, 2023 • edited Loading

Choose a reason for hiding this comment

niklasad1 Aug 2, 2023 • edited Loading

Choose a reason for hiding this comment

lexnv commented Aug 2, 2023

niklasad1 commented Jun 21, 2023 •

edited

Loading

niklasad1 Jun 27, 2023 •

edited

Loading

jsdw Aug 1, 2023 •

edited

Loading

jsdw Aug 3, 2023 •

edited

Loading

jsdw commented Aug 1, 2023 •

edited

Loading

niklasad1 Aug 2, 2023 •

edited

Loading