distri.lua线程间通信的设计

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› LUA›LUA教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

首先简单介绍下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);       
}