DrSocalkwe3n/LuaVox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Пересмотр асинхронностей

Browse Source
This commit is contained in:
DrSocalkwe3n
2025-03-11 14:55:43 +06:00
parent e99ae2f6ba
commit e190c79d00
12 changed files with 482 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Src/Client/ServerSession.hpp
View File

@@ -24,7 +24,8 @@ struct ParsedPacket {
virtual ~ParsedPacket();
};
class ServerSession : public AsyncObject, public IServerSession, public ISurfaceEventListener {
class ServerSession : public IServerSession, public ISurfaceEventListener {
asio::io_context &IOC;
std::unique_ptr<Net::AsyncSocket> Socket;
IRenderSession *RS = nullptr;
DestroyLock UseLock;
@@ -56,10 +57,10 @@ class ServerSession : public AsyncObject, public IServerSession, public ISurface
public:
// Нужен сокет, на котором только что был согласован игровой протокол (asyncInitGameProtocol)
ServerSession(asio::io_context &ioc, std::unique_ptr<Net::AsyncSocket> &&socket, IRenderSession *rs = nullptr)
: AsyncObject(ioc), Socket(std::move(socket)), RS(rs), NetInputPackets(1024)
: IOC(ioc), Socket(std::move(socket)), RS(rs), NetInputPackets(1024)
{
assert(Socket.get());
co_spawn(run());
asio::co_spawn(IOC, run(), asio::detached);
}
virtual ~ServerSession();

4
Src/Client/Vulkan/Vulkan.cpp
View File

@@ -73,7 +73,7 @@ ByteBuffer loadPNG(std::istream &&read, int &width, int &height, bool &hasAlpha,
}
Vulkan::Vulkan(asio::io_context &ioc)
: AsyncObject(ioc), GuardLock(ioc.get_executor())
: IOC(ioc), GuardLock(ioc.get_executor())
{
Screen.Width = 1920/2;
Screen.Height = 1080/2;
@@ -2075,7 +2075,7 @@ void Vulkan::gui_MainMenu() {
ConnectionProgress.InProgress = true;
ConnectionProgress.Cancel = false;
ConnectionProgress.Progress.clear();
co_spawn(ConnectionProgress.connect(IOC));
asio::co_spawn(IOC, ConnectionProgress.connect(IOC), asio::detached);
}
if(!Game.Server) {

3
Src/Client/Vulkan/Vulkan.hpp
View File

@@ -71,9 +71,10 @@ class Buffer;
Vulkan.reInit();
*/
class Vulkan : public AsyncObject {
class Vulkan {
private:
Logger LOG = "Vulkan";
asio::io_context &IOC;
struct vkInstanceLayer {
std::string LayerName = "nullptr", Description = "nullptr";

160
Src/Common/Async.hpp
View File

@@ -1,14 +1,16 @@
#pragma once
#include "TOSLib.hpp"
#include "boost/asio/async_result.hpp"
#include "boost/asio/deadline_timer.hpp"
#include "boost/date_time/posix_time/posix_time_duration.hpp"
#include <atomic>
#include <boost/asio.hpp>
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/impl/execution_context.hpp>
#include <boost/date_time/posix_time/ptime.hpp>
#include <boost/thread.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/lockfree/queue.hpp>
namespace LV {
@@ -20,152 +22,4 @@ using tcp = asio::ip::tcp;
template<typename T = void>
using coro = asio::awaitable<T>;
class AsyncObject {
protected:
asio::io_context &IOC;
asio::deadline_timer WorkDeadline;
public:
AsyncObject(asio::io_context &ioc)
: IOC(ioc), WorkDeadline(ioc, boost::posix_time::pos_infin)
{
}
inline asio::io_context& EXEC()
{
return IOC;
}
protected:
template<typename Coroutine>
void co_spawn(Coroutine &&coroutine) {
asio::co_spawn(IOC, WorkDeadline.async_wait(asio::use_awaitable) || std::move(coroutine), asio::detached);
}
};
template<typename ValueType>
class AsyncAtomic : public AsyncObject {
protected:
asio::deadline_timer Deadline;
ValueType Value;
boost::mutex Mtx;
public:
AsyncAtomic(asio::io_context &ioc, ValueType &&value)
: AsyncObject(ioc), Deadline(ioc), Value(std::move(value))
{
}
AsyncAtomic& operator=(ValueType &&value) {
boost::unique_lock lock(Mtx);
Value = std::move(value);
Deadline.expires_from_now(boost::posix_time::pos_infin);
return *this;
}
operator ValueType() const {
return Value;
}
ValueType operator*() const {
return Value;
}
AsyncAtomic& operator++() {
boost::unique_lock lock(Mtx);
Value--;
Deadline.expires_from_now(boost::posix_time::pos_infin);
return *this;
}
AsyncAtomic& operator--() {
boost::unique_lock lock(Mtx);
Value--;
Deadline.expires_from_now(boost::posix_time::pos_infin);
return *this;
}
AsyncAtomic& operator+=(ValueType value) {
boost::unique_lock lock(Mtx);
Value += value;
Deadline.expires_from_now(boost::posix_time::pos_infin);
return *this;
}
AsyncAtomic& operator-=(ValueType value) {
boost::unique_lock lock(Mtx);
Value -= value;
Deadline.expires_from_now(boost::posix_time::pos_infin);
return *this;
}
void wait(ValueType oldValue) {
while(true) {
if(oldValue != Value)
return;
boost::unique_lock lock(Mtx);
if(oldValue != Value)
return;
std::atomic_bool flag = false;
Deadline.async_wait([&](boost::system::error_code errc) { flag.store(true); flag.notify_all(); });
lock.unlock();
flag.wait(false);
}
}
void await(ValueType needValue) {
while(true) {
if(needValue == Value)
return;
boost::unique_lock lock(Mtx);
if(needValue == Value)
return;
std::atomic_bool flag = false;
Deadline.async_wait([&](boost::system::error_code errc) { flag.store(true); flag.notify_all(); });
lock.unlock();
flag.wait(false);
}
}
coro<> async_wait(ValueType oldValue) {
while(true) {
if(oldValue != Value)
co_return;
boost::unique_lock lock(Mtx);
if(oldValue != Value)
co_return;
auto coroutine = Deadline.async_wait();
lock.unlock();
try { co_await std::move(coroutine); } catch(...) {}
}
}
coro<> async_await(ValueType needValue) {
while(true) {
if(needValue == Value)
co_return;
boost::unique_lock lock(Mtx);
if(needValue == Value)
co_return;
auto coroutine = Deadline.async_wait();
lock.unlock();
try { co_await std::move(coroutine); } catch(...) {}
}
}
};
}

33
Src/Common/Net.cpp
View File

@@ -17,7 +17,7 @@ bool SocketServer::isStopped() {
coro<void> SocketServer::run(std::function<coro<>(tcp::socket)> onConnect) {
while(true) { // TODO: ловить ошибки на async_accept
try {
co_spawn(onConnect(co_await Acceptor.async_accept()));
asio::co_spawn(IOC, onConnect(co_await Acceptor.async_accept()), asio::detached);
} catch(const std::exception &exc) {
if(const boost::system::system_error *errc = dynamic_cast<const boost::system::system_error*>(&exc);
errc && (errc->code() == asio::error::operation_aborted || errc->code() == asio::error::bad_descriptor))
@@ -31,13 +31,8 @@ AsyncSocket::~AsyncSocket() {
if(SendPackets.Context)
SendPackets.Context->NeedShutdown = true;
{
boost::lock_guard lock(SendPackets.Mtx);
SendPackets.SenderGuard.cancel();
WorkDeadline.cancel();
}
boost::unique_lock lock(SendPackets.Mtx);
if(Socket.is_open())
try { Socket.close(); } catch(...) {}
}
@@ -55,6 +50,8 @@ void AsyncSocket::pushPackets(std::vector<Packet> *simplePackets, std::vector<Sm
// TODO: std::cout << "Передоз пакетами, сокет закрыт" << std::endl;
}
bool wasPackets = SendPackets.SimpleBuffer.size() || SendPackets.SmartBuffer.size();
if(!Socket.is_open()) {
if(simplePackets)
simplePackets->clear();
@@ -86,8 +83,7 @@ void AsyncSocket::pushPackets(std::vector<Packet> *simplePackets, std::vector<Sm
SendPackets.SizeInQueue += addedSize;
if(SendPackets.WaitForSemaphore) {
SendPackets.WaitForSemaphore = false;
if(!wasPackets) {
SendPackets.Semaphore.cancel();
SendPackets.Semaphore.expires_at(boost::posix_time::pos_infin);
}
@@ -147,12 +143,19 @@ coro<> AsyncSocket::runSender(std::shared_ptr<AsyncContext> context) {
while(!context->NeedShutdown) {
{
boost::unique_lock lock(SendPackets.Mtx);
if(SendPackets.SimpleBuffer.empty() && SendPackets.SmartBuffer.empty()) {
SendPackets.WaitForSemaphore = true;
auto coroutine = SendPackets.Semaphore.async_wait();
lock.unlock();
try { co_await std::move(coroutine); } catch(...) {}
if(context->NeedShutdown) {
break;
}
if(SendPackets.SimpleBuffer.empty() && SendPackets.SmartBuffer.empty()) {
auto coroutine = SendPackets.Semaphore.async_wait();
if(SendPackets.SimpleBuffer.empty() && SendPackets.SmartBuffer.empty()) {
lock.unlock();
try { co_await std::move(coroutine); } catch(...) {}
}
continue;
} else {
for(int cycle = 0; cycle < 2; cycle++, NextBuffer++) {

24
Src/Common/Net.hpp
View File

@@ -3,28 +3,28 @@
#include "MemoryPool.hpp"
#include "Async.hpp"
#include "TOSLib.hpp"
#include "boost/asio/io_context.hpp"
#include <atomic>
#include <boost/asio.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/write.hpp>
#include <boost/thread.hpp>
#include <boost/circular_buffer.hpp>
#include <condition_variable>
namespace LV::Net {
class SocketServer : public AsyncObject {
class SocketServer {
protected:
asio::io_context &IOC;
tcp::acceptor Acceptor;
public:
SocketServer(asio::io_context &ioc, std::function<coro<>(tcp::socket)> &&onConnect, uint16_t port = 0)
: AsyncObject(ioc), Acceptor(ioc, tcp::endpoint(tcp::v4(), port))
: IOC(ioc), Acceptor(ioc, tcp::endpoint(tcp::v4(), port))
{
assert(onConnect);
co_spawn(run(std::move(onConnect)));
asio::co_spawn(IOC, run(std::move(onConnect)), asio::detached);
}
bool isStopped();
@@ -178,7 +178,8 @@ protected:
std::function<std::optional<SmartPacket>()> OnSend;
};
class AsyncSocket : public AsyncObject {
class AsyncSocket {
asio::io_context &IOC;
NetPool::Array<32> RecvBuffer, SendBuffer;
size_t RecvPos = 0, RecvSize = 0, SendSize = 0;
bool ReadShutdowned = false;
@@ -196,22 +197,21 @@ protected:
struct SendPacketsObj {
boost::mutex Mtx;
bool WaitForSemaphore = false;
asio::deadline_timer Semaphore, SenderGuard;
asio::deadline_timer Semaphore;
boost::circular_buffer_space_optimized<Packet> SimpleBuffer;
boost::circular_buffer_space_optimized<SmartPacket> SmartBuffer;
size_t SizeInQueue = 0;
std::shared_ptr<AsyncContext> Context;
SendPacketsObj(asio::io_context &ioc)
: Semaphore(ioc, boost::posix_time::pos_infin), SenderGuard(ioc, boost::posix_time::pos_infin)
: Semaphore(ioc, boost::posix_time::pos_infin)
{}
} SendPackets;
public:
AsyncSocket(asio::io_context &ioc, tcp::socket &&socket)
: AsyncObject(ioc), Socket(std::move(socket)), SendPackets(ioc)
{
: IOC(ioc), Socket(std::move(socket)), SendPackets(ioc)
{
SendPackets.SimpleBuffer.set_capacity(512);
SendPackets.SmartBuffer.set_capacity(SendPackets.SimpleBuffer.capacity()/4);
SendPackets.Context = std::make_shared<AsyncContext>();
@@ -221,7 +221,7 @@ protected:
boost::asio::ip::tcp::no_delay optionNoDelay(true); // Отключает попытки объёденить данные в крупные пакеты
Socket.set_option(optionNoDelay);
co_spawn(runSender(SendPackets.Context));
asio::co_spawn(IOC, runSender(SendPackets.Context), asio::detached);
}
~AsyncSocket();

422
Src/Common/async_mutex.hpp Normal file
View File

@@ -0,0 +1,422 @@
// SPDX-FileCopyrightText: 2023 Daniel Vrátil <daniel.vratil@gendigital.com>
// SPDX-FileCopyrightText: 2023 Martin Beran <martin.beran@gendigital.com>
//
// SPDX-License-Identifier: BSL-1.0
#pragma once
#include <boost/asio/associated_executor.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/async_result.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <atomic>
#define ASIO_NS boost::asio
namespace avast::asio {
class async_mutex_lock;
class async_mutex;
/** \internal **/
namespace detail {
/**
* \brief Represents a suspended coroutine that is awaiting lock acquisition.
**/
struct locked_waiter {
/**
* \brief Constructs a new locked_waiter.
* \param next_waiter Pointer to the waiter to prepend this locked_waiter to.
**/
explicit locked_waiter(locked_waiter *next_waiter): next(next_waiter) {}
locked_waiter(locked_waiter &&) = delete;
locked_waiter(const locked_waiter &) = delete;
locked_waiter &operator=(locked_waiter &&) = delete;
locked_waiter &operator=(const locked_waiter &) = delete;
/**
* \brief Destructor.
**/
virtual ~locked_waiter() = default;
/**
* \brief Completes the pending asynchronous operation.
*
* Resumes the currently suspended coroutine with the acquired lock.
**/
virtual void completion() = 0;
/**
* The waiters are held in a linked list. This is a pointer to the next member of the list.
**/
locked_waiter *next = nullptr;
};
/**
* \brief Locked waiter that used `async_mutex::async_lock()` to acquire the lock.
**/
template <typename Token>
struct async_locked_waiter final: public locked_waiter {
/**
* \brief Constructs a new async_locked_waiter.
* \param mutex A mutex that the waiter is trying to acquire a lock for.
* \param next_waiter Pointer to the head of the waiters linked list to prepend this waiter to.
* \param token The complention token to call when the asynchronous operation is completed.
**/
async_locked_waiter([[maybe_unused]] async_mutex *mutex, locked_waiter *next_waiter, Token &&token):
locked_waiter(next_waiter), m_token(std::move(token)) {}
void completion() override {
auto executor = ASIO_NS::get_associated_executor(m_token);
ASIO_NS::post(std::move(executor), [token = std::move(m_token)]() mutable { token(); });
}
private:
Token m_token; //!< The completion token to invoke when the lock is acquired.
};
/**
* \brief Locked waiter that used `async_mutex::async_scoped_lock()` to acquire the lock.
**/
template <typename Token>
struct scoped_async_locked_waiter final: public locked_waiter {
/**
* \brief Constructs a new scoped_async_locked_waiter.
* \param mutex A mutex that the waiter is trying to acquire a lock for.
* \param next_waiter Pointer to the head of the waiters linked list to prepend this waiter to.
* \param token The complention token to call when the asynchronous operation is completed.
**/
scoped_async_locked_waiter(async_mutex *mutex, locked_waiter *next_waiter, Token &&token):
locked_waiter(next_waiter), m_mutex(mutex), m_token(std::move(token)) {}
void completion() override;
private:
async_mutex *m_mutex; //!< The mutex whose lock is being awaited.
Token m_token; //!< The completion token to invoke when the lock is acquired.
};
/**
* \brief An initiator for asio::async_initiate().
**/
template <template <typename Token> typename Waiter>
class async_lock_initiator_base {
public:
/**
* Constructs a new initiator for an operation on the given mutex.
*
* \param mutex A mutex on which the asynchronous lock operation is being initiated.
**/
explicit async_lock_initiator_base(async_mutex *mutex): m_mutex(mutex) {}
/**
* \brief Invoked by boost asio when the asynchronous operation is initiated.
*
* \param handler A completion handler (a callable) to be called when the asynchronous operation
* has completed (in our case, the lock has been acquired).
* \tparam Handler A callable with signature void(T) where T is the type of the object that will be
* returned as a result of `co_await`ing the operation. In our case that's either
* `void` for `async_lock()` or `async_mutex_lock` for `async_scoped_lock()`.
**/
template <typename Handler>
void operator()(Handler &&handler);
protected:
async_mutex *m_mutex; //!< The mutex whose lock is being awaited.
};
/**
* \brief Initiator for the async_lock() operation.
**/
using initiate_async_lock = async_lock_initiator_base<async_locked_waiter>;
/**
* \brief Initiator for the async_scoped_lock() operation.
**/
using initiate_scoped_async_lock = async_lock_initiator_base<scoped_async_locked_waiter>;
} // namespace detail
/** \endinternal **/
/**
* \brief A basic mutex that can acquire lock asynchronously using asio coroutines.
**/
class async_mutex {
public:
/**
* \brief Constructs a new unlocked mutex.
**/
async_mutex() noexcept = default;
async_mutex(const async_mutex &) = delete;
async_mutex(async_mutex &&) = delete;
async_mutex &operator=(const async_mutex &) = delete;
async_mutex &operator=(async_mutex &&) = delete;
/**
* \brief Destroys the mutex.
*
* \warning Destroying a mutex in locked state is undefined.
**/
~async_mutex() {
[[maybe_unused]] const auto state = m_state.load(std::memory_order_relaxed);
assert(state == not_locked || state == locked_no_waiters);
assert(m_waiters == nullptr);
}
/**
* \brief Attempts to acquire lock without blocking.
*
* \return Returns `true` when the lock has been acquired, `false` when the
* lock is already held by someone else.
* **/
[[nodiscard]] bool try_lock() noexcept {
auto old_state = not_locked;
return m_state.compare_exchange_strong(old_state, locked_no_waiters, std::memory_order_acquire,
std::memory_order_relaxed);
}
/**
* \brief Asynchronously acquires as lock.
*
* When the returned awaitable is `co_await`ed it initiates the process
* of acquiring a lock. The awaiter is suspended. Once the lock is acquired
* (which can be immediately if nothing else holds the lock currently) the
* awaiter is resumed and is now holding the lock.
*
* It's awaiter's responsibility to release the lock by calling `unlock()`.
*
* \param token A completion token (`asio::use_awaitable`).
* \tparam LockToken Type of the complention token.
* \return An awaitable which will initiate the async operation when `co_await`ed.
* The result of `co_await`ing the awaitable is void.
**/
#ifdef DOXYGEN
template <typename LockToken>
ASIO_NS::awaitable<> async_lock(LockToken &&token);
#else
template <ASIO_NS::completion_token_for<void()> LockToken>
[[nodiscard]] auto async_lock(LockToken &&token) {
return ASIO_NS::async_initiate<LockToken, void()>(detail::initiate_async_lock(this), token);
}
#endif
/**
* \brief Asynchronously acquires a lock and returns a scoped lock helper.
*
* Behaves exactly as `async_lock()`, except that the result of `co_await`ing the
* returned awaitable is a scoped lock object, which will automatically release the
* lock when destroyed.
*
* \param token A completion token (`asio::use_awaitable`).
* \tparam LockToken Type of the completion token.
* \returns An awaitable which will initiate the async operation when `co_await`ed.
* The result of `co_await`ing the awaitable is `async_mutex_lock` holding
* the acquired lock.
**/
#ifdef DOXYGEN
template <typename LockToken>
ASIO_NS::awaitable<async_mutex_lock> async_scoped_lock(LockToken &&token);
#else
template <ASIO_NS::completion_token_for<void(async_mutex_lock)> LockToken>
[[nodiscard]] auto async_scoped_lock(LockToken &&token) {
return ASIO_NS::async_initiate<LockToken, void(async_mutex_lock)>(detail::initiate_scoped_async_lock(this),
token);
}
#endif
/**
* \brief Releases the lock.
*
* \warning Unlocking and already unlocked mutex is undefined.
**/
void unlock() {
assert(m_state.load(std::memory_order_relaxed) != not_locked);
auto *waiters_head = m_waiters;
if (waiters_head == nullptr) {
auto old_state = locked_no_waiters;
// If old state was locked_no_waiters then transitions to not_locked and returns true,
// otherwise do nothing and returns false.
const bool released_lock = m_state.compare_exchange_strong(old_state, not_locked, std::memory_order_release,
std::memory_order_relaxed);
if (released_lock) {
return;
}
// At least one new waiter. Acquire the list of new waiters atomically
old_state = m_state.exchange(locked_no_waiters, std::memory_order_acquire);
assert(old_state != locked_no_waiters && old_state != not_locked);
// Transfer the list to m_waiters, reversing the list in the process
// so that the head of the list is the first waiter to be resumed
// NOLINTNEXTLINE(performance-no-int-to-ptr)
auto *next = reinterpret_cast<detail::locked_waiter *>(old_state);
do {
auto *temp = next->next;
next->next = waiters_head;
waiters_head = next;
next = temp;
} while (next != nullptr);
}
assert(waiters_head != nullptr);
m_waiters = waiters_head->next;
// Complete the async operation.
waiters_head->completion();
delete waiters_head;
}
private:
template <template <typename Token> typename Waiter>
friend class detail::async_lock_initiator_base;
/**
* \brief Indicates that the mutex is not locked.
**/
static constexpr std::uintptr_t not_locked = 1;
/**
* \brief Indicates that the mutex is locked, but no-one else is attempting to acquire the lock at the moment.
**/
static constexpr std::uintptr_t locked_no_waiters = 0;
/**
* \brief Holds the current state of the lock.
*
* The state can be `not_locked`, `locked_no_waiters` or a pointer to the head of a linked list
* of new waiters (waiters who have attempted to acquire the lock since the last call to unlock().
**/
std::atomic<std::uintptr_t> m_state = {not_locked};
/**
* \brief Linked list of known locked waiters.
**/
detail::locked_waiter *m_waiters = nullptr;
};
/**
* \brief A RAII-style lock for async_mutex which automatically unlocks the mutex when destroyed.
**/
class async_mutex_lock {
public:
using mutex_type = async_mutex;
/**
* Constructs a new async_mutex_lock without any associated mutex.
**/
explicit async_mutex_lock() noexcept = default;
/**
* Constructs a new async_mutex_lock, taking ownership of the \c mutex.
*
* \param mutex Locked mutex to be unlocked when this objectis destroyed.
*
* \warning The \c mutex must be in a locked state.
**/
explicit async_mutex_lock(mutex_type &mutex, std::adopt_lock_t) noexcept: m_mutex(&mutex) {}
/**
* \brief Initializes the lock with contents of other. Leaves other with no associated mutex.
* \param other The moved-from object.
**/
async_mutex_lock(async_mutex_lock &&other) noexcept { swap(other); }
/**
* \brief Move assignment operator.
* Replaces the current mutex with those of \c other using move semantics.
* If \c *this already has an associated mutex, the mutex is unlocked.
*
* \param other The moved-from object.
* \returns *this.
*/
async_mutex_lock &operator=(async_mutex_lock &&other) noexcept {
if (m_mutex != nullptr) {
m_mutex->unlock();
}
m_mutex = std::exchange(other.m_mutex, nullptr);
return *this;
}
/**
* \brief Copy constructor (deleted).
**/
async_mutex_lock(const async_mutex_lock &) = delete;
/**
* \brief Copy assignment operator (deleted).
**/
async_mutex_lock &operator=(const async_mutex_lock &) = delete;
~async_mutex_lock() {
if (m_mutex != nullptr) {
m_mutex->unlock();
}
}
bool owns_lock() const noexcept { return m_mutex != nullptr; }
mutex_type *mutex() const noexcept { return m_mutex; }
/**
* \brief Swaps state with \c other.
* \param other the lock to swap state with.
**/
void swap(async_mutex_lock &other) noexcept { std::swap(m_mutex, other.m_mutex); }
private:
mutex_type *m_mutex = nullptr; //!< The locked mutex being held by the scoped mutex lock.
};
/** \internal **/
namespace detail {
template <typename Token>
void scoped_async_locked_waiter<Token>::completion() {
auto executor = ASIO_NS::get_associated_executor(m_token);
ASIO_NS::post(std::move(executor), [token = std::move(m_token), mutex = m_mutex]() mutable {
token(async_mutex_lock{*mutex, std::adopt_lock});
});
}
template <template <typename Token> typename Waiter>
template <typename Handler>
void async_lock_initiator_base<Waiter>::operator()(Handler &&handler) {
auto old_state = m_mutex->m_state.load(std::memory_order_acquire);
std::unique_ptr<Waiter<Handler>> waiter;
while (true) {
if (old_state == async_mutex::not_locked) {
if (m_mutex->m_state.compare_exchange_weak(old_state, async_mutex::locked_no_waiters,
std::memory_order_acquire, std::memory_order_relaxed))
{
// Lock acquired, resume the awaiter stright away
if (waiter) {
waiter->next = nullptr;
waiter->completion();
} else {
Waiter(m_mutex, nullptr, std::forward<Handler>(handler)).completion();
}
return;
}
} else {
if (!waiter) {
// NOLINTNEXTLINE(performance-no-int-to-ptr)
auto *next_waiter = reinterpret_cast<locked_waiter *>(old_state);
waiter.reset(new Waiter(m_mutex, next_waiter, std::forward<Handler>(handler)));
} else {
// NOLINTNEXTLINE(performance-no-int-to-ptr)
waiter->next = reinterpret_cast<locked_waiter *>(old_state);
}
if (m_mutex->m_state.compare_exchange_weak(old_state, reinterpret_cast<std::uintptr_t>(waiter.get()),
std::memory_order_release, std::memory_order_relaxed))
{
waiter.release();
return;
}
}
}
}
} // namespace detail
/** \endinternal **/
} // namespace avast::asio

4
Src/Server/BinaryResourceManager.cpp
View File

@@ -13,7 +13,7 @@ namespace LV::Server {
BinaryResourceManager::BinaryResourceManager(asio::io_context &ioc,
std::shared_ptr<ResourceFile> zeroResource)
: AsyncObject(ioc), ZeroResource(std::move(zeroResource))
: IOC(ioc), ZeroResource(std::move(zeroResource))
{
}
@@ -88,7 +88,7 @@ ResourceId_t BinaryResourceManager::getResource_Assets(std::string path) {
UpdatedResources.lock_write()->push_back(resId);
} else {
res->IsLoading = true;
co_spawn(checkResource_Assets(resId, iter->second / inDomainPath, res));
asio::co_spawn(IOC, checkResource_Assets(resId, iter->second / inDomainPath, res), asio::detached);
}
}

4
Src/Server/BinaryResourceManager.hpp
View File

@@ -10,13 +10,15 @@
#include <vector>
#include <Common/Async.hpp>
#include "Abstract.hpp"
#include "boost/asio/io_context.hpp"
namespace LV::Server {
namespace fs = std::filesystem;
class BinaryResourceManager : public AsyncObject {
class BinaryResourceManager {
asio::io_context &IOC;
public:
private:

6
Src/Server/GameServer.cpp
View File

@@ -20,8 +20,10 @@ namespace LV::Server {
GameServer::~GameServer() {
shutdown("on ~GameServer");
RunThread.join();
WorkDeadline.cancel();
UseLock.wait_no_use();
LOG.info() << "Сервер уничтожен";
}
@@ -448,7 +450,7 @@ void GameServer::stepPlayers() {
auto lock = External.NewConnectedPlayers.lock_write();
for(std::unique_ptr<RemoteClient> &client : *lock) {
co_spawn(client->run());
asio::co_spawn(IOC, client->run(), asio::detached);
Game.CECs.push_back(std::make_unique<ContentEventController>(std::move(client)));
}

5
Src/Server/GameServer.hpp
View File

@@ -25,7 +25,8 @@ namespace LV::Server {
namespace fs = std::filesystem;
class GameServer : public AsyncObject {
class GameServer {
asio::io_context &IOC;
TOS::Logger LOG = "GameServer";
DestroyLock UseLock;
std::thread RunThread;
@@ -112,7 +113,7 @@ class GameServer : public AsyncObject {
public:
GameServer(asio::io_context &ioc, fs::path worldPath)
: AsyncObject(ioc),
: IOC(ioc),
Content(ioc, nullptr, nullptr, nullptr)
{
init(worldPath);

4
Src/main.cpp
View File

@@ -1,6 +1,8 @@
#include <iostream>
#include <boost/asio.hpp>
#include <Client/Vulkan/Vulkan.hpp>
#include <Common/Async.hpp>
#include <Common/async_mutex.hpp>
namespace LV {
@@ -11,7 +13,7 @@ int main() {
// LuaVox
asio::io_context ioc;
LV::Client::VK::Vulkan vkInst(ioc);
ioc.run();
return 0;