首先简单介绍下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 -;
pth = calloc(,sizeof(*pth));
pth->base.type = CHANNEL;
int tmp[];
if(pipe(tmp) != ){
free(pth);
return -;
}
pth->base.fd = tmp[];
pth->notifyfd = tmp[];
pth->base.on_active = kn_channel_on_active;
pth->base.process = kn_channel_process;
pth->channel = c;
fcntl(tmp[], F_SETFL, O_NONBLOCK | O_RDWR);
fcntl(tmp[], F_SETFL, O_NONBLOCK | O_RDWR);
kn_ref_init(&pth->base.ref,channel_pth_destroy);
if(!= p->Register(p,(kn_fd_t)pth)){
kn_ref_release((kn_ref*)pth);
return -;
}
kn_ref_acquire(&c->ref);
pthread_setspecific(c->t_key,(void*)pth);
kn_procator_putin_active(p,(kn_fd_t)pth);
return ;
}
用于将一个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 = ;
while((msg = kn_channel_getmsg(c)) != NULL && n > ){
c->channel->cb_msg(c->channel,msg->sender,msg->data);
free(msg->data);
free(msg);
--n;
}
if(n <= )
return ;
else
return ;
}
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 = ;
struct msg *msg = calloc(,sizeof(*msg));
msg->sender = from;
msg->data = data;
kn_mutex_lock(to->mtx);
kn_list_pushback(&to->queue,&msg->node);
while(){
tmp = kn_dlist_first(&to->waits);
if(tmp){
//有线程在等待消息，通知它有消息到了
struct channel_pth *pth = (struct channel_pth*)(((char*)tmp)-sizeof(kn_fd));
ret = write(pth->notifyfd,"",);
kn_dlist_pop(&to->waits);
if(!(ret == || (ret < && 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有消息到来.