189 8069 5689
actor的调度由线程池来调度。actor是被调度对象，skynet把所有活跃的actor通过链表串联起来，线程池从actor中取出相等数量的消息进行执行，实现公平调度。
但是，actor消息队列长度可能不一致，会出现部分actor "饿死"现象，skynet通过对线程池的工作线程赋予不同权重来规避这个问题。
二、调度流程源码分析actor=隔离的运行环境+回调函数+消息队列。
2.1、thread_worker()actor是由线程调度，所以从线程入口函数thread_worker开始。线程作为消费者，会不断循环从消息队列中取消息，如果没有消息就进入等待（pthread_cond_wait）。skynet_context_message_dispatch()用来取出消息和消费消息。
/skynet-src/skynet_start.c
static void *
thread_worker(void *p) {struct worker_parm *wp = p;
int id = wp->id;
int weight = wp->weight;
struct monitor *m = wp->m;
struct skynet_monitor *sm = m->m[id];
skynet_initthread(THREAD_WORKER);
struct message_queue * q = NULL;
while (!m->quit) {q = skynet_context_message_dispatch(sm, q, weight);
if (q == NULL) {if (pthread_mutex_lock(&m->mutex) == 0) {++ m->sleep;
// "spurious wakeup" is harmless,
// because skynet_context_message_dispatch() can be call at any time.
if (!m->quit)
pthread_cond_wait(&m->cond, &m->mutex);
-- m->sleep;
if (pthread_mutex_unlock(&m->mutex)) {fprintf(stderr, "unlock mutex error");
exit(1);
}
}
}
}
return NULL;
}
struct skynet_context保存的是actor的上下文信息。
/skynet-src/skynet_server.c
struct skynet_context {void * instance;
struct skynet_module * mod;
void * cb_ud;
skynet_cb cb;
struct message_queue *queue;
ATOM_POINTER logfile;
uint64_t cpu_cost; // in microsec
uint64_t cpu_start; // in microsec
char result[32];
uint32_t handle;
int session_id;
ATOM_INT ref;
int message_count;
bool init;
bool endless;
bool profile;
CHECKCALLING_DECL
};
/skynet-src/skynet_mq.c
struct message_queue {struct spinlock lock;
uint32_t handle;
int cap;
int head;
int tail;
int release;
int in_global;
int overload;
int overload_threshold;
struct skynet_message *queue;
struct message_queue *next;
};
/skynet-src/skynet_server.c
struct message_queue *
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {if (q == NULL) {q = skynet_globalmq_pop();
if (q==NULL)
return NULL;
}
uint32_t handle = skynet_mq_handle(q);
struct skynet_context * ctx = skynet_handle_grab(handle);
if (ctx == NULL) {struct drop_t d = {handle };
skynet_mq_release(q, drop_message, &d);
return skynet_globalmq_pop();
}
int i,n=1;
struct skynet_message msg;
for (i=0;iif (skynet_mq_pop(q,&msg)) {skynet_context_release(ctx);
return skynet_globalmq_pop();
} else if (i==0 && weight >= 0) {n = skynet_mq_length(q);
n >>= weight;
}
int overload = skynet_mq_overload(q);
if (overload) {skynet_error(ctx, "May overload, message queue length = %d", overload);
}
skynet_monitor_trigger(sm, msg.source , handle);
if (ctx->cb == NULL) {skynet_free(msg.data);
} else {dispatch_message(ctx, &msg);
}
skynet_monitor_trigger(sm, 0,0);
}
assert(q == ctx->queue);
struct message_queue *nq = skynet_globalmq_pop();
if (nq) {// If global mq is not empty , push q back, and return next queue (nq)
// Else (global mq is empty or block, don't push q back, and return q again (for next dispatch)
skynet_globalmq_push(q);
q = nq;
}
skynet_context_release(ctx);
return q;
}
dispatch_message()本质上调用回调函数来处理消息，消息内容作为参数；这里就是真正的运行actor了。
/skynet-src/skynet_server.c
static void
dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {assert(ctx->init);
CHECKCALLING_BEGIN(ctx)
pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle));
int type = msg->sz >>MESSAGE_TYPE_SHIFT;
size_t sz = msg->sz & MESSAGE_TYPE_MASK;
FILE *f = (FILE *)ATOM_LOAD(&ctx->logfile);
if (f) {skynet_log_output(f, msg->source, type, msg->session, msg->data, sz);
}
++ctx->message_count;
int reserve_msg;
if (ctx->profile) {ctx->cpu_start = skynet_thread_time();
reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
uint64_t cost_time = skynet_thread_time() - ctx->cpu_start;
ctx->cpu_cost += cost_time;
} else {reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
}
if (!reserve_msg) {skynet_free(msg->data);
}
CHECKCALLING_END(ctx)
}
了解完调度流程，那么如果在c语言的callback函数跳到lua层执行actor呢？
lua可以调用c语言，c语言需要导入一个方法给lua使用，skynet中lcallback()就是c语言导出给lua使用的方法。
在lcallback设置回调函数：skynet_callback()。
/lualib-src/lua-skynet.c
static int
lcallback(lua_State *L) {struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
int forward = lua_toboolean(L, 2);
luaL_checktype(L,1,LUA_TFUNCTION);
lua_settop(L,1);
struct callback_context *cb_ctx = (struct callback_context *)lua_newuserdata(L, sizeof(*cb_ctx));
cb_ctx->L = lua_newthread(L);
lua_pushcfunction(cb_ctx->L, traceback);
lua_setuservalue(L, -2);
lua_setfield(L, LUA_REGISTRYINDEX, "callback_context");
lua_xmove(L, cb_ctx->L, 1);
if (forward) {skynet_callback(context, cb_ctx, forward_cb);
} else {skynet_callback(context, cb_ctx, _cb);
}
return 0;
}
// ...
LUAMOD_API int
luaopen_skynet_core(lua_State *L) {luaL_checkversion(L);
luaL_Reg l[] = {{"send" , lsend },
{"genid", lgenid },
{"redirect", lredirect },
{"command" , lcommand },
{"intcommand", lintcommand },
{"addresscommand", laddresscommand },
{"error", lerror },
{"harbor", lharbor },
{"callback", lcallback },
{"trace", ltrace },
{NULL, NULL },
};
// functions without skynet_context
luaL_Reg l2[] = {{"tostring", ltostring },
{"pack", luaseri_pack },
{"unpack", luaseri_unpack },
{"packstring", lpackstring },
{"trash" , ltrash },
{"now", lnow },
{"hpc", lhpc }, // getHPCounter
{NULL, NULL },
};
lua_createtable(L, 0, sizeof(l)/sizeof(l[0]) + sizeof(l2)/sizeof(l2[0]) -2);
lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");
struct skynet_context *ctx = lua_touserdata(L,-1);
if (ctx == NULL) {return luaL_error(L, "Init skynet context first");
}
luaL_setfuncs(L,l,1);
luaL_setfuncs(L,l2,0);
return 1;
}
当执行lua的skynet.start时会调用c.callback()设置回调函数skynet.dispatch_message，skynet.dispatch_message是一个lua方法；每次消息到来就会调用lua的skynet.dispatch_message，通过它分发消息。
/lualib/skynet.lua
function skynet.start(start_func)
c.callback(skynet.dispatch_message)
init_thread = skynet.timeout(0, function()
skynet.init_service(start_func)
init_thread = nil
end)
end
本专栏知识点是通过<零声教育>的系统学习，进行梳理总结写下文章，对c/c++linux系统提升感兴趣的读者，可以点击链接查看详细的服务：C/C++服务器开发 。
你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