Background
I wanted to write an event loop that supports scheduling, but found that no implementation of priority queue that I know of supports waiting on a condition variable (the idea is that the looping thread will wait on the condition variable and if the wait timed out, it means that it can pop the top of the queue and execute that, otherwise it was interrupted and needs to reconsider sleeping time).
I thought about using plain std::mutexes. Somebody already found out how to do that. But I thought perhaps I could do better. So I ventured forth into a world of atomics ... and there was no end to surprises.
Priority lock
The basic idea was stolen from a bakery lock (I believe it is also called a ticket lock). Normal priority threads take a "ticket" on attempt to enter into critical section. When the critical section has their ticket number, they enter the critical section. The leaving thread will increment the ticket number that should enter the critical section. The only change is that there is an additional atomic flag that is used to control mutual exclusion for the priority thread in case it needs to skip the line.
Guarantees
- Mutual exclusion
Mutual exclusion guarantee is provided by the atomic flag is_lock_held. The entering thread can enter only if it is false, and it acquires the lock via strong compare and exchange (CAS). When the thread that owns the lock exits the critical section, it sets the flag to false.
- Strong order among normal priority threads
This is guaranteed by ticketing system. The thread that enters the wait state earlier in the view of next_label will enter the critical section earlier than others. Only highest priority thread (of which there is only one) can skip the line.
- Deadlock freedom
Unless there was some serious error (process crash or some other way to bypass RAII) the lock is guaranteed to be released in case of exceptions and any other control flow RAII can affect. There is also absence of ways for threads to actively block each other by doing actions that would prevent the other from progress. Only the current owner of the critical section will decide if anybody will progress and then cache effects will determine which thread goes next (since atomics are used, exactly one is guaranteed to progress).
- Only one high priority lock at a time
The implementation does CAS on high priority lock creation (if the counter is 0, replace with 1). The destructor of the high priority lock will reset it back to 0.
Facepalm. I somehow managed to put the reset to zero in the constructor. I guess writing code at night is a bad idea.
The ugly parts
- Potential starvation
As any atomics only algorithm, the lock can starve any thread. The high priority lock might get starved because it is not fast enough to acquire the boolean flag (getting scheduled out of CPU in inopportune moment). The normal priority threads can be starved by design.
Code
#include <cstddef>
#include <atomic>
#include <stdexcept>
#include <immintrin.h>
namespace shino {
/**
* @brief the shared state of the normal priority and highest priority locks. This class is intended to be
* a factory to the mentioned locks, thus not usable directly.
*
* Use `create_normal_priority_lock` and `create_highest_priority_lock` to obtain locks which are akin to
* std::unique_lock (they do provide exception safety by unlocking in the destructor).
*/
class priority_mutex {
std::atomic<std::size_t> current_label = 0;
std::atomic<std::size_t> next_label = 0;
std::atomic<bool> is_lock_held = false;
std::atomic<int> priority_thread_count = 0;
public:
/**
* @brief a lock class intended to be used by normal priority thread.
*
* The mechanism of acquiring lock is by obtaining
* a new label to wait on, and then when the other normal priority thread sets the label of the lock to the label
* obtained by this thread, it will try to lock the boolean which is used to allow high priority thread to
* skip the waiting line and be the next owner of the lock.
*/
class normal_priority_thread_lock {
priority_mutex* shared_state = nullptr;
std::size_t my_label = -1;
bool is_locked = false;
public:
normal_priority_thread_lock(const normal_priority_thread_lock&) = delete;
normal_priority_thread_lock& operator=(const normal_priority_thread_lock&) = delete;
/**
* @brief Lock the priority_mutex using normal priority (will yield to high priority thread if contended),
* the calling thread will be granted exclusive ownership to the critical section guarded by this `priority_mutex`
*
* @pre the lock is in unlocked state (all previous `lock()` calls were followed by `unlock()` in one by one manner)
* @post the lock is in locked state
*/
void lock() {
if (is_locked) {
throw std::logic_error("attempt to deadlock by double locking");
}
my_label = shared_state->next_label++;
while (shared_state->current_label.load(std::memory_order_consume) != my_label) {
_mm_pause();
}
/*
* attempt to reduce starvation of the priority thread. Since pause takes more than one instruction,
* when the pause points don't exactly align, the priority thread might miss the window to lock the
* atomic bool. That is why there is exactly one pause after the loop
*/
_mm_pause();
/*
* don't spin on CAS to decrease cache traffic
*/
bool expected = false;
do {
expected = false;
while (shared_state->is_lock_held.load(std::memory_order_consume) != expected)
{
_mm_pause();
}
} while (!shared_state->is_lock_held.compare_exchange_strong(expected, true, std::memory_order_acq_rel, std::memory_order_consume));
is_locked = true;
}
/**
* @brief Unlock the critical section guarded by this `priority_mutex` thus allowing the next thread to progress.
* If highest_priority_thread_lock contends with other locks, the highest priority one will get priority.
*
* @pre the lock is in locked state (all previous `unlock()` calls were followed by `lock()` in one by one manner)
* @post the lock is in unlocked state
*/
void unlock() {
if (!is_locked) {
throw std::logic_error("attempting to unlock non locked lock");
}
bool expected = true;
if (!shared_state->is_lock_held.compare_exchange_strong(expected, false, std::memory_order_acq_rel, std::memory_order_consume)) {
throw std::logic_error("either double unlock or something wrong with the lock itself");
}
const auto next_in_line = my_label + 1;
auto stored_label = my_label;
if (!shared_state->current_label.compare_exchange_strong(stored_label, next_in_line, std::memory_order_acq_rel, std::memory_order_consume)) {
throw std::logic_error("somebody acquired the lock before this lock unlocked?");
}
is_locked = false;
}
~normal_priority_thread_lock() {
if (is_locked) {
unlock();
}
}
private:
friend priority_mutex;
normal_priority_thread_lock(priority_mutex* shared_state, bool start_locked):
shared_state(shared_state)
{
if (start_locked) {
lock();
}
}
};
/**
* @brief a lock class intended to be used by highest priority thread.
*
* The mechanism of locking is to acquire atomic boolean by setting it to true when it is false. Normal priority
* thread is unlikely to contend if both were waiting on the locked lock, because the boolean will be unlocked
* first and only then the label will be set to the label of the waiting normal priority thread. **Staration is
* possible in theory**.
*/
class highest_priority_thread_lock {
priority_mutex* shared_state = nullptr;
bool is_locked = false;
public:
highest_priority_thread_lock(const highest_priority_thread_lock&) = delete;
highest_priority_thread_lock& operator=(const highest_priority_thread_lock&) = delete;
/**
* @brief Lock the priority_mutex using highest priority (will skip the line if contended on a locked lock),
* the calling thread will be granted exclusive ownership to the critical section guarded by this `priority_mutex`
*
* @pre the lock is in unlocked state (all previous `lock()` calls were followed by `unlock()` in one by one manner)
* @post the lock is in locked state
*/
void lock() {
if (is_locked) {
throw std::logic_error("attempting to deadlock via double locking");
}
/*
* don't spin on CAS to decrease cache traffic
*/
bool expected = false;
do {
expected = false;
while (shared_state->is_lock_held.load(std::memory_order_consume) != expected)
{
_mm_pause();
}
} while (!shared_state->is_lock_held.compare_exchange_strong(expected, true, std::memory_order_acq_rel, std::memory_order_consume));
is_locked = true;
}
/**
* @brief Unlock the critical section guarded by this `priority_mutex` thus allowing the next thread to progress.
* If highest_priority_thread_lock contends with other locks, the highest priority one will get priority.
*
* @pre the lock is in locked state (all previous `unlock()` calls were followed by `lock()` in one by one manner)
* @post the lock is in unlocked state
*/
void unlock() {
if (!is_locked) {
throw std::logic_error("attempting to unlock non-locked lock");
}
bool expected = true;
if (!shared_state->is_lock_held.compare_exchange_strong(expected, false, std::memory_order_acq_rel, std::memory_order_acquire)) {
throw std::logic_error("either double unlock or something wrong with the lock itself");
}
is_locked = false;
}
~highest_priority_thread_lock() {
if (is_locked) {
unlock();
}
}
private:
friend priority_mutex;
highest_priority_thread_lock(priority_mutex* shared_state, bool start_locked):
shared_state(shared_state)
{
if (start_locked) {
lock();
}
shared_state->priority_thread_count.store(0, std::memory_order_release);
}
};
/**
* @brief creates an instance of `normal_priority_thread_lock` that shares inner state with this mutex
* @param start_locked if true the lock will be locked on construction, otherwise it will be constructed in
* unlocked state
* @return instance of `normal_priority_thread_lock`
*/
normal_priority_thread_lock create_normal_priority_lock(bool start_locked = true) {
return {this, start_locked};
}
/**
* @brief creates an instance of `highest_priority_thread_lock` that shares inner state with this mutex.
* There can be only one highest priority lock at a time.
*
* @param start_locked if true the lock will be locked on construction, otherwise it will be constructed in
* unlocked state
* @return instance of `highest_priority_thread_lock`
*/
highest_priority_thread_lock create_highest_priority_lock(bool start_locked = true) {
int desired = 0;
if (priority_thread_count.compare_exchange_strong(desired, 1, std::memory_order_acq_rel)) {
return {this, start_locked};
} else {
throw std::logic_error("attempting to create multiple highest priority locks");
}
}
};
}
/*
* Test: the idea is to have a normal priority thread to be first and lock for a long amount of time,
* and then the priority thread should take the lock even though there are threads waiting before it
*/
#include <iostream>
#include <thread>
void first_thread_function(shino::priority_mutex& mutex, std::atomic<bool>& flag) {
auto normal_lock = mutex.create_normal_priority_lock();
flag.store(true, std::memory_order_release);
std::cout << "print from normal thread\n";
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
void normal_priority_function(shino::priority_mutex& mutex) {
auto normal_lock = mutex.create_normal_priority_lock();
std::cout << "print from normal thread\n";
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
void high_priority_function(shino::priority_mutex& mutex) {
auto priority_lock = mutex.create_highest_priority_lock();
std::cout << "print from priority thread\n";
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
int main() {
for (std::size_t i = 0; i < 4; ++i) {
std::atomic<bool> start_flag = false;
shino::priority_mutex mutex;
auto t0 = std::thread(first_thread_function, std::ref(mutex), std::ref(start_flag));
while (!start_flag.load(std::memory_order_consume)) {_mm_pause();}
auto t1 = std::thread(normal_priority_function, std::ref(mutex));
auto t2 = std::thread(normal_priority_function, std::ref(mutex));
auto t3 = std::thread(high_priority_function, std::ref(mutex));
t0.join();
t1.join();
t2.join();
t3.join();
std::cout << "=======\n";
}
return 0;
}
Why there is insufficient testing?
I only did a sanity check because I am planning on getting the tests reviewed as well separately. The reason for separation is to not overwhelm the reviewers with complexity. Please downvote and comment if you think this approach is wrong.
The output of the test should be
print from normal thread
print from priority thread
print from normal thread
print from normal thread
=======
print from normal thread
print from priority thread
print from normal thread
print from normal thread
=======
print from normal thread
print from priority thread
print from normal thread
print from normal thread
=======
print from normal thread
print from priority thread
print from normal thread
print from normal thread
=======
Build
Since this is a single file, any compiler invocation with C++17 mode will do.
What review I would prefer?
I would prefer a code review that would focus on proper usage of atomics and better mechanisms to reduce starvation. I thought about the layout of the class all falling on the same cache line, but making the class bigger was not the decision I wanted to make.
bool). The create function does CAS (if 0 replace with 1) and if it fails it will throw. The destructor of high priority lock will set it to 0 again. Edited into question. \$\endgroup\$std::atomic<int> priority_thread_count;threw me off, as it seemed to suggest there may be several. \$\endgroup\$