perf(replication): implement async processing during replication#1166
perf(replication): implement async processing during replication#1166ygxio wants to merge 11 commits into
Conversation
|
I am still working on the design for it, just a peak if the direction seems well @levkk |
|
Cool! Would be good to see some benchmark numbers, just to get a sense of how much faster this is. |
|
Do you have any such setup so that i can benchmark it? |
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
This might work: Specifically, see |
|
RUNS = 2
@levkk results seems good but also too good at a point |
|
Async should be a big improvement, so not that surprising. This is very cool. Let me know when you're ready for a review, I'll tag Kiryl. |
|
@levkk ready for review |
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod test { |
There was a problem hiding this comment.
Do you think we should add more unit tests here? I think we should have some for testing the whole flow - prepare -> execute -> await. And also some tests for possible errors
There was a problem hiding this comment.
We still miss some important tests:
- error path - something that will trigger an error response, validate that we get the error properly and we won't block the execution for future calls
- missed rows calculation - just some calls to generate missed rows and make sure we get proper stats for it
|
@meskill i have added a few more could you take a look? |
|
Thanks @ygxio I'm still checking the code more deeply, but anyway looking great so far. Putting my bench results to confirm the improvements: Against mainWith toxi and latency for source/destination for 1ms: SeparatelyAnd the next without comparison with main (compare against prev run) since since it is executing too long to get the result. |
|
i see from 1 to 10 to 100 ms of latency the jump has been pretty high, need to see the cause maybe |
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod test { |
There was a problem hiding this comment.
We still miss some important tests:
- error path - something that will trigger an error response, validate that we get the error properly and we won't block the execution for future calls
- missed rows calculation - just some calls to generate missed rows and make sure we get proper stats for it
8cf5e8b to
0c5844d
Compare
meskill
left a comment
There was a problem hiding this comment.
Looks good, only the small test fix and we'll merge
| async fn prepare_invalid_sql_sync_returns_error() { | ||
| let server = test_server().await; | ||
| let conn = PipelinedConnection::new(server).unwrap(); | ||
|
|
||
| // A single in-transaction prepare with several statements uses | ||
| // ParseAcks(n): the waiter must resolve only after ALL n ParseComplete | ||
| // acks arrive (exercises the n -> 0 countdown, not just n == 1). | ||
| conn.prepare( | ||
| &[ | ||
| Parse::named("__pipe_m1", "SELECT $1::bigint"), | ||
| Parse::named("__pipe_m2", "SELECT $1::text"), | ||
| Parse::named("__pipe_m3", "SELECT $1::bool"), | ||
| ], | ||
| true, | ||
| // Out-of-transaction prepare of invalid SQL: Postgres replies with an | ||
| // ErrorResponse, which is latched and surfaced through the Sync path. | ||
| let err = conn | ||
| .prepare(&[Parse::named("__pipe_bad", "NOT VALID SQL")], false) | ||
| .await | ||
| .unwrap_err(); | ||
| assert!( | ||
| matches!(err, Error::PgError(_)), | ||
| "unexpected error: {err:?}" | ||
| ); | ||
| // The error was taken while surfacing, so nothing remains latched. | ||
| assert!(conn.take_error().is_none()); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn prepare_invalid_sql_flush_returns_error() { | ||
| let server = test_server().await; | ||
| let conn = PipelinedConnection::new(server).unwrap(); | ||
|
|
||
| // In-transaction prepare uses Flush, so Postgres sends no ReadyForQuery | ||
| // on error. The parked ParseAcks waiter can only be released by the | ||
| // 'E' handler's wake_all(): this proves the prepare does not hang. | ||
| let err = conn | ||
| .prepare(&[Parse::named("__pipe_bad", "NOT VALID SQL")], true) | ||
| .await | ||
| .unwrap_err(); | ||
| assert!( | ||
| matches!(err, Error::PgError(_)), | ||
| "unexpected error: {err:?}" | ||
| ); | ||
| assert!(conn.take_error().is_none()); | ||
| } |
There was a problem hiding this comment.
those are identical tests, please verify
There was a problem hiding this comment.
@meskill one it for out of transaction and other is for in transaction, should i merge them into one test with a loop over in_transaction value either false or true











fixes #1036