学步园

前面一篇文章实现了一个抢先式的用户级多线程框架，现在用那个框架编写一个echo服务，

因为只是个实验，所以代码写得比较杂乱，还有很多可能出错的情况也没有处理，这些在今后的进一

步研究中都会慢慢修改，下面是代码:

uthread.h

/*<br /> * brief: 用ucontext实现的用户级线程框架<br /> * author: kenny huang<br /> * date: 2009/10/13<br /> * email: huangweilook@21cn.com<br /> */<br /> #ifndef _UTHREAD_H<br /> #define _UTHREAD_H<br /> #include <ucontext.h><br /> #include <stdio.h><br /> #include <string.h><br /> #include <list><br /> #include <pthread.h><br /> #include <signal.h><br /> #include "socketwrapper.h"</p> <p>#define MAX_UTHREAD 128</p> <p>void int_signal_handler(int sig);<br /> //用户态线程的当前状态<br /> enum thread_status<br /> {<br /> ACTIVED = 0,//可运行的<br /> BLOCKED,//被阻塞<br /> SLEEP,//主动休眠<br /> DIE,//死死亡<br /> };</p> <p>typedef int (*uthread_func)(void*);</p> <p>class Scheduler;<br /> class u_thread;</p> <p>typedef struct<br /> {<br /> int index;//在Scheduler::threads中的下标<br /> u_thread *p_uthread;<br /> ucontext_t *p_context;<br /> }uthread_id;</p> <p>/*<br /> * 用户态线程<br /> */<br /> class u_thread<br /> {<br /> friend class Scheduler;<br /> private:<br /> u_thread(unsigned int ssize,int index,uthread_id parent)<br /> :ssize(ssize),_status(BLOCKED),parent_context(parent.p_context)<br /> {<br /> stack = new char[ssize];<br /> ucontext.uc_stack.ss_sp = stack;<br /> ucontext.uc_stack.ss_size = ssize;<br /> getcontext(&ucontext);<br /> uid.index = index;<br /> uid.p_uthread = this;<br /> uid.p_context = &ucontext;</p> <p> }<br /> ~u_thread()<br /> {<br /> delete []stack;<br /> }<br /> static void star_routine(int uthread,int func,int arg);</p> <p>public:<br /> ucontext_t* GetParentContext()<br /> {<br /> return parent_context;<br /> }<br /> ucontext_t *GetContext()<br /> {<br /> return &ucontext;<br /> }<br /> void SetStatus(thread_status _status)<br /> {<br /> this->_status = _status;<br /> }<br /> thread_status GetStatus()<br /> {<br /> return _status;<br /> }<br /> uthread_id GetUid()<br /> {<br /> return uid;<br /> }</p> <p>private:<br /> ucontext_t ucontext;<br /> ucontext_t *parent_context;//父亲的context<br /> char *stack;//coroutine使用的栈<br /> unsigned int ssize;//栈的大小<br /> thread_status _status;<br /> uthread_id uid;<br /> };</p> <p>void BeginRun();<br /> bool cSpawn(uthread_func func,void *arg,unsigned int stacksize);<br /> int cRecv(int sock,char *buf,int len);<br /> int cSend(int sock,char *buf,int len);<br /> int cListen(int sock);<br /> void cSleep(int t);</p> <p>class beat;</p> <p>/*<br /> * 任务调度器<br /> */<br /> class Scheduler<br /> {<br /> friend class beat;<br /> friend class u_thread;<br /> friend void BeginRun();<br /> friend bool cSpawn(uthread_func func,void *arg,unsigned int stacksize);<br /> friend int cRecv(int sock,char *buf,int len);<br /> friend int cSend(int sock,char *buf,int len);<br /> friend int cListen(int sock);<br /> friend void cSleep(int t);<br /> public:<br /> static void scheduler_init();<br /> static void int_sig();</p> <p>private:</p> <p> //休眠time时间<br /> static void sleep(int t);<br /> static void check_Network();<br /> static void schedule();<br /> static bool spawn(uthread_func func,void *arg,unsigned int stacksize);<br /> static int recv(int sock,char *buf,int len);<br /> static int send(int sock,char *buf,int len);<br /> static int listen(int sock);</p> <p>private:<br /> static std::list<u_thread*> activeList;//可运行uthread列表</p> <p> static std::list<std::pair<u_thread*,time_t> > sleepList;//正在睡眠uthread列表<br /> static volatile bool block_signal;<br /> static char stack[4096];</p> <p> static ucontext_t ucontext;//Scheduler的context</p> <p> static uthread_id uid_current;//当前正获得运行权的context</p> <p> static uthread_id uid_self;</p> <p> static u_thread *threads[MAX_UTHREAD];<br /> static int total_count;<br /> static int epollfd;<br /> const static int maxsize = 10;<br /> };<br /> /*心跳发射器，发射器必须运行在一个独立的线程中，以固定的间隔<br /> * 往所有运行着coroutine的线程发送中断信号<br /> */<br /> class beat<br /> {<br /> public:<br /> beat(unsigned int interval):interval(interval)<br /> {}<br /> void setThread(pthread_t id)<br /> {<br /> thread_scheduler = id;<br /> }<br /> void loop()<br /> {<br /> while(true)<br /> {<br /> //每隔固定时间向所有线程发中断信号<br /> ::usleep(1000 * interval);<br /> while(1)<br /> {<br /> if(!Scheduler::block_signal)<br /> {<br /> pthread_kill(thread_scheduler,SIGUSR1);<br /> break;<br /> }<br /> }</p> <p> }<br /> }<br /> private:<br /> unsigned int interval;//发送中断的间隔(豪秒)<br /> pthread_t thread_scheduler;<br /> };</p> <p>bool initcoroutine(unsigned int interval);</p> <p>#endif<br />  

