mirror of
https://github.com/skyline-emu/skyline.git
synced 2024-12-23 12:41:50 +01:00
Rework circular queue locking
Should now be (hopefully) race-free, also switch to a spinlock to avoid any locking overhead.
This commit is contained in:
parent
5d527cb965
commit
2f6d27e8d7
@ -15,6 +15,7 @@
|
|||||||
#include <string>
|
#include <string>
|
||||||
#include <memory>
|
#include <memory>
|
||||||
#include <compare>
|
#include <compare>
|
||||||
|
#include <condition_variable>
|
||||||
#include <boost/container/small_vector.hpp>
|
#include <boost/container/small_vector.hpp>
|
||||||
#include <common/exception.h>
|
#include <common/exception.h>
|
||||||
#include <common/span.h>
|
#include <common/span.h>
|
||||||
|
@ -4,6 +4,7 @@
|
|||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
#include <common/trace.h>
|
#include <common/trace.h>
|
||||||
|
#include <common/spin_lock.h>
|
||||||
#include <common.h>
|
#include <common.h>
|
||||||
|
|
||||||
namespace skyline {
|
namespace skyline {
|
||||||
@ -14,12 +15,12 @@ namespace skyline {
|
|||||||
class CircularQueue {
|
class CircularQueue {
|
||||||
private:
|
private:
|
||||||
std::vector<u8> vector; //!< The internal vector holding the circular queue's data, we use a byte vector due to the default item construction/destruction semantics not being appropriate for a circular buffer
|
std::vector<u8> vector; //!< The internal vector holding the circular queue's data, we use a byte vector due to the default item construction/destruction semantics not being appropriate for a circular buffer
|
||||||
Type *start{reinterpret_cast<Type *>(vector.begin().base())}; //!< The start/oldest element of the queue
|
std::atomic<Type *> start{reinterpret_cast<Type *>(vector.begin().base())}; //!< The start/oldest element of the queue
|
||||||
Type *end{reinterpret_cast<Type *>(vector.begin().base())}; //!< The end/newest element of the queue
|
std::atomic<Type *> end{reinterpret_cast<Type *>(vector.begin().base())}; //!< The end/newest element of the queue
|
||||||
std::mutex consumptionMutex;
|
SpinLock consumptionMutex;
|
||||||
std::condition_variable consumeCondition;
|
std::condition_variable_any consumeCondition;
|
||||||
std::mutex productionMutex;
|
SpinLock productionMutex;
|
||||||
std::condition_variable produceCondition;
|
std::condition_variable_any produceCondition;
|
||||||
|
|
||||||
public:
|
public:
|
||||||
/**
|
/**
|
||||||
@ -57,14 +58,14 @@ namespace skyline {
|
|||||||
|
|
||||||
while (true) {
|
while (true) {
|
||||||
if (start == end) {
|
if (start == end) {
|
||||||
std::unique_lock lock{consumptionMutex};
|
std::unique_lock productionLock{productionMutex};
|
||||||
|
|
||||||
TRACE_EVENT_END("containers");
|
TRACE_EVENT_END("containers");
|
||||||
preWait();
|
preWait();
|
||||||
produceCondition.wait(lock, [this]() { return start != end; });
|
produceCondition.wait(productionLock, [this]() { return start != end; });
|
||||||
TRACE_EVENT_BEGIN("containers", "CircularQueue::Process");
|
TRACE_EVENT_BEGIN("containers", "CircularQueue::Process");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
std::scoped_lock comsumptionLock{consumptionMutex};
|
||||||
while (start != end) {
|
while (start != end) {
|
||||||
auto next{start + 1};
|
auto next{start + 1};
|
||||||
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
||||||
@ -77,46 +78,50 @@ namespace skyline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
Type Pop() {
|
Type Pop() {
|
||||||
std::unique_lock lock(productionMutex);
|
{
|
||||||
produceCondition.wait(lock, [this]() { return start != end; });
|
std::unique_lock productionLock{productionMutex};
|
||||||
|
produceCondition.wait(productionLock, [this]() { return start != end; });
|
||||||
|
}
|
||||||
|
|
||||||
|
std::scoped_lock comsumptionLock{consumptionMutex};
|
||||||
auto next{start + 1};
|
auto next{start + 1};
|
||||||
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
||||||
Type item{*next};
|
Type item{*next};
|
||||||
start = next;
|
start = next;
|
||||||
|
|
||||||
if (start == end)
|
|
||||||
consumeCondition.notify_one();
|
consumeCondition.notify_one();
|
||||||
|
|
||||||
return item;
|
return item;
|
||||||
}
|
}
|
||||||
|
|
||||||
void Push(const Type &item) {
|
void Push(const Type &item) {
|
||||||
std::unique_lock lock(productionMutex);
|
Type *waitNext{};
|
||||||
|
|
||||||
|
while (true) {
|
||||||
|
if (waitNext) {
|
||||||
|
std::unique_lock consumeLock{consumptionMutex};
|
||||||
|
consumeCondition.wait(consumeLock, [=]() { return waitNext != start; });
|
||||||
|
waitNext = nullptr;
|
||||||
|
}
|
||||||
|
|
||||||
|
std::scoped_lock lock{productionMutex};
|
||||||
auto next{end + 1};
|
auto next{end + 1};
|
||||||
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
||||||
if (next == start) {
|
if (next == start) {
|
||||||
std::unique_lock consumeLock(consumptionMutex);
|
waitNext = next;
|
||||||
consumeCondition.wait(consumeLock, [=]() { return next != start; });
|
continue;
|
||||||
}
|
}
|
||||||
*next = item;
|
*next = item;
|
||||||
end = next;
|
end = next;
|
||||||
produceCondition.notify_one();
|
produceCondition.notify_one();
|
||||||
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
void Append(span <Type> buffer) {
|
|
||||||
std::unique_lock lock(productionMutex);
|
|
||||||
for (const auto &item : buffer) {
|
|
||||||
auto next{end + 1};
|
|
||||||
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
|
||||||
if (next == start) {
|
|
||||||
std::unique_lock consumeLock(consumptionMutex);
|
|
||||||
consumeCondition.wait(consumeLock, [=]() { return next != start; });
|
|
||||||
}
|
}
|
||||||
*next = item;
|
|
||||||
end = next++;
|
void Append(span<Type> buffer) {
|
||||||
}
|
for (const auto &item : buffer)
|
||||||
produceCondition.notify_one();
|
Push(item);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -125,18 +130,8 @@ namespace skyline {
|
|||||||
*/
|
*/
|
||||||
template<typename TransformedType, typename Transformation>
|
template<typename TransformedType, typename Transformation>
|
||||||
void AppendTranform(TransformedType &container, Transformation transformation) {
|
void AppendTranform(TransformedType &container, Transformation transformation) {
|
||||||
std::unique_lock lock(productionMutex);
|
for (const auto &item : container)
|
||||||
for (auto &item : container) {
|
Push(transformation(item));
|
||||||
auto next{end + 1};
|
|
||||||
next = (next == reinterpret_cast<Type *>(vector.end().base())) ? reinterpret_cast<Type *>(vector.begin().base()) : next;
|
|
||||||
if (next == start) {
|
|
||||||
std::unique_lock consumeLock(consumptionMutex);
|
|
||||||
consumeCondition.wait(consumeLock, [=]() { return next != start; });
|
|
||||||
}
|
|
||||||
*next = transformation(item);
|
|
||||||
end = next;
|
|
||||||
}
|
|
||||||
produceCondition.notify_one();
|
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user