Implement reentrant mutexes and make stdio use them

write_fmt calls write for each formatted field. The default implementation of write_fmt is used,
which will call write on not-yet-locked stdout (and write locking after), therefore making print!
in multithreaded environment still interleave contents of two separate prints.

This patch implements reentrant mutexes, changes stdio handles to use these mutexes and overrides
write_fmt to lock the stdio handle for the whole duration of the call.

Author: nagisa

src/libstd/io/stdio.rs

@@ -16,7 +16,7 @@ use cmp;
 use fmt;
 use io::lazy::Lazy;
 use io::{self, BufReader, LineWriter};
-use sync::{Arc, Mutex, MutexGuard};
+use sync::{Arc, ReentrantMutex, ReentrantMutexGuard};
 use sys::stdio;
 
 /// Stdout used by print! and println! macros
@@ -96,7 +96,7 @@ impl Write for StderrRaw {
 /// of `Stdin` must be executed with care.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub struct Stdin {
-    inner: Arc<Mutex<BufReader<StdinRaw>>>,
+    inner: Arc<ReentrantMutex<BufReader<StdinRaw>>>,
 }
 
 /// A locked reference to the a `Stdin` handle.
@@ -105,7 +105,7 @@ pub struct Stdin {
 /// constructed via the `lock` method on `Stdin`.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub struct StdinLock<'a> {
-    inner: MutexGuard<'a, BufReader<StdinRaw>>,
+    inner: ReentrantMutexGuard<'a, BufReader<StdinRaw>>,
 }
 
 /// Create a new handle to the global standard input stream of this process.
@@ -119,17 +119,17 @@ pub struct StdinLock<'a> {
 /// locked version, `StdinLock`, implements both `Read` and `BufRead`, however.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub fn stdin() -> Stdin {
-    static INSTANCE: Lazy<Mutex<BufReader<StdinRaw>>> = lazy_init!(stdin_init);
+    static INSTANCE: Lazy<ReentrantMutex<BufReader<StdinRaw>>> = lazy_init!(stdin_init);
     return Stdin {
         inner: INSTANCE.get().expect("cannot access stdin during shutdown"),
     };
 
