skynet源码分析之skynet_server

2018/01/19 11:31
阅读数 1.7K

skynet是以服务为主体进行运作的,服务称作为skynet_context(简称ctx),是一个c结构,是skynet里最重要的结构,整个skynet的运作都是围绕ctx进行的。skynet_server提供的api主要分两大类:

1.对ctx的一系列操作,比如创建,删除ctx等

2.如何发送消息和处理自身的消息

1.对ctx的一系列操作

ctx的结构如下,创建一个服务时,会构建一个skynet上下文skynet_context,然后把该ctx存放到handle_storage(skynet_handle.c)里。ctx有一个引用计数(ctx->ref)控制其生命周期,当引用计数为0,删除ctx,释放内存。

struct skynet_context { //一个skynet服务ctx的结构
        void * instance; //每个ctx自己的数据块,不同类型ctx有不同数据结构,相同类型ctx数据结构相同,但具体的数据不一样,由指定module的create函数返回
        struct skynet_module * mod; //保存module的指针,方便之后调用create,init,signal,release
        void * cb_ud; //给callback函数调用第二个参数,可以是NULL
        skynet_cb cb; //消息回调函数指针,通常在module的init里设置
        struct message_queue *queue; //ctx自己的消息队列指针
        FILE * logfile; //日志句柄
        uint64_t cpu_cost;      // in microsec
        uint64_t cpu_start;     // in microsec
        char result[32]; //保存skynet_command操作后的结果
        uint32_t handle; //标识唯一的ctx id
        int session_id; //本方发出请求会设置一个对应的session,当收到对方消息返回时,通过session匹配是哪一个请求的返回
        int ref; //引用计数,当为0,可以删除ctx
        int message_count; //累计收到的消息数量
        bool init; //标记是否完成初始化
        bool endless; //标记消息是否堵住
        bool profile; //标记是否需要开启性能监测

        CHECKCALLING_DECL // 自旋锁
};

为了统一对ctx操作的接口,采用指令的格式,定义了一系列指令(cmd_xxx),cmd_launch创建一个新服务,cmd_exit服务自身退出,cmd_kill杀掉一个服务等,上层统一调用skynet_command接口即可执行这些操作。对ctx操作,通常会先调用skynet_context_grab将引用计数+1,操作完调用skynet_context_release将引用计数-1,以保证操作ctx过程中,不会被其他线程释放掉。下面介绍几个常见的操作:

struct command_func { //skynet指令结构
        const char *name;
        const char * (*func)(struct skynet_context * context, const char * param);
};

static struct command_func cmd_funcs[] = { //skynet可接收的一系列指令
        { "TIMEOUT", cmd_timeout },
        { "REG", cmd_reg },
        { "QUERY", cmd_query },
        { "NAME", cmd_name },
        { "EXIT", cmd_exit },
        { "KILL", cmd_kill },
        { "LAUNCH", cmd_launch },
        { "GETENV", cmd_getenv },
        { "SETENV", cmd_setenv },
        { "STARTTIME", cmd_starttime },
        { "ABORT", cmd_abort },
        { "MONITOR", cmd_monitor },
        { "STAT", cmd_stat },
        { "LOGON", cmd_logon },
        { "LOGOFF", cmd_logoff },
        { "SIGNAL", cmd_signal },
        { NULL, NULL },
};

(1). cmd_launch,启动一个新服务,最终会通过skynet_context_new创建一个ctx,初始化ctx中各个数据。

struct skynet_context * 
skynet_context_new(const char * name, const char *param) { //启动一个新服务ctx
        struct skynet_module * mod = skynet_module_query(name); //从skynet_module获取对应的模板
        ...
        void *inst = skynet_module_instance_create(mod); //ctx独有的数据块,最终会调用c服务里的xxx_create
        ...
        ctx->mod = mod;
        ctx->instance = inst;
        ctx->ref = 2; //初始化完成会调用skynet_context_release将引用计数-1,ref变成1而不会被释放掉
        ...
        // Should set to 0 first to avoid skynet_handle_retireall get an uninitialized handle
        ctx->handle = 0;        
        ctx->handle = skynet_handle_register(ctx); //从skynet_handle获得唯一的标识id
        struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle); //初始化次级消息队列
        ...
        CHECKCALLING_BEGIN(ctx)
        int r = skynet_module_instance_init(mod, inst, ctx, param);//初始化ctx独有的数据块
        CHECKCALLING_END(ctx)
}        

