Rust Tokio task 누수로 RAM 폭증 진단법

// Cargo.toml
// tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
// tokio-metrics = "0.3"

use tokio_metrics::RuntimeMonitor;
use std::time::Duration;

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    let handle = tokio::runtime::Handle::current();
    let monitor = RuntimeMonitor::new(&handle);

    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(5));
        loop {
            interval.tick().await;
            let m = monitor.cumulative();
            // 핵심: total_tasks_count, active_tasks_count 같은 지표를 로그/메트릭으로
            eprintln!(
                "tasks_total={} tasks_active={} park_count={} busy_duration={:?}",
                m.total_tasks_count,
                m.active_tasks_count,
                m.total_park_count,
                m.total_busy_duration
            );
        }
    });

    // 서버 로직...
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;
    }
}

use tracing::{info, Instrument};
use std::sync::atomic::{AtomicU64, Ordering};

static TASK_SEQ: AtomicU64 = AtomicU64::new(1);

fn next_task_id() -> u64 {
    TASK_SEQ.fetch_add(1, Ordering::Relaxed)
}

async fn spawn_labeled<F>(name: &'static str, fut: F) -> tokio::task::JoinHandle<()>
where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    let id = next_task_id();
    tokio::spawn(
        async move {
            info!(task.id = id, task.name = name, "task_start");
            fut.await;
            info!(task.id = id, task.name = name, "task_end");
        }
        .instrument(tracing::info_span!("task", task.id = id, task.name = name)),
    )
}

use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

struct TaskGroup {
    cancel: CancellationToken,
    handles: Vec<JoinHandle<()>>,
}

impl TaskGroup {
    fn new() -> Self {
        Self { cancel: CancellationToken::new(), handles: Vec::new() }
    }

    fn token(&self) -> CancellationToken {
        self.cancel.clone()
    }

    fn spawn(&mut self, h: JoinHandle<()>) {
        self.handles.push(h);
    }

    async fn shutdown(mut self) {
        self.cancel.cancel();
        for h in self.handles.drain(..) {
            let _ = h.await;
        }
    }
}

use tokio::select;

async fn worker(mut rx: tokio::sync::mpsc::Receiver<Vec<u8>>) {
    loop {
        select! {
            msg = rx.recv() => {
                // rx가 닫히면 msg는 None이 되지만, 이를 처리하지 않으면 루프가 계속 돈다
                if let Some(_m) = msg {
                    // 처리
                }
            }
            _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {
                // 주기 작업
            }
        }
    }
}

async fn worker(
    mut rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
    cancel: tokio_util::sync::CancellationToken,
) {
    loop {
        tokio::select! {
            _ = cancel.cancelled() => {
                break;
            }
            msg = rx.recv() => {
                match msg {
                    Some(_m) => { /* 처리 */ }
                    None => break, // 채널 종료 시 worker 종료
                }
            }
        }
    }
}

use tokio::time::{timeout, Duration};

async fn call_with_timeout() -> anyhow::Result<()> {
    let res = timeout(Duration::from_secs(2), async {
        // 외부 호출
        Ok::<_, anyhow::Error>(())
    }).await;

    match res {
        Ok(inner) => inner,
        Err(_) => anyhow::bail!("timeout"),
    }
}

let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1024);

// 생산자
tokio::spawn(async move {
    loop {
        let payload = vec![0u8; 4096];
        if tx.send(payload).await.is_err() {
            break; // 소비자 종료 시 생산자도 종료
        }
    }
});

// 소비자
tokio::spawn(async move {
    while let Some(_msg) = rx.recv().await {
        // 처리
    }
});

use std::sync::{Arc, Weak};

struct Hub {
    // 구독자 목록 등을 Weak로 보관
    subs: Vec<Weak<Subscriber>>,
}

struct Subscriber {
    // ...
}

use tokio::task::JoinSet;
use tokio::time::{sleep, Duration};

async fn run_supervisor() {
    let mut set = JoinSet::new();

    for _ in 0..100 {
        set.spawn(async {
            // 작업
        });
    }

    while let Some(res) = set.join_next().await {
        if res.is_err() {
            sleep(Duration::from_millis(200)).await; // 최소 backoff
        }
    }
}

use tokio::sync::Semaphore;
use std::sync::Arc;

async fn limited_blocking_work(sem: Arc<Semaphore>) {
    let _permit = sem.acquire().await.unwrap();
    let _ = tokio::task::spawn_blocking(move || {
        // CPU/IO heavy
    }).await;
}

#[tokio::test]
async fn worker_exits_on_channel_close() {
    let (tx, rx) = tokio::sync::mpsc::channel::<u8>(8);
    let cancel = tokio_util::sync::CancellationToken::new();

    let h = tokio::spawn(async move {
        super::worker(rx, cancel).await;
    });

    drop(tx); // 채널 닫기

    // 일정 시간 내 종료 보장
    tokio::time::timeout(std::time::Duration::from_secs(1), h)
        .await
        .expect("worker did not exit");
}