-    fn stdin_init() -> Arc<Mutex<BufReader<StdinRaw>>> {
+    fn stdin_init() -> Arc<ReentrantMutex<BufReader<StdinRaw>>> {
         // 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.
-        Arc::new(Mutex::new(if cfg!(windows) {
+        Arc::new(ReentrantMutex::new(if cfg!(windows) {
             BufReader::with_capacity(8 * 1024, stdin_raw())
         } else {
             BufReader::new(stdin_raw())
@@ -210,7 +210,7 @@ pub struct Stdout {
     // FIXME: this should be LineWriter or BufWriter depending on the state of
     //        stdout (tty or not). Note that if this is not line buffered it
     //        should also flush-on-panic or some form of flush-on-abort.
-    inner: Arc<Mutex<LineWriter<StdoutRaw>>>,
+    inner: Arc<ReentrantMutex<LineWriter<StdoutRaw>>>,
 }
 
 /// A locked reference to the a `Stdout` handle.
@@ -219,7 +219,7 @@ pub struct Stdout {
 /// method on `Stdout`.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub struct StdoutLock<'a> {
-    inner: MutexGuard<'a, LineWriter<StdoutRaw>>,
+    inner: ReentrantMutexGuard<'a, LineWriter<StdoutRaw>>,
 }
 
 /// Constructs a new reference to the standard output of the current process.
@@ -231,13 +231,13 @@ pub struct StdoutLock<'a> {
 /// The returned handle implements the `Write` trait.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub fn stdout() -> Stdout {
-    static INSTANCE: Lazy<Mutex<LineWriter<StdoutRaw>>> = lazy_init!(stdout_init);
+    static INSTANCE: Lazy<ReentrantMutex<LineWriter<StdoutRaw>>> = lazy_init!(stdout_init);
     return Stdout {
         inner: INSTANCE.get().expect("cannot access stdout during shutdown"),
     };
 
-    fn stdout_init() -> Arc<Mutex<LineWriter<StdoutRaw>>> {
-        Arc::new(Mutex::new(LineWriter::new(stdout_raw())))
+    fn stdout_init() -> Arc<ReentrantMutex<LineWriter<StdoutRaw>>> {
+        Arc::new(ReentrantMutex::new(LineWriter::new(stdout_raw())))
     }
 }
 
@@ -264,8 +264,9 @@ impl Write for Stdout {
     fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
         self.lock().write_all(buf)
     }
-    // Don't override write_fmt as it's possible to run arbitrary code during a
-    // write_fmt, allowing the possibility of a recursive lock (aka deadlock)
+    fn write_fmt(&mut self, args: fmt::Arguments) -> io::Result<()> {
+        self.lock().write_fmt(args)
+    }
 }
 #[stable(feature = "rust1", since = "1.0.0")]
 impl<'a> Write for StdoutLock<'a> {
@@ -280,7 +281,7 @@ impl<'a> Write for StdoutLock<'a> {
 /// For more information, see `stderr`
 #[stable(feature = "rust1", since = "1.0.0")]
 pub struct Stderr {
-    inner: Arc<Mutex<StderrRaw>>,
+    inner: Arc<ReentrantMutex<StderrRaw>>,
 }
 
 /// A locked reference to the a `Stderr` handle.
@@ -289,7 +290,7 @@ pub struct Stderr {
 /// method on `Stderr`.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub struct StderrLock<'a> {
-    inner: MutexGuard<'a, StderrRaw>,
+    inner: ReentrantMutexGuard<'a, StderrRaw>,
 }
 
 /// Constructs a new reference to the standard error stream of a process.
@@ -300,13 +301,13 @@ pub struct StderrLock<'a> {
 /// The returned handle implements the `Write` trait.
 #[stable(feature = "rust1", since = "1.0.0")]
 pub fn stderr() -> Stderr {
-    static INSTANCE: Lazy<Mutex<StderrRaw>> = lazy_init!(stderr_init);
+    static INSTANCE: Lazy<ReentrantMutex<StderrRaw>> = lazy_init!(stderr_init);
     return Stderr {
         inner: INSTANCE.get().expect("cannot access stderr during shutdown"),
     };
 
-    fn stderr_init() -> Arc<Mutex<StderrRaw>> {
-        Arc::new(Mutex::new(stderr_raw()))
+    fn stderr_init() -> Arc<ReentrantMutex<StderrRaw>> {
+        Arc::new(ReentrantMutex::new(stderr_raw()))
     }
 }
 
@@ -333,7 +334,9 @@ impl Write for Stderr {
     fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
         self.lock().write_all(buf)
     }
-    // Don't override write_fmt for the same reasons as Stdout
+    fn write_fmt(&mut self, args: fmt::Arguments) -> io::Result<()> {
+        self.lock().write_fmt(args)
+    }
 }
 #[stable(feature = "rust1", since = "1.0.0")]
 impl<'a> Write for StderrLock<'a> {

src/libstd/sync/mod.rs

@@ -20,15 +20,16 @@
 pub use alloc::arc::{Arc, Weak};
 pub use core::atomic;
 
-pub use self::mutex::{Mutex, MutexGuard, StaticMutex};
-pub use self::mutex::MUTEX_INIT;
-pub use self::rwlock::{RwLock, StaticRwLock, RW_LOCK_INIT};
-pub use self::rwlock::{RwLockReadGuard, RwLockWriteGuard};
+pub use self::barrier::{Barrier, BarrierWaitResult};
 pub use self::condvar::{Condvar, StaticCondvar, CONDVAR_INIT};
+pub use self::mutex::MUTEX_INIT;
+pub use self::mutex::{Mutex, MutexGuard, StaticMutex};
 pub use self::once::{Once, ONCE_INIT};
-pub use self::semaphore::{Semaphore, SemaphoreGuard};
-pub use self::barrier::{Barrier, BarrierWaitResult};
 pub use self::poison::{PoisonError, TryLockError, TryLockResult, LockResult};
+pub use self::remutex::{ReentrantMutex, ReentrantMutexGuard};
+pub use self::rwlock::{RwLockReadGuard, RwLockWriteGuard};
+pub use self::rwlock::{RwLock, StaticRwLock, RW_LOCK_INIT};
+pub use self::semaphore::{Semaphore, SemaphoreGuard};
 
 pub use self::future::Future;
 
@@ -40,5 +41,6 @@ mod future;
 mod mutex;
 mod once;
 mod poison;
+mod remutex;
 mod rwlock;
 mod semaphore;

src/libstd/sync/mutex.rs

@@ -212,7 +212,7 @@ impl<T> Mutex<T> {
 
     /// Attempts to acquire this lock.
     ///
-    /// If the lock could not be acquired at this time, then `None` is returned.
+    /// If the lock could not be acquired at this time, then `Err` is returned.
     /// Otherwise, an RAII guard is returned. The lock will be unlocked when the
     /// guard is dropped.
     ///

src/libstd/sys/unix/mutex.rs

@@ -12,6 +12,7 @@ use prelude::v1::*;
 
 use cell::UnsafeCell;
 use sys::sync as ffi;
+use mem;
 
 pub struct Mutex { inner: UnsafeCell<ffi::pthread_mutex_t> }
 
@@ -67,3 +68,38 @@ impl Mutex {
         debug_assert!(r == 0 || r == libc::EINVAL);
     }
 }
+
+pub struct ReentrantMutex { inner: UnsafeCell<ffi::pthread_mutex_t> }
+
+unsafe impl Send for ReentrantMutex {}
+unsafe impl Sync for ReentrantMutex {}
+
+impl ReentrantMutex {
+    pub unsafe fn new() -> ReentrantMutex {
+        let mut mutex: ffi::pthread_mutex_t = mem::uninitialized();
+        let mut attr: ffi::pthread_mutexattr_t = mem::uninitialized();
+        debug_assert_eq!(ffi::pthread_mutexattr_init(&mut attr as *mut _), 0);
+        debug_assert_eq!(ffi::pthread_mutexattr_settype(&mut attr as *mut _,
+                                                        ffi::PTHREAD_MUTEX_RECURSIVE), 0);
+        debug_assert_eq!(ffi::pthread_mutex_init(&mut mutex as *mut _, &attr as *const _), 0);
+        debug_assert_eq!(ffi::pthread_mutexattr_destroy(&mut attr as *mut _), 0);
+        ReentrantMutex { inner: UnsafeCell { value: mutex } }
+    }
+
+    pub unsafe fn lock(&self) {
+        debug_assert_eq!(ffi::pthread_mutex_lock(self.inner.get()), 0);
+    }
+
+    #[inline]
+    pub unsafe fn try_lock(&self) -> bool {
+        ffi::pthread_mutex_trylock(self.inner.get()) == 0
+    }
+
+    pub unsafe fn unlock(&self) {
+        debug_assert_eq!(ffi::pthread_mutex_unlock(self.inner.get()), 0);
+    }
+
+    pub unsafe fn destroy(&self) {
+        debug_assert_eq!(ffi::pthread_mutex_destroy(self.inner.get()), 0);
+    }
+}

src/libstd/sys/unix/sync.rs

@@ -12,17 +12,25 @@
 
 use libc;
 
-pub use self::os::{PTHREAD_MUTEX_INITIALIZER, pthread_mutex_t};
+pub use self::os::{PTHREAD_MUTEX_INITIALIZER, PTHREAD_MUTEX_RECURSIVE, pthread_mutex_t,
+                   pthread_mutexattr_t};
 pub use self::os::{PTHREAD_COND_INITIALIZER, pthread_cond_t};
 pub use self::os::{PTHREAD_RWLOCK_INITIALIZER, pthread_rwlock_t};
 
 extern {
     // mutexes
+    pub fn pthread_mutex_init(lock: *mut pthread_mutex_t, attr: *const pthread_mutexattr_t)
+                            -> libc::c_int;
     pub fn pthread_mutex_destroy(lock: *mut pthread_mutex_t) -> libc::c_int;
     pub fn pthread_mutex_lock(lock: *mut pthread_mutex_t) -> libc::c_int;
     pub fn pthread_mutex_trylock(lock: *mut pthread_mutex_t) -> libc::c_int;
     pub fn pthread_mutex_unlock(lock: *mut pthread_mutex_t) -> libc::c_int;
 
+    pub fn pthread_mutexattr_init(attr: *mut pthread_mutexattr_t) -> libc::c_int;
+    pub fn pthread_mutexattr_destroy(attr: *mut pthread_mutexattr_t) -> libc::c_int;
+    pub fn pthread_mutexattr_settype(attr: *mut pthread_mutexattr_t, _type: libc::c_int)
+                                    -> libc::c_int;
+
     // cvars
     pub fn pthread_cond_wait(cond: *mut pthread_cond_t,
                              lock: *mut pthread_mutex_t) -> libc::c_int;
@@ -52,12 +60,14 @@ mod os {
     use libc;
 
     pub type pthread_mutex_t = *mut libc::c_void;
+    pub type pthread_mutexattr_t = *mut libc::c_void;
     pub type pthread_cond_t = *mut libc::c_void;
     pub type pthread_rwlock_t = *mut libc::c_void;
 
     pub const PTHREAD_MUTEX_INITIALIZER: pthread_mutex_t = 0 as *mut _;
     pub const PTHREAD_COND_INITIALIZER: pthread_cond_t = 0 as *mut _;
     pub const PTHREAD_RWLOCK_INITIALIZER: pthread_rwlock_t = 0 as *mut _;
+    pub const PTHREAD_MUTEX_RECURSIVE: libc::c_int = 2;
 }
 
 #[cfg(any(target_os = "macos", target_os = "ios"))]
@@ -95,6 +105,11 @@ mod os {
         __opaque: [u8; __PTHREAD_MUTEX_SIZE__],
     }
     #[repr(C)]
+    pub struct pthread_mutexattr_t {
+        __sig: libc::c_long,
+        __opaque: [u8; 4],
+    }
+    #[repr(C)]
     pub struct pthread_cond_t {
         __sig: libc::c_long,
         __opaque: [u8; __PTHREAD_COND_SIZE__],
@@ -117,6 +132,8 @@ mod os {
         __sig: _PTHREAD_RWLOCK_SIG_INIT,
         __opaque: [0; __PTHREAD_RWLOCK_SIZE__],
     };
+
+    pub const PTHREAD_MUTEX_RECURSIVE: libc::c_int = 8;
 }
 
 #[cfg(target_os = "linux")]
@@ -161,6 +178,11 @@ mod os {
         size: [u8; __SIZEOF_PTHREAD_MUTEX_T],
     }
     #[repr(C)]
+    pub struct pthread_mutexattr_t {
+        __align: libc::c_longlong,
+        size: [u8; 4],
+    }
+    #[repr(C)]
     pub struct pthread_cond_t {
         __align: libc::c_longlong,
         size: [u8; __SIZEOF_PTHREAD_COND_T],
@@ -183,6 +205,7 @@ mod os {
         __align: 0,
         size: [0; __SIZEOF_PTHREAD_RWLOCK_T],
     };
+    pub const PTHREAD_MUTEX_RECURSIVE: libc::c_int = 1;
 }
 #[cfg(target_os = "android")]
 mod os {
@@ -191,6 +214,8 @@ mod os {
     #[repr(C)]
     pub struct pthread_mutex_t { value: libc::c_int }
     #[repr(C)]
+    pub struct pthread_mutexattr_t { value: libc::c_long }
+    #[repr(C)]
     pub struct pthread_cond_t { value: libc::c_int }
     #[repr(C)]
     pub struct pthread_rwlock_t {
@@ -218,4 +243,5 @@ mod os {
         pendingWriters: 0,
         reserved: [0 as *mut _; 4],
     };
+    pub const PTHREAD_MUTEX_RECURSIVE: libc::c_int = 1;
 }

src/libstd/sys/windows/mutex.rs

@@ -57,3 +57,35 @@ impl Mutex {
         // ...
     }
 }
+
+pub struct ReentrantMutex { inner: UnsafeCell<ffi::CRITICAL_SECTION> }
+
+unsafe impl Send for ReentrantMutex {}
+unsafe impl Sync for ReentrantMutex {}
+
+impl ReentrantMutex {
+    pub unsafe fn new() -> ReentrantMutex {
+        let section = CRITICAL_SECTION_INIT;
+        ffi::InitializeCriticalSection(section);
+        ReentrantMutex {
+            inner: UnsafeCell { value: section }
+        }
+    }
+
+    pub unsafe fn lock(&self) {
+        ffi::EnterCriticalSection(self.inner.get());
+    }
+
+    #[inline]
+    pub unsafe fn try_lock(&self) -> bool {
+        ffi::TryEnterCriticalSection(self.inner.get()) != 0
+    }
+
+    pub unsafe fn unlock(&self) {
+        ffi::LeaveCriticalSection(self.inner.get());
+    }
+
+    pub unsafe fn destroy(&self) {
+        ffi::DeleteCriticalSection(self.inner.get());
+    }
+}

src/libstd/sys/windows/sync.rs

@@ -19,11 +19,14 @@ pub type ULONG = c_ulong;
 pub struct CONDITION_VARIABLE { pub ptr: LPVOID }
 #[repr(C)]
 pub struct SRWLOCK { pub ptr: LPVOID }
+#[repr(C)]
+pub struct CRITICAL_SECTION { pub ptr: LPVOID }
 
 pub const CONDITION_VARIABLE_INIT: CONDITION_VARIABLE = CONDITION_VARIABLE {
     ptr: 0 as *mut _,
 };
 pub const SRWLOCK_INIT: SRWLOCK = SRWLOCK { ptr: 0 as *mut _ };
+pub const CRITICAL_SECTION_INIT: CRITICAL_SECTION = CRITICAL_SECTION { ptr: 0 as *mut _ };
 
 extern "system" {
     // condition variables
@@ -41,4 +44,10 @@ extern "system" {
     pub fn ReleaseSRWLockShared(SRWLock: PSRWLOCK);
     pub fn TryAcquireSRWLockExclusive(SRWLock: PSRWLOCK) -> BOOLEAN;
     pub fn TryAcquireSRWLockShared(SRWLock: PSRWLOCK) -> BOOLEAN;
+
+    pub fn InitializeCriticalSection(CriticalSection: CRITICAL_SECTION);
+    pub fn EnterCriticalSection(CriticalSection: CRITICAL_SECTION);
+    pub fn TryEnterCriticalSection(CriticalSection: CRITICAL_SECTION) -> BOOLEAN;
+    pub fn LeaveCriticalSection(CriticalSection: CRITICAL_SECTION);
+    pub fn DeleteCriticalSection(CriticalSection: CRITICAL_SECTION);
 }