技术系列，在本人的认知范围内，探讨系统级别的核心技术，文章描述的都是技术本身，最终目的是建立公用的模块库，比如含消息队列的线程模块、状态机模块、定时器模块、对象池模块、log模块、网络层模块、管理接口模块等等。

从个人的角度考虑，开发公用模块，可以加深对系统整体的了解，能有更多的时间去了解新的技术、去认识业务的重要性、去考虑系统划分的合理性，不要陷入技术繁琐细节的泥潭，技术仅仅是工具而已，它们背后的原理都很简单。

从团队的角度考虑，开发公有模块，可以增加系统的可维护性、减少对新员工的培训成本、使具有不同业务的系统在技术细节上呈现一致性，形成开发的可持续性，减少对个人技术的依赖，不会因为某个核心人员的离职而造成后续开发的戛然而止。

公有模块开发导致的后果是减少对个人的依赖，因此有的技术人员可能对开发这些模块产生抵触心理。个人认为，技术是没有边际的，单个人是永远也不会掌握所有技术，单纯的技术也不是正确的职业规划，固步自封只能导致自己技术水平的落后。以开放的心态、职业化的心态去做实际的开发、去交流沟通，最终受益的是自己。

业务系列的文章写点本人接触过的业务，以公有协议为主线，mgcp、sip、rtsp、rtp、rtcp、加密框架等，协议本身、应用场景等，想想可写的东西也不多，和技术一样，都是很少很简单的东西。

系统系列，则从业务的角度出发，描述一些通用系统，也就是纯共有协议业务系统。当然不会涉及到公司的系统，呵呵。比如说个媒体负载子系统，该子系统可以独立于信令协议，一个信令系统下可以挂载n个媒体子系统，可协同信令系统完成nat穿透、负载均衡、媒体流加解密等功能，使用私有协议和信令系统通讯，可定时比如60秒发一次负载个数信令到信令系统，一是作为心跳，二是作为负载均衡的依据。再比如一些顶层协议无关的proxy系统设计等。
从团队、公司的角度来考虑，开发公用系统的重要性不言而喻。开发公用系统也即开发工作由项目驱动转化为产品驱动、根据所做一系列项目的固有特点，找出事物的内在规律，合理划分子系统。
开发公用系统，可以降低开发的成本开销，便于公司去开发更多的业务，涉足更多的业务市场，从而形成不断扩大的业务市场，而不是以防守者的姿态，维护自己已有的甚至是逐渐萎缩的业务市场份额。

测试系列。测试团队的开发还是集中于业务层面的测试，白盒测试、稳定性测试还是要研发自己来做。写点自己的测试心得。虽然本blog上的技术文章远多于测试，但个人认为测试的重要性大于技术，不重视测试的研发不是一个好研发，也不会成为一个好研发，顶多是一个夸夸其谈的华而不实者。

团队管理系列，也或许不会写，毕竟不在管理位置，不会谋其职。但从职业发展的角度看，还是应当总结点，在不同的公司感受不同的管理，看看各个公司的相同点不同点，管理的初衷等。团队管理部分，个人看呈现下面几个部分:公司福利/公司制度、项目管理、技术培训、对员工的人文关怀、个人管理等。

int select(int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

其中参数n表示监控的所有fd中最大值＋1。
和select模型紧密结合的四个宏，含义不解释了：

FD_CLR(int fd, fd_set *set);
FD_ISSET(int fd, fd_set *set);
FD_SET(int fd, fd_set *set);
FD_ZERO(fd_set *set);

理解select模型的关键在于理解fd_set,为说明方便，取fd_set长度为1字节，fd_set中的每一bit可以对应一个文件描述符fd。则1字节长的fd_set最大可以对应8个fd。
（1）执行fd_set set; FD_ZERO(&set);则set用位表示是0000,0000。
（2）若fd＝5,执行FD_SET(fd,&set);后set变为0001,0000(第5位置为1)
（3）若再加入fd＝2，fd=1,则set变为0001,0011
（4）执行select(6,&set,0,0,0)阻塞等待
（5）若fd=1,fd=2上都发生可读事件，则select返回，此时set变为0000,0011。注意：没有事件发生的fd=5被清空。

基于上面的讨论，可以轻松得出select模型的特点：
（1)可监控的文件描述符个数取决与sizeof(fd_set)的值。我这边服务器上sizeof(fd_set)＝512，每bit表示一个文件描述符，则我服务器上支持的最大文件描述符是512*8=4096。据说可调，另有说虽然可调，但调整上限受于编译内核时的变量值。本人对调整fd_set的大小不太感兴趣，参考http://www.cppblog.com/CppExplore/archive/2008/03/21/45061.html中的模型2（1）可以有效突破select可监控的文件描述符上限。
（2）将fd加入select监控集的同时，还要再使用一个数据结构array保存放到select监控集中的fd，一是用于再select返回后，array作为源数据和fd_set进行FD_ISSET判断。二是select返回后会把以前加入的但并无事件发生的fd清空，则每次开始select前都要重新从array取得fd逐一加入（FD_ZERO最先），扫描array的同时取得fd最大值maxfd，用于select的第一个参数。
（3）可见select模型必须在select前循环array（加fd，取maxfd），select返回后循环array（FD_ISSET判断是否有时间发生）。

