c++20协程实现actor模型
!!没有在大功能里测试过,打算边做边改
思路和遇到的问题
首先协程的定义, 一个函数体内部有co_await co_return co_yield这三个之一,这个函数被视为协程,创建时将整个函数放在堆上,恢复时再从堆上加载到栈上执行,所以可以在其他任意地方任意线程恢复它,看起来协程本身好像没有实体一样,但是这个协程函数返回一个对象,其实返回的这个对象就是协程句柄,这个对象有诸多要求,只不过这个句柄很自由,让人看起来无从下手
反过来说,定义协程句柄类A,类A需要含有内置类型名promise_type,并且promise_type需要内含几个指定名字的函数
接下来写一个函数返回类型是A, 内含任意co_await co_return co_yield,调用该函数就会以协程的概念的运行
关于promise_type内指定名字的函数,应该是通过概念约束的方式,满足该概念即被视为满足部分协程要求
做的时候我问ai有没有什么办法来判断一个函数是协程,ai给我写了一个概念约束,约束就是该函数返回值类型内含promise_type,并且promise_type包含指定名称的函数,但是没办法判断函数体内是否有co_await co_return co_yield,所以没办法百分百保证
Ai给出的约束
template<typename F>
concept IsCoroutine = requires {
// 验证存在promise_type
typename decltype(get_coroutine_traits(args_tuple{}))::promise_type;
// ...其他略
};
template <typename P>
struct is_valid_promise<P, std::void_t<
decltype(std::declval<P>().get_return_object()),
decltype(std::declval<P>().initial_suspend()),
decltype(std::declval<P>().final_suspend()),
decltype(std::declval<P>().unhandled_exception())
>> : std::true_type {};
消息传递格式
一开始消息传递想要做各种类型安全保障,但是各种实现要么代价很高,要么代码复杂
使用基类派生类,虚函数,虚函数在高频消息传递中代价不低,放弃
后来试图通过概念约束的方式将虚函数从运行期挪到编译期,但是概念本身没有载体,又需要std::ant std::variant这种东西作为载体,感觉还不如虚函数
std::typeid和数据指针一起传递?开始神志不清
AI写一个,模板套了七八层看不懂
最后躺平了,传递原始指针算了,使用者自己强转,自己管理生命周期,go的actor调用也是自己转,不过人家有反射能检查,c+转坏了就崩,后来加了一个协程崩溃重启的机制
异步调用
异步调用很好做,加锁,消息队列就可以实现了
定时调用
定时调用是在管理类中开了一个线程定时检查哪些actor到时间了去调用,一开始是在该线程中直接调用注册的定时函数,后来考虑到actor的创建大概率在不同线程,如果别的线程中恢复调用了协程,这个定时线程中又恢复一次,会出现未定义行为
定时调用的实现是异步调用,然后投递一个-1的消息,这样在处理异步消息时检查到-1消息就去执行一下定时的逻辑,定时逻辑本身记录了执行时间,然后计算误差修正下一次的执行时间
同步调用
同步调用和定时调用出现了同样的竞争问题,同步调用使用std::future包裹调用等待一个返回值,该包裹调用也投递到异步调用的逻辑中,然后立刻唤醒,优先执行同步调用然后返回结果,
Actor和Actor互相调用锁
一开始没有处理这个,实际用的时候
给A投递消息,锁住全局Actor列表,然后立刻唤醒了A, 这时全局列表还没释放
A处理消息,内部又给B投递消息,B尝试锁定全局Actor列表
发生死锁
解决:缩小锁的范围,先从全局列表取出A,然后释放锁,为了防止取出A之后A被释放,改用sharedptr防止析构
完整代码
#ifndef _ACTOR_H_
#define _ACTOR_H_
#include <string>
#include <coroutine>
#include <queue>
#include <mutex>
#include <unordered_map>
#include <unordered_set>
#include <atomic>
#include <functional>
#include <memory>
#include <thread>
#include <future>
#include "BaseType.h"
using std::string;
using std::lock_guard;
using std::mutex;
using std::unordered_map;
using std::unordered_set;
using std::shared_ptr;
using std::thread;
using std::atomic;
using std::vector;
class C_Actor;
struct CoroutineMsg
{
INT32 MsgId;
void* pData;
};
// 全局Actor管理
class C_ActorMgr
{
public:
static C_ActorMgr& Instance()
{
static C_ActorMgr instance;
static std::once_flag flag;
std::call_once(flag, [&] { instance.StartTimer(); });
return instance;
}
~C_ActorMgr() { StopTimer(); }
unordered_map<string, std::shared_ptr<C_Actor>> AllActor; // 所有Actor列表
unordered_set<string> ActorTimer; // 有定时任务要跑的actor
mutex Mtx; // 锁上面这两个
// 定时唤醒有 定时任务的actor
thread TimerTherad;
atomic<bool> TimerTheradIsRun{ false };
void StartTimer();
void StopTimer();
void ProcessTimers();
static C_Actor* CreateActor(string ActorId, INT32 ActorLevel);
static bool HasActor(string ActorId);
static void* SendMessage_Sync(C_Actor* pActor, string DstActorId, INT32 MsgId, void* pMs);
static bool SendMessage_ASync(string ActorId, INT32 MsgId, void* pMsg);
static void RemoveActor(string ActorId);
};
// 协程句柄, 给C_Actor控制协程挂起恢复
struct ActorHandle {
struct promise_type;
using Handle = std::coroutine_handle<promise_type>;
Handle pHhandle;
ActorHandle() = delete;
ActorHandle(Handle h) : pHhandle(h) {}
~ActorHandle() { if (pHhandle) pHhandle.destroy(); }
ActorHandle(ActorHandle&& other) noexcept;
ActorHandle& operator=(ActorHandle&& other) noexcept;
struct promise_type {
ActorHandle get_return_object() { return ActorHandle(Handle::from_promise(*this)); }
std::suspend_always initial_suspend() noexcept { return {}; }
std::suspend_always final_suspend() noexcept { return {}; }
void return_void() {}
void unhandled_exception() { std::terminate(); }
};
};
// Actor类
class C_Actor
{
friend class C_ActorMgr;
public:
using ActorMsgHandle_Async = std::function<void(void*)>;
using ActorMsgHandle_Sync = std::function<void*(void*)>;
using TimerHandler = std::function<void()>;
struct St_ActorTimer { // 定时函数用的结构对象
TimerHandler Func;
INT64 NextTime = 0; // 下一次的调用时间
INT32 Inteval = 10000; // 调用间隔, 毫秒
};
struct SyncRequest { // 同步调用Actor的封装
std::promise<void*> Promise;
INT32 MsgId;
void* pData;
};
enum CoroState {
Suspended,
Running,
Dead
};
string Id;
INT32 Level; // 级别, 越小越高
std::queue<CoroutineMsg> MsgQueue1; // 同步消息列表
std::queue<CoroutineMsg> MsgQueue2; // 同步消息列表
std::queue<SyncRequest> MsgQueue_Sync; // 异步消息队列
mutex Mtx_AsyncMsg; // 同步消息队列锁, 锁队列1, 消息进入1和从1出来进入2时
mutex Mtx_SyncMsg; // 异步消息队列锁
mutex Mtx_Except; // 协程本身的锁, 用于异常恢复时锁定
std::atomic<bool> IsRunning = true; // 协程停止标记
std::atomic<INT32> ErrorCount = 0; // 崩溃次数
static constexpr INT32 ErrorCountMax = -1; // 协程崩溃多少次删除actor
std::optional<ActorHandle> pActorHandle; // 协程句柄
std::atomic<CoroState> Status = Suspended; // 协程状态, 防止重复唤醒
vector<St_ActorTimer> TimerDeal; // 定时任务列表
unordered_map<INT32, ActorMsgHandle_Sync> MsgDeal_Sync; // 同步调用的消息注册列表
unordered_map<INT32, ActorMsgHandle_Async> MsgDeal_Async; // 异步同步调用的消息注册列表
ActorHandle Run(); // 协程入口函数
void ProcessMessages(); // 处理消息
void Stop(); // 停止Actor
void ResumeIfSuspended(); // 唤醒协程
void Recv(INT32 MsgId, void* pMsg); // 由C_ActorMgr调用, 接收消息
void RegisterHandle_Timer(TimerHandler handler, INT32 IntervalMillSec); // 注册定时任务
void RegisterHandle_Sync(INT32 msgType, ActorMsgHandle_Sync handler); // 注册同步消息
void RegisterHandle_Async(INT32 msgType, ActorMsgHandle_Async handler); // 注册异步消息
};
#endif
#include "Actor.h"
using namespace std::chrono;
void C_ActorMgr::StartTimer()
{
TimerTheradIsRun = true;
TimerTherad = thread([this]
{
while (TimerTheradIsRun.load(std::memory_order_acquire))
{
ProcessTimers();
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
});
}
void C_ActorMgr::StopTimer()
{
TimerTheradIsRun = false;
if (TimerTherad.joinable())
TimerTherad.join();
}
void C_ActorMgr::ProcessTimers()
{
auto MillNow = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
vector<string> TimerTmp;
{
lock_guard<mutex> lock(Mtx);
TimerTmp.assign(ActorTimer.begin(), ActorTimer.end());
}
for (const auto& ActorId : TimerTmp) {
std::shared_ptr<C_Actor> pActor;
{
lock_guard<mutex> lock(Mtx);
auto it = AllActor.find(ActorId);
if (it == AllActor.end())
continue;
pActor = it->second;
}
bool NeedFlag = false;
for (auto& timer : pActor->TimerDeal) {
if (timer.NextTime <= MillNow) {
NeedFlag = true;
break;
}
}
if (NeedFlag) {
pActor->Recv(-1, nullptr);
}
}
}
C_Actor* C_ActorMgr::CreateActor(string ActorId, INT32 ActorLevel)
{
auto pNewActor = std::make_shared<C_Actor>();
pNewActor->Id = ActorId;
pNewActor->Level = ActorLevel;
pNewActor->pActorHandle.emplace(pNewActor->Run());
pNewActor->pActorHandle->pHhandle.resume();
auto ptr = pNewActor.get();
{
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
C_ActorMgr::Instance().AllActor.emplace(ActorId, std::move(pNewActor));
}
return ptr;
}
bool C_ActorMgr::HasActor(string ActorId)
{
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
auto It = C_ActorMgr::Instance().AllActor.find(ActorId);
return It != C_ActorMgr::Instance().AllActor.end();
}
void* C_ActorMgr::SendMessage_Sync(C_Actor* pActor, string DstActorId, INT32 MsgId, void* pMsg)
{
std::shared_ptr<C_Actor> pDst;
{
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
auto it = C_ActorMgr::Instance().AllActor.find(DstActorId);
if (it == C_ActorMgr::Instance().AllActor.end())
return nullptr;
if (pActor->Level > it->second->Level)
return nullptr;
pDst = it->second;
}
auto promise = std::make_shared<std::promise<void*>>();
auto future = promise->get_future();
C_Actor::SyncRequest Req{ std::move(*promise), MsgId, pMsg };
{
lock_guard<mutex> Lock(pDst->Mtx_SyncMsg);
pDst->MsgQueue_Sync.push(std::move(Req));
}
pDst->Recv(-2, nullptr);
auto b = future.get();
return b;
}
bool C_ActorMgr::SendMessage_ASync(string ActorId, INT32 MsgId, void* pMsg)
{
std::shared_ptr<C_Actor> pDst;
{
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
auto it = C_ActorMgr::Instance().AllActor.find(ActorId);
if (it == C_ActorMgr::Instance().AllActor.end())
return false;
pDst = it->second;
}
pDst->Recv(MsgId, pMsg);
}
void C_ActorMgr::RemoveActor(string ActorId)
{
std::shared_ptr<C_Actor> actor;
{
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
auto it = C_ActorMgr::Instance().AllActor.find(ActorId);
if (it == C_ActorMgr::Instance().AllActor.end())
return;
actor = it->second; // 保持引用,防止析构时锁问题
actor->Stop();
C_ActorMgr::Instance().AllActor.erase(it);
C_ActorMgr::Instance().ActorTimer.erase(ActorId);
}
}
ActorHandle::ActorHandle(ActorHandle&& other) noexcept
: pHhandle(other.pHhandle)
{
other.pHhandle = nullptr;
}
ActorHandle& ActorHandle::operator=(ActorHandle&& other) noexcept
{
if (this != &other) {
if (pHhandle) pHhandle.destroy();
pHhandle = other.pHhandle;
other.pHhandle = nullptr;
}
return *this;
}
ActorHandle C_Actor::Run() {
Status = Running;
try {
while (IsRunning) {
ProcessMessages();
Status = Suspended;
co_await std::suspend_always{};
Status = Running;
}
Status = Dead;
}
catch (...) {
Status = Dead;
lock_guard<mutex> lock(Mtx_Except);
std::exception_ptr eptr = std::current_exception();
ErrorCount++;
pActorHandle.reset();
}
co_return;
}
void C_Actor::ProcessMessages() {
// 先处理同步调用消息
{
lock_guard<mutex> lock(Mtx_SyncMsg);
while (!MsgQueue_Sync.empty()) {
auto& req = MsgQueue_Sync.front();
if (auto it = MsgDeal_Sync.find(req.MsgId); it != MsgDeal_Sync.end()) {
try {
auto result = it->second(req.pData);
req.Promise.set_value(result);
}
catch (...) {
req.Promise.set_exception(std::current_exception());
}
}
else {
req.Promise.set_value(nullptr);
}
MsgQueue_Sync.pop();
}
}
// 同步消息双缓冲队列去除
{
lock_guard<mutex> lock(Mtx_SyncMsg);
INT32 Protect = 0;
while (!MsgQueue1.empty())
{
if (++Protect > 300)
break;
MsgQueue2.push(MsgQueue1.front());
MsgQueue1.pop();
}
}
// 处理同步消息
bool HasTimerFlag = false;
INT32 Protect2 = 0;
while (!MsgQueue2.empty()) {
if (++Protect2 > 2000)
break;
auto pMsg = MsgQueue2.front();
MsgQueue2.pop();
if (pMsg.MsgId == -1) {
HasTimerFlag = true;
}
auto it = MsgDeal_Async.find(pMsg.MsgId);
if (it == MsgDeal_Async.end())
continue;
try {
it->second(pMsg.pData); // 捕获单个消息处理的异常
}
catch (...) {
// 报错日志
}
}
// 如果有定时任务唤醒协程, 处理定时任务
if (HasTimerFlag)
{
auto MillNow = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
for (auto& TimerIt : TimerDeal) {
if (TimerIt.NextTime > MillNow)
continue;
INT32 Devi = MillNow - TimerIt.NextTime; // 计算误差修正, 防止调用间隔导致误差越来越多
TimerIt.NextTime = MillNow + TimerIt.Inteval - Devi;
TimerIt.Func();
}
}
}
void C_Actor::Stop()
{
IsRunning = false;
ResumeIfSuspended(); // 唤醒退出
}
void C_Actor::ResumeIfSuspended()
{
CoroState expected = Suspended;
if (Status.compare_exchange_strong(expected, Running)) {
if (pActorHandle.has_value() &&
!pActorHandle->pHhandle.done())
{
pActorHandle->pHhandle.resume();
}
}
if (Status == Dead) {
lock_guard<mutex> lock(Mtx_Except);
if (ErrorCountMax > 0 && ErrorCount > ErrorCountMax) {
C_ActorMgr::RemoveActor(Id);
}
else {
pActorHandle.emplace(Run());
pActorHandle->pHhandle.resume();
ErrorCount++;
}
}
}
void C_Actor::Recv(INT32 MsgId, void* pMsg)
{
{
lock_guard<mutex> lock(Mtx_SyncMsg);
MsgQueue1.push({MsgId, pMsg});
}
ResumeIfSuspended(); // 尝试恢复协程处理新消息
}
void C_Actor::RegisterHandle_Timer(C_Actor::TimerHandler handler, INT32 IntervalMillSec)
{
auto MillNow = duration_cast<milliseconds>(system_clock::now().time_since_epoch()) .count();
C_Actor::St_ActorTimer Tmp;
TimerDeal.push_back({ handler, MillNow + IntervalMillSec, IntervalMillSec });
lock_guard<mutex> lock(C_ActorMgr::Instance().Mtx);
C_ActorMgr::Instance().ActorTimer.insert(this->Id);
}
void C_Actor::RegisterHandle_Sync(INT32 msgType, ActorMsgHandle_Sync handler)
{
MsgDeal_Sync[msgType] = handler;
}
void C_Actor::RegisterHandle_Async(INT32 msgType, ActorMsgHandle_Async handler)
{
MsgDeal_Async[msgType] = handler;
}
测试代码
INT32 IntervalMillSec = 50;
auto pA = C_ActorMgr::CreateActor("actor", 1);
pA->RegisterHandle_Async(1, [](void* a) {
printf("Async %d\n", *((INT32*)a)); }
);
pA->RegisterHandle_Sync(2, [](void* a) {
return new INT32(114515);
}
);
pA->RegisterHandle_Timer([]() {
printf("Timer %llu\n", duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count()); }
, IntervalMillSec);
auto TmpThread1 = thread([pA, IntervalMillSec]
{
INT32 A = 0;
while (A++ < 50)
{
printf("Sync %d\n", *((INT32*)C_ActorMgr::SendMessage_Sync(pA, "actor", 2, new INT32(114514))));
std::this_thread::sleep_for(std::chrono::milliseconds(IntervalMillSec));
}
});
auto TmpThread2 = thread([IntervalMillSec]
{
INT32 A = 0;
while (A++ < 50)
{
C_ActorMgr::SendMessage_ASync("actor", 1, new INT32(114514));
std::this_thread::sleep_for(std::chrono::milliseconds(IntervalMillSec));
}
});
TmpThread1.join();
TmpThread2.join();
运行

魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。
更多推荐


所有评论(0)