fn clone(&self) -> Self {

        Self {

            context: self.context.clone(),

            tasks: self.tasks.cloned(),

            shared_tasks: self.shared_tasks.cloned(),

        }

    }

    fn deref(&self) -> &Self::Target {

        &self.context

    }

    pub fn spawn<Status, F: Future + 'static>(&self, future: F) -> TaskHandle<Status, F::Output>

    where

        Status: Container<SpawnedTaskStatus<F::Output>>,

    {

        self.shared_tasks.map_locked(|tasks| {

            let status = Status::new(SpawnedTaskStatus::default());

            let task_id = tasks.next_task_id;

            tasks.next_task_id = tasks.next_task_id.checked_add(1).expect("u64 wrapped");

            // TODO two allocations isn't ideal. Can we do pin projection through

            // dynamic dispatch? I think it's "safe" because it seems to fit the

            // definition of structural pinning?

            tasks.new.push_back(Box::new(SpawnedTask {

                id: task_id,

                future: Some(Box::pin(future)),

                status: status.cloned(),

                waker: budget_waker::for_current_thread(self.shared_tasks.cloned(), Some(task_id)),

            }));

            TaskHandle {

                status,

                _output: PhantomData,

            }

        })

    }

#[derive(Default)]

#[derive(Default)]

fn run_with_budget<Budget, Tasks, SharedTasks, Backing, F>(

    future: impl FnOnce(Runtime<Budget, Tasks, SharedTasks, Backing>) -> F,

    initial_budget: Budget,

) -> Progress<Budget, Tasks, SharedTasks, Backing, F>

where

    Budget: Budgetable,

    Tasks: Container<RunningTasks>,

    SharedTasks: Container<PendingTasks>,

    F: Future,

    Backing: Container<BudgetContextData<Budget>>,

{

    let runtime = Runtime {

        context: BudgetContext {

            data: Backing::new(crate::BudgetContextData {

                budget: initial_budget,

                paused_future: None,

            }),

            _budget: PhantomData,

        },

        tasks: Tasks::new(RunningTasks::default()),

        shared_tasks: SharedTasks::new(PendingTasks::default()),

    };

    let waker = budget_waker::for_current_thread(runtime.shared_tasks.cloned(), None);

    execute_future(Box::pin(future(runtime.clone())), waker, runtime, false)

}

fn execute_future<Budget, Tasks, SharedTasks, Backing, F>(

    mut future: Pin<Box<F>>,

    waker: Waker,

    runtime: Runtime<Budget, Tasks, SharedTasks, Backing>,

    mut wait_for_budget: bool,

) -> Progress<Budget, Tasks, SharedTasks, Backing, F>

where

    Budget: Budgetable,

    Tasks: Container<RunningTasks>,

    SharedTasks: Container<PendingTasks>,

    F: Future,

    Backing: Container<BudgetContextData<Budget>>,