下面给一个伪码说明基本select模型的服务器模型：

array[slect_len];
nSock=0;
array[nSock++]=listen_fd;(之前listen port已绑定并listen)
maxfd=listen_fd;
while{
   FD_ZERO(&set);
   foreach (fd in array) 
   {
       fd大于maxfd，则maxfd=fd
       FD_SET(fd,&set)
   }
   res=select(maxfd+1,&set,0,0,0)；
   if(FD_ISSET(listen_fd,&set))
   {
       newfd=accept(listen_fd);
       array[nsock++]=newfd;
            if(--res<=0) continue
   }
   foreach 下标1开始 (fd in array) 
   {
       if(FD_ISSET(fd,&set))
          执行读等相关操作
          如果错误或者关闭，则要删除该fd，将array中相应位置和最后一个元素互换就好，nsock减一
             if(--res<=0) continue

   }
}

二、poll模型
poll原型:

int poll(struct pollfd *ufds, unsigned int nfds, int timeout);
struct pollfd {
                       int fd;           /**//* file descriptor */
                       short events;     /**//* requested events */
                       short revents;    /**//* returned events */
               };

和select相比，两大改进：
（1）不再有fd个数的上限限制，可以将参数ufds想象成栈低指针，nfds是栈中元素个数，该栈可以无限制增长
（2）引入pollfd结构，将fd信息、需要监控的事件、返回的事件分开保存，则poll返回后不会丢失fd信息和需要监控的事件信息，也就省略了select模型中前面的循环操作，返回后的循环仍然不可避免。另每次poll阻塞操作都会自动把上次的revents清空。
poll的服务器模型伪码：

struct pollfd fds[POLL_LEN];
unsigned int nfds=0;
fds[0].fd=server_sockfd;
fds[0].events=POLLIN|POLLPRI;
nfds++;
while{
  res=poll(fds,nfds,-1);
  if(fds[0].revents&(POLLIN|POLLPRI)){执行accept并加入fds中,if(--res<=0)continue}
  循环之后的fds，if(fds[i].revents&(POLLIN|POLLERR )){操作略if(--res<=0)continue}
}

三、epoll模型

epoll阻塞操作的原型：

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout)
epoll引入了新的结构epoll_event。
typedef union epoll_data {
                 void *ptr;
                 int fd;
                 __uint32_t u32;
                 __uint64_t u64;
            } epoll_data_t;

            struct epoll_event {
                 __uint32_t events;  /**//* Epoll events */
                 epoll_data_t data;  /**//* User data variable */
            };

与以上模型的优点：
（1）它保留了poll的两个相对与select的优点
（2）epoll_wait的参数events作为出参，直接返回了有事件发生的fd，epoll_wait的返回值既是发生事件的个数，省略了poll中返回之后的循环操作。
（3）不再象select、poll一样将标识符局限于fd，epoll中可以将标识符扩大为指针，大大增加了epoll模型下的灵活性。
epoll的服务器模型伪码：

epollfd=epoll_create(EPOLL_LEN);
epoll_ctl(epollfd,EPOLL_CTL_ADD,server_sockfd,&ev)
struct epoll_event events[EPOLL_MAX_EVENT];
while
{
nfds=epoll_wait(epollfd,events,EPOLL_MAX_EVENT,-1);
循环nfds，是server_sockfd则accept，否则执行响应操作
}

