summaryrefslogtreecommitdiffstats
path: root/vendor/tokio/src/runtime/tests/task.rs
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/tokio/src/runtime/tests/task.rs')
-rw-r--r--vendor/tokio/src/runtime/tests/task.rs159
1 files changed, 159 insertions, 0 deletions
diff --git a/vendor/tokio/src/runtime/tests/task.rs b/vendor/tokio/src/runtime/tests/task.rs
new file mode 100644
index 000000000..7c2012523
--- /dev/null
+++ b/vendor/tokio/src/runtime/tests/task.rs
@@ -0,0 +1,159 @@
+use crate::runtime::task::{self, Schedule, Task};
+use crate::util::linked_list::{Link, LinkedList};
+use crate::util::TryLock;
+
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+#[test]
+fn create_drop() {
+ let _ = super::joinable::<_, Runtime>(async { unreachable!() });
+}
+
+#[test]
+fn schedule() {
+ with(|rt| {
+ let (task, _) = super::joinable(async {
+ crate::task::yield_now().await;
+ });
+
+ rt.schedule(task);
+
+ assert_eq!(2, rt.tick());
+ })
+}
+
+#[test]
+fn shutdown() {
+ with(|rt| {
+ let (task, _) = super::joinable(async {
+ loop {
+ crate::task::yield_now().await;
+ }
+ });
+
+ rt.schedule(task);
+ rt.tick_max(1);
+
+ rt.shutdown();
+ })
+}
+
+fn with(f: impl FnOnce(Runtime)) {
+ struct Reset;
+
+ impl Drop for Reset {
+ fn drop(&mut self) {
+ let _rt = CURRENT.try_lock().unwrap().take();
+ }
+ }
+
+ let _reset = Reset;
+
+ let rt = Runtime(Arc::new(Inner {
+ released: task::TransferStack::new(),
+ core: TryLock::new(Core {
+ queue: VecDeque::new(),
+ tasks: LinkedList::new(),
+ }),
+ }));
+
+ *CURRENT.try_lock().unwrap() = Some(rt.clone());
+ f(rt)
+}
+
+#[derive(Clone)]
+struct Runtime(Arc<Inner>);
+
+struct Inner {
+ released: task::TransferStack<Runtime>,
+ core: TryLock<Core>,
+}
+
+struct Core {
+ queue: VecDeque<task::Notified<Runtime>>,
+ tasks: LinkedList<Task<Runtime>, <Task<Runtime> as Link>::Target>,
+}
+
+static CURRENT: TryLock<Option<Runtime>> = TryLock::new(None);
+
+impl Runtime {
+ fn tick(&self) -> usize {
+ self.tick_max(usize::MAX)
+ }
+
+ fn tick_max(&self, max: usize) -> usize {
+ let mut n = 0;
+
+ while !self.is_empty() && n < max {
+ let task = self.next_task();
+ n += 1;
+ task.run();
+ }
+
+ self.0.maintenance();
+
+ n
+ }
+
+ fn is_empty(&self) -> bool {
+ self.0.core.try_lock().unwrap().queue.is_empty()
+ }
+
+ fn next_task(&self) -> task::Notified<Runtime> {
+ self.0.core.try_lock().unwrap().queue.pop_front().unwrap()
+ }
+
+ fn shutdown(&self) {
+ let mut core = self.0.core.try_lock().unwrap();
+
+ for task in core.tasks.iter() {
+ task.shutdown();
+ }
+
+ while let Some(task) = core.queue.pop_back() {
+ task.shutdown();
+ }
+
+ drop(core);
+
+ while !self.0.core.try_lock().unwrap().tasks.is_empty() {
+ self.0.maintenance();
+ }
+ }
+}
+
+impl Inner {
+ fn maintenance(&self) {
+ use std::mem::ManuallyDrop;
+
+ for task in self.released.drain() {
+ let task = ManuallyDrop::new(task);
+
+ // safety: see worker.rs
+ unsafe {
+ let ptr = task.header().into();
+ self.core.try_lock().unwrap().tasks.remove(ptr);
+ }
+ }
+ }
+}
+
+impl Schedule for Runtime {
+ fn bind(task: Task<Self>) -> Runtime {
+ let rt = CURRENT.try_lock().unwrap().as_ref().unwrap().clone();
+ rt.0.core.try_lock().unwrap().tasks.push_front(task);
+ rt
+ }
+
+ fn release(&self, task: &Task<Self>) -> Option<Task<Self>> {
+ // safety: copying worker.rs
+ let task = unsafe { Task::from_raw(task.header().into()) };
+ self.0.released.push(task);
+ None
+ }
+
+ fn schedule(&self, task: task::Notified<Self>) {
+ self.0.core.try_lock().unwrap().queue.push_back(task);
+ }
+}