sanitizer_common: add new mutex

[dvyukov]

Summary

We currently have 3 different mutexes:

• RWMutex
• BlockingMutex
• __tsan::Mutex

RWMutex and __tsan::Mutex are roughly the same,
except that tsan version supports deadlock detection.
BlockingMutex degrades better under heavy contention
from lots of threads (blocks in OS), but much slower
for light contention and has non-portable performance
and has larger static size and is not reader-writer.

Add a new mutex that combines all advantages of these
mutexes: it's reader-writer, has fast non-contended path,
supports blocking to gracefully degrade under higher contention,
has portable size/performance.

For now it's named Mutex2 for incremental submission. The plan is to:

• land this change
• then move deadlock detection logic from tsan
• then rename it to Mutex and remove tsan Mutex
• then typedef RWMutex/BlockingMutex to this mutex

SpinMutex stays as separate type because it has faster fast path:
1 atomic RMW per lock/unlock as compared to 2 for this mutex.

[melver]

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

91

Can we make it have a constexpr constructor?
Probably also needs initializing state_ inline.

106

s/We've spin/We've spun/

157

Would this be cleaner if it was:

do {
....
} while (UNLIKELY(!atomic_compare_exchange_weak(...)));
if (UNLIKELY(wake_writer))
  writers_.Post();
else if (UNLIKELY(wake_readers))
  readers_.Post(wake_readers);

And similar for the other loops in Read*() that do an explicit 'return;' at the end.

That's my preference, because it makes it visually more obvious what is actually repeatedly executed. But of course either is fine if you don't like it.

196

Should there be a check (new_state & kReaderLockMask) == 0 to decide if to wake a waiting writer?

Or is there a performance benefit to waking the writer early, in case there are more readers still locked, and that writer then spinning and hopefully not going back to sleep and more quickly seeing the last reader unlocking?

217

Should this only check & (kReaderLockMask | kWriterLock) ?

I think for the purpose of testing, this would be more accurate and potentially catch those bits not set even though we've entered a C-S.

For the tests, I'd even go so far and add a separate CheckWriterLocked() and CheckReaderLocked() in addition to this.

235

"woken" is used in present perfect tense ("writer has woken up" -> woke in the past, and still awake) or could be passive ("was woken up by other thread").

Does this bit indicate the writer is "awake and spinning", or rather that it "was woken up by somebody else"?

If I read it right, I think it means the former. If that's the case, perhaps "a writer is awake and spin-waiting" is clearer (with flag "kWriterSpinWait" or similar)?

238

s/worken/woken/
?

compiler-rt/lib/sanitizer_common/tests/sanitizer_mutex_test.cpp

38

Can this keep calling mtx_.CheckLocked() (CheckWriterLocked()) ?

58

Call mtx_.CheckLocked() ? Or perhaps just mtx_.CheckReaderLocked() if it can be added.

[dvyukov]

addressed comments

[dvyukov]

PTAL

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

91

I did not make it constexpr b/c then somebody will use it as a global object, but then we will also need to remove DCHECK in destructor. I don't know what's better.

196

Ha, good point! I intended to check (new_state & kReaderLockMask) == 0.
The point re unblocking a writer earlier is also good. I don't know what's better. Since I intended to check kReaderLockMask and you asked about it, let's go with that additional check. It looks more straightforward.

[melver]

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

91

It's still usable as a global object, compiler won't complain, unless that global has attribute [[clang::require_constant_initialization]]. Without it, the only way to discourage that is to make the constructor private and instead have a factory function, but I think that's not something we should do for a Mutex.

If used as a global object, the DCHECK may then crash in other spectacular ways, but at least it'd crash/warn somehow if state_!=0 (and there's a bug :-)). In production builds the DCHECK doesn't exist.

Perhaps a comment, if it really shouldn't be used as a global object, but I think as-is (modulo DCHECK), there is no technical limitation that makes it unusable as a global object.

So I think it's still safer + more efficient to make it constexpr.

133

s/spinned/spun/

[vitalybuka]

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

157

I slightly prefer previous version. Uninitialized variable now concern me more than continue and return from loop. It requires more mental efforts to track values. Also prev version was similar to Lock() implementation.

[dvyukov]

made ctor constexpr

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

157

I can change it again, but, folks, you need to decide :)
Marco?

