美文网首页服务器开发
一起来写web server 08 -- 多线程+非阻塞IO+e

一起来写web server 08 -- 多线程+非阻塞IO+e

作者: Yihulee | 来源:发表于2016-11-01 22:55 被阅读33次

    一起来写web server 08 -- 多线程+非阻塞IO+epoll


    到了多线程,一些东西就变得耐人寻味了.

    这个版本是在前面单线程epoll的基础上引入了线程池,当然不是前面玩具一样的线程池,而是一个通用的组件,生产者消费者队列.

    生产者消费者队列

    生产者消费者问题是操作系统中一个很经典的同步互斥问题,已经有了很不错的解决方案,将它的解决方案拓展一下,就可以用于我们的实践啦.

    我自己写了一个生产者消费者的队列,然后发现muduo中已经内置了这种模型,而且使用起来比我写的更加顺手,所以我就引用它的实现,我这里稍微来讲解一下它的实现,然后我会顺带讲解一下我的思路.

    muduo库的生产者消费者模型

    这是ThreadPool类的一个声明:

    class ThreadPool : noncopyable
    {
    public:
        typedef boost::function<void()> Task; /* 需要执行的任务 */
    private:
        bool isFull();
        Task take();
        size_t queueSize();
        int threadNum_; /* 线程的数目 */
        int maxQueueSize_;
        std::list<Task> queue_; /* 工作队列 */
        MutexLock mutex_;
        Condition notEmpty_;
        Condition notFull_;
    };
    

    这里的boost::function其实在cpp 11标准中已经加入了,你如果没有安装boost库的话,可以缓存std的版本,效果是一样的.因为boost本来就是cppstd的一个备选库.

    为什么要使用boost::function不用我多说,你可以查看这里:http://blog.csdn.net/solstice/article/details/3066268

    我们来看一下代码的实现,首先是构造函数:

    ThreadPool::ThreadPool(int threadNum, int maxQueueSize)
        : threadNum_(threadNum)
        , maxQueueSize_(maxQueueSize)
        , mutex_()
        , notEmpty_(mutex_)
        , notFull_(mutex_)
    
    {
        assert(threadNum >= 1 && maxQueueSize >= 1);
        /* 接下来要构建threadNum个线程 */
        pthread_t tid_t;
        for (int i = 0; i < threadNum; i++) {
            Pthread_create(&tid_t, NULL, startThread, this);
        }
    }
    

    这里ThreadPool有两个条件变量,一个是notEmpty_,一个是notFull_,构造函数接受两个参数,一个是线程的数目,一个是最大的队列的大小.

    接下来是所有的线程都运行的函数startThread:

    void* ThreadPool::startThread(void* obj)
    { /* 工作者线程 */
        Pthread_detach(Pthread_self());
        ThreadPool* pool = static_cast<ThreadPool*>(obj);
        pool->run();
        return pool;
    }
    

    它们都开始调用run函数:

    void ThreadPool::run()
    {
        for ( ; ; ) { /* 一直运行下去 */
            Task task(take());
            if (task) {
                //mylog("task run!");
                task();
            }
            //mylog("task over!");
        }
    }
    

    run函数非常简单,就是不断从队列中取出任务,然后运行任务,没有任务的话,会阻塞在那里.

    我们来看take函数:

    ThreadPool::Task ThreadPool::take()
    {
        MutexLockGuard lock(mutex_); /* 加锁 */
        while (queue_.empty()) { /* 如果队列为空 */
            notEmpty_.wait(); /* 等待 */
        }
        Task task;
        if (!queue_.empty()) {
            task = queue_.front();
            queue_.pop_front();
            if (maxQueueSize_ > 0) { /* 通知生产者队列有空位置了 */
                notFull_.notify();
            }
        }
        //mylog("threadpool take 1 task!");
        return task;
    }
    

    对于生产者而言,有一个非常重要的函数,那就是append:

    bool ThreadPool::append(Task&& task)
    { /* 使用了右值引用 */
        {
            MutexLockGuard lock(mutex_); /* 首先加锁 */
            while (isFull()) { /* 如果队列已满 */
                notFull_.wait(); /* 等待queue有空闲位置 */
            }
            assert(!isFull());
            queue_.push_back(std::move(task)); /* 直接用move语义,提高了效率 */
            //mylog("put task onto queue!");
        }
        notEmpty_.notify(); /* 通知消费者有任务可做了 */
    }
    

    生产者消费者队列的代码就是这么简单,但是muduo库写的确实很漂亮.

    我的思路

    其实代码基本上和前面的类似,不同的是,我压根就没有考虑过使用boost::funcitonboost::bind这对神器,因为我之前也压根就没有这样编过码.

    如果不用boost::funcitonboost::bind这两样东西,我们要实现类似的代码的话,可能的一个解决方案是使用模版(template).

    队列里面放的是T类型,然后消费者取出一个T类型,调用T类型的一个run或者别的什么不带参数的方法.这样以来,对T类型就有了限制,要求T类型必须实现run之类的方法.

    而且代码变得不太容易读.加了模版的玩意总是不容易读,不是吗?所以要积极使用cpp的新特性.

    主程序变成了生产者

    这一次的代码变得简洁多了,

    int main(int argc, char *argv[])
    {
        int listenfd = Open_listenfd(8080); /* 8080号端口监听 */
        epoll_event events[MAXEVENTNUM];
        sockaddr clnaddr;
        socklen_t clnlen = sizeof(clnaddr);
    
        block_sigpipe(); /* 首先要将SIGPIPE消息阻塞掉 */
    
        int epollfd = Epoll_create(1024); /* 10基本上没有什么用处 */
        addfd(epollfd, listenfd, false); /* epollfd要监听listenfd上的可读事件 */
        ThreadPool pools(10, 30000); /* 10个线程,300个任务 */
        HttpHandle::setEpollfd(epollfd);
        HttpHandle handle[2000];
    
        for ( ; ;) {
            int eventnum = Epoll_wait(epollfd, events, MAXEVENTNUM, -1);
            for (int i = 0; i < eventnum; ++i) {
                int sockfd = events[i].data.fd;
                if (sockfd == listenfd) { /* 有连接到来 */
                    //mylog("connection comes!");
                    for ( ; ; ) {
                        int connfd = accept(listenfd, &clnaddr, &clnlen);
                        if (connfd == -1) {
                            if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) { /* 将连接已经建立完了 */
                                break;
                            }
                            unix_error("accept error");
                        }
                        handle[connfd].init(connfd); /* 初始化 */
                        addfd(epollfd, connfd, false); /* 加入监听 */
                    }
                }
                else { /* 有数据可读或者可写 */
                    pools.append(boost::bind(&HttpHandle::process, &handle[sockfd]));
                }
            }
        }
        return 0;
    }
    

    注意最后的一句boost::bind(&HttpHandle::process, &handle[sockfd]),直接将对象往函数上一绑定,就往队列里面扔.非常爽.

    这一次,我们终于将SIGPIPE消息给忽略掉了,主要是调用下面这个函数:

    void block_sigpipe()
    {
        sigset_t signal_mask;
        sigemptyset(&signal_mask);
        sigaddset(&signal_mask, SIGPIPE);
        int rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
        if (rc != 0) {
            printf("block sigpipe error\n");
        }
    }
    

    shared_ptr并不是线程安全的

    正如文章开头所讲的,多线程一来,很多事情就变得莫名奇妙了,比如说shared_ptr,因为这个玩意的线程不安全性,我调了半天bug,才发现原来是Cache的查找函数出了问题,下面是修改过后的线程安全版本的函数:

    /* 线程安全版本的getFileAddr */
        void getFileAddr(std::string fileName, int fileSize, boost::shared_ptr<FileInfo>& ptr) {
            /*-
            * shared_ptr并不是线程安全的,对其的读写都需要加锁.
            */
            MutexLockGuard lock(mutex_);
            if (cache_.end() != cache_.find(fileName)) { /* 如果在cache中找到了 */
                ptr = cache_[fileName];
                return;
            }
            if (cache_.size() >= MAX_CACHE_SIZE) { /* 文件数目过多,需要删除一个元素 */
                cache_.erase(cache_.begin()); /* 直接移除掉最前一个元素 */
            }
            boost::shared_ptr<FileInfo> fileInfo(new FileInfo(fileName, fileSize));
            cache_[fileName] = fileInfo;
            ptr = std::move(fileInfo); /* 直接使用move语义 */
        }
    

    至于为什么不安全,可以查看这里,写的再好不过了:http://blog.csdn.net/solstice/article/details/8547547

    多线程的调试

    原谅我到了这接近尾声的时候,才提起多线程的调试,首先要说一句的是,多线程真的不太好调,因为很难重现错误,但是我在这里稍稍介绍一下我的技巧.

    打印

    打印算是屡试不爽的一种方法,对于我们这个简陋的web server,我封装了一个日志函数mylog:

    void mlog(pthread_t tid, const char *fileName, int lineNum, const char *func, const char *log_str, ...)
    {
        va_list vArgList; //定义一个va_list型的变量,这个变量是指向参数的指针.
        char buf[1024];
        va_start(vArgList, log_str); //用va_start宏初始化变量,这个宏的第二个参数是第一个可变参数的前一个参数,是一个固定的参数
        vsnprintf(buf, 1024, log_str, vArgList); //注意,不要漏掉前面的_
        va_end(vArgList);  //用va_end宏结束可变参数的获取
        printf("%lu:%s:%d:%s --> %s\n", tid, fileName, lineNum, func, buf);
    }
    

    然后定义了一个宏,方便使用这个函数:

    #define mylog(formatPM, args...)\
      mlog(pthread_self(), __FILE__, __LINE__, __FUNCTION__,  (formatPM) , ##args)
    

    需要日志的时候,可以像printf函数一样使用:

    mylog("My simple web server! %d, %s\n", 1, "hello, workd!");
    

    这个宏展开后会调用mlog函数,打印出行,文件名,函数名等信息,对付我们这个小玩意足够了.

    用VS来调试

    VS其实也内置了线程的调试,你可以结合Visual Gdb一起来调试linux下的代码.一两个线程问题倒是不大,不过线程多了的话,这个玩意就不好调了,要我说,最好的方法还是分析日志.

    总结

    这个版本已经算是比较强劲的一个版本了,修复了前面的一些bug,但是引入了新的bug,这个bug我也是折腾了很久才弄出来.

    一般在单线程下不可能出现这样的bug,只有在多线程的条件下,这样的代码才变成了bug,正如前面见到的,每个HttpHandle处理一个连接,试想这样一种情形:客户端不知道因为什么原因,第一次发送了这样的数据:

    GET /
    

    隔了很短时间才会发送余下的数据.这时,第一次发送的数据正在被另外一个线程处理,在多线程条件下,对于第二次到来的数据,这个HttpHandle会交由另外一个线程处理,也就是说,有两个线程在不加锁地使用同一个HttpHandle,不出问题才怪.

    解决方案是有的,那就是EPOLLONESHOT参数.不过那是下一个版本的故事啦.

    和之前类似的,代码在这里:https://github.com/lishuhuakai/Spweb

    相关文章

      网友评论

        本文标题:一起来写web server 08 -- 多线程+非阻塞IO+e

        本文链接:https://www.haomeiwen.com/subject/lsfvuttx.html