Memory Issues in Multiple Tokio MultiThread Runtimes Scenario

[TimeMST]

tokio = {version = "1.33", features = ["full"]}
glibc 2.17

We have a scenario where multiple runtimes are being used.
Each runtime allocates some objects, which are then deallocated, so there is no memory leak in each runtime. However, each runtime seems to exclusively occupy the memory it has allocated (and deallocated), preventing other runtimes from using that memory. Similarly, other runtimes also exclusively occupy their respective allocated (and deallocated) memory. This ultimately leads to program crashes.
Here is a simple example where a linked list is used for memory allocation purposes. Additionally, we have test jemalloc with an aggressive configuration in this situation, but similar problems still persist. We would like to understand why this is happening, how the tokio runtimes manage their memory. Furthermore, is there a way to share memory between multiple runtimes?

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::runtime::Runtime;
struct AllocCounter {
    count: AtomicUsize,
}
unsafe impl GlobalAlloc for AllocCounter {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        self.count.fetch_add(layout.size(), Ordering::Relaxed);
        System.alloc(layout)
    }
    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        self.count.fetch_sub(layout.size(), Ordering::Relaxed);
        System.dealloc(ptr, layout)
    }
    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
        self.count.fetch_add(layout.size(), Ordering::Relaxed);
        System.alloc_zeroed(layout)
    }
}
#[global_allocator]
static A: AllocCounter = AllocCounter {
    count: AtomicUsize::new(0),
};
fn rt() -> Runtime {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
}
fn log() {
    let process = psutil::process::Process::current().unwrap();  // psutil = "3.2"
    let curr = process.memory_info().unwrap().rss() as i64;
    println!("RSS is {:.3} MB", curr as f64 / 1024.0 / 1024.0);
}

fn main() {
    let rt1 = rt();
    let rt2 = rt();
    let rt3 = rt();
    log();
    println!("Current memory: {}", A.count.load(Ordering::Relaxed));
    rt1.block_on(use_tokio_spawn(128 * 1024 * 1024));  // first runtime
    println!("Current memory: {}", A.count.load(Ordering::Relaxed));
    log();
    rt1.block_on(use_tokio_spawn(128 * 1024 * 1024));  // first again
    println!("Current memory: {}", A.count.load(Ordering::Relaxed));
    log();
    rt2.block_on(use_tokio_spawn(128 * 1024 * 1024));  // second
    println!("Current memory: {}", A.count.load(Ordering::Relaxed));
    log();
    rt3.block_on(use_tokio_spawn(128 * 1024 * 1024));  // third
    println!("Current memory: {}", A.count.load(Ordering::Relaxed));
    log();
}

async fn use_tokio_spawn(cap: usize) {
    let j = tokio::spawn(async move { alloc(cap) });
    j.await.unwrap();
}
fn alloc(cap: usize) {
    let size_of_node = std::mem::size_of::<Node<usize>>();
    let cap = cap / size_of_node;
    let mut list = List::<usize>::new();
    for i in 0..cap {
        list.push(i + 1);
    }
    println!("The memory: {}", A.count.load(Ordering::Relaxed));
}

type Link<T> = Option<Box<Node<T>>>;
struct Node<T> {
    data: T,
    next: Link<T>,
}
struct List<T> {
    head: Link<T>,
}
impl<T> List<T> {
    pub fn new() -> Self {
        Self { head: None }
    }

    pub fn push(&mut self, t: T) {
        let new_node = Box::new(Node {
            data: t,
            next: self.head.take(),
        });
        self.head = Some(new_node);
    }

    pub fn pop(&mut self) -> Option<T> {
        self.head.take().map(|node| {
            self.head = node.next;
            node.data
        })
    }
}
impl<T> Drop for List<T> {
    fn drop(&mut self) {
        let mut head = self.head.take();
        while let Some(mut node) = head {
            head = node.next.take();
        }
    }
}

And there is the output:

RSS is 15.168 MB  // initial RSS
Current memory: 742429  // initial alloc

The memory: 134960325  // after 128MB list allocated in runtime1
Current memory: 742469  // after list deallocated in runtime1
RSS is 271.172 MB  // after first runtime1

The memory: 134960325  // after 128MB list allocated in runtime1 (second time)
Current memory: 742469 // after list deallocated in runtime1 (second time)
RSS is 271.172 MB // after first runtime1 (second time), we believe there is no memory leak here

The memory: 134960325   // after 128MB list allocated in runtime2
Current memory: 742469  // after list deallocated in runtime2
RSS is 526.918 MB  // after runtime2，RSS increased unexpectedly

The memory: 134960325    // after 128MB list allocated in runtime3
Current memory: 742469    // after list deallocated in runtime3
RSS is 782.664 MB // after runtime3，RSS increased unexpectedly

The following is a counterexample against tokio::spawn:

fn use_thread(cap: usize) {
    let t1 = std::thread::spawn(|| alloc(128 * 1024 * 1024));
    t1.join().unwrap();
}

use_thread(128 * 1024 * 1024);  // directly call the function, no tokio runtime

and the output:

RSS is 0.848 MB
Current memory: 1165