通过指定名称查找对应的module,并获取不同c服务独有的数据块(struct snlua, struct logger,  struct gate等),赋值给ctx->mod,ctx->instance。ctx->ref=2是为了保证创建ctx期间不会被其他线程释放掉,创建完成后会调用skynet_context_release将引用计数-1。将ctx保存在handle_storage,并获得一个唯一的标识id(ctx->handle = skynet_handle_register(ctx)),每个ctx都有一个唯一的id对应起来。创建ctx的次级消息队列(skynet_mq_create)。最后再调用skynet_module_instance_init初始化独有的数据块,放到skynet_mq_create之后是因为init过程中可能会发送消息。至此就是一个ctx的整个创建流程。

(2). cmd_exit,服务主动退出;cmd_kill杀掉某个服务(被动),都会调用到handle_exit,然后调用到skynet_handle_retire,回收ctx->handle,供之后创建新的ctx使用,并将引用计数-1,如果ctx没有在其他地方引用,ctx->ref此时是0,所以可以删掉,delete_context主要是做一些清理回收工作。如果其他地方有引用,在下一次skynet_context_release时删掉。注:这时还不能释放ctx->queue,只能做个标记(skyne_module_instance_release),原因参考 http://www.cnblogs.com/RainRill/p/8252480.html

struct skynet_context * 
skynet_context_release(struct skynet_context *ctx) { //ctx引用计数-1
        if (ATOM_DEC(&ctx->ref) == 0) {
                delete_context(ctx);
                return NULL;
        }
        return ctx;
}

static void 
delete_context(struct skynet_context *ctx) {//删除ctx,释放内存
        if (ctx->logfile) {
                fclose(ctx->logfile);
        }
        skynet_module_instance_release(ctx->mod, ctx->instance);
        skynet_mq_mark_release(ctx->queue);
        CHECKCALLING_DESTROY(ctx)
        skynet_free(ctx);
        context_dec();
}

(3). cmd_reg,给自身起一个名字(支持多个),cmd_name给指定ctx起一个名字,即将ctx->handle绑定一个名称(skynet_handle_namehandle)

(4). cmd_query,通过名字查找对应的handle,发送消息前先要找到对应的ctx,才能给ctx发送消息

(5). cmd_stat,查看ctx的内部状态信息,比如查看当前的消息队列长度,查看累计消耗CPU时间,查看消息是否阻塞等

(6). cmd_setenv,cmd_getenv,设置/获取skynet环境变量,是key-value结构,所有ctx共享的

(7). cmd_signal,在skynet控制台,可以给指定的ctx发信号以完成相应的命令

2. 如理发送消息和处理消息

ctx之间通过消息进行通信,调用skynet_send向对方发送消息(skynet_sendname最终也会调用skynet_send)。

int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * data, size_t sz) {
        if ((sz & MESSAGE_TYPE_MASK) != sz) {
                skynet_error(context, "The message to %x is too large", destination);
                if (type & PTYPE_TAG_DONTCOPY) {
                        skynet_free(data);
                }
                return -1;
        }
        _filter_args(context, type, &session, (void **)&data, &sz); //预处理消息数据块

        if (source == 0) {
                source = context->handle;
        }

        if (destination == 0) {
                return session;
        }
        if (skynet_harbor_message_isremote(destination)) { //不是同一skynet节点里服务,交给harbor
                struct remote_message * rmsg = skynet_malloc(sizeof(*rmsg));
                rmsg->destination.handle = destination;
                rmsg->message = data;
                rmsg->sz = sz;
                skynet_harbor_send(rmsg, source, session);
        } else {
                struct skynet_message smsg;
                smsg.source = source;
                smsg.session = session;
                smsg.data = data;
                smsg.sz = sz;

                if (skynet_context_push(destination, &smsg)) { //push给目的ctx
                        skynet_free(data);
                        return -1;
                }
        }
        return session;
}

