首先简单介绍下distri.lua中的线程设计方案.
distri.lua提供一个API函数fork用于创建新的C线程,这个C线程运行独立的lua虚拟机,为了在各线程之间通信
每个线程都会创建一个channel,用于接收其它线程发送过来的消息.
这个channel内部实现为单向链表,为了将channel的处理与网络消息处理接口合并,channel使用tls为每个单
独的 线程创建一个管道,这个管道被添加到proactor中监听,如果一个线程尝试从channel中读消息,而消息队
列为空,就 将这个管道添加到channel的等待列表中.其它线程向channel添加消息后会检查等待列表,如果等
待列表中有元素 则弹出一个,并向弹出的管道中写入一个字节以通知接收者有消息可读.向管道中写入数据之
后,接收线程的proactor 发现一个管道被激活,事件处理循环触发从那个管道对应的channel中继续读取消息.
下面分析下关键代码：
struct channel_pth{ kn_fd base; kn_dlist_node node; int notifyfd; kn_list local_que; kn_channel_t channel; };
每个期望从一个channel对象中接收消息的线程都会产生一个对应的struct
channel_pth,struct
channel_pth
继承自kn_fd,也就是说struct
channel_pth可以被添加到cproctor中监听.notifyfd则用于其它线程发送通知消息,
以通知管道中有新消息到来.
int kn_channel_bind(struct kn_proactor *p,kn_channel_t c){ struct channel_pth *pth = (struct channel_pth*)pthread_getspecific(c->t_key); if(pth) return -1; pth = calloc(1,sizeof(*pth)); pth->base.type = CHANNEL; int tmp[2]; if(pipe(tmp) != 0){ free(pth); return -1; } pth->base.fd = tmp[0]; pth->notifyfd = tmp[1]; pth->base.on_active = kn_channel_on_active; pth->base.process = kn_channel_process; pth->channel = c; fcntl(tmp[0], F_SETFL, O_NONBLOCK | O_RDWR); fcntl(tmp[1], F_SETFL, O_NONBLOCK | O_RDWR); kn_ref_init(&pth->base.ref,channel_pth_destroy); if(0!= p->Register(p,(kn_fd_t)pth)){ kn_ref_release((kn_ref*)pth); return -1; } kn_ref_acquire(&c->ref); pthread_setspecific(c->t_key,(void*)pth); kn_procator_putin_active(p,(kn_fd_t)pth); return 0; }
用于将一个channel绑定到proactor，可以看到，在绑定时首先为当前线程产生一个struct
channel_pth 对象,
并正始化相关的管道.然后将管道的读端添加到proactor中.要注意的是管道的监听模式也是edge trigger. 绑定完成
之后将这个struct
channel_pth添加到激活队列中,这样在proactor的主循环中将不断的从对应的 channel中提
取消息,之后消息为空,才从活队列移除.
static int8_t kn_channel_process(kn_fd_t s){ struct channel_pth* c = (struct channel_pth*)s; struct msg *msg; int n = 1024; while((msg = kn_channel_getmsg(c)) != NULL && n > 0){ c->channel->cb_msg(c->channel,msg->sender,msg->data); free(msg->data); free(msg); --n; } if(n <= 0) return 1; else return 0; }
proactor主循环中对每个 struct
channel_pth的处理,不断的尝试从channel中提取消息, 调用回调函数处理消息.
这里要注意其中的一个条件值1024.这个值的设定是为了防止一个channel 中消息过多,cpu时间全都被用于处理这个
channel,网络事件和其它的channel都每机会执行.所以, 对每个channel在一次处理中最多只提取1024个消息,如果还
有剩余到下一个循环再继续处理.
static inline struct msg* kn_channel_getmsg(struct channel_pth *c){ struct msg *msg = (struct msg*)kn_list_pop(&c->local_que); if(msg) return msg; kn_mutex_lock(c->channel->mtx); if(!kn_list_size(&c->channel->queue)){ kn_dlist_push(&c->channel->waits,&c->node); kn_mutex_unlock(c->channel->mtx); return NULL; }else{ kn_list_swap(&c->local_que,&c->channel->queue); } kn_mutex_unlock(c->channel->mtx); return (struct msg*)kn_list_pop(&c->local_que); }
从channel中提取消息,如果channel为空将当前channel对应的struct
channel_pth添加 到等待队列中.
void kn_channel_putmsg(kn_channel_t to,kn_channel_t from,void *data) { kn_dlist_node *tmp = NULL; int ret = 0; struct msg *msg = calloc(1,sizeof(*msg)); msg->sender = from; msg->data = data; kn_mutex_lock(to->mtx); kn_list_pushback(&to->queue,&msg->node); while(1){ tmp = kn_dlist_first(&to->waits); if(tmp){ //有线程在等待消息，通知它有消息到了 struct channel_pth *pth = (struct channel_pth*)(((char*)tmp)-sizeof(kn_fd)); ret = write(pth->notifyfd,"",1); kn_dlist_pop(&to->waits); if(!(ret == 0 || (ret < 0 && errno != EAGAIN))) break; /*if(ret == 0 || (ret < 0 && errno != EAGAIN)){ //对端关闭 kn_dlist_pop(&to->waits); }else break;*/ }else break; }; kn_mutex_unlock(to->mtx); }
向channel投递消息,首先将消息写入到channel中,然后查看等待列表看看是否有线正在等待消息,如果有
则从等待列表中弹出一个等待者,并向那个等待者对应的管道写入一个字节以通知channel有消息到来.