uthread.cpp

#include "uthread.h"<br /> #include <assert.h><br /> #include <stdlib.h><br /> #include <time.h><br /> #include <netinet/in.h><br /> #include <sys/socket.h><br /> #include <sys/time.h><br /> #include <netdb.h><br /> #include <stdio.h><br /> #include <stdlib.h><br /> #include <string.h><br /> #include <unistd.h><br /> #include <sys/epoll.h><br /> #include "thread.h"</p> <p>ucontext_t Scheduler::ucontext;</p> <p>char Scheduler::stack[4096];</p> <p>uthread_id Scheduler::uid_current;</p> <p>uthread_id Scheduler::uid_self;</p> <p>u_thread *Scheduler::threads[MAX_UTHREAD];</p> <p>int Scheduler::total_count = 0;</p> <p>int Scheduler::epollfd = 0;</p> <p>volatile bool Scheduler::block_signal = true; </p> <p>std::list<u_thread*> Scheduler::activeList;</p> <p>std::list<std::pair<u_thread*,time_t> > Scheduler::sleepList;</p> <p>struct sock_struct<br /> {<br /> int sockfd;<br /> u_thread *uThread;<br /> };</p> <p>void int_signal_handler(int sig)<br /> {<br /> Scheduler::int_sig();<br /> }</p> <p>void u_thread::star_routine(int uthread,int func,int arg)<br /> {<br /> u_thread *self_uthread = (u_thread *)uthread;<br /> assert(self_uthread);</p> <p> self_uthread->SetStatus(ACTIVED);<br /> ucontext_t *self_context = self_uthread->GetContext();<br /> swapcontext(self_context,self_uthread->GetParentContext());<br /> Scheduler::block_signal = false;</p> <p> uthread_func _func = (uthread_func)func;<br /> void *_arg = (void*)arg;<br /> int ret = _func(_arg);<br /> self_uthread->SetStatus(DIE);<br /> }<br /> void Scheduler::scheduler_init()<br /> {<br /> for(int i = 0; i < MAX_UTHREAD; ++i)<br /> threads[i] = 0;<br /> getcontext(&ucontext);<br /> ucontext.uc_stack.ss_sp = stack;<br /> ucontext.uc_stack.ss_size = sizeof(stack);<br /> ucontext.uc_link = NULL;</p> <p> //scheduler占用下标0<br /> makecontext(&ucontext,schedule, 0); </p> <p> uid_self.index = 0;<br /> uid_self.p_uthread = 0;<br /> uid_self.p_context = &ucontext;<br /> uid_current = uid_self;</p> <p>}<br /> int Scheduler::listen(int sock)<br /> {<br /> u_thread *self_thread = uid_current.p_uthread;</p> <p> epoll_event ev;<br /> sock_struct *ss = new sock_struct;<br /> ss->uThread = self_thread;<br /> ss->sockfd = sock;<br /> ev.data.ptr = (void*)ss;<br /> ev.events = EPOLLIN;<br /> int ret;<br /> TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));<br /> if(ret != 0)<br /> {<br /> return -1;<br /> }</p> <p> self_thread->SetStatus(BLOCKED);<br /> Scheduler::block_signal = true;<br /> swapcontext(uid_current.p_context,&Scheduler::ucontext);<br /> Scheduler::block_signal = false;<br /> return 1;<br /> }</p> <p>int Scheduler::recv(int sock,char *buf,int len)<br /> {<br /> if(!buf || !(len > 0))<br /> return -1;<br /> u_thread *self_thread = uid_current.p_uthread;<br /> sock_struct *ss = new sock_struct;<br /> ss->uThread = self_thread;<br /> ss->sockfd = sock;<br /> epoll_event ev;<br /> ev.data.ptr = (void*)ss;<br /> ev.events = EPOLLIN;<br /> int ret;<br /> TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));<br /> if(ret != 0)<br /> return -1;</p> <p> self_thread->SetStatus(BLOCKED);<br /> Scheduler::block_signal = true;<br /> swapcontext(uid_current.p_context,&Scheduler::ucontext);<br /> printf("recv return/n");<br /> Scheduler::block_signal = false;<br /> ret = read(sock,buf,len);<br /> return ret;<br /> }<br /> int Scheduler::send(int sock,char *buf,int len)<br /> {<br /> if(!buf || !(len > 0))<br /> return -1;</p> <p> u_thread *self_thread = uid_current.p_uthread; </p> <p> sock_struct *ss = new sock_struct;<br /> ss->uThread = self_thread;<br /> ss->sockfd = sock;<br /> epoll_event ev;<br /> ev.data.ptr = (void*)ss;<br /> ev.events = EPOLLOUT;<br /> int ret;<br /> TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));<br /> if(ret != 0)<br /> return -1;<br /> self_thread->SetStatus(BLOCKED);<br /> Scheduler::block_signal = true;<br /> swapcontext(uid_current.p_context,&Scheduler::ucontext);<br /> Scheduler::block_signal = false;<br /> ret = write(sock,buf,len);<br /> return ret;<br /> }<br /> void Scheduler::check_Network()<br /> {<br /> epoll_event events[maxsize];<br /> sock_struct *ss;<br /> int nfds = TEMP_FAILURE_RETRY(epoll_wait(epollfd,events,maxsize,0));<br /> for( int i = 0 ; i < nfds ; ++i)<br /> {<br /> //套接口可读<br /> if(events[i].events & EPOLLIN)<br /> {<br /> ss = (sock_struct*)events[i].data.ptr;<br /> printf("a sock can read/n");<br /> ss->uThread->SetStatus(ACTIVED);<br /> epoll_event ev;<br /> ev.data.fd = ss->sockfd;<br /> if(0 != TEMP_FAILURE_RETRY(epoll_ctl(epollfd,EPOLL_CTL_DEL,ev.data.fd,&ev)))<br /> {<br /> printf("error here/n");<br /> exit(0);<br /> }<br /> delete ss;<br /> continue;<br /> }<br /> //套接口可写<br /> if(events[i].events & EPOLLOUT)<br /> {<br /> ss = (sock_struct*)events[i].data.ptr;<br /> printf("a sock can write/n");<br /> ss->uThread->SetStatus(ACTIVED);<br /> epoll_event ev;<br /> ev.data.fd = ss->sockfd;<br /> TEMP_FAILURE_RETRY(epoll_ctl(epollfd,EPOLL_CTL_DEL,ev.data.fd,&ev));<br /> delete ss;<br /> continue;<br /> }<br /> }<br /> }<br /> void Scheduler::schedule()<br /> {<br /> epollfd = TEMP_FAILURE_RETRY(epoll_create(maxsize)); </p> <p> if(epollfd<= 0)<br /> {<br /> printf("epoll init error/n");<br /> return;<br /> }</p> <p> while(total_count > 0)<br /> {</p> <p> //首先执行active列表中的uthread<br /> std::list<u_thread*>::iterator it = activeList.begin();<br /> std::list<u_thread*>::iterator end = activeList.end();<br /> for( ; it != end; ++it)<br /> {<br /> if(*it && (*it)->GetStatus() == ACTIVED)<br /> {<br /> uid_current = (*it)->GetUid();<br /> swapcontext(&ucontext,uid_current.p_context);<br /> uid_current = uid_self;</p> <p> int index = (*it)->GetUid().index;<br /> if((*it)->GetStatus() == DIE)<br /> {<br /> printf("%d die/n",index);<br /> delete threads[index];<br /> threads[index] = 0;<br /> --total_count;<br /> activeList.erase(it);<br /> break;<br /> }<br /> else if((*it)->GetStatus() == SLEEP)<br /> {<br /> printf("%d sleep/n",index);<br /> activeList.erase(it);<br /> break;<br /> }<br /> }<br /> }<br /> //检查网络，看看是否有套接口可以操作<br /> check_Network();<br /> //看看Sleep列表中是否有uthread该醒来了<br /> std::list<std::pair<u_thread*,time_t> >::iterator its = sleepList.begin();<br /> std::list<std::pair<u_thread*,time_t> >::iterator ends = sleepList.end();<br /> time_t now = time(NULL);<br /> for( ; its != ends; ++its)<br /> {<br /> //可以醒来了<br /> if(now >= its->second)<br /> {<br /> u_thread *uthread = its->first;<br /> uthread->SetStatus(ACTIVED);<br /> activeList.push_back(uthread);<br /> sleepList.erase(its);<br /> break;<br /> }<br /> }<br /> }<br /> printf("scheduler end/n");<br /> }</p> <p>bool Scheduler::spawn(uthread_func func,void *arg,unsigned int stacksize)<br /> {<br /> printf("uthread_create/n");<br /> if(total_count >= MAX_UTHREAD)<br /> return false;<br /> int i = 1;<br /> for( ; i < MAX_UTHREAD; ++i)<br /> {<br /> if(threads[i] == 0)<br /> {<br /> threads[i] = new u_thread(stacksize,i,uid_current);<br /> ++total_count;<br /> ucontext_t *cur_context = threads[i]->GetContext();<br /> activeList.push_back(threads[i]);</p> <p> cur_context->uc_link = &ucontext;<br /> makecontext(cur_context,(void (*)())u_thread::star_routine, 3,(int)&(*threads[i]),(int)func,(int)arg);<br /> swapcontext(uid_current.p_context, cur_context);<br /> printf("return from parent/n");<br /> return true;<br /> }<br /> }</p> <p> return false;<br /> }<br /> void Scheduler::sleep(int t)<br /> {<br /> u_thread *self_thread = uid_current.p_uthread; </p> <p> time_t now = time(NULL);<br /> now += t;<br /> //插入到sleep列表中<br /> sleepList.push_back(std::make_pair(self_thread,now));</p> <p> //保存当前上下文切换回scheduler<br /> self_thread->SetStatus(SLEEP);</p> <p> ucontext_t *cur_context = self_thread->GetContext();<br /> Scheduler::block_signal = true;<br /> swapcontext(cur_context,&Scheduler::ucontext);<br /> Scheduler::block_signal = false;<br /> }<br /> void Scheduler::int_sig()<br /> {<br /> //printf("Scheduler::int_sig()%x/n",uid_current.p_context);<br /> Scheduler::block_signal = true;<br /> swapcontext(uid_current.p_context,&Scheduler::ucontext);<br /> Scheduler::block_signal = false;<br /> }</p> <p>class HeartBeat : public runnable<br /> {<br /> public:<br /> HeartBeat(unsigned int interval)<br /> {<br /> _beat = new beat(interval);<br /> _beat->setThread(pthread_self());<br /> }</p> <p> ~HeartBeat()<br /> {<br /> delete _beat;<br /> }</p> <p> bool run()<br /> {<br /> _beat->loop();<br /> return true;<br /> }</p> <p>private:<br /> beat *_beat;<br /> };</p> <p>bool initcoroutine(unsigned int interval)<br /> {<br /> //初始化信号<br /> struct sigaction sigusr1;<br /> sigusr1.sa_flags = 0;<br /> sigusr1.sa_handler = int_signal_handler;<br /> sigemptyset(&sigusr1.sa_mask);<br /> int status = sigaction(SIGUSR1,&sigusr1,NULL);<br /> if(status == -1)<br /> {<br /> printf("error sigaction/n");<br /> return false;<br /> }</p> <p> //首先初始化调度器<br /> Scheduler::scheduler_init();</p> <p> //启动心跳<br /> static HeartBeat hb(interval);<br /> static Thread c(&hb);<br /> c.start();<br /> return true;<br /> }</p> <p>void BeginRun()<br /> {<br /> Scheduler::schedule();<br /> }</p> <p>bool cSpawn(uthread_func func,void *arg,unsigned int stacksize)<br /> {<br /> return Scheduler::spawn(func,arg,stacksize);<br /> }</p> <p>int cRecv(int sock,char *buf,int len)<br /> {<br /> return Scheduler::recv(sock,buf,len);<br /> }</p> <p>int cSend(int sock,char *buf,int len)<br /> {<br /> return Scheduler::send(sock,buf,len);<br /> }</p> <p>int cListen(int sock)<br /> {<br /> return Scheduler::listen(sock);<br /> }</p> <p>void cSleep(int t)<br /> {<br /> return Scheduler::sleep(t);<br /> }</p> <p> 

