/** brief: 用ucontext实现的用户级线程框架* author: kenny huang* date: 2009/10/13* email: huangweilook@21cn.com*/#ifndef _UTHREAD_H#define _UTHREAD_H#include <ucontext.h>#include <stdio.h>#include <string.h>#include <list>#include <pthread.h>#include <signal.h>#include "socketwrapper.h"

#define MAX_UTHREAD 128

void int_signal_handler(int sig);//用户态线程的当前状态enum thread_status{    ACTIVED = 0,//可运行的    BLOCKED,//被阻塞    SLEEP,//主动休眠    DIE,//死死亡};

typedef int (*uthread_func)(void*);

class Scheduler;class u_thread;

typedef struct{int index;//在Scheduler::threads中的下标    u_thread *p_uthread;    ucontext_t *p_context; }uthread_id;

/** 用户态线程*/class u_thread{    friend class Scheduler;private:    u_thread(unsigned int ssize,int index,uthread_id parent)        :ssize(ssize),_status(BLOCKED),parent_context(parent.p_context)        {            stack = new char[ssize];            ucontext.uc_stack.ss_sp = stack;            ucontext.uc_stack.ss_size = ssize;            getcontext(&ucontext);            uid.index = index;            uid.p_uthread = this;            uid.p_context = &ucontext;

        }    ~u_thread()    {        delete []stack;    }static void star_routine(int uthread,int func,int arg);

public:    ucontext_t* GetParentContext()    {return parent_context;    }    ucontext_t *GetContext()    {return &ucontext;    }void SetStatus(thread_status _status)    {this->_status = _status;    }    thread_status GetStatus()    {return _status;    }    uthread_id GetUid()    {return uid;    }

private:    ucontext_t ucontext;    ucontext_t *parent_context;//父亲的context    char *stack;//coroutine使用的栈    unsigned int ssize;//栈的大小    thread_status _status;    uthread_id uid;};

void BeginRun();bool cSpawn(uthread_func func,void *arg,unsigned int stacksize);int  cRecv(int sock,char *buf,int len);int  cSend(int sock,char *buf,int len);int  cListen(int sock);void cSleep(int t);

class beat;

/** 任务调度器*/class Scheduler{    friend class beat;    friend class u_thread;    friend void BeginRun();    friend bool cSpawn(uthread_func func,void *arg,unsigned int stacksize);    friend int  cRecv(int sock,char *buf,int len);    friend int  cSend(int sock,char *buf,int len);    friend int  cListen(int sock);    friend void cSleep(int t);public:static void scheduler_init();static void int_sig();

private:

//休眠time时间    static void sleep(int t);static void check_Network();static void schedule();static bool spawn(uthread_func func,void *arg,unsigned int stacksize);static int recv(int sock,char *buf,int len);static int send(int sock,char *buf,int len);static int listen(int sock);

private:static std::list<u_thread*> activeList;//可运行uthread列表

static std::list<std::pair<u_thread*,time_t> > sleepList;//正在睡眠uthread列表    static volatile bool block_signal; static  char stack[4096];

static     ucontext_t ucontext;//Scheduler的context

static uthread_id uid_current;//当前正获得运行权的context

static uthread_id uid_self;

static  u_thread *threads[MAX_UTHREAD];static  int total_count;static  int epollfd;const static int maxsize = 10;};/*心跳发射器，发射器必须运行在一个独立的线程中，以固定的间隔* 往所有运行着coroutine的线程发送中断信号*/class beat{public:    beat(unsigned int interval):interval(interval)    {}void setThread(pthread_t id)    {        thread_scheduler = id;    }void loop()    {while(true)        {//每隔固定时间向所有线程发中断信号            ::usleep(1000 * interval);while(1)            {if(!Scheduler::block_signal)                {                    pthread_kill(thread_scheduler,SIGUSR1);break;                }            }

        }    }private:    unsigned int interval;//发送中断的间隔(豪秒)    pthread_t thread_scheduler;};

bool initcoroutine(unsigned int interval);

#endif

#include "uthread.h"#include <assert.h>#include <stdlib.h>#include <time.h>#include <netinet/in.h>#include <sys/socket.h>#include <sys/time.h>#include <netdb.h>#include <stdio.h>#include <stdlib.h>#include <string.h>#include <unistd.h>#include <sys/epoll.h>#include "thread.h"

ucontext_t Scheduler::ucontext;

char Scheduler::stack[4096];

uthread_id Scheduler::uid_current;

uthread_id Scheduler::uid_self;

u_thread *Scheduler::threads[MAX_UTHREAD];

int Scheduler::total_count = 0;

int Scheduler::epollfd = 0;

volatile bool Scheduler::block_signal = true; 

std::list<u_thread*> Scheduler::activeList;

std::list<std::pair<u_thread*,time_t> > Scheduler::sleepList;

struct sock_struct{int sockfd;    u_thread *uThread;};

void int_signal_handler(int sig){    Scheduler::int_sig();}

void u_thread::star_routine(int uthread,int func,int arg){    u_thread *self_uthread = (u_thread *)uthread;    assert(self_uthread);

    self_uthread->SetStatus(ACTIVED);    ucontext_t *self_context = self_uthread->GetContext();    swapcontext(self_context,self_uthread->GetParentContext());    Scheduler::block_signal = false;

    uthread_func _func = (uthread_func)func;void *_arg = (void*)arg;int ret = _func(_arg);    self_uthread->SetStatus(DIE);}void Scheduler::scheduler_init(){for(int i = 0; i < MAX_UTHREAD; ++i)        threads[i] = 0;    getcontext(&ucontext);    ucontext.uc_stack.ss_sp = stack;    ucontext.uc_stack.ss_size = sizeof(stack);    ucontext.uc_link = NULL;

//scheduler占用下标0    makecontext(&ucontext,schedule, 0);    

    uid_self.index = 0;    uid_self.p_uthread = 0;    uid_self.p_context = &ucontext;    uid_current = uid_self;

}int Scheduler::listen(int sock){    u_thread *self_thread = uid_current.p_uthread;

    epoll_event ev;    sock_struct *ss = new sock_struct;    ss->uThread = self_thread;    ss->sockfd = sock;    ev.data.ptr = (void*)ss;    ev.events = EPOLLIN;int ret;    TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));if(ret != 0)    {return -1;    }

    self_thread->SetStatus(BLOCKED);    Scheduler::block_signal = true;    swapcontext(uid_current.p_context,&Scheduler::ucontext);    Scheduler::block_signal = false;return 1;}

int Scheduler::recv(int sock,char *buf,int len){if(!buf || !(len > 0))return -1;    u_thread *self_thread = uid_current.p_uthread;     sock_struct *ss = new sock_struct;    ss->uThread = self_thread;    ss->sockfd = sock;    epoll_event ev;    ev.data.ptr = (void*)ss;    ev.events = EPOLLIN;int ret;    TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));if(ret != 0)return -1;

    self_thread->SetStatus(BLOCKED);    Scheduler::block_signal = true;    swapcontext(uid_current.p_context,&Scheduler::ucontext);    printf("recv return/n");    Scheduler::block_signal = false;    ret = read(sock,buf,len);return ret;}int Scheduler::send(int sock,char *buf,int len){if(!buf || !(len > 0))return -1;

    u_thread *self_thread = uid_current.p_uthread; 

    sock_struct *ss = new sock_struct;    ss->uThread = self_thread;    ss->sockfd = sock;    epoll_event ev;    ev.data.ptr = (void*)ss;    ev.events = EPOLLOUT;int ret;    TEMP_FAILURE_RETRY(ret = epoll_ctl(epollfd,EPOLL_CTL_ADD,sock,&ev));if(ret != 0)return -1;    self_thread->SetStatus(BLOCKED);    Scheduler::block_signal = true;    swapcontext(uid_current.p_context,&Scheduler::ucontext);    Scheduler::block_signal = false;    ret = write(sock,buf,len);return ret;    }void Scheduler::check_Network(){        epoll_event events[maxsize];    sock_struct *ss;int nfds = TEMP_FAILURE_RETRY(epoll_wait(epollfd,events,maxsize,0));for( int i = 0 ; i < nfds ; ++i)    {//套接口可读        if(events[i].events & EPOLLIN)        {                 ss = (sock_struct*)events[i].data.ptr;            printf("a sock can read/n");            ss->uThread->SetStatus(ACTIVED);            epoll_event ev;            ev.data.fd = ss->sockfd;if(0 != TEMP_FAILURE_RETRY(epoll_ctl(epollfd,EPOLL_CTL_DEL,ev.data.fd,&ev)))            {                printf("error here/n");                exit(0);            }            delete ss;continue;        }//套接口可写        if(events[i].events & EPOLLOUT)        {            ss = (sock_struct*)events[i].data.ptr;            printf("a sock can write/n");            ss->uThread->SetStatus(ACTIVED);            epoll_event ev;            ev.data.fd = ss->sockfd;            TEMP_FAILURE_RETRY(epoll_ctl(epollfd,EPOLL_CTL_DEL,ev.data.fd,&ev));            delete ss;continue;        }    }}void Scheduler::schedule(){    epollfd = TEMP_FAILURE_RETRY(epoll_create(maxsize)); 

if(epollfd<= 0)    {        printf("epoll init error/n");return;    }

while(total_count > 0)    {

//首先执行active列表中的uthread        std::list<u_thread*>::iterator it = activeList.begin();         std::list<u_thread*>::iterator end = activeList.end();for( ; it != end; ++it)        {if(*it && (*it)->GetStatus() == ACTIVED)            {                uid_current = (*it)->GetUid();                swapcontext(&ucontext,uid_current.p_context);                uid_current = uid_self;

int index = (*it)->GetUid().index;if((*it)->GetStatus() == DIE)                {                    printf("%d die/n",index);                    delete threads[index];                    threads[index] = 0;                    --total_count;                    activeList.erase(it);break;                }else if((*it)->GetStatus() == SLEEP)                {                    printf("%d sleep/n",index);                    activeList.erase(it);break;                }            }        }//检查网络，看看是否有套接口可以操作        check_Network();//看看Sleep列表中是否有uthread该醒来了        std::list<std::pair<u_thread*,time_t> >::iterator its = sleepList.begin();        std::list<std::pair<u_thread*,time_t> >::iterator ends = sleepList.end();        time_t now = time(NULL);for( ; its != ends; ++its)        {//可以醒来了            if(now >= its->second)            {                    u_thread *uthread = its->first;                uthread->SetStatus(ACTIVED);                activeList.push_back(uthread);                sleepList.erase(its);break;            }        }    }    printf("scheduler end/n");}

bool Scheduler::spawn(uthread_func func,void *arg,unsigned int stacksize){    printf("uthread_create/n");if(total_count >= MAX_UTHREAD)return false;int i = 1;for( ; i < MAX_UTHREAD; ++i)    {if(threads[i] == 0)        {            threads[i] = new u_thread(stacksize,i,uid_current);            ++total_count;            ucontext_t *cur_context = threads[i]->GetContext();            activeList.push_back(threads[i]);

            cur_context->uc_link = &ucontext;            makecontext(cur_context,(void (*)())u_thread::star_routine, 3,(int)&(*threads[i]),(int)func,(int)arg);            swapcontext(uid_current.p_context, cur_context);            printf("return from parent/n");return true;        }    }

return false;}void Scheduler::sleep(int t){    u_thread *self_thread = uid_current.p_uthread; 

    time_t now = time(NULL);    now += t;//插入到sleep列表中    sleepList.push_back(std::make_pair(self_thread,now));

//保存当前上下文切换回scheduler    self_thread->SetStatus(SLEEP);

    ucontext_t *cur_context = self_thread->GetContext();    Scheduler::block_signal = true;    swapcontext(cur_context,&Scheduler::ucontext);    Scheduler::block_signal = false;}void Scheduler::int_sig(){//printf("Scheduler::int_sig()%x/n",uid_current.p_context);    Scheduler::block_signal = true;    swapcontext(uid_current.p_context,&Scheduler::ucontext);    Scheduler::block_signal = false;}

class HeartBeat : public runnable{public:    HeartBeat(unsigned int interval)    {        _beat = new beat(interval);        _beat->setThread(pthread_self());    }

    ~HeartBeat()    {        delete _beat;    }

bool run()    {        _beat->loop();return true;    }

private:    beat *_beat;};

bool initcoroutine(unsigned int interval){//初始化信号    struct sigaction sigusr1;    sigusr1.sa_flags = 0;    sigusr1.sa_handler = int_signal_handler;    sigemptyset(&sigusr1.sa_mask);int status = sigaction(SIGUSR1,&sigusr1,NULL);if(status == -1)    {        printf("error sigaction/n");return false;    }

//首先初始化调度器    Scheduler::scheduler_init();

//启动心跳    static HeartBeat hb(interval);static Thread c(&hb);    c.start();return true;}

void BeginRun(){    Scheduler::schedule();}

bool cSpawn(uthread_func func,void *arg,unsigned int stacksize){return Scheduler::spawn(func,arg,stacksize);    }

int cRecv(int sock,char *buf,int len){return Scheduler::recv(sock,buf,len);    }

int cSend(int sock,char *buf,int len){return Scheduler::send(sock,buf,len);    }

int cListen(int sock){return Scheduler::listen(sock);}

void cSleep(int t){return Scheduler::sleep(t);}

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

        }    }    }int main(int argc, char **argv){    port = atoi(argv[1]);if(!initcoroutine(20))return 0;    cSpawn(listener,0,4096);char name[10] = "test";    cSpawn(test,name,4096);

    printf("create finish/n");

//开始调度线程的运行    BeginRun();return 0;}