Utilities for creating a lock hierarchy?
Yes, lock hierarchies can effectively prevent deadlocks; of course whether you can actually define a hierarchy for your program (especially, in the presence of plugins) is another matter entirely.
The basic blocks are simple:
- Each mutex should have a level (either determined at compile-time or run-time)
- Each thread should only ever acquire mutex in ascending or descending level (decide once)
I hope I can do the idea justice, please consider the example implementation below a sketch; it has never been compiled/tested.
A basic mutex:
template <typename Mutex, size_t Level>
class HierarchicalMutex {
public:
friend class LevelManager;
void lock() {
LevelManager::Lock(*this);
}
void unlock() {
LevelManager::Unlock(*this);
}
private:
size_t previous;
Mutex mutex;
}; // class HierarchicalMutex
template <typename Mutex, size_t Level>
size_t level(HierarchicalMutex<Mutex,Level> const&) { return Level; }
The LevelManager
's role is simply to ensure that level transitions happen in correct order.
class LevelManager {
public:
//
// Single Mutex locking
//
template <typename M>
static void Lock(M& m) {
m.previous = LevelUp(level(m));
m.mutex.lock();
}
template <typename M>
static void Unlock(M& m) {
m.mutex.unlock();
LevelDown(level(m), m.previous);
}
//
// Multiple Mutexes Group Locking
//
// Note: those should expose a "size_t level(M const&)" function,
// and calls to lock/unlock should appropriately call
// this manager to raise/lower the current level.
//
// Note: mutexes acquired as a group
// should be released with the same group.
//
template <typename M>
static void Lock(std::array_ref<M*> mutexes) { // I wish this type existed
using std::begin; using std::end;
auto begin = begin(mutexes);
auto end = end(mutexes);
end = std::remove_if(begin, end, [](M const* m) { return m == 0; });
if (begin == end) { return; }
Sort(begin, end);
size_t const previous = LevelUp(level(*std::prev(end)));
for (; begin != end; ++begin) {
begin->previous = previous;
begin->mutex.lock();
}
}
template <typename M>
static void Unlock(std::array_ref<M*> mutexes) {
using std::begin; using std::end;
auto begin = begin(mutexes);
auto end = end(mutexes);
end = std::remove_if(begin, end, [](M const* m) { return m == 0; });
if (begin == end) { return; }
Sort(begin, end);
std::reverse(begin, end);
for (auto it = begin; it != end; ++it) { it->mutex.unlock(); }
LevelDown(level(*begin), begin->previous);
}
private:
static __thread size_t CurrentLevel = 0;
template <typename It>
static void Sort(It begin, It end) {
using Ref = typename std::iterator_traits<It>::const_reference;
auto const sorter = [](Ref left, Ref right) {
return std::tie(level(left), left) < std::tie(level(right), right);
};
std::sort(begin, end, sorter);
}
static size_t LevelUp(size_t const to) {
if (CurrentLevel >= to) { throw LockHierarchyViolation(); }
CurrentLevel = to;
}
static void LevelDown(size_t const from, size_t const to) {
if (CurrentLevel != from) { throw LockHierarchyViolation(); }
CurrentLevel = to;
}
}; // class LevelManager
For kicks, I implemented the possibility to lock multiples locks of the same level in a single shot.
No need for a separate class to manage the hierarchy. A nice solution can be found in C++ Concurrency in Action, by Anthony Williams (ISBN 9781933988771):
#include <mutex>
#include <stdexcept>
class hierarchical_mutex
{
std::mutex internal_mutex;
unsigned long const hierarchy_value;
unsigned long previous_hierarchy_value;
static thread_local unsigned long this_thread_hierarchy_value;
void check_for_hierarchy_violation()
{
if(this_thread_hierarchy_value <= hierarchy_value)
{
throw std::logic_error("mutex hierarchy violated");
}
}
void update_hierarchy_value()
{
previous_hierarchy_value=this_thread_hierarchy_value;
this_thread_hierarchy_value=hierarchy_value;
}
public:
explicit hierarchical_mutex(unsigned long value):
hierarchy_value(value),
previous_hierarchy_value(0)
{}
void lock()
{
check_for_hierarchy_violation();
internal_mutex.lock();
update_hierarchy_value();
}
void unlock()
{
this_thread_hierarchy_value=previous_hierarchy_value;
internal_mutex.unlock();
}
bool try_lock()
{
check_for_hierarchy_violation();
if(!internal_mutex.try_lock())
return false;
update_hierarchy_value();
return true;
}
};
thread_local unsigned long
hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);
int main()
{
hierarchical_mutex m1(42);
hierarchical_mutex m2(2000);
}
The main thing you can do in this sort of case is just make sure your locks are always hierarchically applied (meaning nested). That way you can't access the level 3 lock without possessing the level 2 lock, which you can't access without first possessing the level 1 lock. You won't even be able to get to 3 without first getting to 1 and 2, so that should prevent major problems.
Can you be more specific in some of the deadlock cases that do arise? Maybe we can find a workaround for some of the particularly complicated things that may not be as easy to manipulate as I described above.