Task in LocalSet not being invoked at all

help
1

In this code there's two async tasks that keep incrementing a shared counter. One is spawned in a local set while the other is spawned using tokio::spawn. It seems only the latter is being scheduled continuously.

I have a larger app where I am trying to use this, but isolated my issue into this example

use std::sync::Arc;
use std::time::Duration;
use tokio::task::{self, JoinHandle};
use tokio::sync::Mutex;

struct State {
    count: u32
}

type SharedState = Arc<Mutex<State>>;

#[tokio::main]
async fn main() {
    let local_set = task::LocalSet::new();
    
    let state = Arc::new(Mutex::new(State{
        count: 0
    }));
    
    let state_clone_1 = state.clone();

    // Call start_loop_local in local set
    let loop_1 = local_set.run_until(async move {
        start_loop_local(state_clone_1, Duration::from_millis(100), "local set").await
    }).await;
    
    
    // Spawn one in runtime block
    let state_clone_2 = state.clone();
    
    // Call start_loop as a normal task
    let loop_2 = tokio::spawn(async move {
        start_loop(state_clone_2, Duration::from_millis(500), "top level set").await
    }).await.expect("Failed to spawn second task");

    // let j = tokio::join!(loop_1, loop_2);

    // Confirming if the local set task is early returning for some reason.
    loop_1.await.expect("Failed to wait on first task");

    println!("Local set task finished");

    loop_2.await.expect("Failed to wait on second task");

    println!("Top level task finished");
}

async fn start_loop_local(state: SharedState, interval: Duration, name: &str) -> JoinHandle<()> {
    let name = name.to_string();
    tokio::task::spawn_local(async move {
        println!("Calling keep_incrementing");
        keep_incrementing(state, interval, name).await;
    })
}

async fn start_loop(state: SharedState, interval: Duration, name: &str) -> JoinHandle<()> {
    let name = name.to_string();
    tokio::spawn(async move {
        println!("Calling keep_incrementing");
        keep_incrementing(state, interval, name).await;
    })
}


async fn keep_incrementing(state: SharedState, interval: Duration, name: String) {
    let interval = tokio::time::interval(interval);
    tokio::pin!(interval);
    
    loop {
        interval.as_mut().tick().await;
        let mut guard = state.lock().await;
        guard.count += 1;
        
        println!("{}: Incremented to {}", name, guard.count);
    }
}
2

I tested it locally, not sure if it is what you want.

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::task::{self, JoinHandle};

struct State {
    count: u32,
}

type SharedState = Arc<Mutex<State>>;

#[tokio::main]
async fn main() {
    let local_set = task::LocalSet::new();

    let state = Arc::new(Mutex::new(State { count: 0 }));

    let state_clone_1 = state.clone();

    // Call start_loop_local in local set
    let loop_1 = local_set.run_until(async move {
        start_loop_local(state_clone_1, Duration::from_millis(100), "local set").await
    });

    // Spawn one in runtime block
    let state_clone_2 = state.clone();

    // Call start_loop as a normal task
    let loop_2 = tokio::spawn(async move {
        start_loop(state_clone_2, Duration::from_millis(500), "top level set").await
    });

    match tokio::join!(loop_1, loop_2) {
        (Ok(_), Ok(_)) => println!("Both tasks finished"),
        (Err(e), _) => println!("Error in first task: {:?}", e),
        (_, Err(e)) => println!("Error in second task: {:?}", e),
    }
}

fn start_loop_local(state: SharedState, interval: Duration, name: &str) -> JoinHandle<()> {
    let name = name.to_string();
    tokio::task::spawn_local(async move {
        println!("Calling keep_incrementing 1");
        keep_incrementing(state, interval, name).await;
    })
}

fn start_loop(state: SharedState, interval: Duration, name: &str) -> JoinHandle<()> {
    let name = name.to_string();
    tokio::spawn(async move {
        println!("Calling keep_incrementing 2");
        keep_incrementing(state, interval, name).await;
    })
}

async fn keep_incrementing(state: SharedState, interval: Duration, name: String) {
    let interval = tokio::time::interval(interval);
    tokio::pin!(interval);

    loop {
        interval.as_mut().tick().await;
        let mut guard = state.lock().await;
        guard.count += 1;

        println!("{}: Incremented to {}", name, guard.count);
    }
}

result:

Calling keep_incrementing 1
Calling keep_incrementing 2    
local set: Incremented to 1    
top level set: Incremented to 2
local set: Incremented to 3
local set: Incremented to 4
local set: Incremented to 5
local set: Incremented to 6
top level set: Incremented to 7
local set: Incremented to 8
local set: Incremented to 9
local set: Incremented to 10
local set: Incremented to 11
local set: Incremented to 12
top level set: Incremented to 13
...
3

Yes, this was the issue. I put the await on the local set run_until before creating the normal task, so the latter code was not reached at all since the local set is being run in the current thread itself and the task never returns as its a loop.