uthreadtest.cpp

// kcoroutine.cpp : 定义控制台应用程序的入口点。<br /> //<br /> #include "uthread.h"<br /> #include "thread.h"<br /> int port;<br /> int test(void *arg)<br /> {<br /> char *name = (char*)arg;<br /> unsigned long c = 0;<br /> while(1)<br /> {<br /> if(c % 10000 == 0)<br /> {<br /> printf("%d/n",c);<br /> cSleep(1);<br /> }<br /> ++c;<br /> }<br /> }<br /> int echo(void *arg)<br /> {<br /> int sock = *(int*)arg;<br /> while(1)<br /> {<br /> char buf[1024];<br /> int ret = cRecv(sock,buf,1024);<br /> if(ret > 0)<br /> {<br /> printf("%s/n",buf);<br /> ret = cSend(sock,buf,ret);<br /> }<br /> }<br /> }<br /> int listener(void *arg)<br /> {<br /> struct sockaddr_in servaddr;<br /> int listenfd;<br /> if(0 > (listenfd = Tcp_Listen("127.0.0.1",port,servaddr,5)))<br /> {<br /> printf("listen error/n");<br /> return 0;<br /> }<br /> while(1)<br /> {<br /> if(cListen(listenfd) > 0)<br /> {<br /> printf("a user comming/n");<br /> struct sockaddr_in cliaddr;<br /> socklen_t len;<br /> int sock = Accept(listenfd,(struct sockaddr*)NULL,NULL);<br /> if(sock >= 0)<br /> {<br /> cSpawn(echo,&sock,4096);<br /> }</p> <p> }<br /> }<br /> }<br /> int main(int argc, char **argv)<br /> {<br /> port = atoi(argv[1]);<br /> if(!initcoroutine(20))<br /> return 0;<br /> cSpawn(listener,0,4096);<br /> char name[10] = "test";<br /> cSpawn(test,name,4096);</p> <p> printf("create finish/n");</p> <p> //开始调度线程的运行<br /> BeginRun();<br /> return 0;<br /> }<br />  

运行后会看到控制台中不断的输出1，那是runable_test在工作，

telnet几个客户端上去，就可以看到echo的效果了，总体来看效果还不错，

不过context的切换效率还没有测试过.