ucontext实现的用户级多线程框架3（实现echo服务器）

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

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

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

uthread.h

/*
* 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

uthread.cpp

#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);
}

echoserver.c

// 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;
}

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

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

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