current_thread

pub(crate) struct CurrentThread {
    /// Core scheduler data is acquired by a thread entering `block_on`.
    core: AtomicCell<Core>,

    /// Notifier for waking up other threads to steal the
    /// driver.
    notify: Notify,
}

struct Core {
    /// Scheduler run queue
    tasks: VecDeque<Notified>,

    /// Current tick
    tick: u32,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    // ...

    /// How often to check the global queue
    global_queue_interval: u32,

    // ...
}

struct Shared {
    /// Remote run queue
    inject: Inject<Arc<Handle>>,

    /// Collection of all active tasks spawned onto this executor.
    owned: OwnedTasks<Arc<Handle>>,

    /// Indicates whether the blocked on thread was woken.
    woken: AtomicBool,

    // ...
}

crate::runtime::context::enter_runtime(handle, false, |blocking| {
    let handle = handle.as_current_thread();

    // Attempt to steal the scheduler core and block_on the future if we can
    // there, otherwise, lets select on a notification that the core is
    // available or the future is complete.
    loop {
        if let Some(core) = self.take_core(handle) {
            // ...
            return core.block_on(future);
        } else {
            let notified = self.notify.notified();
            pin!(notified);

            if let Some(out) = blocking
                .block_on(poll_fn(|cx| {
                    if notified.as_mut().poll(cx).is_ready() {
                        return Ready(None);
                    }

                    if let Ready(out) = future.as_mut().poll(cx) {
                        return Ready(Some(out));
                    }

                    Pending
                }))
                .expect("Failed to `Enter::block_on`")
            {
                return out;
            }
        }
    }
})

let waker = Handle::waker_ref(&context.handle);
let mut cx = std::task::Context::from_waker(&waker);

'outer: loop {
    let handle = &context.handle;

    if handle.reset_woken() {
        let (c, res) = context.enter(core, || {
            crate::task::coop::budget(|| future.as_mut().poll(&mut cx))
        });

        core = c;

        if let Ready(v) = res {
            return (core, Some(v));
        }
    }

    for _ in 0..handle.shared.config.event_interval {
        // Make sure we didn't hit an unhandled_panic
        if core.unhandled_panic {
            return (core, None);
        }

        core.tick();

        let entry = core.next_task(handle);

        let task = match entry {
            Some(entry) => entry,
            None => {
                // ...
                core = if context.has_pending_work(&core) {
                    context.park_yield(core, handle)
                } else {
                    context.park(core, handle)
                };

                // Try polling the `block_on` future next
                continue 'outer;
            }
        };

        let task = context.handle.shared.owned.assert_owner(task);

        let c = context.run_task(task, core);

        core = c;
    }

    // Yield to the driver, this drives the timer and pulls any
    // pending I/O events.
    core = context.park_yield(core, handle);
}

fn next_task(&mut self, handle: &Handle) -> Option<Notified> {
    if self.tick % self.global_queue_interval == 0 {
        handle
            .next_remote_task()
            .or_else(|| self.next_local_task(handle))
    } else {
        self.next_local_task(handle)
            .or_else(|| handle.next_remote_task())
    }
}

fn schedule(&self, task: task::Notified<Self>) {
    use scheduler::Context::CurrentThread;

    // ...

    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;
use std::time::Duration;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let h = tokio::spawn(async {
    println!("spawned task on {:?}", thread::current().id());
  });
  println!("main on {:?}", thread::current().id());
  thread::sleep(Duration::from_millis(100));
  println!("main slept 100ms");
  h.await.unwrap();
  println!("done");
}

main on ThreadId(1)
main slept 100ms
spawned task on ThreadId(1)
done

fn shutdown2(mut core: Box<Core>, handle: &Handle) -> Box<Core> {
    // Drain the OwnedTasks collection. This call also closes the
    // collection, ensuring that no tasks are ever pushed after this
    // call returns.
    handle.shared.owned.close_and_shutdown_all(0);

    // Drain local queue
    // We already shut down every task, so we just need to drop the task.
    while let Some(task) = core.next_local_task(handle) {
        drop(task);
    }

    // Close the injection queue
    handle.shared.inject.close();

    // Drain remote queue
    while let Some(task) = handle.shared.inject.pop() {
        drop(task);
    }

    assert!(handle.shared.owned.is_empty());

    // ...

    // Shutdown the resource drivers
    if let Some(driver) = core.driver.as_mut() {
        driver.shutdown(&handle.driver);
    }

    core
}

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct CountPolls<F> {
  inner: Pin<Box<F>>,
  polls: u32,
}

impl<F: Future> Future for CountPolls<F> {
  type Output = u32;

  fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
    self.polls += 1;
    match self.inner.as_mut().poll(cx) {
      Poll::Ready(_) => Poll::Ready(self.polls),
      Poll::Pending => Poll::Pending,
    }
  }
}

fn main() {
  let rt = tokio::runtime::Builder::new_current_thread()
    .enable_all()
    .build()
    .unwrap();
  let polls = rt.block_on(CountPolls {
    inner: Box::pin(async {
      let h = tokio::spawn(async {
        for _ in 0..100 {
          tokio::task::yield_now().await;
        }
      });
      h.await.unwrap();
    }),
    polls: 0,
  });
  println!("main future polled {polls} times");
}

main future polled 2 times