forked from andreasbuhr/cppcoro
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathasync_mutex.cpp
More file actions
122 lines (105 loc) · 3.22 KB
/
Copy pathasync_mutex.cpp
File metadata and controls
122 lines (105 loc) · 3.22 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
///////////////////////////////////////////////////////////////////////////////
// Copyright (c) Lewis Baker
// Licenced under MIT license. See LICENSE.txt for details.
///////////////////////////////////////////////////////////////////////////////
#include <cppcoro/async_mutex.hpp>
#include <cassert>
cppcoro::async_mutex::async_mutex() noexcept
: m_state(not_locked)
, m_waiters(nullptr)
{}
cppcoro::async_mutex::~async_mutex()
{
[[maybe_unused]] auto state = m_state.load(std::memory_order_relaxed);
assert(state == not_locked || state == locked_no_waiters);
assert(m_waiters == nullptr);
}
bool cppcoro::async_mutex::try_lock() noexcept
{
// Try to atomically transition from nullptr (not-locked) -> this (locked-no-waiters).
auto oldState = not_locked;
return m_state.compare_exchange_strong(
oldState,
locked_no_waiters,
std::memory_order_acquire,
std::memory_order_relaxed);
}
cppcoro::async_mutex_lock_operation cppcoro::async_mutex::lock_async() noexcept
{
return async_mutex_lock_operation{ *this };
}
cppcoro::async_mutex_scoped_lock_operation cppcoro::async_mutex::scoped_lock_async() noexcept
{
return async_mutex_scoped_lock_operation{ *this };
}
void cppcoro::async_mutex::unlock()
{
assert(m_state.load(std::memory_order_relaxed) != not_locked);
async_mutex_lock_operation* waitersHead = m_waiters;
if (waitersHead == nullptr)
{
auto oldState = locked_no_waiters;
const bool releasedLock = m_state.compare_exchange_strong(
oldState,
not_locked,
std::memory_order_release,
std::memory_order_relaxed);
if (releasedLock)
{
return;
}
// At least one new waiter.
// Acquire the list of new waiter operations atomically.
oldState = m_state.exchange(locked_no_waiters, std::memory_order_acquire);
assert(oldState != locked_no_waiters && oldState != not_locked);
// Transfer the list to m_waiters, reversing the list in the process so
// that the head of the list is the first to be resumed.
auto* next = reinterpret_cast<async_mutex_lock_operation*>(oldState);
do
{
auto* temp = next->m_next;
next->m_next = waitersHead;
waitersHead = next;
next = temp;
} while (next != nullptr);
}
assert(waitersHead != nullptr);
m_waiters = waitersHead->m_next;
// Resume the waiter.
// This will pass the ownership of the lock on to that operation/coroutine.
waitersHead->m_awaiter.resume();
}
bool cppcoro::async_mutex_lock_operation::await_suspend(cppcoro::coroutine_handle<> awaiter) noexcept
{
m_awaiter = awaiter;
std::uintptr_t oldState = m_mutex.m_state.load(std::memory_order_acquire);
while (true)
{
if (oldState == async_mutex::not_locked)
{
if (m_mutex.m_state.compare_exchange_weak(
oldState,
async_mutex::locked_no_waiters,
std::memory_order_acquire,
std::memory_order_relaxed))
{
// Acquired lock, don't suspend.
return false;
}
}
else
{
// Try to push this operation onto the head of the waiter stack.
m_next = reinterpret_cast<async_mutex_lock_operation*>(oldState);
if (m_mutex.m_state.compare_exchange_weak(
oldState,
reinterpret_cast<std::uintptr_t>(this),
std::memory_order_release,
std::memory_order_relaxed))
{
// Queued operation to waiters list, suspend now.
return true;
}
}
}
}