Client not responding #13
Comments
|
Thanks for filing a great bug repro @dbstratta . I tried to repro this locally and unfortunately I wasn't able to reproduce it. That being said, I had to make some changes to your code, including a change to the type of This was the code I used: use std::time::Duration;
use futures::stream::StreamExt;
use futures::pin_mut;
use fred::prelude::*;
use fred::pool::StaticRedisPool;
#[tokio::main]
async fn main() -> Result<(), RedisError> {
let pool_config = RedisConfig {
server: ServerConfig::default_centralized(),
..Default::default()
};
let pool = StaticRedisPool::new(pool_config, 5)?;
pool.connect(Some(ReconnectPolicy::default()));
pool.wait_for_connect().await?;
let mut stream = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(Duration::from_millis(100)))
.then(move |_| {
let pool = pool.clone();
async move {
let value: Option<String> = pool.get("key").await?; // This call never responds.
Ok::<_, RedisError>(value)
}
});
pin_mut!(stream);
for value in stream.next().await {
println!("{:?}", value);
}
Ok(())
}The most notable change is the switch from the I'll do a little digging to see what the differences are and whether those traits are compatible, but in the meantime can you try your code using the |
|
Thank you for your quick reply! Yes, some changes were needed to the repro code, I'm sorry for that. I updated it with your changes. I tried to reproduce it with the updated example and couldn't either. Earlier I was using a slightly modified version of it, in a subscription handler of I'm going to try to see how |
|
I just took a look at the tokio-stream docs and one thing jumped out at me here: https://docs.rs/tokio-stream/0.1.8/tokio_stream/#iterating-over-a-stream It might be worth trying to use |
|
You're right. I updated the example above with |
|
This is a good callout though. I'm going to update the docs to make it more clear which |
|
The |
|
I took a look at the I kept trying things and I noticed that when I disable pipelining the client doesn't get stuck. |
|
Huh that's odd. I tried your updated repro code (only adding an import for Can you try adding |
|
I couldn't reproduce it with the code from the original comment either, although it keeps happening in the I'll send the logs later today. |
|
This is a log leading to a halt. It was using a After this point there were no more logs and no more communication to the Redis server. I started the stream again and the first batch of logs of a client for the next request contained a Let me know if I can provide you with anything else. |
|
Ah ok that's interesting. The beginning of the first set of logs you sent probably contain the same The way this works under the hood is that each time you send a command fred creates a That log line is emitted when the client receives an error calling send on the oneshot Is it possible that in your code the future returned from |
|
Oh.... https://github.com/tokio-rs/tokio/releases/tag/tokio-1.13.1. That's unfortunate. Can you try changing your tokio version to 1.14? |
|
What a coincidence! 😂 Although I updated What makes me think that the |
|
Ah damn, that's too bad. Yeah I can't imagine why pipelining would affect this since from the logs you showed the request is sent and the response is received even when pipelining is enabled, and all the non-pipeline logic happens before that point. I also have a test harness that covers pretty much everything in the client and runs all the tests several times with both pipelined and non-pipelined clients, but who knows, anything is possible. At this point I'm most concerned about the fact that we can't find a minimal repro that shows this happening. I'm sorry to ask this, but would it be possible to see more of your code (even if it uses graphql)? I'm happy to chat over a less public forum if you want to keep some of that code private, but at this point I think I need some form a repro to dig further into this. |
|
Sure, I'll send you a repro with the graphql code tomorrow. |
|
Here's the minimal code I could reproduce the issue with: use std::time::Duration;
use actix_web::{
get, guard, post, web, App, HttpRequest, HttpResponse, HttpServer, Responder, Result,
};
use async_graphql::http::{playground_source, GraphQLPlaygroundConfig};
use async_graphql::{Context, EmptyMutation, Object, Subscription};
use async_graphql_actix_web::{GraphQLRequest, GraphQLResponse, GraphQLSubscription};
use fred::pool::StaticRedisPool;
use fred::prelude::*;
use futures::{Stream, StreamExt};
const GRAPHQL_ENDPOINT_URL: &str = "/graphql";
type Schema = async_graphql::Schema<Query, EmptyMutation, Subscription>;
#[actix_web::main]
async fn main() {
fred::globals::set_default_command_timeout(1000);
let redis_pool = StaticRedisPool::new(
RedisConfig {
server: ServerConfig::new_centralized("127.0.0.1", 6379),
pipeline: true,
..Default::default()
},
2,
)
.unwrap();
redis_pool.connect(Some(ReconnectPolicy::default()));
redis_pool.wait_for_connect().await.unwrap();
HttpServer::new(move || {
App::new()
.app_data(web::Data::new(
Schema::build(Query, EmptyMutation, Subscription)
.data(redis_pool.clone())
.finish(),
))
.service(
web::resource(GRAPHQL_ENDPOINT_URL)
.guard(guard::Header("upgrade", "websocket"))
.route(web::get().to(graphql_ws)),
)
.service(playground)
.service(graphql)
})
.bind(("127.0.0.1", 4000))
.unwrap()
.run()
.await
.unwrap();
}
#[post("/graphql")]
async fn graphql(graphql_request: GraphQLRequest, schema: web::Data<Schema>) -> GraphQLResponse {
schema.execute(graphql_request.into_inner()).await.into()
}
async fn graphql_ws(
request: HttpRequest,
schema: web::Data<Schema>,
payload: web::Payload,
) -> Result<HttpResponse> {
GraphQLSubscription::new(Schema::clone(&*schema)).start(&request, payload)
}
#[get("/graphql")]
async fn playground() -> impl Responder {
HttpResponse::Ok()
.content_type("text/html")
.body(playground_source(
GraphQLPlaygroundConfig::new(GRAPHQL_ENDPOINT_URL)
.subscription_endpoint(GRAPHQL_ENDPOINT_URL),
))
}
pub struct Subscription;
#[Subscription]
impl Subscription {
async fn tick(&self, ctx: &Context<'_>) -> impl Stream<Item = Option<bool>> {
let redis_pool = ctx.data_unchecked::<StaticRedisPool>().clone();
tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(Duration::from_millis(
100,
)))
.then(move |_| {
let redis_pool = redis_pool.clone();
async move {
let value = redis_pool.get::<Option<bool>, _>("key").await.unwrap();
value
}
})
}
}
pub struct Query;
#[Object]
impl Query {
async fn hello(&self) -> &str {
"world"
}
}With [package]
name = "repro"
version = "0.0.1"
edition = "2021"
[dependencies]
actix-web = { version = "= 4.0.0-beta.11", default-features = false }
async-graphql = { version = "3.0.4" }
async-graphql-actix-web = { version = "3.0.4" }
fred = { version = "4.2.1", default-features = false, features = [
"pool-prefer-active",
"ignore-auth-error",
] }
futures = { version = "0.3.17", default-features = false }
tokio = { version = "1.14.0", features = ["rt", "macros", "time"] }
tokio-stream = "0.1.8"Go to http://localhost:4000/graphql, in the playground enter the following text and start the subscription (with the play button): subscription {
tick
}That will start the stream and start responding (on the right panel of the playground) with: {
"data": {
"tick": null
}
}Let it be subscribed for 10 seconds, stop it (with the stop button in the playground) and start it again for 10 seconds. Repeat this until it crashes because of the
|
|
Hi @dbstratta First off thanks a lot for taking the time to write all that up. I tried your example and was able to successfully repro the issue with 4.2.1 of fred after a few tries, but I did also have to change the actix version to This morning I released 4.2.2 for fred which addressed #14. I had hoped that would also address this issue since both issues used the same static pool code paths, but unfortunately that doesn't seem to be the case. As I was writing this comment I saw a repro even on 4.2.2 with your code above, so I'll keep looking and update you when I know more. Possibly unrelated - is it normal to see this in the console? I'll sometimes see that after starting the subscriber on the client, but it happens without any issues/panics from fred. |
|
Hi @dbstratta I have some good news and some bad news, depending on how you look at it. The good news is that I was able to repro this issue without fred in play at all, and the bad news is that I think this is an actix issue. Given that they're still in beta for 4.x that may be somewhat expected, but unfortunately means I can't help too much here. The logs from your original repro showed that the receiver half of the oneshot sender used by fred (which is returned to the caller) was being dropped before fred could send a response. I modified your repro to simulate how fred works, but just using vanilla tokio tasks and channels with fake data. I also removed all the timeout management logic, as well as This was the resulting minimum repro case I could come up with: use actix_web::{get, guard, post, web, App, HttpRequest, HttpResponse, HttpServer, Responder, Result};
use async_graphql::http::{playground_source, GraphQLPlaygroundConfig};
use async_graphql::{Context, EmptyMutation, Object, Subscription};
use async_graphql_actix_web::{GraphQLRequest, GraphQLResponse, GraphQLSubscription};
use fred::error::RedisError; // i still use fred's error types, but that's it
use futures::stream::repeat;
use futures::{Stream, StreamExt};
use tokio::sync::oneshot::channel;
const GRAPHQL_ENDPOINT_URL: &str = "/graphql";
type Schema = async_graphql::Schema<Query, EmptyMutation, Subscription>;
#[derive(Clone)]
struct RedisClient;
impl RedisClient {
async fn simulate_redis(&self) -> Result<Option<bool>, RedisError> {
// this is essentially how fred works under the hood for all requests
let (tx, rx) = channel();
tokio::spawn(async move {
if let Err(e) = tx.send(Ok(None)) {
println!("Error sending fake response: {:?}", e);
}
});
rx.await?
}
}
#[actix_web::main]
async fn main() {
pretty_env_logger::init();
let client = RedisClient;
HttpServer::new(move || {
App::new()
.app_data(web::Data::new(
Schema::build(Query, EmptyMutation, Subscription)
.data(client.clone())
.finish(),
))
.service(
web::resource(GRAPHQL_ENDPOINT_URL)
.guard(guard::Header("upgrade", "websocket"))
.route(web::get().to(graphql_ws)),
)
.service(playground)
.service(graphql)
})
.bind(("127.0.0.1", 4000))
.unwrap()
.run()
.await
.unwrap();
}
#[post("/graphql")]
async fn graphql(graphql_request: GraphQLRequest, schema: web::Data<Schema>) -> GraphQLResponse {
schema.execute(graphql_request.into_inner()).await.into()
}
async fn graphql_ws(request: HttpRequest, schema: web::Data<Schema>, payload: web::Payload) -> Result<HttpResponse> {
GraphQLSubscription::new(Schema::clone(&*schema)).start(&request, payload)
}
#[get("/graphql")]
async fn playground() -> impl Responder {
HttpResponse::Ok().content_type("text/html").body(playground_source(
GraphQLPlaygroundConfig::new(GRAPHQL_ENDPOINT_URL).subscription_endpoint(GRAPHQL_ENDPOINT_URL),
))
}
pub struct Subscription;
#[Subscription]
impl Subscription {
async fn tick(&self, ctx: &Context<'_>) -> impl Stream<Item = Option<bool>> {
let pool = ctx.data_unchecked::<RedisClient>().clone();
repeat(0).scan(pool, move |pool, _| {
let pool = pool.clone();
async move { pool.simulate_redis().await.ok() }
})
}
}
pub struct Query;
#[Object]
impl Query {
async fn hello(&self) -> &str {
"world"
}
}This log line in particular is what shows the error occurring: println!("Error sending fake response: {:?}", e);I was able to repro this consistently after making these changes, pointing to an issue with actix most likely dropping the I'm not too familiar with actix so unfortunately I can't be much help here in the short term, but I'll see if I can slim down the repro even further and submit an issue there. Hopefully this is something they already know about or is something that they can give some guidance on. |
|
@aembke Thank you so much for your help!!! Good to know that fred isn't the problem here. I'll too try to slim down the repro to find the root cause and open an issue on the corresponding project. I'll link it here to let you know.
I don't remember getting it, but I'll pay attention to it next time. |
|
No problem. Feel free to tag me if you do open up an issue, and I'll do the same if I'm able to get a smaller repro working. The first 2 issues since I open sourced this have been related to getting it to work with actix, and while I don't personally have a lot of experience with it I keep hearing great things. It was actually really interesting and compelling that you're able to spin up a non-trivial API as easily as you showed in the repro here too. This is probably a sign that I should get more familiar with it and likely add some examples on how to get things working well with fred + actix. In the meantime I'm happy to help out filing or debugging this in actix as well since it seems several people are trying to use these two libraries together. |
|
I'm going to close this out in the meantime but feel free to add to this you learn more, and i'll do the same. |
Description
Hi! Thanks for the awesome library.
I'm having a problem with the Redis client getting stuck before sending the command to the Redis server (it happens with any command).
This is happening in a
Stream, but didn't try to replicate outside of one.The pattern of getting stuck is random, sometimes it runs for tens of stream values and gets stuck and other times it gets stuck on the first value of the stream.
What I tried
I tried to track to the best of my abilities where it was getting stuck and I think it's at
wait_for_response(rx)in src/utils.rs, but I'm not sure.When setting a default timeout it gets hit, otherwise it waits forever.
I also ran
redis-cli monitoron the Redis server and confirmed that the client gets stuck before sending the command.Reproducibility
The minimal code to reproduce it is (UPDATE: this doesn't reproduce the issue, see #13 (comment)):
Cargo.toml:
OS and versions
Rust: 1.56.0
fred: 4.2.1
Redis: 6.2.6 inside Docker
OS: Ubuntu 21.04
The text was updated successfully, but these errors were encountered: