spawn 全路径

#[track_caller]
pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,
{
    let fut_size = std::mem::size_of::<F>();
    if AutoBox::<F>::SHOULD_BOX {
        spawn_inner(Box::pin(future), SpawnMeta::new_unnamed(fut_size))
    } else {
        spawn_inner(future, SpawnMeta::new_unnamed(fut_size))
    }
}

/// Boundary value to prevent stack overflow caused by a large-sized
/// Future being placed in the stack.
pub(crate) const BOX_FUTURE_THRESHOLD: usize = if cfg!(debug_assertions)  {
    2048
} else {
    16384
};

let id = task::Id::next();
let task = crate::util::trace::task(future, "task", meta, id.as_u64());

match context::with_current(|handle| handle.spawn(task, id, meta.spawned_at)) {
    Ok(join_handle) => join_handle,
    Err(e) => panic!("{}", e),
}

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()),
    }
}

let (handle, notified) = me.shared.owned.bind(future, me.clone(), id, spawned_at);

me.task_hooks.spawn(&TaskMeta {
    id,
    spawned_at,
    #[cfg(all(tokio_unstable, feature = "schedule-latency"))]
    schedule_latency: None,
    _phantom: Default::default(),
});

me.schedule_option_task_without_yield(notified);

handle

with_current(|maybe_cx| {
    if let Some(cx) = maybe_cx {
        // Make sure the task is part of the **current** scheduler.
        if self.ptr_eq(&cx.worker.handle) {
            // And the current thread still holds a core
            if let Some(core) = cx.core.borrow_mut().as_mut() {
                self.schedule_local(core, task, is_yield);
                return;
            }
        }
    }

    // Otherwise, use the inject queue.
    self.push_remote_task(task);
    self.notify_parked_remote();
});

context::with_scheduler(|maybe_cx| match maybe_cx {
    Some(CurrentThread(cx)) if Arc::ptr_eq(self, &cx.handle) => {
        let mut core = cx.core.borrow_mut();

        // If `None`, the runtime is shutting down, so there is no need
        // to schedule the task.
        if let Some(core) = core.as_mut() {
            core.push_task(self, task);
        }
        self.driver.unpark_self();
    }
    _ => {
        // Track that a task was scheduled from **outside** of the runtime.
        self.shared.scheduler_metrics.inc_remote_schedule_count();

        // Schedule the task
        self.shared.inject.push(task);
        self.driver.unpark();
    }
});

use std::thread;

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

fn main() {
  let rt = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(4)
    .thread_name_fn(|| {
      use std::sync::atomic::{AtomicUsize, Ordering};
      static N: AtomicUsize = AtomicUsize::new(0);
      format!("worker-{}", N.fetch_add(1, Ordering::Relaxed))
    })
    .build()
    .unwrap();

  rt.block_on(async {
    println!("block_on body runs on: {}", name());
    // spawn from the block_on thread (not a worker): goes to the inject queue
    let t = tokio::spawn(async { name() }).await.unwrap();
    println!("spawned from main, first polled on: {t}");

    // spawn 8 children from inside a worker task, 200 rounds
    let mut same = [0u32; 8];
    for _ in 0..200 {
      let res = tokio::spawn(async {
        let parent = name();
        let hs: Vec<_> = (0..8).map(|_| tokio::spawn(async { name() })).collect();
        let mut out = [false; 8];
        for (i, h) in hs.into_iter().enumerate() {
          out[i] = h.await.unwrap() == parent;
        }
        out
      })
      .await
      .unwrap();
      for i in 0..8 {
        if res[i] {
          same[i] += 1;
        }
      }
    }
    for i in 0..8 {
      println!("child #{i}: ran on parent's thread in {}/200 rounds", same[i]);
    }
  });
}

block_on body runs on: main
spawned from main, first polled on: worker-0
child #0: ran on parent's thread in 200/200 rounds
child #1: ran on parent's thread in 200/200 rounds
child #2: ran on parent's thread in 199/200 rounds
child #3: ran on parent's thread in 197/200 rounds
child #4: ran on parent's thread in 197/200 rounds
child #5: ran on parent's thread in 146/200 rounds
child #6: ran on parent's thread in 126/200 rounds
child #7: ran on parent's thread in 200/200 rounds

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;

struct Rec;
static ON: AtomicBool = AtomicBool::new(false);
static SIZES: Mutex<Vec<usize>> = Mutex::new(Vec::new());

unsafe impl GlobalAlloc for Rec {
  unsafe fn alloc(&self, l: Layout) -> *mut u8 {
    if l.size() >= 64 && ON.swap(false, Ordering::SeqCst) {
      SIZES.lock().unwrap().push(l.size());
      ON.store(true, Ordering::SeqCst);
    }
    unsafe { System.alloc(l) }
  }
  unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
    unsafe { System.dealloc(p, l) }
  }
}

#[global_allocator]
static A: Rec = Rec;

fn spawn_sized<const N: usize>(rt: &tokio::runtime::Runtime) {
  let buf = [1u8; N];
  let fut = async move {
    std::hint::black_box(&buf);
  };
  let size = std::mem::size_of_val(&fut);
  SIZES.lock().unwrap().clear();
  ON.store(true, Ordering::SeqCst);
  let h = rt.spawn(fut);
  ON.store(false, Ordering::SeqCst);
  let sizes = SIZES.lock().unwrap().clone();
  println!("future {size:>5} bytes -> allocations >= 64B in spawn: {sizes:?}");
  rt.block_on(h).unwrap();
}

fn main() {
  let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
  spawn_sized::<16000>(&rt);
  spawn_sized::<16383>(&rt);
  spawn_sized::<16384>(&rt);
  spawn_sized::<17000>(&rt);
}

future 16001 bytes -> allocations >= 64B in spawn: [16128]
future 16384 bytes -> allocations >= 64B in spawn: [16512]
future 16385 bytes -> allocations >= 64B in spawn: [16385, 128]
future 17001 bytes -> allocations >= 64B in spawn: [17001, 128]

future 16001 bytes -> allocations >= 64B in spawn: [16001, 128]
future 16384 bytes -> allocations >= 64B in spawn: [16384, 128]
future 16385 bytes -> allocations >= 64B in spawn: [16385, 128]
future 17001 bytes -> allocations >= 64B in spawn: [17001, 128]

current_thread: 1, 2
multi_thread(4): 7
current_thread again: 8, 9