use core::fmt;
use core::future::Future;
use core::marker::PhantomData;
use core::mem;
use core::ptr::NonNull;
use core::sync::atomic::Ordering;
use core::task::Waker;
use crate::header::Header;
use crate::raw::RawTask;
use crate::state::*;
use crate::JoinHandle;
pub fn spawn<F, R, S, T>(future: F, schedule: S, tag: T) -> (Task<T>, JoinHandle<R, T>)
where
F: Future<Output = R> + Send + 'static,
R: Send + 'static,
S: Fn(Task<T>) + Send + Sync + 'static,
T: Send + Sync + 'static,
{
let raw_task = if mem::size_of::<F>() >= 2048 {
let future = alloc::boxed::Box::pin(future);
RawTask::<_, R, S, T>::allocate(future, schedule, tag)
} else {
RawTask::<F, R, S, T>::allocate(future, schedule, tag)
};
let task = Task {
raw_task,
_marker: PhantomData,
};
let handle = JoinHandle {
raw_task,
_marker: PhantomData,
};
(task, handle)
}
pub fn spawn_local<F, R, S, T>(future: F, schedule: S, tag: T) -> (Task<T>, JoinHandle<R, T>)
where
F: Future<Output = R> + 'static,
R: 'static,
S: Fn(Task<T>) + Send + Sync + 'static,
T: Send + Sync + 'static,
{
extern crate std;
use std::mem::ManuallyDrop;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::thread::{self, ThreadId};
use std::thread_local;
#[inline]
fn thread_id() -> ThreadId {
thread_local! {
static ID: ThreadId = thread::current().id();
}
ID.try_with(|id| *id)
.unwrap_or_else(|_| thread::current().id())
}
struct Checked<F> {
id: ThreadId,
inner: ManuallyDrop<F>,
}
impl<F> Drop for Checked<F> {
fn drop(&mut self) {
assert!(
self.id == thread_id(),
"local task dropped by a thread that didn't spawn it"
);
unsafe {
ManuallyDrop::drop(&mut self.inner);
}
}
}
impl<F: Future> Future for Checked<F> {
type Output = F::Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
assert!(
self.id == thread_id(),
"local task polled by a thread that didn't spawn it"
);
unsafe { self.map_unchecked_mut(|c| &mut *c.inner).poll(cx) }
}
}
let future = Checked {
id: thread_id(),
inner: ManuallyDrop::new(future),
};
let raw_task = if mem::size_of::<F>() >= 2048 {
let future = alloc::boxed::Box::pin(future);
RawTask::<_, R, S, T>::allocate(future, schedule, tag)
} else {
RawTask::<_, R, S, T>::allocate(future, schedule, tag)
};
let task = Task {
raw_task,
_marker: PhantomData,
};
let handle = JoinHandle {
raw_task,
_marker: PhantomData,
};
(task, handle)
}
pub struct Task<T> {
pub(crate) raw_task: NonNull<()>,
pub(crate) _marker: PhantomData<T>,
}
unsafe impl<T> Send for Task<T> {}
unsafe impl<T> Sync for Task<T> {}
impl<T> Task<T> {
pub fn schedule(self) {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
mem::forget(self);
unsafe {
((*header).vtable.schedule)(ptr);
}
}
pub fn run(self) -> bool {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
mem::forget(self);
unsafe { ((*header).vtable.run)(ptr) }
}
pub fn cancel(&self) {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
unsafe {
(*header).cancel();
}
}
pub fn tag(&self) -> &T {
let offset = Header::offset_tag::<T>();
let ptr = self.raw_task.as_ptr();
unsafe {
let raw = (ptr as *mut u8).add(offset) as *const T;
&*raw
}
}
pub fn into_raw(self) -> *const T {
let offset = Header::offset_tag::<T>();
let ptr = self.raw_task.as_ptr();
mem::forget(self);
unsafe { (ptr as *mut u8).add(offset) as *const T }
}
pub unsafe fn from_raw(raw: *const T) -> Task<T> {
let offset = Header::offset_tag::<T>();
let ptr = (raw as *mut u8).sub(offset) as *mut ();
Task {
raw_task: NonNull::new_unchecked(ptr),
_marker: PhantomData,
}
}
pub fn waker(&self) -> Waker {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
unsafe {
let raw_waker = ((*header).vtable.clone_waker)(ptr);
Waker::from_raw(raw_waker)
}
}
}
impl<T> Drop for Task<T> {
fn drop(&mut self) {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
unsafe {
(*header).cancel();
((*header).vtable.drop_future)(ptr);
let state = (*header).state.fetch_and(!SCHEDULED, Ordering::AcqRel);
if state & AWAITER != 0 {
(*header).notify(None);
}
((*header).vtable.drop_task)(ptr);
}
}
}
impl<T: fmt::Debug> fmt::Debug for Task<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let ptr = self.raw_task.as_ptr();
let header = ptr as *const Header;
f.debug_struct("Task")
.field("header", unsafe { &(*header) })
.field("tag", self.tag())
.finish()
}
}