diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index ee32b954a..057055e26 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -69,7 +69,6 @@ timer = ["slab"] udp = ["io", "net-driver"] uds = ["io", "net-driver", "mio-uds", "libc"] process = [ - "crossbeam-queue", "io", "libc", "mio-named-pipes", @@ -101,7 +100,6 @@ pin-project = { version = "0.4", optional = true } slab = { version = "0.4.1", optional = true } [target.'cfg(unix)'.dependencies] -crossbeam-queue = { version = "0.1.2", optional = true } mio-uds = { version = "0.6.5", optional = true } libc = { version = "0.2.42", optional = true } signal-hook-registry = { version = "1.1.1", optional = true } diff --git a/tokio/src/process/unix/mod.rs b/tokio/src/process/unix/mod.rs index 46177caca..dfb69f21d 100644 --- a/tokio/src/process/unix/mod.rs +++ b/tokio/src/process/unix/mod.rs @@ -24,7 +24,7 @@ mod orphan; mod reap; -use self::orphan::{AtomicOrphanQueue, OrphanQueue, Wait}; +use self::orphan::{OrphanQueue, OrphanQueueImpl, Wait}; use self::reap::Reaper; use super::SpawnedChild; use crate::net::util::PollEvented; @@ -59,7 +59,7 @@ impl Kill for process::Child { } lazy_static::lazy_static! { - static ref ORPHAN_QUEUE: AtomicOrphanQueue = AtomicOrphanQueue::new(); + static ref ORPHAN_QUEUE: OrphanQueueImpl = OrphanQueueImpl::new(); } struct GlobalOrphanQueue; diff --git a/tokio/src/process/unix/orphan.rs b/tokio/src/process/unix/orphan.rs index 7a52575d5..5cfdd1f6c 100644 --- a/tokio/src/process/unix/orphan.rs +++ b/tokio/src/process/unix/orphan.rs @@ -1,6 +1,6 @@ -use crossbeam_queue::SegQueue; use std::io; use std::process::ExitStatus; +use std::sync::Mutex; /// An interface for waiting on a process to exit. pub(crate) trait Wait { @@ -39,50 +39,51 @@ impl> OrphanQueue for &O { } } -/// An atomic implementation of `OrphanQueue`. +/// An implementation of `OrphanQueue`. #[derive(Debug)] -pub(crate) struct AtomicOrphanQueue { - queue: SegQueue, +pub(crate) struct OrphanQueueImpl { + queue: Mutex>, } -impl AtomicOrphanQueue { +impl OrphanQueueImpl { pub(crate) fn new() -> Self { Self { - queue: SegQueue::new(), + queue: Mutex::new(Vec::new()), } } + + #[cfg(test)] + fn len(&self) -> usize { + self.queue.lock().unwrap().len() + } } -impl OrphanQueue for AtomicOrphanQueue { +impl OrphanQueue for OrphanQueueImpl { fn push_orphan(&self, orphan: T) { - self.queue.push(orphan) + self.queue.lock().unwrap().push(orphan) } fn reap_orphans(&self) { - let len = self.queue.len(); + let mut queue = self.queue.lock().unwrap(); + let queue = &mut *queue; - if len == 0 { - return; - } - - let mut orphans = Vec::with_capacity(len); - while let Ok(mut orphan) = self.queue.pop() { - match orphan.try_wait() { + let mut i = 0; + while i < queue.len() { + match queue[i].try_wait() { Ok(Some(_)) => {} Err(_) => { // TODO: bubble up error some how. Is this an internal bug? // Shoudl we panic? Is it OK for this to be silently // dropped? } - // Still not done yet, we need to put it back in the queue - // when were done draining it, so that we don't get stuck - // in an infinite loop here - Ok(None) => orphans.push(orphan), + // Still not done yet + Ok(None) => { + i += 1; + continue; + } } - } - for orphan in orphans { - self.queue.push(orphan); + queue.remove(i); } } } @@ -90,7 +91,7 @@ impl OrphanQueue for AtomicOrphanQueue { #[cfg(all(test, not(loom)))] mod test { use super::Wait; - use super::{AtomicOrphanQueue, OrphanQueue}; + use super::{OrphanQueue, OrphanQueueImpl}; use std::cell::Cell; use std::io; use std::os::unix::process::ExitStatusExt; @@ -156,30 +157,30 @@ mod test { let third_waits = third_orphan.total_waits.clone(); let fourth_waits = fourth_orphan.total_waits.clone(); - let orphanage = AtomicOrphanQueue::new(); + let orphanage = OrphanQueueImpl::new(); orphanage.push_orphan(first_orphan); orphanage.push_orphan(third_orphan); orphanage.push_orphan(second_orphan); orphanage.push_orphan(fourth_orphan); - assert_eq!(orphanage.queue.len(), 4); + assert_eq!(orphanage.len(), 4); orphanage.reap_orphans(); - assert_eq!(orphanage.queue.len(), 2); + assert_eq!(orphanage.len(), 2); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 1); assert_eq!(third_waits.get(), 1); assert_eq!(fourth_waits.get(), 1); orphanage.reap_orphans(); - assert_eq!(orphanage.queue.len(), 1); + assert_eq!(orphanage.len(), 1); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 2); assert_eq!(third_waits.get(), 2); assert_eq!(fourth_waits.get(), 1); orphanage.reap_orphans(); - assert_eq!(orphanage.queue.len(), 0); + assert_eq!(orphanage.len(), 0); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 2); assert_eq!(third_waits.get(), 3);