线程本地上下文

struct Context {
    /// Uniquely identifies the current thread
    #[cfg(feature = "rt")]
    thread_id: Cell<Option<ThreadId>>,

    /// Handle to the runtime scheduler running on the current thread.
    #[cfg(feature = "rt")]
    current: current::HandleCell,

    /// Handle to the scheduler's internal "context"
    #[cfg(feature = "rt")]
    scheduler: Scoped<scheduler::Context>,

    #[cfg(feature = "rt")]
    current_task_id: Cell<Option<Id>>,

    /// Tracks if the current thread is currently driving a runtime.
    /// Note, that if this is set to "entered", the current scheduler
    /// handle may not reference the runtime currently executing. This
    /// is because other runtime handles may be set to current from
    /// within a runtime.
    #[cfg(feature = "rt")]
    runtime: Cell<EnterRuntime>,

    #[cfg(any(feature = "rt", feature = "macros"))]
    rng: Cell<Option<FastRand>>,

    /// Tracks the amount of "work" a task may still do before yielding back to
    /// the scheduler
    budget: Cell<coop::Budget>,

    // taskdump 用的 trace 字段，只在 Linux + tokio_unstable 下存在
    // ...
}

tokio_thread_local! {
    static CONTEXT: Context = const {
        Context {
            // ...
            runtime: Cell::new(EnterRuntime::NotEntered),
            // ...
            budget: Cell::new(coop::Budget::unconstrained()),
            // ...
        }
    }
}

pub(super) struct HandleCell {
    /// Current handle
    handle: RefCell<Option<scheduler::Handle>>,

    /// Tracks the number of nested calls to `try_set_current`.
    depth: Cell<usize>,
}

pub(super) fn set_current(&self, handle: &scheduler::Handle) -> SetCurrentGuard {
    let old_handle = self.current.handle.borrow_mut().replace(handle.clone());
    let depth = self.current.depth.get();

    assert!(depth != usize::MAX, "reached max `enter` depth");

    let depth = depth + 1;
    self.current.depth.set(depth);

    SetCurrentGuard {
        prev: old_handle,
        depth,
        _p: PhantomData,
    }
}

impl Drop for SetCurrentGuard {
    fn drop(&mut self) {
        CONTEXT.with(|ctx| {
            let depth = ctx.current.depth.get();

            if depth != self.depth {
                if !std::thread::panicking() {
                    panic!(
                        "`EnterGuard` values dropped out of order. Guards returned by \
                         `tokio::runtime::Handle::enter()` must be dropped in the reverse \
                         order as they were acquired."
                    );
                } else {
                    // Just return... this will leave handles in a wonky state though...
                    return;
                }
            }

            *ctx.current.handle.borrow_mut() = self.prev.take();
            ctx.current.depth.set(depth - 1);
        });
    }
}

pub struct EnterGuard<'a> {
    _guard: context::SetCurrentGuard,
    _handle_lifetime: PhantomData<&'a Handle>,
}

pub(crate) fn with_current<F, R>(f: F) -> Result<R, TryCurrentError>
where
    F: FnOnce(&scheduler::Handle) -> R,
{
    match CONTEXT.try_with(|ctx| ctx.current.handle.borrow().as_ref().map(f)) {
        Ok(Some(ret)) => Ok(ret),
        Ok(None) => Err(TryCurrentError::new_no_context()),
        Err(_access_error) => Err(TryCurrentError::new_thread_local_destroyed()),
    }
}

there is no reactor running, must be called from the context of a Tokio 1.x runtime

let handle = scheduler::Handle::current();
// Panic if the time driver is not enabled (backwards compat)
_ = handle.driver().time();

pub(crate) fn enter_runtime<F, R>(handle: &scheduler::Handle, allow_block_in_place: bool, f: F) -> R
where
    F: FnOnce(&mut BlockingRegionGuard) -> R,
{
    let maybe_guard = CONTEXT.with(|c| {
        if c.runtime.get().is_entered() {
            None
        } else {
            // Set the entered flag
            c.runtime.set(EnterRuntime::Entered {
                allow_block_in_place,
            });

            // Generate a new seed
            let rng_seed = handle.seed_generator().next_seed();

            // Swap the RNG seed
            let mut rng = c.rng.get().unwrap_or_else(FastRand::new);
            let old_seed = rng.replace_seed(rng_seed);
            c.rng.set(Some(rng));

            Some(EnterRuntimeGuard {
                blocking: BlockingRegionGuard::new(),
                handle: c.set_current(handle),
                old_seed,
            })
        }
    });

    if let Some(mut guard) = maybe_guard {
        return f(&mut guard.blocking);
    }

    panic!(
        "Cannot start a runtime from within a runtime. This happens \
            because a function (like `block_on`) attempted to block the \
            current thread while the thread is being used to drive \
            asynchronous tasks. If you are in an async function, use \
            `.await` on the future instead of blocking on it."
    );
}

pub(crate) fn block_on<F: Future>(f: F) -> F::Output {
    let mut e = crate::runtime::context::try_enter_blocking_region().expect(
        "Cannot block the current thread from within a runtime. This \
        happens because a function attempted to block the current \
        thread while the thread is being used to drive asynchronous \
        tasks."
    );
    e.block_on(f).unwrap()
}

pub(super) fn with_scheduler<R>(f: impl FnOnce(Option<&scheduler::Context>) -> R) -> R {
    let mut f = Some(f);
    CONTEXT.try_with(|c| {
        let f = f.take().unwrap();
        if matches!(c.runtime.get(), EnterRuntime::Entered { .. }) {
            c.scheduler.with(f)
        } else {
            f(None)
        }
    })
        .unwrap_or_else(|_| (f.take().unwrap())(None))
}

