Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Replace linked list with VecDeque #14

Merged
merged 4 commits into from
Sep 23, 2024
+25 −287
Merged

Replace linked list with VecDeque #14

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0a6ea46
Replace linked list with VecDeque
Sep 21, 2024
f67e5f1
Use the ?
Sep 21, 2024
9c60b7a
Use fut field instead of is_in_queue
Sep 21, 2024
d0de899
Small fixes.
dragostis Sep 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
298 changes: 22 additions & 276 deletions src/job.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use std::{
cell::{Cell, UnsafeCell},
collections::VecDeque,
mem::ManuallyDrop,
panic::{self, AssertUnwindSafe},
ptr::NonNull,
Expand Down Expand Up @@ -132,8 +133,7 @@ impl<F> JobStack<F> {
pub struct Job<T = ()> {
stack: NonNull<JobStack>,
harness: unsafe fn(&mut Scope<'_>, NonNull<JobStack>, NonNull<Future>),
prev: Cell<Option<NonNull<Self>>>,
fut_or_next: Cell<Option<NonNull<Future<T>>>>,
fut: Cell<Option<NonNull<Future<T>>>>,
}

impl<T> Job<T> {
Expand Down Expand Up @@ -171,13 +171,12 @@ impl<T> Job<T> {
Self {
stack: NonNull::from(stack).cast(),
harness: harness::<F, T>,
prev: Cell::new(None),
fut_or_next: Cell::new(None),
fut: Cell::new(None),
}
}

pub fn is_in_queue(&self) -> bool {
self.prev.get().is_some()
pub fn is_waiting(&self) -> bool {
self.fut.get().is_none()
}

pub fn eq(&self, other: &Job) -> bool {
Expand All @@ -187,7 +186,7 @@ impl<T> Job<T> {
/// SAFETY:
/// It should only be called after being popped from a `JobQueue`.
pub unsafe fn poll(&self) -> bool {
self.fut_or_next
self.fut
.get()
.map(|fut| {
// SAFETY:
Expand All @@ -202,7 +201,7 @@ impl<T> Job<T> {
/// SAFETY:
/// It should only be called after being popped from a `JobQueue`.
pub unsafe fn wait(&self) -> Option<thread::Result<T>> {
self.fut_or_next.get().and_then(|fut| {
self.fut.get().and_then(|fut| {
// SAFETY:
// Before being popped, the `JobQueue` allocates and stores a
// `Future` in `self.fur_or_next` that should get passed here.
Expand All @@ -222,7 +221,7 @@ impl<T> Job<T> {
/// It should only be called in the case where the job has been popped
/// from the front and will not be `Job::Wait`ed.
pub unsafe fn drop(&self) {
if let Some(fut) = self.fut_or_next.get() {
if let Some(fut) = self.fut.get() {
// SAFETY:
// Before being popped, the `JobQueue` allocates and store a
// `Future` in `self.fur_or_next` that should get passed here.
Expand All @@ -242,7 +241,7 @@ impl Job {
// Before being popped, the `JobQueue` allocates and store a
// `Future` in `self.fur_or_next` that should get passed here.
unsafe {
(self.harness)(scope, self.stack, self.fut_or_next.get().unwrap());
(self.harness)(scope, self.stack, self.fut.get().unwrap());
}
}
}
Expand All @@ -253,285 +252,32 @@ impl Job {
// sent across threads.
unsafe impl Send for Job {}

#[derive(Debug)]
pub struct JobQueue {
sentinel: NonNull<Job>,
tail: NonNull<Job>,
len: u32,
}

impl Default for JobQueue {
fn default() -> Self {
let root = Box::leak(Box::new(Job {
stack: NonNull::dangling(),
harness: |_, _, _| (),
prev: Cell::new(None),
fut_or_next: Cell::new(None),
}))
.into();

Self {
sentinel: root,
tail: root,
len: 0,
}
}
}

impl Drop for JobQueue {
fn drop(&mut self) {
// SAFETY:
// `self.sentinel` never gets written over, so it contains the original
// `leak`ed `Box` that gets allocated in `JobQueue::default`.
unsafe {
drop(Box::from_raw(self.sentinel.as_ptr()));
}
}
}
#[derive(Debug, Default)]
pub struct JobQueue(VecDeque<NonNull<Job>>);

impl JobQueue {
pub fn len(&self) -> u32 {
self.len
pub fn len(&self) -> usize {
self.0.len()
}

/// SAFETY:
/// Any `Job` pushed onto the queue should alive at least until it gets
/// popped.
pub unsafe fn push_back<T>(&mut self, job: &Job<T>) {
// SAFETY:
// The tail can either be the root `Box::leak`ed in the default
// constructor or a `Job` that has been pushed previously and which is
// still alive.
let current_tail = unsafe { self.tail.as_ref() };
// SAFETY:
// This effectively casts the `Job`'s `fut_or_next` from `Future<T>` to
// `Future<()>` which casts the `Future`'s `Box<T>` to a `Box<()>`.
//
// This box will not be access until the pointer gets passed in the
// `harness` where it gets cast back to `T`.
let next_tail = unsafe { &*(job as *const Job<T> as *const Job) };

current_tail
.fut_or_next
.set(Some(NonNull::from(next_tail).cast()));
next_tail.prev.set(Some(current_tail.into()));

self.len += 1;

self.tail = next_tail.into();
self.0.push_back(NonNull::from(next_tail).cast());
}

/// SAFETY:
/// The last `Job` in the queue must still be alive.
pub unsafe fn pop_back(&mut self) {
// SAFETY:
// The tail can either be the root `Box::leak`ed in the default
// constructor or a `Job` that has been pushed previously and which is
// still alive.
let current_tail = unsafe { self.tail.as_ref() };
if let Some(prev_tail) = current_tail.prev.get() {
// SAFETY:
// `Job`'s `prev` pointer can only be set by `JobQueue::push_back`
// to the previous tail which should still be alive or by
// `JobQueue::pop_front` when it's set to `self.sentinel` which is
// alive for the entirety of `self`.
let prev_tail = unsafe { prev_tail.as_ref() };

current_tail.prev.set(None);
prev_tail.fut_or_next.set(None);

self.len -= 1;

self.tail = prev_tail.into();
}
pub fn pop_back(&mut self) {
self.0.pop_back();
}

/// SAFETY:
/// The first `Job` in the queue must still be alive.
pub unsafe fn pop_front(&mut self) -> Option<Job> {
pub fn pop_front(&mut self) -> Option<Job> {
let val = self.0.pop_front()?;
// SAFETY:
// `self.sentinel` is alive for the entirety of `self`.
let sentinel = unsafe { self.sentinel.as_ref() };

sentinel.fut_or_next.get().map(|next| {
// SAFETY:
// `self.sentinel`'s `fut_or_next` pointer can only be set by
// `JobQueue::push_back` or by `JobQueue::pop_front` when it's set
// to a job that was previous set by `JobQueue::push_back` and
// should still be alive.
let head: &Job = unsafe { next.cast().as_ref() };

if let Some(next) = head.fut_or_next.get() {
sentinel.fut_or_next.set(Some(next.cast()));

// SAFETY:
// `Job`'s `fut_or_next` pointer can only be set by
// `JobQueue::push_back` or by `JobQueue::pop_front` when it's set
// to a job that was previous set by `JobQueue::push_back` and
// should still be alive.
//
// It can also be set to a `Future`, but that can only happen after
// the job was removed from the queue.
let next: &Job = unsafe { next.cast().as_ref() };
next.prev.set(Some(sentinel.into()));
} else {
sentinel.fut_or_next.set(None);
self.tail = sentinel.into();
}

// SAFETY:
// `self.sentinel`'s `fut_or_next` pointer can only be set by
// `JobQueue::push_back` or by `JobQueue::pop_front` when it's set
// to a job that was previous set by `JobQueue::push_back` and
// should still be alive.
let head: &Job<Future> = unsafe { next.cast().as_ref() };

head.prev.set(None);
head.fut_or_next
.set(Some(Box::leak(Box::new(Future::default())).into()));

self.len -= 1;

// SAFETY:
// `self.sentinel`'s `fut_or_next` pointer can only be set by
// `JobQueue::push_back` or by `JobQueue::pop_front` when it's set
// to a job that was previous set by `JobQueue::push_back` and
// should still be alive.
unsafe { next.cast::<Job>().as_ref().clone() }
})
}
}

#[cfg(test)]
mod tests {
use super::*;

impl Job {
pub fn from_usize(val: &'static usize) -> Self {
Self {
stack: NonNull::from(val).cast(),
harness: |_, _, _| (),
prev: Cell::new(None),
fut_or_next: Cell::new(None),
}
}

pub fn as_usize(&self) -> usize {
unsafe { *self.stack.cast().as_ref() }
}
}

#[test]
fn push_pop_back() {
let mut queue = JobQueue::default();

assert_eq!(queue.sentinel, queue.tail);

let job1 = Job::from_usize(&1);

unsafe {
queue.push_back(&job1);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 1);

unsafe {
queue.pop_back();
}
assert_eq!(queue.sentinel, queue.tail);
}

#[test]
fn push2_pop2_back() {
let mut queue = JobQueue::default();

assert_eq!(queue.sentinel, queue.tail);

let job1 = Job::from_usize(&1);
let job2 = Job::from_usize(&2);

unsafe {
queue.push_back(&job1);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 1);

unsafe {
queue.push_back(&job2);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 2);

unsafe {
queue.pop_back();
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 1);

unsafe {
queue.pop_back();
}
assert_eq!(queue.sentinel, queue.tail);
}

#[test]
fn push_pop_front() {
let mut queue = JobQueue::default();

assert_eq!(queue.sentinel, queue.tail);

let job1 = Job::from_usize(&1);

unsafe {
queue.push_back(&job1);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 1);

let job = unsafe { queue.pop_front().unwrap() };
assert_eq!(job.as_usize(), 1);
assert!(job.prev.get().is_none());
assert!(job.fut_or_next.get().is_some());

unsafe {
job.drop();
}

assert_eq!(queue.sentinel, queue.tail);
}

#[test]
fn push2_pop2_front() {
let mut queue = JobQueue::default();

assert_eq!(queue.sentinel, queue.tail);

let job1 = Job::from_usize(&1);
let job2 = Job::from_usize(&2);

unsafe {
queue.push_back(&job1);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 1);

unsafe {
queue.push_back(&job2);
}
assert_eq!(unsafe { queue.tail.as_ref().as_usize() }, 2);

let job = unsafe { queue.pop_front().unwrap() };
assert_eq!(job.as_usize(), 1);
assert!(job.prev.get().is_none());
assert!(job.fut_or_next.get().is_some());

unsafe {
job.drop();
}

let job = unsafe { queue.pop_front().unwrap() };
assert_eq!(job.as_usize(), 2);
assert!(job.prev.get().is_none());
assert!(job.fut_or_next.get().is_some());

unsafe {
job.drop();
}

assert_eq!(queue.sentinel, queue.tail);
// `Job` is still alive as per contract in `push_back`
let job = unsafe { val.as_ref() };
job.fut.set(Some(Box::leak(Box::new(Future::default())).into()));
Some(job.clone())
}
}
2 changes: 1 addition & 1 deletion src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -383,7 +383,7 @@ impl<'s> Scope<'s> {

let rb = b(self);

if job.is_in_queue() {
if job.is_waiting() {
// SAFETY:
// `job` is alive until the end of this scope and there has been no
// other pop up to this point.
Expand Down
Loading