epoll使用中的问题：
（1）epoll_ctl的EPOLL_CTL_DEL操作中，最后一个参数是无意义的，但是在小版本号过低的2.6内核下要求最后一个参数一定非NULL，否则返回失败，并且返回的errno在man epoll_ctl中不存在，因此安全期间，保证epoll_ctl的最后一个参数总非NULLL。
（2）如果一个fd（比如管道）的事件导致了另一个fd2的删除，则必须扫描返回结果集中是否有fd2，有则在结果集中删除，避免冲突。
（3）有文章说epoll在G网环境下性能会低于poll/select，看有些测试，给出的拐点在2w/s并发之后，我本人的工作范围不可能达到这么高的并发，个人在测试性能的时候最大也是取的1w/s的并发，一个是因为系统单进程允许打开的文件描述符最大值，4w的数字太高了，另一个就是我这边服务器的性能达不到那么高的性能，极限1.7w/s的响应，那测试的数据竟然在2w并发的时候还有2w的响应，不知道是什么硬件配置。或许等有了G网的环境，会关注epoll高并发下的性能下降

。
（4）epoll的LT和ET性能的差异，我测试的数据表明两者性能相当，“使用epoll就是为了高性能，就是要使用ET模式”这个说法是站不住脚的。个人倾向于使用LT模式，编程简单、安全。

四、port模型
port则和epoll非常接近，不需要前后的两次扫描，直接返回有事件的结果，可以象epoll一样绑定指针，不同点是
（1）epoll可以返回多个事件，而port一次只返回一个（port_getn可以返回多个，但是在不到指定的n值时，等待直到达到n个）
（2）port返回的结果会自动port_dissociate，如果要再次监控，需要重新port_associate
这个就不多说了。

可以看出select-->poll-->epoll/port的演化路线：
（1）从readset、writeset等分离到 将读写事件集中到统一的结构
（2）从阻塞操作前后的两次循环 到 之后的一次循环  到精确返回有事件发生的fd
（3）从只能绑定fd信息，到可以绑定指针结构信息

五、抽象接口
综合以上多路复用函数的特点，可以进行统一的封装，这里给出我封装的接口，也算是给一个思路：

 virtual int init()=0;
 virtual int wait()=0;
 virtual void * next_result()=0;
 virtual void delete_from_results(void * data)=0;
 virtual void * get_data(void * event)=0;
 virtual int get_event(void * event)=0;
 virtual int add_data(int fd,XPollData * data)=0;
 virtual int delete_data(int fd,XPollData *data)=0;
 virtual int change_data(int fd,XPollData *data)=0;
 virtual int reset_data(int fd,XPollData *data)=0;

使用的时候就是先init，再wait，再循环执行next_result直到空，每个result，使用get_data和get_event挨个处理，如果某个fd引起另一个fd关闭，调delete_from_results（除epoll，其它都直接return），处理完reset_data（select和port用，poll/epoll直接return）。

timeradd(struct timeval *p1,struct timeval *p2,struct timeval *result);
timersub(struct timeval *p1,struct timeval *p2,struct timeval *result);
timercmp(struct timeval *p1,struct timeval *p2,operator op);

对struct timespec同样有timespecadd/timespecsub/timespeccmp，另外还有两者的转换宏，使用函数原型的方式理解就是：

TIMEVAL_TO_TIMESPEC(struct timeval *p1,struct timespec *result);
TIMESPEC_TO_TIMEVAL(struct timespec *p1,struct timeval *result);

如果系统的头文件中找不到，也可以自己实现下，明白这两个结构的细节结构，实现也很简单，这里拿timeradd举例，其它不说了。

#define    timeradd(tvp, uvp, vvp)                        \
    do {                                \
        (vvp)->tv_sec = (tvp)->tv_sec + (uvp)->tv_sec;        \
        (vvp)->tv_usec = (tvp)->tv_usec + (uvp)->tv_usec;    \
        if ((vvp)->tv_usec >= 1000000) {            \
            (vvp)->tv_sec++;                \
            (vvp)->tv_usec -= 1000000;            \
        }                            \
    } while (0)