[melver]

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

157

I think it's on you now, whatever you prefer.

( Personally, I'm not worried about declaring variables and a few lines below clearly initialize them, perhaps because I've been staring at too much Linux kernel code and this style is all over the place ... so I'd do that and get the nicer-to-follow control-flow. But there are good arguments the other way around as well. But in the end this is all just bikeshedding. :-) )

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

157

OK, I will go with the current version then just because that's the code I have now (and I am a bit afraid of introducing bugs while changing this code).

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

public:
 
   void Wait();
   void Post(u32 count = 1);
 
  private:
   atomic_uint32_t state_ = {0};
 };
 
+// Reader-writer mutex.
+class MUTEX Mutex2 {
+ public:
+  constexpr Mutex2() {}
+
+  void Lock() ACQUIRE() {
+    u64 reset_mask = ~0ull;
+    u64 state = atomic_load_relaxed(&state_);
+    const uptr kMaxSpinIters = 1500;
+    for (uptr spin_iters = 0;; spin_iters++) {
+      u64 new_state;
+      bool locked = (state & (kWriterLock | kReaderLockMask)) != 0;
+      if (LIKELY(!locked)) {
+        // The mutex is not read-/write-locked, try to lock.
+        new_state = (state | kWriterLock) & reset_mask;
+      } else if (spin_iters > kMaxSpinIters) {
+        // We've spun enough, increment waiting writers count and block.
+        // The counter will be decremented by whoever wakes us.
+        new_state = (state + kWaitingWriterInc) & reset_mask;
+      } else if ((state & kWriterSpinWait) == 0) {
+        // Active spinning, but denote our presence so that unlocking
+        // thread does not wake up other threads.
+        new_state = state | kWriterSpinWait;
+      } else {
+        // Active spinning.
+        state = atomic_load(&state_, memory_order_relaxed);
+        continue;
+      }
+      if (UNLIKELY(!atomic_compare_exchange_weak(&state_, &state, new_state,
+                                                 memory_order_acquire)))
+        continue;
+      if (LIKELY(!locked))
+        return;  // We've locked the mutex.
+      if (spin_iters > kMaxSpinIters) {
+        // We've incremented waiting writers, so now block.
+        writers_.Wait();
+        spin_iters = 0;
+        state = atomic_load(&state_, memory_order_relaxed);
+        DCHECK_NE(state & kWriterSpinWait, 0);
+      } else {
+        // We've set kWriterSpinWait, but we are still in active spinning.
+      }
+      // We either blocked and were unblocked,
+      // or we just spun but set kWriterSpinWait.
+      // Either way we need to reset kWriterSpinWait
+      // next time we take the lock or block again.
+      reset_mask = ~kWriterSpinWait;
+    }
+  }
+
+  void Unlock() RELEASE() {
+    bool wake_writer;
+    u64 wake_readers;
+    u64 new_state;
+    u64 state = atomic_load_relaxed(&state_);
+    do {
+      DCHECK_NE(state & kWriterLock, 0);
+      DCHECK_EQ(state & kReaderLockMask, 0);
+      new_state = state & ~kWriterLock;
+      wake_writer =
+          (state & kWriterSpinWait) == 0 && (state & kWaitingWriterMask) != 0;
+      if (wake_writer)
+        new_state = (new_state - kWaitingWriterInc) | kWriterSpinWait;
+      wake_readers =
+          (state & (kWriterSpinWait | kWaitingWriterMask)) != 0
+              ? 0
+              : ((state & kWaitingReaderMask) >> kWaitingReaderShift);
+      if (wake_readers)
+        new_state = (new_state & ~kWaitingReaderMask) +
+                    (wake_readers << kReaderLockShift);
+    } while (UNLIKELY(!atomic_compare_exchange_weak(&state_, &state, new_state,
+                                                    memory_order_release)));
+    if (UNLIKELY(wake_writer))
+      writers_.Post();
+    else if (UNLIKELY(wake_readers))
+      readers_.Post(wake_readers);
+  }
+
+  void ReadLock() ACQUIRE_SHARED() {
+    bool locked;
+    u64 new_state;
+    u64 state = atomic_load_relaxed(&state_);
+    do {
+      locked =
+          (state & kReaderLockMask) == 0 &&
+          (state & (kWriterLock | kWriterSpinWait | kWaitingWriterMask)) != 0;
+      if (LIKELY(!locked))
+        new_state = state + kReaderLockInc;
+      else
+        new_state = state + kWaitingReaderInc;
+    } while (UNLIKELY(!atomic_compare_exchange_weak(&state_, &state, new_state,
+                                                    memory_order_acquire)));
+    if (UNLIKELY(locked))
+      readers_.Wait();
+    DCHECK_EQ(atomic_load_relaxed(&state_) & kWriterLock, 0);
+    DCHECK_NE(atomic_load_relaxed(&state_) & kReaderLockMask, 0);
+  }
+
+  void ReadUnlock() RELEASE_SHARED() {
+    bool wake;
+    u64 new_state;
+    u64 state = atomic_load_relaxed(&state_);
+    do {
+      DCHECK_NE(state & kReaderLockMask, 0);
+      DCHECK_EQ(state & (kWaitingReaderMask | kWriterLock), 0);
+      new_state = state - kReaderLockInc;
+      wake = (new_state & (kReaderLockMask | kWriterSpinWait)) == 0 &&
+             (new_state & kWaitingWriterMask) != 0;
+      if (wake)
+        new_state = (new_state - kWaitingWriterInc) | kWriterSpinWait;
+    } while (UNLIKELY(!atomic_compare_exchange_weak(&state_, &state, new_state,
+                                                    memory_order_release)));
+    if (UNLIKELY(wake))
+      writers_.Post();
+  }
+
+  // This function does not guarantee an explicit check that the calling thread
+  // is the thread which owns the mutex. This behavior, while more strictly
+  // correct, causes problems in cases like StopTheWorld, where a parent thread
+  // owns the mutex but a child checks that it is locked. Rather than
+  // maintaining complex state to work around those situations, the check only
+  // checks that the mutex is owned.
+  void CheckWriteLocked() const CHECK_LOCKED() {
+    CHECK(atomic_load(&state_, memory_order_relaxed) & kWriterLock);
+  }
+
+  void CheckLocked() const CHECK_LOCKED() { CheckWriteLocked(); }
+
+  void CheckReadLocked() const CHECK_LOCKED() {
+    CHECK(atomic_load(&state_, memory_order_relaxed) & kReaderLockMask);
+  }
+
+ private:
+  atomic_uint64_t state_ = {0};
+  Semaphore writers_;
+  Semaphore readers_;
+
+  // The state has 3 counters:
+  //  - number of readers holding the lock,
+  //    if non zero, the mutex is read-locked
+  //  - number of waiting readers,
+  //    if not zero, the mutex is write-locked
+  //  - number of waiting writers,
+  //    if non zero, the mutex is read- or write-locked
+  // And 2 flags:
+  //  - writer lock
+  //    if set, the mutex is write-locked
+  //  - a writer is awake and spin-waiting
+  //    the flag is used to prevent thundering herd problem
+  //    (new writers are not woken if this flag is set)
+  //
+  // Writer support active spinning, readers does not.
+  // But readers are more aggressive and always take the mutex
+  // if there are any other readers.
+  // Writers hand off the mutex to readers: after wake up readers
+  // already assume ownership of the mutex (don't need to do any
+  // state updates). But the mutex is not handed off to writers,
+  // after wake up writers compete to lock the mutex again.
+  // This is needed to allow repeated write locks even in presence
+  // of other blocked writers.
+  static constexpr u64 kCounterWidth = 20;
+  static constexpr u64 kReaderLockShift = 0;
+  static constexpr u64 kReaderLockInc = 1ull << kReaderLockShift;
+  static constexpr u64 kReaderLockMask = ((1ull << kCounterWidth) - 1)
+                                         << kReaderLockShift;
+  static constexpr u64 kWaitingReaderShift = kCounterWidth;
+  static constexpr u64 kWaitingReaderInc = 1ull << kWaitingReaderShift;
+  static constexpr u64 kWaitingReaderMask = ((1ull << kCounterWidth) - 1)
+                                            << kWaitingReaderShift;
+  static constexpr u64 kWaitingWriterShift = 2 * kCounterWidth;
+  static constexpr u64 kWaitingWriterInc = 1ull << kWaitingWriterShift;
+  static constexpr u64 kWaitingWriterMask = ((1ull << kCounterWidth) - 1)
+                                            << kWaitingWriterShift;
+  static constexpr u64 kWriterLock = 1ull << (3 * kCounterWidth);
+  static constexpr u64 kWriterSpinWait = 1ull << (3 * kCounterWidth + 1);
+
+  Mutex2(const Mutex2 &) = delete;
+  void operator=(const Mutex2 &) = delete;
+};
+
 void FutexWait(atomic_uint32_t *p, u32 cmp);
 void FutexWake(atomic_uint32_t *p, u32 count);
 
 class MUTEX BlockingMutex {
  public:
   explicit constexpr BlockingMutex(LinkerInitialized)
       : opaque_storage_ {0, }, owner_ {0} {}
   BlockingMutex();

compiler-rt/lib/sanitizer_common/tests/sanitizer_mutex_test.cpp

explicit TestData(MutexType *mtx)
     for (int i = 0; i < kSize; i++)
       data_[i] = 0;
   }
 
   void Write() {
     Lock l(mtx_);
     T v0 = data_[0];
     for (int i = 0; i < kSize; i++) {
+      mtx_->CheckLocked();
       CHECK_EQ(data_[i], v0);
       data_[i]++;
     }
   }
 
   void TryWrite() {
     if (!mtx_->TryLock())
       return;
     T v0 = data_[0];
     for (int i = 0; i < kSize; i++) {
+      mtx_->CheckLocked();
       CHECK_EQ(data_[i], v0);
       data_[i]++;
     }
     mtx_->Unlock();
   }
 
