Builder、Runtime 与 Handle

pub fn new_multi_thread() -> Builder {
    // The number `61` is fairly arbitrary. I believe this value was copied from golang.
    Builder::new(Kind::MultiThread, 61)
}

fn get_cfg(&self) -> driver::Cfg {
    driver::Cfg {
        enable_pause_time: match self.kind {
            Kind::CurrentThread => true,
            #[cfg(feature = "rt-multi-thread")]
            Kind::MultiThread => false,
        },
        enable_io: self.enable_io,
        enable_time: self.enable_time,
        start_paused: self.start_paused,
        nevents: self.nevents,
        nevents_busy: self.nevents_busy,
        timer_flavor: self.timer_flavor,
    }
}

let mut cfg = self.get_cfg();
cfg.timer_flavor = TimerFlavor::Traditional;
let (driver, driver_handle) = driver::Driver::new(cfg)?;

// Blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads, 0);
let blocking_spawner = blocking_pool.spawner().clone();

// Generate a rng seed for this runtime.
let seed_generator_1 = self.seed_generator.next_generator();
let seed_generator_2 = self.seed_generator.next_generator();

// And now put a single-threaded scheduler on top of the timer. When
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let (scheduler, handle) = CurrentThread::new(
    driver,
    driver_handle,
    blocking_spawner,
    seed_generator_2,
    Config { /* ... */ },
    local_tid,
    self.name.clone(),
);

let handle = Handle {
    inner: scheduler::Handle::CurrentThread(handle),
};

pub struct Runtime {
    /// Task scheduler
    scheduler: Scheduler,

    /// Handle to runtime, also contains driver handles
    handle: Handle,

    /// Blocking pool handle, used to signal shutdown
    blocking_pool: BlockingPool,
}

let _enter = self.enter();

match &self.scheduler {
    Scheduler::CurrentThread(exec) => exec.block_on(&self.handle.inner, future),
    #[cfg(feature = "rt-multi-thread")]
    Scheduler::MultiThread(exec) => exec.block_on(&self.handle.inner, future),
}

impl Drop for Runtime {
    fn drop(&mut self) {
        match &mut self.scheduler {
            Scheduler::CurrentThread(current_thread) => {
                // This ensures that tasks spawned on the current-thread
                // runtime are dropped inside the runtime's context.
                let _guard = context::try_set_current(&self.handle.inner);
                current_thread.shutdown(&self.handle.inner);
            }
            #[cfg(feature = "rt-multi-thread")]
            Scheduler::MultiThread(multi_thread) => {
                // The threaded scheduler drops its tasks on its worker threads, which is
                // already in the runtime's context.
                multi_thread.shutdown(&self.handle.inner);
            }
        }
    }
}

#[derive(Debug, Clone)]
pub struct Handle {
    pub(crate) inner: scheduler::Handle,
}

#[derive(Debug, Clone)]
pub(crate) enum Handle {
    #[cfg(feature = "rt")]
    CurrentThread(Arc<current_thread::Handle>),

    #[cfg(feature = "rt-multi-thread")]
    MultiThread(Arc<multi_thread::Handle>),

    // TODO: This is to avoid triggering "dead code" warnings many other places
    // in the codebase. Remove this during a later cleanup
    #[cfg(not(feature = "rt"))]
    #[allow(dead_code)]
    Disabled,
}

match self {
    Handle::CurrentThread(h) => current_thread::Handle::spawn(h, future, id, spawned_at),

    #[cfg(feature = "rt-multi-thread")]
    Handle::MultiThread(h) => multi_thread::Handle::spawn(h, future, id, spawned_at),
}

size_of::<Handle>() = 16

// Enter the runtime context. This sets the current driver handles and
// prevents blocking an existing runtime.
context::enter_runtime(&self.inner, true, |blocking| {
    blocking.block_on(future).expect("failed to park thread")
})

use std::thread;
use std::time::{Duration, Instant};
use tokio::runtime::{Builder, Handle};

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

fn main() {
  println!("size_of::<Handle>() = {}", std::mem::size_of::<Handle>());

  // multi_thread：Handle 拿到别的线程上用
  let rt = Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap();
  let handle = rt.handle().clone();
  let t = thread::Builder::new().name("plain-thread".into()).spawn(move || {
    let jh = handle.spawn(async { name() });
    let on = handle.block_on(async move {
      let spawned_on = jh.await.unwrap();
      (name(), spawned_on)
    });
    println!("[mt] block_on future ran on {}, spawned task ran on {}", on.0, on.1);
  }).unwrap();
  t.join().unwrap();

  // Runtime drop 之后，Handle 还在
  let handle = rt.handle().clone();
  drop(rt);
  let jh = handle.spawn(async { 42 });
  let r = handle.block_on(jh);
  println!("[mt] after drop(rt): spawn -> {:?}", r.map_err(|e| e.is_cancelled()));

  // current_thread：只有 Runtime::block_on 能驱动定时器
  let rt = Builder::new_current_thread().enable_all().build().unwrap();
  let handle = rt.handle().clone();
  let start = Instant::now();
  let t = thread::spawn(move || {
    handle.block_on(async {
      tokio::time::sleep(Duration::from_millis(10)).await;
    });
    println!("[ct] other thread: 10ms sleep finished at {:?}", start.elapsed());
  });
  thread::sleep(Duration::from_millis(300));
  println!("[ct] main: 300ms passed, other thread finished = {}", t.is_finished());
  rt.block_on(async {
    tokio::time::sleep(Duration::from_millis(1)).await;
  });
  t.join().unwrap();
  println!("[ct] main: Runtime::block_on returned at {:?}", start.elapsed());
}

size_of::<Handle>() = 16
[mt] block_on future ran on plain-thread, spawned task ran on tokio-rt-worker
[mt] after drop(rt): spawn -> Err(true)
[ct] main: 300ms passed, other thread finished = false
[ct] other thread: 10ms sleep finished at 306.88275ms
[ct] main: Runtime::block_on returned at 309.395334ms

use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use tokio::runtime::Builder;

static RAN: AtomicUsize = AtomicUsize::new(0);

fn spawn_three(rt: &tokio::runtime::Runtime) {
  for _ in 0..3 {
    rt.spawn(async {
      RAN.fetch_add(1, Relaxed);
    });
  }
}

fn main() {
  // A：block_on 一个立刻完成的 future
  let rt = Builder::new_current_thread().build().unwrap();
  spawn_three(&rt);
  rt.block_on(async {});
  println!("A: block_on(async {{}})        -> ran {}", RAN.swap(0, Relaxed));
  drop(rt);

  // B：block_on 里 yield 一次
  let rt = Builder::new_current_thread().build().unwrap();
  spawn_three(&rt);
  rt.block_on(tokio::task::yield_now());
  println!("B: block_on(yield_now())      -> ran {}", RAN.swap(0, Relaxed));
  drop(rt);

  // C：多线程运行时，spawn 后什么也不做，sleep 50ms
  let rt = Builder::new_multi_thread().worker_threads(2).build().unwrap();
  spawn_three(&rt);
  std::thread::sleep(std::time::Duration::from_millis(50));
  println!("C: multi_thread, no block_on  -> ran {}", RAN.swap(0, Relaxed));
}

A: block_on(async {})        -> ran 0
B: block_on(yield_now())      -> ran 3
C: multi_thread, no block_on  -> ran 3