From 7da00f38323304a8ab5141268054bf5cfd53d02d Mon Sep 17 00:00:00 2001 From: Michal 'vorner' Vaner Date: Thu, 26 Jan 2017 21:22:19 +0100 Subject: [PATCH] signal: Use IDs that don't run out Replace the sequential counting (which might be exhausted) by an address of an object (in a box, so it doesn't change). This is also a unique, so it is acceptable ID. --- src/unix.rs | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/src/unix.rs b/src/unix.rs index b004c1b79..2021feee1 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -15,7 +15,7 @@ use std::io::prelude::*; use std::io; use std::mem; use std::os::unix::prelude::*; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering, ATOMIC_USIZE_INIT}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Mutex, Once, ONCE_INIT}; use futures::future; @@ -39,7 +39,7 @@ const SIGNUM: usize = 32; struct SignalInfo { pending: AtomicBool, // The ones interested in this signal - recipients: Mutex)>>, + recipients: Mutex>>>, init: Once, initialized: UnsafeCell, @@ -271,7 +271,7 @@ impl Driver { // actually want to get woken up to continue sending a // message. Let's optimise it later on though, as we know // this works. - match recipients[i].1.start_send(signum) { + match recipients[i].start_send(signum) { Ok(AsyncSink::Ready) => {} Ok(AsyncSink::NotReady(_)) => {} Err(_) => { recipients.swap_remove(i); } @@ -317,10 +317,19 @@ impl Driver { /// alleviate some of these limitations if possible! pub struct Signal { signal: c_int, - token: usize, + // Used only as an identifier. We place the real sender into a Box, so it + // stays on the same address forever. That gives us a unique pointer, so we + // can use this to identify the sender in a Vec and delete it when we are + // dropped. + id: *const Sender, rx: Receiver, } +// The raw pointer prevents the compiler from determining it as Send +// automatically. But the only thing we use the raw pointer for is to identify +// the correct Box to delete, not manipulate any data through that. +unsafe impl Send for Signal {} + impl Signal { /// Creates a new stream which will receive notifications when the current /// process receives the signal `signal`. @@ -341,8 +350,6 @@ impl Signal { /// multiple times. When a signal is received then all the associated /// channels will receive the signal notification. pub fn new(signal: c_int, handle: &Handle) -> IoFuture { - static TOKENS: AtomicUsize = ATOMIC_USIZE_INIT; - let result = (|| { // Turn the signal delivery on once we are ready for it try!(signal_enable(signal)); @@ -360,12 +367,13 @@ impl Signal { // One wakeup in a queue is enough, no need for us to buffer up any // more. let (tx, rx) = channel(1); - let token = TOKENS.fetch_add(1, Ordering::SeqCst); + let tx = Box::new(tx); + let id: *const _ = &*tx; let idx = signal as usize; - globals().signals[idx].recipients.lock().unwrap().push((token, tx)); + globals().signals[idx].recipients.lock().unwrap().push(tx); Ok(Signal { rx: rx, - token: token, + id: id, signal: signal, }) })(); @@ -388,6 +396,6 @@ impl Drop for Signal { fn drop(&mut self) { let idx = self.signal as usize; let mut list = globals().signals[idx].recipients.lock().unwrap(); - list.retain(|pair| pair.0 != self.token); + list.retain(|sender| &**sender as *const _ != self.id); } }