while(manager->m_state==TIMER_MANAGER_START){
    tm.tv_sec=manager->m_interval.tv_sec;
    tm.tv_usec=manager->m_interval.tv_usec;
    while(select(0,0,0,0,&tm)<0&&errno==EINTR);
    gettimeofday(&now,0);
/**//*加上误差补偿时间*/
    timeradd(&now,&manager－>m_repair,&stand);
    pthread_mutex_lock(&manager->m_mutex);
    TAILQ_FOREACH(item, &(manager->list_), entry_){
/**//*取修正后的时间逐个和定时器中的超时时间点相比，遇到不超时对象则退出扫描*/
        if(timercmp(&item->m_endtime,&stand,<)){
                 if(item->m_func)
                        item->m_func(item,item->m_data);
             if(item->m_type==CTimer::TIMER_ONCE){
                       manager->remove_timer_(item);
                      item->m_state=CTimer::TIMER_TIMEOUT;
                }
            else if(item->m_type==CTimer::TIMER_CIRCLE){
/**//*循环性的要保证链表的顺序性，如要重新插入，保存entry_的原因，是执行新的插入后不要影响当前进行的循环*/
                    tmpTimer.entry_=item->entry_;
                    manager->remove_timer_(item);
                    manager->add_timer_(item);
                   item=&tmpTimer;}
                 }
          else break;
        }
        pthread_mutex_unlock(&manager->m_mutex);
    }

pthread_mutex_t mux=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;

//wait操作：
pthread_mutex_lock(&mux);
pthread_cond_wait(&cond,&mux);//睡眠前将内部会执行pthread_mutex_unlock，醒来时内部会执行pthread_mutex_lock
pthread_mutex_unlock(&mux);

//signal操作
pthread_mutex_lock(&mux);
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mux);

（4）sem_t类型的操作。同样:man sem_init 这个系列一般是用不到的，太重量级了，也是最强大的一种。

二、linux2.6内核的线程库。2.6内核的默认安装的是redhat公司的NPTL（原生posix线程库），以前内核安装的是LinuxThreads库，两者的简单介绍可以看http://www.ibm.com/developerworks/cn/linux/l-threading.html。不过对于应用者，分析两者的区别和优劣也没什么大意义。这里特别提下NPTL的futex机制。借助该机制，pthread_mutex的性能大大提高，只要不进入竞争态，进程就不会陷入内核态，则意味着可以放心使用pthread_mutex，只要不进入竞争态，就象int i;i++之类的语句一样高效。这点可以自己写示例程序，通过strace -c 跟踪进程的系统调用得以证实，另外还可以证实总是进入内核态的操作有pthread_cond_signal和sem_post。

三、实用的线程消息队列实现。下面设计一个具有线程消息队列的线程封装类。通过上面的分析，我们可以有如下结论：
（1）减少pthread_cond_signal和sem_post的调用，只在有必要的时候调用；
（2）尽量避免pthread_mutex进入竞争态。增大消息队列的大小，可以有效减少竞态条件的出现。

下面给出一个实用的线程消息队列的实现类，这个类也将是以后《网络模型》文章中用到的线程消息队列类，代码注释请看对私有属性的注释：

class CThreadQueue
{

public:
    CThreadQueue(int queueSize=1024):
        sizeQueue(queueSize),lput(0),lget(0),nFullThread(0),nEmptyThread(0),nData(0)
    {
        pthread_mutex_init(&mux,0);
        pthread_cond_init(&condGet,0);
        pthread_cond_init(&condPut,0);
        buffer=new (void *)[sizeQueue];
    }
    virtual ~CThreadQueue()
    {
        delete[] buffer;
    }
    void * getq()
    {
        void *data;
        pthread_mutex_lock(&mux);
        while(lget==lput&&nData==0)//此处循环判断的原因如下：假设2个线程在getq阻塞，然后两者都被激活，而其中一个线程运行比较块，快速消耗了2个数据，另一个线程醒来的时候已经没有新数据可以消耗了。另一点，man pthread_cond_wait可以看到，该函数可以被信号中断返回，此时返回EINTR。为避免以上任何一点，都必须醒来后再次判断睡眠条件。
        {
            nEmptyThread++;
            pthread_cond_wait(&condGet,&mux);
            nEmptyThread--;     
        }
            
        data=buffer[lget++];
        nData--;
        if(lget==sizeQueue)
        {
            lget=0;
        }
        if(nFullThread) //必要时才进行signal操作，勿总是signal
        {
            pthread_cond_signal(&condPut);    
        }
        pthread_mutex_unlock(&mux);
        return data;
    }
    void putq(void *data)
    {
        pthread_mutex_lock(&mux);
        while(lput==lget&&nData)
        { 
            nFullThread++;
            pthread_cond_wait(&condPut,&mux);
            nFullThread--;
        }
        buffer[lput++]=data;
        nData++;
        if(lput==sizeQueue)
        {
            lput=0;
        }
        if(nEmptyThread)
        {
            pthread_cond_signal(&condGet);
        }
<