Rx.recv doesn't work but rx.try_recv do

help
1

Hello as the title says i am using recv in a tokio spawn funciton but it doesn't work.

 tokio::spawn(async move {
     while let Some(update) = rx.recv().await {
                info!("received {:?}", update);
            }
}

when I change it to this:

tokio::spawn(async move {
     loop {
                match rx.try_recv() {
                    Ok(update) => {
                        info!("received {:?}", update);

                    }
                    Err(err) => (),
                }
            }
}

it works correctly , anyone got an idea of why ?
It shouldn't be an issue on the sender side because it works correctly whith try_recv and I tested it with try_send.

2

Your try_recv example is going to block the runtime, i.e. not await for a relatively long amount of time, which can produce unexpected behavior. The fact it works suggests me that you might also be blocking the runtime somewhere else. Is this all your code?

3

here is all my code , it contain both cases ( I know it doesn't execute because of the loops , i just keep it here to change whenever i'll find the solution ).

tokio::spawn(async move {
            info!("task spawned");
            info!("{}", rx.is_closed());
            loop { //this one works
                match rx.try_recv() {
                    Ok(update) => {
                        match update {
                            MarketUpdate::Bid(tick, price) => {
                                // Handle bid price update
                                let mut actions_unlocked = actions.lock().await;

                                if let Some(action) = actions_unlocked.get_mut(&tick) {
                                    action.update_bid(price);
                                    info!(
                                        "updated ask price for {} with value : {:?}",
                                        action.symbol, action.bid
                                    )
                                }
                            }
                            MarketUpdate::Ask(tick, price) => {
                                let mut actions_unlocked = actions.lock().await;

                                if let Some(action) = actions_unlocked.get_mut(&tick) {
                                    action.update_ask(price);
                                    info!(
                                        "updated ask price for {} with value : {:?}",
                                        action.symbol, action.ask
                                    )
                                }
                                // Handle ask price update
                            }
                            MarketUpdate::BidSize(tick, size) => {
                                // Handle bid size update
                            }
                            MarketUpdate::AskSize(tick, size) => {
                                // Handle ask size update
                            }
                        }
                    }
                    Err(err) => (),
                }
            }
            while let Some(update) = rx.recv().await { //this one doesn't
                info!("received {:?}", update);
                match update {
                    MarketUpdate::Bid(tick, price) => {
                        // Handle bid price update
                        let mut actions_unlocked = actions.lock().await;

                        if let Some(action) = actions_unlocked.get_mut(&tick) {
                            action.update_bid(price);
                            info!(
                                "updated ask price for {} with value : {:?}",
                                action.symbol, action.bid
                            )
                        }
                    }
                    MarketUpdate::Ask(tick, price) => {
                        let mut actions_unlocked = actions.lock().await;

                        if let Some(action) = actions_unlocked.get_mut(&tick) {
                            action.update_ask(price);
                            info!(
                                "updated ask price for {} with value : {:?}",
                                action.symbol, action.ask
                            )
                        }
                        // Handle ask price update
                    }
                    MarketUpdate::BidSize(tick, size) => {
                        // Handle bid size update
                    }
                    MarketUpdate::AskSize(tick, size) => {
                        // Handle ask size update
                    }
                }
            }
            info!("task ended");
        });
4

I meant code other than the one doing the recv. In particular you should look for code that might stuck somewhere never reaching a .await

5

no this is all my code in this spawn function

6

Async executors expect all spawned tasks to be cooperative (doing some small amount of work then yielding back to the executor via await). Something outside of this spawn can be blocking instead of awaiting and affect the behavior of the block you shared.

7

Any blocking tokio::spawn can create issues for any other task.

8

why ? It doesn't realy make sens for me aren't they suppose to work in a different thread ? Maybe I understood something wrong about it.

Here is my other spanw :

 spawn(async move {
            let subscription = subscription_struct.borrow_subscription();
            while let Some(tick) = subscription.next() {
                // info!("recevied {:?}", tick);
                match tick {
                    TickTypes::PriceSize(tick_price) => match tick_price.price_tick_type {
                        TickType::Bid => {
                            info!("sending bid");
                            let _ = tx.send(MarketUpdate::Bid(action.clone(), tick_price.price));
                        }
                        TickType::Ask => {
                            info!("sending ask");

                            let _ = tx.send(MarketUpdate::Ask(action.clone(), tick_price.price));
                        }
                        TickType::DelayedAsk => {
                            info!("sending ask");

                            let _ = tx
                                .send(MarketUpdate::Ask(action.clone(), tick_price.price))
                                .await
                                .and_then(|_| {
                                    info!("ask sent");
                                    Ok(())
                                });
                        }
                        TickType::DelayedBid => {
                            let _ = tx.send(MarketUpdate::Bid(action.clone(), tick_price.price));
                        }
                        _ => {}
                    },
                    TickTypes::SnapshotEnd => subscription.cancel(),
                    _ => {}
                }
            }
        });

I also tried without await.and_then() in both cases it only work with try_recv.

9

Nope, all tokio::spawns work on the same threadpool, so they can definitely end up interleaving on the same thread.