协程与任务泄漏

概念概述

协程与任务泄漏（Goroutine and Task Leaks）指的是：任务泄漏是因永远等待通道或条件而无法结束的协程，会逐步占用内存与资源，通常源于缺少取消路径。

从更严格的定义看，A goroutine leak is a permanently blocked lightweight task that holds memory and possibly other resources because nothing will ever unblock it. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It appears in servers that spawn tasks per request or subscription when cancellation paths are missing. 它与 I/O Multiplexing with epoll and kqueue、Atomic Operations、Seqlocks 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• leaked tasks are harmless because they are cheap
• the garbage collector reclaims blocked tasks

需要避免的反模式

• starting a task that sends to an unbuffered channel nobody reads
• missing a context cancellation on a long-lived subscription

实践要点

1. A task blocked on a channel send or receive stays reachable and keeps its stack, so a leak grows unboundedly under sustained traffic and eventually exhausts memory.
2. Every task that waits must have a path to exit, typically by selecting on a cancellation signal alongside the data channel.
3. Leak detectors and runtime task counts reveal the pattern, because the count climbs monotonically and never falls back after load subsides.

参考来源

https://go.dev/doc/go1.22

代码示例

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

async fn run_task(id: u32) -> u32 {
    sleep(Duration::from_millis(20 * id as u64)).await;
    id * id
}

async fn supervise(ids: &[u32], deadline: Duration) -> Vec<u32> {
    let mut set = JoinSet::new();
    for &id in ids {
        set.spawn(run_task(id));
    }

    let mut results = Vec::new();
    let outcome = {
        let collect = async {
            while let Some(joined) = set.join_next().await {
                match joined {
                    Ok(value) => results.push(value),
                    Err(error) => eprintln!("task failed: {error}"),
                }
            }
        };
        timeout(deadline, collect).await
    };

    if outcome.is_err() {
        eprintln!("deadline reached, aborting {} tasks", set.len());
        set.abort_all();
    }

    results.sort_unstable();
    results
}

#[tokio::main]
async fn main() {
    let finished = supervise(&[1, 2, 3], Duration::from_secs(1)).await;
    println!("finished: {finished:?}");

    let partial = supervise(&[1, 2, 30], Duration::from_millis(200)).await;
    println!("partial: {partial:?}");
}

import pLimit from 'p-limit';

const limit = pLimit(4);

async function syncAll(ids) {
  const tasks = ids.map((id) =>
    limit(() => fetch(`https://api.example.com/items/${id}`).then((r) => r.json())),
  );
  const results = await Promise.all(tasks);
  return results;
}

const started = [];
const tracker = pLimit(2);
const jobs = [1, 2, 3, 4].map((n) =>
  tracker(async () => {
    started.push(n);
    console.log('running', n, 'active:', started.length);
    await new Promise((r) => setTimeout(r, 100));
    started.pop();
  }),
);
await Promise.all(jobs);