一起来写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本来就是cpp的std的一个备选库.
为什么要使用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::funciton和boost::bind这对神器,因为我之前也压根就没有这样编过码.
如果不用boost::funciton和boost::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,不出问题才怪.
解决方案是有的,那就是EPOLL的ONESHOT参数.不过那是下一个版本的故事啦.
和之前类似的,代码在这里:https://github.com/lishuhuakai/Spweb
