Allow the async connection to have multiple in flight requests #143

Marwes · 2018-02-02T14:48:13Z

Based on #141 (see 8fba6cf)

This is an extension of #141 where the reading and writing to the socket is done in a separate future which understands how to pipeline requests. This makes it possible to Clone the SharedConnection and do multiple queries concurrently, without any manual pipelining.

~~The downsides are that the connection is not Send or Sync (could be alleviated with sync types but I am not sure it is worth it) and there is not (currently) a way to run explicit pipelines.~~

The connection is now Send + Sync

Marwes · 2018-04-24T08:14:43Z

Rebased on top of #141 so that it works without tokio-uds https://github.com/mitsuhiko/redis-rs/pull/143/files/83e82b87702f3161dc78b11aac514756fae3e7f2..21a1b0007ccaa6acae18a5323c8dbeb32c7de6b4 .

Also changed SharedConnection to be Send + Sync.

This prevents us from doing small writes when there are multiple messages already ready to be send through the pipeline.

Version 0.2 were released a few weeks ago so we might as well enable it immediately

Marwes · 2018-06-26T13:05:19Z

@badboy Rebased this on top of master. Non-rustfmt changes are at https://github.com/mitsuhiko/redis-rs/pull/143/files/6e8492be1214d831a73628ebdf5d7f2398b14729..a3eefd4b35677df42a05b879388d90c388043c79.

I find this PR makes it significantly nicer to work with the async API so I think it would be nice to get in as well before making any release :) (I could maybe extract this outside of this crate as well though so it is not as critical I think).

badboy · 2018-06-26T13:08:50Z

Lined it up for review, planning to get to it by the end of the week.

badboy · 2018-06-27T19:02:16Z

Could you rebase this?

Marwes · 2018-06-27T19:27:06Z

@badboy I merged from master to avoid breaking the git dependency I have, can rebase this before merging (or just squash and merge, none of the commits document anything important)

badboy · 2018-06-27T19:32:51Z

Works for me!

Marwes · 2018-07-11T11:52:56Z

@badboy Have you had the time to look at this yet?

badboy · 2018-07-11T12:03:06Z

Unfortunately no.
But looking at my calendar, I simply block 2 hours tomorrow to get it done.

badboy

Hey!

First of all, thanks for patiently waiting for my long-overdue review. This work is really appreciated.
Now I just want to make sure we actually get good code and long-term maintainable (I have to admit I still haven't actually used futures too much).

I already left some small comments inline.
Now to the bigger picture:
I have some trouble following the intention of the code, especially with things like VecDeque<Vec<oneshot::Sender<Result<T::Item, T::Error>>>> and Vec<oneshot::Sender<Result<T::Item, T::Error>>>.

Why do we need nested Vec(Deque)s here? Why do we reverse the inner Vec later and pop items off? Is this effectively emulating an ordered stream?

Can you give me a short high-level overview of what the code achieves? I think it then might make sense to include such documentation with the code as well.

Thanks again for your work, it would be great to have these things in the code base.

badboy · 2018-07-12T15:33:37Z

Cargo.toml

@@ -25,11 +25,17 @@ with-system-unix-sockets = []
 sha1 = ">= 0.2, < 0.7"
 url = "1.2"
 rustc-serialize = { version = "0.3.16", optional = true }
+


Let's remove these newlines. Also some below.

badboy · 2018-07-12T16:40:32Z

examples/basic.rs

-                 .unwrap_or(&0));
+    println!(
+        "  max-intset:        {}",
+        config.get("set-max-intset-entries").unwrap_or(&0)


I'm not too happy about this formatting, but I'm ok with accepting it in this patch. I wonder if we can tell rustfmt to keep the string on the same line as println! (can be done later)

badboy · 2018-07-12T16:45:59Z

src/async.rs

+            };
+            self.send_result(item);
+        }
+    }


Needs a newline

badboy · 2018-07-12T16:46:26Z

src/async.rs

