Is there something like JoinSet, but will drop completed task automatically?

[lz1998]

I want to use this feature: When the JoinSet is dropped, all tasks in the JoinSet are immediately aborted.

I don't care what tasks return, and I hope those completed task to be dropped automatically from memory.

pub struct Client {
    pub tasks: JoinSet
}

impl Client {
    pub fn start(&self) {
        loop {
            self.tasks.spawn(async {
                // new task
                // when completed, remove it from tasks automatically. If I use JoinSet, the memory usage will grow.
            })
        }
    }
}

// when client is dropped, all the tasks are dropped.

[Darksonn]

You could remove all completed tasks from the set every time you spawn?

[Darksonn]

So you are saying that it worked?

[lz1998]

It can be compiled and runned successfully, but the memory usage doesn't decline after while let Some(Some(_)) = join_set.join_next().now_or_never() {}. The memory usage is expected to decline after removing completed tasks.

When 1 is printed, it's 325MB. When 2 is printed it's still 325MB.

[Darksonn]

Unfortunately, you cannot measure a program's real memory usage that way. The standard memory allocator will keep the memory around as a performance optimization, since it is much cheaper to reuse the memory than to give it back to the OS, and then ask for the OS to give it back.

Please try using the stats_alloc crate.

[Darksonn]

Note that it may help to wrap the calls in tokio::task::unconstrained.

pub fn clean(&mut self) {
    while let Some(Some(_)) = tokio::task::unconstrained(self.tasks.join_next()).now_or_never() {}
}

[lz1998]

// futures = "0.3"
// tokio = { version = "1", features = ["full"] }
// stats_alloc = "0.1.10"
extern crate stats_alloc;

use futures::executor::ThreadPool;
use futures::future::FutureObj;
use futures::stream::FuturesUnordered;
use futures::task::SpawnExt;
use futures::{FutureExt, StreamExt};
use stats_alloc::{Region, StatsAlloc, INSTRUMENTED_SYSTEM};
use std::alloc::System;
use std::future::Future;

#[global_allocator]
static GLOBAL: &StatsAlloc<System> = &INSTRUMENTED_SYSTEM;

#[derive(Default)]
pub struct Client {
    pub tasks: FuturesUnordered<FutureObj<'static, ()>>,
}

impl Client {
    pub fn spawn<F, T>(&mut self, f: F)
    where
        F: Future<Output = T>,
        F: Send + 'static,
        T: Send,
    {
        self.clean();
        self.tasks
            .spawn(async {
                f.await;
            })
            .expect("failed to spawn");
    }

    pub fn clean(&mut self) {
        while let Some(Some(_)) = self.tasks.next().now_or_never() {}
    }
}

#[tokio::main]
async fn main() {
    let reg = Region::new(&GLOBAL);
    let mut cli = Client::default();

    for _ in 0..1000000 {
        cli.spawn(async {
            tokio::time::sleep(std::time::Duration::from_secs(20)).await;
        });
    }
    println!("Stats at 1: {:#?}", reg.change());
    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
    cli.clean();
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    println!("Stats at 2: {:#?}", reg.change());
    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
    println!("Stats at 3: {:#?}", reg.change());
    drop(cli);
    println!("Stats at 4: {:#?}", reg.change());
    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
}

Stats at 1: Stats {
    allocations: 2000003,
    deallocations: 0,
    reallocations: 0,
    bytes_allocated: 320000216,
    bytes_deallocated: 0,
    bytes_reallocated: 0,
}
Stats at 2: Stats {
    allocations: 2000005,
    deallocations: 254,
    reallocations: 0,
    bytes_allocated: 320001304,
    bytes_deallocated: 40640,
    bytes_reallocated: 0,
}
Stats at 3: Stats {
    allocations: 2000005,
    deallocations: 254,
    reallocations: 0,
    bytes_allocated: 320001304,
    bytes_deallocated: 40640,
    bytes_reallocated: 0,
}
Stats at 4: Stats {
    allocations: 2000005,
    deallocations: 2000002,
    reallocations: 0,
    bytes_allocated: 320001304,
    bytes_deallocated: 320000152,
    bytes_reallocated: 0,
}

[lz1998]

It works, thanks 👍