std: Re-enable at_exit()

The new semantics of this function are that the callbacks are run when the *main
thread* exits, not when all threads have exited. This implies that other threads
may still be running when the `at_exit` callbacks are invoked and users need to
be prepared for this situation.

Users in the standard library have been audited in accordance to these new rules
as well.

Closes rust-lang#20012

Author: alexcrichton

src/liblog/lib.rs

@@ -177,7 +177,7 @@ use std::mem;
 use std::os;
 use std::rt;
 use std::slice;
-use std::sync::{Once, ONCE_INIT};
+use std::sync::{Once, ONCE_INIT, StaticMutex, MUTEX_INIT};
 
 use regex::Regex;
 
@@ -193,6 +193,8 @@ pub const MAX_LOG_LEVEL: u32 = 255;
 /// The default logging level of a crate if no other is specified.
 const DEFAULT_LOG_LEVEL: u32 = 1;
 
+static LOCK: StaticMutex = MUTEX_INIT;
+
 /// An unsafe constant that is the maximum logging level of any module
 /// specified. This is the first line of defense to determining whether a
 /// logging statement should be run.
@@ -281,9 +283,18 @@ impl Drop for DefaultLogger {
 pub fn log(level: u32, loc: &'static LogLocation, args: fmt::Arguments) {
     // Test the literal string from args against the current filter, if there
     // is one.
-    match unsafe { FILTER.as_ref() } {
-        Some(filter) if !filter.is_match(args.to_string()[]) => return,
-        _ => {}
+    unsafe {
+        let _g = LOCK.lock();
+        match FILTER as uint {
+            0 => {}
+            1 => panic!("cannot log after main thread has exited"),
+            n => {
+                let filter = mem::transmute::<_, &Regex>(n);
+                if !filter.is_match(args.to_string().as_slice()) {
+                    return
+                }
+            }
+        }
     }
 
     // Completely remove the local logger from TLS in case anyone attempts to
@@ -401,9 +412,15 @@ pub fn mod_enabled(level: u32, module: &str) -> bool {
 
     // This assertion should never get tripped unless we're in an at_exit
     // handler after logging has been torn down and a logging attempt was made.
-    assert!(unsafe { !DIRECTIVES.is_null() });
 
-    enabled(level, module, unsafe { (*DIRECTIVES).iter() })
+    let _g = LOCK.lock();
+    unsafe {
+        assert!(DIRECTIVES as uint != 0);
+        assert!(DIRECTIVES as uint != 1,
+                "cannot log after the main thread has exited");
+
+        enabled(level, module, (*DIRECTIVES).iter())
+    }
 }
 
 fn enabled(level: u32,
@@ -459,14 +476,15 @@ fn init() {
 
         // Schedule the cleanup for the globals for when the runtime exits.
         rt::at_exit(move |:| {
+            let _g = LOCK.lock();
             assert!(!DIRECTIVES.is_null());
             let _directives: Box<Vec<directive::LogDirective>> =
                 mem::transmute(DIRECTIVES);
-            DIRECTIVES = 0 as *const Vec<directive::LogDirective>;
+            DIRECTIVES = 1 as *const Vec<directive::LogDirective>;
 
             if !FILTER.is_null() {
                 let _filter: Box<Regex> = mem::transmute(FILTER);
-                FILTER = 0 as *const _;
+                FILTER = 1 as *const _;
             }
         });
     }

src/libstd/io/stdio.rs

@@ -26,29 +26,19 @@
 //! ```
 
 use self::StdSource::*;
+use prelude::*;
 
-use boxed::Box;
 use cell::RefCell;
-use clone::Clone;
 use failure::LOCAL_STDERR;
 use fmt;
-use io::{Reader, Writer, IoResult, IoError, OtherIoError, Buffer,
-         standard_error, EndOfFile, LineBufferedWriter, BufferedReader};
-use kinds::{Sync, Send};
+use io::{IoResult, IoError, OtherIoError};
+use io::{standard_error, EndOfFile, LineBufferedWriter, BufferedReader};
 use libc;
 use mem;
-use option::Option;
-use option::Option::{Some, None};
-use ops::{Deref, DerefMut, FnOnce};
-use result::Result::{Ok, Err};
 use rt;
-use slice::SliceExt;
-use str::StrExt;
-use string::String;
 use sys::{fs, tty};
-use sync::{Arc, Mutex, MutexGuard, Once, ONCE_INIT};
+use sync::{Arc, Mutex, MutexGuard, StaticMutex, MUTEX_INIT};
 use uint;
-use vec::Vec;
 
 // And so begins the tale of acquiring a uv handle to a stdio stream on all
 // platforms in all situations. Our story begins by splitting the world into two
@@ -215,14 +205,15 @@ impl Reader for StdinReader {
 pub fn stdin() -> StdinReader {
     // We're following the same strategy as kimundi's lazy_static library
     static mut STDIN: *const StdinReader = 0 as *const StdinReader;
-    static ONCE: Once = ONCE_INIT;
+    static LOCK: StaticMutex = MUTEX_INIT;
 
     unsafe {
-        ONCE.doit(|| {
-            // The default buffer capacity is 64k, but apparently windows doesn't like
-            // 64k reads on stdin. See #13304 for details, but the idea is that on
-            // windows we use a slightly smaller buffer that's been seen to be
-            // acceptable.
+        let _g = LOCK.lock();
+        if STDIN as uint == 0 {
+            // The default buffer capacity is 64k, but apparently windows
+            // doesn't like 64k reads on stdin. See #13304 for details, but the
+            // idea is that on windows we use a slightly smaller buffer that's
+            // been seen to be acceptable.
             let stdin = if cfg!(windows) {
                 BufferedReader::with_capacity(8 * 1024, stdin_raw())
             } else {
@@ -235,11 +226,15 @@ pub fn stdin() -> StdinReader {
 
             // Make sure to free it at exit
             rt::at_exit(|| {
-                mem::transmute::<_, Box<StdinReader>>(STDIN);
-                STDIN = 0 as *const _;
+                let g = LOCK.lock();
+                let stdin = STDIN;
+                STDIN = 1 as *const _;
+                drop(g);
+                mem::transmute::<_, Box<StdinReader>>(stdin);
             });
-        });
-
+        } else if STDIN as uint == 1 {
+            panic!("accessing stdin after the main thread has exited")
+        }
         (*STDIN).clone()
     }
 }

src/libstd/lib.rs

@@ -228,13 +228,13 @@ pub mod thread;
 pub mod sync;
 pub mod comm;
 
+#[path = "sys/common/mod.rs"] mod sys_common;
+
 #[cfg(unix)]
 #[path = "sys/unix/mod.rs"] mod sys;
 #[cfg(windows)]
 #[path = "sys/windows/mod.rs"] mod sys;
 
-#[path = "sys/common/mod.rs"] mod sys_common;
-
 pub mod rt;
 mod failure;
 

src/libstd/rt/at_exit_imp.rs

@@ -29,6 +29,8 @@ type Queue = Vec<Thunk>;
 static LOCK: Mutex = MUTEX_INIT;
 static mut QUEUE: *mut Queue = 0 as *mut Queue;
 
+const DTOR_RUN_ITERS: uint = 10;
+
 unsafe fn init() {
     if QUEUE.is_null() {
         let state: Box<Queue> = box Vec::new();
@@ -49,7 +51,7 @@ pub fn cleanup() {
     unsafe {
         LOCK.lock();
         let queue = QUEUE;
-        QUEUE = 1 as *mut _;
+        QUEUE = 1u as *mut _;
         LOCK.unlock();
 
         // make sure we're not recursively cleaning up

src/libstd/rt/mod.rs

@@ -92,9 +92,7 @@ fn lang_start(main: *const u8, argc: int, argv: *const *const u8) -> int {
         // but we just do this to name the main thread and to give it correct
         // info about the stack bounds.
         let thread: Thread = NewThread::new(Some("<main>".to_string()));
-        thread_info::set((my_stack_bottom, my_stack_top),
-                         sys::thread::guard::main(),
-                         thread);
+        thread_info::set(sys::thread::guard::main(), thread);
 
         // By default, some platforms will send a *signal* when a EPIPE error
         // would otherwise be delivered. This runtime doesn't install a SIGPIPE
@@ -133,20 +131,14 @@ fn lang_start(main: *const u8, argc: int, argv: *const *const u8) -> int {
     }
 }
 
-/// Enqueues a procedure to run when the runtime is cleaned up
-///
-/// The procedure passed to this function will be executed as part of the
-/// runtime cleanup phase. For normal rust programs, this means that it will run
-/// after all other threads have exited.
-///
-/// The procedure is *not* executed with a local `Thread` available to it, so
-/// primitives like logging, I/O, channels, spawning, etc, are *not* available.
-/// This is meant for "bare bones" usage to clean up runtime details, this is
-/// not meant as a general-purpose "let's clean everything up" function.
+/// Enqueues a procedure to run when the main thread exits.
 ///
 /// It is forbidden for procedures to register more `at_exit` handlers when they
 /// are running, and doing so will lead to a process abort.
-pub fn at_exit<F:FnOnce()+Send>(f: F) {
+///
+/// Note that other threads may still be running when `at_exit` routines start
+/// running.
+pub fn at_exit<F: FnOnce() + Send>(f: F) {
     at_exit_imp::push(Thunk::new(f));
 }
 
@@ -162,8 +154,5 @@ pub fn at_exit<F:FnOnce()+Send>(f: F) {
 pub unsafe fn cleanup() {
     args::cleanup();
     sys::stack_overflow::cleanup();
-    // FIXME: (#20012): the resources being cleaned up by at_exit
-    // currently are not prepared for cleanup to happen asynchronously
-    // with detached threads using the resources; for now, we leak.
-    // at_exit_imp::cleanup();
+    at_exit_imp::cleanup();
 }

src/libstd/rt/unwind.rs

@@ -68,7 +68,7 @@ use intrinsics;
 use libc::c_void;
 use mem;
 use sync::atomic;
-use sync::{Once, ONCE_INIT};
+use sys_common::mutex::{Mutex, MUTEX_INIT};
 
 use rt::libunwind as uw;
 
@@ -587,11 +587,20 @@ pub fn begin_unwind<M: Any + Send>(msg: M, file_line: &(&'static str, uint)) ->
 /// Doing this split took the LLVM IR line counts of `fn main() { panic!()
 /// }` from ~1900/3700 (-O/no opts) to 180/590.
 #[inline(never)] #[cold] // this is the slow path, please never inline this
-fn begin_unwind_inner(msg: Box<Any + Send>, file_line: &(&'static str, uint)) -> ! {
+fn begin_unwind_inner(msg: Box<Any + Send>,
+                      file_line: &(&'static str, uint)) -> ! {
     // Make sure the default failure handler is registered before we look at the
     // callbacks.
-    static INIT: Once = ONCE_INIT;
-    INIT.doit(|| unsafe { register(failure::on_fail); });
+    unsafe {
+        static LOCK: Mutex = MUTEX_INIT;
+        static mut INIT: bool = false;
+        LOCK.lock();
+        if !INIT {
+            register(failure::on_fail);
+            INIT = true;
+        }
+        LOCK.unlock();
+    }
 
     // First, invoke call the user-defined callbacks triggered on thread panic.
     //

src/libstd/sys/common/helper_thread.rs

@@ -20,6 +20,8 @@
 //! can be created in the future and there must be no active timers at that
 //! time.
 
+#![macro_escape]
+
 use prelude::*;
 
 use cell::UnsafeCell;
@@ -68,6 +70,17 @@ struct RaceBox(helper_signal::signal);
 unsafe impl Send for RaceBox {}
 unsafe impl Sync for RaceBox {}
 
+macro_rules! helper_init { (static $name:ident: Helper<$m:ty>) => (
+    static $name: Helper<$m> = Helper {
+        lock: ::sync::MUTEX_INIT,
+        cond: ::sync::CONDVAR_INIT,
+        chan: ::cell::UnsafeCell { value: 0 as *mut Sender<$m> },
+        signal: ::cell::UnsafeCell { value: 0 },
+        initialized: ::cell::UnsafeCell { value: false },
+        shutdown: ::cell::UnsafeCell { value: false },
+    };
+) }
+
 impl<M: Send> Helper<M> {
     /// Lazily boots a helper thread, becoming a no-op if the helper has already
     /// been spawned.
@@ -84,7 +97,7 @@ impl<M: Send> Helper<M> {
     {
         unsafe {
             let _guard = self.lock.lock().unwrap();
-            if !*self.initialized.get() {
+            if *self.chan.get() as uint == 0 {
                 let (tx, rx) = channel();
                 *self.chan.get() = mem::transmute(box tx);
                 let (receive, send) = helper_signal::new();
@@ -93,15 +106,17 @@ impl<M: Send> Helper<M> {
                 let receive = RaceBox(receive);
 
                 let t = f();
-                Thread::spawn(move |:| {
+                Thread::spawn(move || {
                     helper(receive.0, rx, t);
                     let _g = self.lock.lock().unwrap();
                     *self.shutdown.get() = true;
                     self.cond.notify_one()
                 }).detach();
 
-                rt::at_exit(move|:| { self.shutdown() });
+                rt::at_exit(move || { self.shutdown() });
                 *self.initialized.get() = true;
+            } else if *self.chan.get() as uint == 1 {
+                panic!("cannot continue usage after shutdown");
             }
         }
     }
@@ -116,7 +131,9 @@ impl<M: Send> Helper<M> {
             // Must send and *then* signal to ensure that the child receives the
             // message. Otherwise it could wake up and go to sleep before we
             // send the message.
-            assert!(!self.chan.get().is_null());
+            assert!(*self.chan.get() as uint != 0);
+            assert!(*self.chan.get() as uint != 1,
+                    "cannot continue usage after shutdown");
             (**self.chan.get()).send(msg);
             helper_signal::signal(*self.signal.get() as helper_signal::signal);
         }
@@ -129,9 +146,13 @@ impl<M: Send> Helper<M> {
             // returns.
             let mut guard = self.lock.lock().unwrap();
 
+            let ptr = *self.chan.get();
+            if ptr as uint == 1 {
+                panic!("cannot continue usage after shutdown");
+            }
             // Close the channel by destroying it
             let chan: Box<Sender<M>> = mem::transmute(*self.chan.get());
-            *self.chan.get() = 0 as *mut Sender<M>;
+            *self.chan.get() = 1 as *mut Sender<M>;
             drop(chan);
             helper_signal::signal(*self.signal.get() as helper_signal::signal);
 

src/libstd/sys/common/mod.rs

@@ -9,7 +9,7 @@
 // except according to those terms.
 
 #![allow(missing_docs)]
-#![allow(dead_code)]
+#![macro_escape]
 
 use io::{mod, IoError, IoResult};
 use prelude::*;

src/libstd/sys/common/mutex.rs

@@ -29,6 +29,7 @@ impl Mutex {
     /// Behavior is undefined if the mutex is moved after the first method is
     /// called on the mutex.
     #[inline]
+    #[allow(dead_code)] // sys is not exported yet
     pub unsafe fn new() -> Mutex { Mutex(imp::Mutex::new()) }
 
     /// Lock the mutex blocking the current thread until it is available.

src/libstd/sys/common/net.rs

@@ -22,7 +22,9 @@ use io::{IoResult, IoError};
 use sys::{mod, retry, c, sock_t, last_error, last_net_error, last_gai_error, close_sock,
           wrlen, msglen_t, os, wouldblock, set_nonblocking, timer, ms_to_timeval,
           decode_error_detailed};
-use sync::{Mutex, MutexGuard};
+use sync::Mutex;
+#[cfg(not(target_os = "linux"))]
+use sync::MutexGuard;
 use sys_common::{mod, keep_going, short_write, timeout};
 use prelude::*;
 use cmp;
@@ -573,11 +575,13 @@ impl Drop for Inner {
     fn drop(&mut self) { unsafe { close_sock(self.fd); } }
 }
 
+#[cfg(not(target_os = "linux"))]
 pub struct Guard<'a> {
     pub fd: sock_t,
     pub guard: MutexGuard<'a, ()>,
 }
 
+#[cfg(not(target_os = "linux"))]
 #[unsafe_destructor]
 impl<'a> Drop for Guard<'a> {
     fn drop(&mut self) {

src/libstd/sys/common/rwlock.rs

@@ -26,6 +26,7 @@ impl RWLock {
     /// Usage of an RWLock is undefined if it is moved after its first use (any
     /// function calls below).
     #[inline]
+    #[allow(dead_code)] // sys is not exported yet
     pub unsafe fn new() -> RWLock { RWLock(imp::RWLock::new()) }
 
     /// Acquire shared access to the underlying lock, blocking the current