接口参数:

struct skynet_context *context: 源服务的ctx,可以为NULL,drop_message时这个参数为NULL

uint32_t  source:源服务的地址,通常设置为0即可,api里会设置成ctx->handle,当context为NULL时,需指定source

uint32_t destination:目的服务地址

int type:消息类型, skynet定义了多种消息,PTYPE_TEXT,PTYPE_CLIENT,PTYPE_RESPONSE等(详情见skynet.h)

int session:如果在type里设上allocsession的tag(PTYPE_TAG_ALLOCSESSION),api会忽略掉传入的session参数,重新生成一个新的唯一的

void *data, size_t sz:消息包和长度

返回: int session:源服务保存这个session,同时约定,目的服务处理完这个消息后,把这个session原样发送回来(skynet_message结构里带有一个session字段),源服务就知道是哪个请求的返回,从而正确调用对应的回调函数。

首先检查sz是否合法,然后在filter_args里进行预处理,如果type设上PTYPE_TAG_ALLOCSESSION,则需要生成一个唯一的session(skynet_context_newsession)。如果type没有设置PTYPE_TAG_DONTCOPY,则需要拷贝一份消息包。把type打包在sz的高8位*sz |= (size_t)type<<MESSAGE_TYPE_SHIFT,因为 消息包的长度限制在24位(16M),是个合理的限制。判断目的地址和源地址是否在一个skynet节点内,若在,则构建一个skynet_message结构,push给目的服务(skynet_context_push);若不在,则交给harbor,让harbor去转发。

 

skynet可以启动多个工作线程,调用skynet_context_message_dispatch这个api不断的分发消息,有三个参数:skynet_monitor监测消息是否堵住,message_queue(mq)需要分发的次级消息队列,weight权重(之后会详细说明),api返回下一个要分发的mq。

struct message_queue * 
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {//工作线程分发消息,处理完返回全局队列的下一条次级消息队列,供下一帧调用
        if (q == NULL) {
                q = skynet_globalmq_pop(); //从全局消息队列里pop一条次级消息队列
                if (q==NULL)
                        return NULL;
        }

        uint32_t handle = skynet_mq_handle(q);

        struct skynet_context * ctx = skynet_handle_grab(handle); //获取次级消息队列的ctx
        if (ctx == NULL) { //ctx不存在了
                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;i<n;i++) {
                if (skynet_mq_pop(q,&msg)) { //如果ctx的次级消息队列为空,返回
                        skynet_context_release(ctx);
                        return skynet_globalmq_pop();
                } else if (i==0 && weight >= 0) { //weight:-1只处理一条消息,0处理完q中所有消息,>0处理长度右移weight位(1/(2*weight))条消息
                        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;
}                

如果传入的mq为空,则从全局消息队列里pop一个(skynet_globalmq_pop)。查找mq对应的ctx,若ctx不存在,需要删除该mq(skynet_mq_release)。接下来就是分发mq中的消息包,一帧分发多少条消息取决于weight参数(稍后讨论)。调用dispatch_message分发消息包,最终调用ctx->cb消息回调函数,在消息回调函数处理消息包,返回0表示回调成功,至此完成一个消息包的处理流程。最后把mq push回全局消息队列,供下次工作线程调度(前提是mq不为空,且global mq也不为空)。

static void
dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { //处理一条ctx消息,即调用ctx的消息回调函数
       ...
       reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);//消息回调函数
        ...
}

关于weight参数,在启动工作线程时,会为每个线程设置一个weight,即线程0-3是-1,4-7是0,8-15是1,16-23是2,24-31是3。-1表示每帧只处理mq中一个消息包,0表示每帧处理mq中所有消息包,1表示每帧处理mq长度的1/2条(>>1右移一位)消息,2表示每帧处理mq长度的1/4(右移2位),3表示每帧处理mq长度的1/8(右移3位)。

//skynet_start.c
static
int weight[] = { -1, -1, -1, -1, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, };
展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部