use std::panic;
use tokio::runtime::{Builder, Handle};

// 只打印 panic 的消息，不打印位置和 backtrace 提示
fn quiet_hook() {
  panic::set_hook(Box::new(|info| {
    let msg = info
      .payload()
      .downcast_ref::<String>()
      .cloned()
      .or_else(|| info.payload().downcast_ref::<&str>().map(|s| s.to_string()))
      .unwrap_or_default();
    println!("  panic: {msg}");
  }));
}

fn main() {
  quiet_hook();

  println!("1. 运行时之外 Handle::try_current()");
  match Handle::try_current() {
    Ok(_) => println!("  ok"),
    Err(e) => println!("  Err: {e} (is_missing_context = {})", e.is_missing_context()),
  }

  println!("2. 运行时之外 tokio::spawn");
  let _ = panic::catch_unwind(|| {
    tokio::spawn(async {});
  });

  let rt = Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap();

  println!("3. rt.enter() 之后 Handle::current() 和 block_on");
  {
    let _guard = rt.enter();
    let h = Handle::current();
    println!("  current flavor = {:?}", h.runtime_flavor());
    let v = rt.block_on(async { 1 + 1 });
    println!("  block_on inside enter() -> {v}");
  }

  println!("4. 在 block_on 的 future 里再 block_on");
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    rt.block_on(async {
      Handle::current().block_on(async {});
    });
  }));

  println!("5. 在 spawn 出去的任务里 blocking_recv");
  let r = rt.block_on(async {
    tokio::spawn(async {
      let (_tx, mut rx) = tokio::sync::mpsc::channel::<u8>(1);
      let _ = rx.blocking_recv();
    })
    .await
  });
  println!("  JoinError is_panic = {}", r.unwrap_err().is_panic());

  println!("6. spawn_blocking 线程里 Handle::current().block_on");
  let v = rt.block_on(async {
    tokio::task::spawn_blocking(|| Handle::current().block_on(async { "fine" })).await.unwrap()
  });
  println!("  -> {v}");

  println!("7. sleep 在 block_on 之外构造");
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    rt.block_on(tokio::time::sleep(std::time::Duration::from_millis(1)));
  }));
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    rt.block_on(async { tokio::time::sleep(std::time::Duration::from_millis(1)).await });
    println!("  sleep built inside block_on: ok");
  }));

  println!("8. EnterGuard 乱序 drop");
  let rt2 = Builder::new_current_thread().build().unwrap();
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    let g1 = rt.enter();
    let g2 = rt2.enter();
    drop(g1);
    drop(g2);
  }));
}

1. 运行时之外 Handle::try_current()
  Err: there is no reactor running, must be called from the context of a Tokio 1.x runtime (is_missing_context = true)
2. 运行时之外 tokio::spawn
  panic: there is no reactor running, must be called from the context of a Tokio 1.x runtime
3. rt.enter() 之后 Handle::current() 和 block_on
  current flavor = MultiThread
  block_on inside enter() -> 2
4. 在 block_on 的 future 里再 block_on
  panic: Cannot start a runtime from within a runtime. This happens because a function (like `block_on`) attempted to block the current thread while the thread is being used to drive asynchronous tasks. If you are in an async function, use `.await` on the future instead of blocking on it.
5. 在 spawn 出去的任务里 blocking_recv
  panic: Cannot block the current thread from within a runtime. This happens because a function attempted to block the current thread while the thread is being used to drive asynchronous tasks.
  JoinError is_panic = true
6. spawn_blocking 线程里 Handle::current().block_on
  -> fine
7. sleep 在 block_on 之外构造
  panic: there is no reactor running, must be called from the context of a Tokio 1.x runtime
  sleep built inside block_on: ok
8. EnterGuard 乱序 drop
  panic: `EnterGuard` values dropped out of order. Guards returned by `tokio::runtime::Handle::enter()` must be dropped in the reverse order as they were acquired.

use std::panic;
use std::thread;
use tokio::runtime::{Builder, Runtime};

fn name() -> String {
  thread::current().name().unwrap_or("?").to_string()
}

fn rt(n: &str) -> Runtime {
  Builder::new_multi_thread().worker_threads(1).thread_name(n).build().unwrap()
}

fn main() {
  panic::set_hook(Box::new(|info| {
    let msg = info.payload().downcast_ref::<&str>().map(|s| s.to_string())
      .or_else(|| info.payload().downcast_ref::<String>().cloned()).unwrap_or_default();
    println!("   panic: {}", &msg[..msg.find('.').unwrap_or(msg.len())]);
  }));
  let rt_a = rt("rt-a");
  let rt_b = rt("rt-b");

  rt_a.block_on(async {
    println!("A: {}", tokio::spawn(async { name() }).await.unwrap());
    {
      let _g = rt_b.enter();
      println!("B: {}", tokio::spawn(async { name() }).await.unwrap());
      let r = panic::catch_unwind(panic::AssertUnwindSafe(|| rt_b.block_on(async { 1 })));
      println!("C: rt_b.block_on is_err = {}", r.is_err());
    }
    println!("D: {}", tokio::spawn(async { name() }).await.unwrap());
  });
}

A: rt-a
B: rt-b
   panic: Cannot start a runtime from within a runtime
C: rt_b.block_on is_err = true
D: rt-a