+        match sender.send(item) {
+            Ok(_) => (),
+            Err(_) => {
+                // `Err` means that the receiver were dropped in which case it does not


the receiver was dropped

badboy · 2018-07-12T17:03:42Z

src/async.rs

+    T::Error: ::std::fmt::Debug,
+{
+    fn new(sink_stream: T) -> Self {
+        let (sender, receiver) = mpsc::unbounded();


Unbounded sounds scary.
Do we have any other limitations to ensure we don't actually grow unbounded?
Or is there any way we could actually have a bounded channel (for some reasonable buffer size?)

Fixed, 50 was chosen by a fair dice roll. I think in the long term this could be implemented with https://github.com/tower-rs/tower and we could get this (and other niceties) for free.

badboy · 2018-07-12T17:06:26Z

src/async.rs

+        Box::new(
+            self_
+                .send((item, senders))
+                .map_err(|_| panic!("Unexpected close of redis mpsc sender"))


Panics in the middle of code always make me worry.
Under what circumstances will this be hit? How easy is this to trigger?
Is this really unrecoverable?

Nice catch. It was supposed to be unreachable barring programming bugs but it ended up being reachable and should have been mapped to a closed connection. Fixed

badboy · 2018-07-12T17:07:07Z

src/async.rs

+            let pipeline = match con.con {
+                ActualConnection::Tcp(tcp) => ActualPipeline::Tcp(Pipeline::new(
+                    ValueCodec::default().framed(tcp.into_inner()),
+                )),


The formatting here is a bit weird.
Can we just move the ActualPipeline part into a block on the next line?

badboy · 2018-07-12T17:16:50Z

src/async.rs

+        let self_ = self.0.clone();
+
+        let mut senders = Vec::new();
+        let mut receivers = Vec::new();


We do know the size here, so Vec::with_capacity() seems like a good idea.

👍 Thought about using https://github.com/servo/rust-smallvec here since 1 element vectors will be common. Seems like premature optimization though.

By reversing the senders vector before it is sent we don't need the extra intermediate step

No actual reason to use an unbounded channel here and its better to have some sort of backpressure. Ideally this should be configurable and this would also apply to the messages currently in flight on the `Sink + Stream` (perhaps with https://github.com/tower-rs/tower if it gets support for this)

Marwes · 2018-07-14T22:06:06Z

I hope I added enough context to the pipeline stuff so it is understandable!

badboy · 2018-07-18T08:41:57Z

@Marwes Much better! I give it another look

badboy · 2018-07-31T14:24:34Z

I'm sorry it took me another 2 weeks to take a look at it. But I think it's ready to merge now.

…s#143)

Marwes force-pushed the shared_async branch 2 times, most recently from ce80ce1 to 41a4406 Compare February 5, 2018 11:39

Marwes mentioned this pull request Feb 12, 2018

Add an async interface using futures-rs #141

Merged

Marwes force-pushed the shared_async branch from db64666 to c9fd0a8 Compare February 12, 2018 16:05

Marwes mentioned this pull request Mar 1, 2018

Create Redis session backend gotham-rs/gotham#69

Open

Marwes force-pushed the shared_async branch from c9fd0a8 to c8a3505 Compare March 18, 2018 19:22

Marwes changed the title ~~[WIP] Allow the async connection to have multiple in flight requests~~ Allow the async connection to have multiple in flight requests Mar 18, 2018

Marwes force-pushed the shared_async branch 3 times, most recently from 5ca319e to 21a1b00 Compare April 24, 2018 08:05

Marwes force-pushed the shared_async branch from 21a1b00 to c89bd33 Compare April 25, 2018 09:56

Marwes force-pushed the shared_async branch 2 times, most recently from cb16409 to 601fd3c Compare May 30, 2018 13:24

Marwes force-pushed the shared_async branch 2 times, most recently from 9a7d88a to bcb0cff Compare June 26, 2018 12:43

Markus Westerlind added 6 commits June 26, 2018 15:00

Run rustfmt

6e8492b

feat: Add automatic pipelining for async connections

35b892e

Add a benchmark for using the shared connection

11b8e39

perf: Start sending multiple messages before calling poll_complete

c03099e

This prevents us from doing small writes when there are multiple messages already ready to be send through the pipeline.

Re-enable tokio_uds

1ca3908

Version 0.2 were released a few weeks ago so we might as well enable it immediately

Fix deprecation warnings from tokio_io

a3eefd4

Marwes force-pushed the shared_async branch from bcb0cff to a3eefd4 Compare June 26, 2018 13:01

badboy self-requested a review June 26, 2018 13:08

Marwes mentioned this pull request Jun 27, 2018

Cannot send RedisFuture between thread #156

Closed

Merge branch 'master' into shared_async

39236bc

badboy requested changes Jul 12, 2018

View reviewed changes

Marwes added 7 commits July 14, 2018 22:50

doc: Explain Pipeline better

ca07cad

refactor: Skip the senders field in pipelining

89d3850

By reversing the senders vector before it is sent we don't need the extra intermediate step

test: Check that the shared connection handles errors from redis

0ad119f

a

ec5ab40

Don't use tokio::run as it suppresses panics

0ca1f17

Don't panic on closed shared connections

4ffd3e9

badboy approved these changes Jul 31, 2018

View reviewed changes

badboy merged commit 48f6505 into redis-rs:master Jul 31, 2018

barshaul pushed a commit to barshaul/redis-rs that referenced this pull request Jun 9, 2024

Added script to gather command info and updated base_routing (redis-r…

e28a978

…s#143)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Allow the async connection to have multiple in flight requests #143

Allow the async connection to have multiple in flight requests #143

Marwes commented Feb 2, 2018 •

edited

Loading

Marwes commented Apr 24, 2018 •

edited

Loading

Marwes commented Jun 26, 2018

badboy commented Jun 26, 2018

badboy commented Jun 27, 2018

Marwes commented Jun 27, 2018 •

edited

Loading

badboy commented Jun 27, 2018

Marwes commented Jul 11, 2018

badboy commented Jul 11, 2018

badboy left a comment

badboy Jul 12, 2018

badboy Jul 12, 2018

badboy Jul 12, 2018

badboy Jul 12, 2018

Marwes Jul 14, 2018

badboy Jul 12, 2018

Marwes Jul 14, 2018

badboy Jul 12, 2018

Marwes Jul 14, 2018

badboy Jul 12, 2018

badboy Jul 12, 2018

Marwes Jul 14, 2018

Marwes commented Jul 14, 2018

badboy commented Jul 18, 2018

badboy commented Jul 31, 2018

Allow the async connection to have multiple in flight requests #143

Allow the async connection to have multiple in flight requests #143

Conversation

Marwes commented Feb 2, 2018 • edited Loading

Marwes commented Apr 24, 2018 • edited Loading

Marwes commented Jun 26, 2018

badboy commented Jun 26, 2018

badboy commented Jun 27, 2018

Marwes commented Jun 27, 2018 • edited Loading

badboy commented Jun 27, 2018

Marwes commented Jul 11, 2018

badboy commented Jul 11, 2018

badboy left a 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

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

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

Marwes commented Jul 14, 2018

badboy commented Jul 18, 2018

badboy commented Jul 31, 2018

Marwes commented Feb 2, 2018 •

edited

Loading

Marwes commented Apr 24, 2018 •

edited

Loading

Marwes commented Jun 27, 2018 •

edited

Loading