{

    let mut pinned_future = Pin::new(&mut future);

    let mut cx = Context::from_waker(&waker);

        let poll_result = pinned_future.poll(&mut cx);

        let (ran_out_of_budget, budget_remaining_after_completion) =

            runtime.context.data.map_locked(|data| {

                data.budget.remove_waker(cx.waker());

                let ran_out_of_budget = data.paused_future.take().is_some();

                let budget_remaining_after_completion = if poll_result.is_ready() {

                    Some(data.budget.clone())

                    if !ran_out_of_budget || wait_for_budget {

                        data.budget.add_waker(cx.waker());

                    }

                    None

                (ran_out_of_budget, budget_remaining_after_completion)

            });

        if let Poll::Ready(output) = poll_result {

            return Progress::Complete(BudgetResult {

                output,

                balance: budget_remaining_after_completion

                    .expect("should always be present when Ready returns"),

            });

        }

        if ran_out_of_budget && !wait_for_budget {

            return Progress::NoBudget(IncompleteFuture {

                future,

                waker,

                runtime,

            });

        }

        pinned_future = Pin::new(&mut future);

        // If we have our own tasks to run, execute them. Otherwise, park

        // the thread.

        runtime.tasks.map_locked(|tasks| {

            runtime.shared_tasks.map_locked(|shared| {

                tasks.woke.extend(shared.new.drain(..));

                for task_id in shared.tasks_to_wake.drain() {

                    if let Some(task) = tasks.pending.remove(&task_id) {

                        tasks.woke.push_back(task);

                    }

            });

            if tasks.woke.is_empty() {

                wait_for_budget = false;

                runtime

                    .context

                    .data

                    .map_locked(|data| data.budget.park_for_budget());

            } else {

                for mut task in tasks.woke.drain(..) {

                    let result = task.poll();

                    match result {

                        Poll::Ready(_) => {

                            // The future is complete.

                        }

                        Poll::Pending => {

                            // Move the future to the pending queue.

                            tasks.pending.insert(task.id(), task);

                        }

        });

}

    pub fn continue_with_additional_budget(

        self,

        additional_budget: usize,

    ) -> Progress<Budget, Tasks, SharedTasks, Backing, F> {

        let Self {

            future,

            waker,

            runtime,

            ..

        } = self;

        runtime

            .context

            .data

            .map_locked(|data| data.budget.replenish(additional_budget));

        execute_future(future, waker, runtime, false)

    }

    pub fn wait_for_budget(self) -> Progress<Budget, Tasks, SharedTasks, Backing, F> {

        let Self {

            future,

            waker,

            runtime,

            ..

        } = self;

        execute_future(future, waker, runtime, true)

    }

            if let Some(task_id) = self.task_id {

                self.tasks

                    .map_locked(|tasks| tasks.tasks_to_wake.insert(task_id));

                self.thread.unpark();

            } else {

                // The main task is unblocked, we always unpark.

                self.thread.unpark();

            }

        }

    pub(crate) fn for_current_thread<Tasks>(tasks: Tasks, task_id: Option<u64>) -> Waker

    where

        Tasks: Container<PendingTasks>,

    {

        let arc_thread = Arc::new(WakerData {

            tasks,

            thread: std::thread::current(),

            task_id,

        });

        let arc_thread = Arc::into_raw(arc_thread);

        unsafe { Waker::from_raw(RawWaker::new(arc_thread.cast::<()>(), vtable::<Tasks>())) }

    }

    unsafe fn clone<Tasks>(arc_thread: *const ()) -> RawWaker

    where

        Tasks: Container<PendingTasks>,

    {

        let arc_thread: Arc<WakerData<Tasks>> =

            Arc::from_raw(arc_thread.cast::<WakerData<Tasks>>());

        let cloned = arc_thread.clone();

        let cloned = Arc::into_raw(cloned);

        let _ = Arc::into_raw(arc_thread);

        RawWaker::new(cloned.cast::<()>(), vtable::<Tasks>())

    }

    fn vtable<Tasks>() -> &'static RawWakerVTable

    where

        Tasks: Container<PendingTasks>,

    {

        let task_type = TypeId::of::<Tasks>();

        let sync_type = TypeId::of::<SyncContainer<PendingTasks>>();

        let not_sync_type = TypeId::of::<NotSyncContainer<PendingTasks>>();

        if task_type == sync_type {

            &SYNC_VTABLE

            &NOT_SYNC_VTABLE

    }

    unsafe fn wake_consuming<Tasks>(arc_thread: *const ())

    where

        Tasks: Container<PendingTasks>,

    {

        let arc_thread: Arc<WakerData<Tasks>> = Arc::from_raw(arc_thread as *mut WakerData<Tasks>);

        arc_thread.wake();

    }

    unsafe fn wake_by_ref<Tasks>(arc_thread: *const ())

    where

        Tasks: Container<PendingTasks>,

    {

        let arc_thread: Arc<WakerData<Tasks>> = Arc::from_raw(arc_thread as *mut WakerData<Tasks>);

        arc_thread.wake();

        let _ = Arc::into_raw(arc_thread);

    }

    unsafe fn drop_waker<Tasks>(arc_thread: *const ()) {

        let arc_thread: Arc<WakerData<Tasks>> = Arc::from_raw(arc_thread as *mut WakerData<Tasks>);

        drop(arc_thread);

    }

    fn default() -> Self {

        Self {

            output: TaskOutput::NotCompleted,

            waker: None,

        }

    }

    fn id(&self) -> u64 {

        self.id

    }

    fn poll(&mut self) -> Poll<()> {

        match self.future.as_mut() {

            Some(future) => {

                let pinned_future = Pin::new(future);

                match pinned_future.poll(&mut Context::from_waker(&self.waker)) {

                    Poll::Ready(output) => {

                        self.status.map_locked(|status| {

                            status.output = TaskOutput::Completed(Some(output));

                            if let Some(waker) = status.waker.take() {

                                waker.wake();

                            }

                        });

                        Poll::Ready(())

                    Poll::Pending => Poll::Pending,

    }

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

        self.status.map_locked(|status| {

            match &mut status.output {

                TaskOutput::Completed(output) => {

                    Poll::Ready(output.take().expect("already polled completion"))

                    status.waker = Some(cx.waker().clone());

                    Poll::Pending

        })

    }

            #[derive(Debug, Clone)]

                pub fn run_with_budget<F>(

                    future: impl FnOnce(Self) -> F,

                    initial_budget: Budget,

                ) -> Progress<Budget, F>

                where

                    Budget: Budgetable,

                    F: Future,

                {

                    Progress::from(super::run_with_budget(

                        |rt| future(Self(rt)),

                        initial_budget,

                    ))

                }

                pub fn spawn<F: Future + 'static>(&self, future: F) -> TaskHandle<F::Output> {

                    TaskHandle(self.0.spawn(future))

                }

                pub fn spend(&self, amount: usize) -> SpendBudget<'_, Budget> {

                    // How do we re-export SpendBudget since it's sahrd with async too. crate-level module?

                    SpendBudget::from(self.0.spend(amount))

                }

                pub fn wait_until_complete(self) -> BudgetResult<F::Output, Budget> {

                    let mut progress = self;

                    loop {

                        match progress {

                            Progress::NoBudget(incomplete) => {

                                progress = incomplete.wait_for_budget();

                            }

                            Progress::Complete(result) => break result,

                        }

                    }

                }

                fn from(

                    progress: super::Progress<

                        Budget,

                        crate::$backing<RunningTasks>,

                        crate::$backing<PendingTasks>,

                        crate::$backing<BudgetContextData<Budget>>,

                        F,

                    >,

                ) -> Self {

                    match progress {

                        super::Progress::NoBudget(incomplete) => {

                            Self::NoBudget(IncompleteFuture(incomplete))

                        super::Progress::Complete(result) => Self::Complete(result),

                }

                pub fn continue_with_additional_budget(

                    self,

                    additional_budget: usize,

                ) -> Progress<Budget, F> {

                    Progress::from(self.0.continue_with_additional_budget(additional_budget))

                }

                pub fn wait_for_budget(self) -> Progress<Budget, F> {

                    Progress::from(self.0.wait_for_budget())

                }

                fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

                    let inner = Pin::new(&mut self.0);

                    inner.poll(cx)

                }

define_public_interface!(

    threadsafe,

    SyncContainer,

    "A threadsafe (`Send + Sync`), blocking budgeting implementation that is runtime agnostic.\n\nThe only difference between this module and the [`singlethreaded`] module is that this one uses [`std::sync::Arc`] and [`std::sync::Mutex`] instead of [`std::rc::Rc`] and [`std::cell::RefCell`]."

);

define_public_interface!(

    singlethreaded,

    NotSyncContainer,

    "A single-threaded (`!Send + !Sync`), blocking budgeting implementation that is runtime agnostic.\n\nThe only difference between this module and the [`threadsafe`] module is that this one uses [`std::rc::Rc`] and [`std::cell::RefCell`] instead of [`std::sync::Arc`] and [`std::sync::Mutex`]."

);