+  void Read() {
+    ReadLock l(mtx_);
+    T v0 = data_[0];
+    for (int i = 0; i < kSize; i++) {
+      mtx_->CheckReadLocked();
+      CHECK_EQ(data_[i], v0);
+    }
+  }
+
   void Backoff() {
     volatile T data[kSize] = {};
     for (int i = 0; i < kSize; i++) {
       data[i]++;
       CHECK_EQ(data[i], 1);
     }
   }
 
  private:
   typedef GenericScopedLock<MutexType> Lock;
+  typedef GenericScopedReadLock<MutexType> ReadLock;
   static const int kSize = 64;
   typedef u64 T;
   MutexType *mtx_;
   char pad_[kCacheLineSize];
   T data_[kSize];
 };
 
 const int kThreads = 8;
static void *try_thread(void *param) {
   TestData<MutexType> *data = (TestData<MutexType>*)param;
   for (int i = 0; i < kIters; i++) {
     data->TryWrite();
     data->Backoff();
   }
   return 0;
 }
 
-template<typename MutexType>
+template <typename MutexType>
+static void *read_write_thread(void *param) {
+  TestData<MutexType> *data = (TestData<MutexType> *)param;
+  for (int i = 0; i < kIters; i++) {
+    if ((i % 10) == 0)
+      data->Write();
+    else
+      data->Read();
+    data->Backoff();
+  }
+  return 0;
+}
+
+template<typename MutexType>
 static void check_locked(MutexType *mtx) {
   GenericScopedLock<MutexType> l(mtx);
   mtx->CheckLocked();
 }
 
 TEST(SanitizerCommon, SpinMutex) {
   SpinMutex mtx;
   mtx.Init();
TEST(SanitizerCommon, BlockingMutex) {
   pthread_t threads[kThreads];
   for (int i = 0; i < kThreads; i++)
     PTHREAD_CREATE(&threads[i], 0, lock_thread<BlockingMutex>, &data);
   for (int i = 0; i < kThreads; i++)
     PTHREAD_JOIN(threads[i], 0);
   check_locked(mtx);
 }
 
+TEST(SanitizerCommon, Mutex2) {
+  Mutex2 mtx;
+  TestData<Mutex2> data(&mtx);
+  pthread_t threads[kThreads];
+  for (int i = 0; i < kThreads; i++)
+    PTHREAD_CREATE(&threads[i], 0, read_write_thread<Mutex2>, &data);
+  for (int i = 0; i < kThreads; i++) PTHREAD_JOIN(threads[i], 0);
+}
+
 struct SemaphoreData {
   Semaphore *sem;
   bool done;
 };
 
 void *SemaphoreThread(void *arg) {
   auto data = static_cast<SemaphoreData *>(arg);
   data->sem->Wait();