The memory: 134219117
Current memory: 1165
RSS is 256.863 MB

The memory: 134219117
Current memory: 1165
RSS is 256.863 MB

The memory: 134219117
Current memory: 1165
RSS is 256.863 MB

The memory: 134219117
Current memory: 1165
RSS is 256.863 MB

It can be observed that after the initial allocation, RSS no longer changes.

Below is output of test on jemalloc on multiple tokio runtimes, It is similar to the results of Unix system allocator.

RSS is 845.062 MB  // initial RSS
Current memory: 741917  // initial alloc

The memory: 134960069  // after 128MB list allocated in runtime1
Current memory: 742213  // after list deallocated in runtime1
RSS is 961.102 MB  // after first runtime1

The memory: 134960069  // after 128MB list allocated in runtime1 (second time)
Current memory: 742213 // after list deallocated in runtime1 (second time)
RSS is 1007.945 MB // after first runtime1 (second time), slightly different with Unix System allocator.

The memory: 134960069   // after 128MB list allocated in runtime2
Current memory: 742213  // after list deallocated in runtime2
RSS is 1125.953 MB  // after runtime2，RSS increased unexpectedly

The memory: 134960069    // after 128MB list allocated in runtime3
Current memory: 742213    // after list deallocated in runtime3
RSS is 1256.957 MB // after runtime3，RSS increased unexpectedly

If additional information is required, please tell us. We need your help!

[Darksonn]

This sounds like memory fragmentation.

Tokio doesn't do anything special when allocating memory. For example, when spawning a task, Tokio just uses Box::new to allocate memory for it.

[TimeMST]

Thanks for your response!

But what bother me is why first runtime can reuse memory it has allocated, while other runtime can't use that memory fragmentation of first runtime.

in our Unix system allocator test case, when first runtime do same allocate operation again, there is no RSS and allocate count increasing.

Does the runtime know which memory fragmentation belong to itself and which belong to other runtimes?

[Darksonn]

All Tokio runtimes use the same memory. They just use Box::new. There's no separation going on.

Could you try to see what happens if you delete these lines?

tokio/tokio/src/runtime/task/core.rs

Lines 69 to 113 in 6871084

	#[cfg_attr(
	any(
	target_arch = "arm",
	target_arch = "mips",
	target_arch = "mips64",
	target_arch = "sparc",
	target_arch = "hexagon",
	),
	repr(align(32))
	)]
	// m68k has 16-byte cache line size.
	//
	// Sources:
	// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/m68k/include/asm/cache.h#L9
	#[cfg_attr(target_arch = "m68k", repr(align(16)))]
	// s390x has 256-byte cache line size.
	//
	// Sources:
	// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_s390x.go#L7
	// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/s390/include/asm/cache.h#L13
	#[cfg_attr(target_arch = "s390x", repr(align(256)))]
	// x86, riscv, wasm, and sparc64 have 64-byte cache line size.
	//
	// Sources:
	// - https://github.com/golang/go/blob/dda2991c2ea0c5914714469c4defc2562a907230/src/internal/cpu/cpu_x86.go#L9
	// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_wasm.go#L7
	// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/sparc/include/asm/cache.h#L19
	// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/riscv/include/asm/cache.h#L10
	//
	// All others are assumed to have 64-byte cache line size.
	#[cfg_attr(
	not(any(
	target_arch = "x86_64",
	target_arch = "aarch64",
	target_arch = "powerpc64",
	target_arch = "arm",
	target_arch = "mips",
	target_arch = "mips64",
	target_arch = "sparc",
	target_arch = "hexagon",
	target_arch = "m68k",
	target_arch = "s390x",
	)),
	repr(align(64))
	)]

[TimeMST]

Of course! I have removed the mentioned code, but it seems that there has been no noticeable change in above three test case.

Unix System Allocator: no change

RSS is 15.168 MB
Current memory: 742429

The memory: 134960325
Current memory: 742469
RSS is 271.172 MB

The memory: 134960325
Current memory: 742469
RSS is 271.172 MB

The memory: 134960325
Current memory: 742469
RSS is 526.918 MB

The memory: 134960325
Current memory: 742469
RSS is 782.664 MB

thread case (counterexample): initial RSS is changed from 0.848MB to 0.844MB, and below RSS s

RSS is 0.844 MB
Current memory: 1165

The memory: 134219117
Current memory: 1165
RSS is 256.859 MB

The memory: 134219117
Current memory: 1165
RSS is 256.859 MB

The memory: 134219117
Current memory: 1165
RSS is 256.859 MB

The memory: 134219117
Current memory: 1165
RSS is 256.859 MB

jemalloc: I found output of jemalloc case varies each time it runs. But the increments between two alloc operations in one execution are close to previous executions.

RSS is 818.969 MB
Current memory: 741405

The memory: 134959941
Current memory: 742085
RSS is 955.055 MB

The memory: 134959941
Current memory: 742085
RSS is 1020.074 MB

The memory: 134959941
Current memory: 742085
RSS is 1153.078 MB

The memory: 134959941
Current memory: 742085
RSS is 1284.082 MB