make shared_mutex fair

Author: Kirillog

runtime/include/blocking_primitives.h

@@ -38,7 +38,24 @@ struct mutex {
   friend struct condition_variable;
 };
 
-struct shared_mutex {
+struct condition_variable {
+  as_atomic void wait(std::unique_lock<ltest::mutex>& lock) {
+    addr = lock.mutex()->state.addr;
+    lock.unlock();
+    this_coro->SetBlocked({addr, 1});
+    CoroYield();
+    lock.lock();
+  }
+
+  as_atomic void notify_one() { block_manager.UnblockOn(addr, 1); }
+
+  as_atomic void notify_all() { block_manager.UnblockAllOn(addr); }
+
+ private:
+  std::intptr_t addr;
+};
+
+struct shared_mutex_r {
   as_atomic void lock() {
     while (locked != 0) {
       this_coro->SetBlocked(state);
@@ -67,21 +84,47 @@ struct shared_mutex {
   BlockState state{reinterpret_cast<std::intptr_t>(&locked), locked};
 };
 
-struct condition_variable {
-  as_atomic void wait(std::unique_lock<ltest::mutex>& lock) {
-    addr = lock.mutex()->state.addr;
-    lock.unlock();
-    this_coro->SetBlocked({addr, 1});
-    CoroYield();
-    lock.lock();
+struct shared_mutex {
+  as_atomic void lock() {
+    std::unique_lock lock{mutex_};
+    while (write_) {
+      write_entered_.wait(lock);
+    }
+    write_ = true;
+    while (reader_count_ > 0) {
+      no_readers_.wait(lock);
+    }
   }
 
-  as_atomic void notify_one() { block_manager.UnblockOn(addr, 1); }
+  as_atomic void unlock() {
+    std::unique_lock lock{mutex_};
+    write_ = false;
+    write_entered_.notify_all();
+  }
 
-  as_atomic void notify_all() { block_manager.UnblockAllOn(addr); }
+  as_atomic void lock_shared() {
+    std::unique_lock lock{mutex_};
+    while (write_) {
+      write_entered_.wait(lock);
+    }
+    ++reader_count_;
+  }
+  as_atomic void unlock_shared() {
+    std::unique_lock lock{mutex_};
+    --reader_count_;
+    if (write_ && reader_count_ == 0) {
+      no_readers_.notify_one();
+    }
+  }
 
  private:
-  std::intptr_t addr;
+  int reader_count_{0};
+  bool write_{false};
+
+  ltest::condition_variable write_entered_;
+  ltest::condition_variable no_readers_;
+
+  ltest::mutex mutex_;
 };
 
 }  // namespace ltest

runtime/include/pct_strategy.h

@@ -20,6 +20,37 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
     PrepareForDepth(current_depth, 1);
   }
 
+  void AvoidLivelock(size_t& index, size_t& prior) {
+    auto& threads = this->threads;
+    if (fair_stage > 0) [[unlikely]] {
+      for (size_t attempt = 0; attempt < threads.size(); ++attempt) {
+        auto i = (++last_chosen) % threads.size();
+        if (!threads[i].empty() && threads[i].back()->IsBlocked()) {
+          continue;
+        }
+        index = i;
+        prior = priorities[i];
+        break;
+      }
+      // debug(stderr, "round robin choose: %d\n", index_of_max);
+      if (fair_start == index) {
+        --fair_stage;
+      }
+    }
+
+    // TODO: Choose wiser constant
+    if (count_chosen_same == 1000 && index == last_chosen) [[unlikely]] {
+      fair_stage = 5;
+      fair_start = index;
+    }
+
+    if (index == last_chosen) {
+      ++count_chosen_same;
+    } else {
+      count_chosen_same = 1;
+    }
+  }
+
   std::optional<size_t> NextThreadId() override {
     auto& threads = this->threads;
     size_t max = std::numeric_limits<size_t>::min();
@@ -45,33 +76,7 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
       }
     }
 
-    if (round_robin_stage > 0) [[unlikely]] {
-      for (size_t attempt = 0; attempt < threads.size(); ++attempt) {
-        auto i = (++last_chosen) % threads.size();
-        if (!threads[i].empty() && threads[i].back()->IsBlocked()) {
-          continue;
-        }
-        index_of_max = i;
-        max = priorities[i];
-        break;
-      }
-      // debug(stderr, "round robin choose: %d\n", index_of_max);
-      if (round_robin_start == index_of_max) {
-        --round_robin_stage;
-      }
-    }
-
-    // TODO: Choose wiser constant
-    if (count_chosen_same == 1000 && index_of_max == last_chosen) [[unlikely]] {
-      round_robin_stage = 5;
-      round_robin_start = index_of_max;
-    }
-
-    if (index_of_max == last_chosen) {
-      ++count_chosen_same;
-    } else {
-      count_chosen_same = 1;
-    }
+    AvoidLivelock(index_of_max, max);
 
     if (max == std::numeric_limits<size_t>::min()) [[unlikely]] {
       return std::nullopt;
@@ -117,7 +122,7 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
       }
     }
 
-    if (round_robin_stage > 0) [[unlikely]] {
+    if (fair_stage > 0) [[unlikely]] {
       for (size_t attempt = 0; attempt < threads.size(); ++attempt) {
         auto i = (++last_chosen) % threads.size();
         int task_index = this->GetNextTaskInThread(i);
@@ -130,15 +135,15 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
         break;
       }
       // debug(stderr, "round robin choose: %d\n", index_of_max);
-      if (round_robin_start == index_of_max) {
-        --round_robin_stage;
+      if (fair_start == index_of_max) {
+        --fair_stage;
       }
     }
 
     // TODO: Choose wiser constant
     if (count_chosen_same == 1000 && index_of_max == last_chosen) [[unlikely]] {
-      round_robin_stage = 5;
-      round_robin_start = index_of_max;
+      fair_stage = 5;
+      fair_start = index_of_max;
     }
 
     if (index_of_max == last_chosen) {
@@ -194,7 +199,7 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
     k_statistics.push_back(current_schedule_length);
     current_schedule_length = 0;
     count_chosen_same = 0;
-    round_robin_stage = 0;
+    fair_stage = 0;
 
     // current_depth have been increased
     size_t new_k = std::reduce(k_statistics.begin(), k_statistics.end()) /
@@ -227,8 +232,8 @@ struct PctStrategy : public BaseStrategyWithThreads<TargetObj, Verifier> {
   // original article)
   size_t count_chosen_same;
   size_t last_chosen;
-  size_t round_robin_start;
-  size_t round_robin_stage{0};
+  size_t fair_start;
+  size_t fair_stage{0};
   std::vector<int> priorities;
   std::vector<size_t> priority_change_points;
   std::mt19937 rng;