muduo源码分析之EventLoop::runInLoop()函数
2016-11-27 21:22
495 查看
前面所学的一些内容,从最早的什么都不做的
输出结果如下所示:
success write to efd, write 8 bytes(4) at 1328805612s 21939us
success read from efd, read 8 bytes(4) at 1328805612s 21997us
success write to efd, write 8 bytes(4) at 1328805613s 22247us
success read from efd, read 8 bytes(4) at 1328805613s 22287us
success write to efd, write 8 bytes(4) at 1328805614s 22462us
success read from efd, read 8 bytes(4) at 1328805614s 22503us
success write to efd, write 8 bytes(4) at 1328805615s 22688us
success read from efd, read 8 bytes(4) at 1328805615s 22726us
success write to efd, write 8 bytes(4) at 1328805616s 22973us
success read from efd, read 8 bytes(4) at 1328805616s 23007us
epoll wait timed out
上述例子,首先使用eventfd创建描述符,并在线程里面使用epoll管理这个描述符,当在主线程中write时,线程中的epoll返回,描述符可读。
不难发现,通过eventfd创建的描述符,读/写大小为
这个函数很简单:
比如说现在线程1的EventLoop对象正在loop中,并且没有可读的描述符,那么它底层的poll调用将一直不停的timeout(用strace调试一个什么都不做的EventLoop可以很清楚的看到)。那么runInLoop()顾名思义,我们要怎么在这个loop中去执行用户回调呢? 基于我们的前提是线程1已经正在loop,因此,通过某些方式,我们从其他线程获得线程1中EventLoop对象的指针,并在该线程中执行其runInLoop方法,此时进入上述函数的eles分支:
我们将这些函数放在一个vector中,事宜的时候唤醒正在loop中的线程,再来看看被唤醒后线程做的事情:
由于EventFd可读,poll返回,线程1将执行doPendingFunctors,该函数就是将上述vector中的回调执行。
这里通过EventLoopThread对象创建线程,并在线程中创建了EventLoop对象,并通过startLoop()将EventLoop对象的指针返回给主线程,并在主线程中调用runInLoop实现了跨线程调用的例子。
这里需要关注的是EventLoopThread的工作原理。
在构造函数中,先绑定了一个线程函数,我们来看这个函数:
当线程启动时,将由loop_成员变量获取EventLoop对象的指针,并唤醒条件变量:
在startLoop()中启动线程,然后startLoop将wait(),直到threadFunc对其唤醒,之后startLoop将返回loop_,主线程就获得了EventLoop的指针。这么做的原因在于,可以保证在threadFunc中loop_获得loop对象后才唤醒startLoop()对指针进行返回,可以保证loop_的有效性。
2.http://blog.csdn.net/yusiguyuan/article/details/15026941
3.http://blog.csdn.net/jnu_simba/article/details/14517953
EventLoop开始,到后面的定时器,功能不断在丰富,不过一直都是单线程下的。也就是说
EventLoop对象在主线程中进行事件循环。今天花了一天时间所学习的
EventLoop::runInLoop()就打开muduo多线程编程的大门。
1.eventfd唤醒线程
先来看看这个eventfd的用法,直接上示例:#include <stdio.h> #include <unistd.h> #include <sys/time.h> #include <stdint.h> #include <pthread.h> #include <sys/eventfd.h> #include <sys/epoll.h> int efd = -1; void *read_thread(void *dummy) { int ret = 0; uint64_t count = 0; int ep_fd = -1; struct epoll_event events[10]; if (efd < 0) { printf("efd not inited.\n"); goto fail; } ep_fd = epoll_create(1024); if (ep_fd < 0) { perror("epoll_create fail: "); goto fail; } { struct epoll_event read_event; read_event.events = EPOLLHUP | EPOLLERR | EPOLLIN; read_event.data.fd = efd; ret = epoll_ctl(ep_fd, EPOLL_CTL_ADD, efd, &read_event); if (ret < 0) { perror("epoll ctl failed:"); goto fail; } } while (1) { ret = epoll_wait(ep_fd, &events[0], 10, 5000); if (ret > 0) { int i = 0; for (; i < ret; i++) { if (events[i].events & EPOLLHUP) { printf("epoll eventfd has epoll hup.\n"); goto fail; } else if (events[i].events & EPOLLERR) { printf("epoll eventfd has epoll error.\n"); goto fail; } else if (events[i].events & EPOLLIN) { int event_fd = events[i].data.fd; ret = read(event_fd, &count, sizeof(count)); if (ret < 0) { perror("read fail:"); goto fail; } else { struct timeval tv; gettimeofday(&tv, NULL); printf("success read from efd, read %d bytes(%llu) at %lds %ldus\n", ret, count, tv.tv_sec, tv.tv_usec); } } } } else if (ret == 0) { /* time out */ printf("epoll wait timed out.\n"); break; } else { perror("epoll wait error:"); goto fail; } } fail: if (ep_fd >= 0) { close(ep_fd); ep_fd = -1; } return NULL; } int main(int argc, char *argv[]) { pthread_t pid = 0; uint64_t count = 0; int ret = 0; int i = 0; efd = eventfd(0, 0); if (efd < 0) { perror("eventfd failed."); goto fail; } ret = pthread_create(&pid, NULL, read_thread, NULL); if (ret < 0) { perror("pthread create:"); goto fail; } for (i = 0; i < 5; i++) { count = 4; ret = write(efd, &count, sizeof(count)); if (ret < 0) { perror("write event fd fail:"); goto fail; } else { struct timeval tv; gettimeofday(&tv, NULL); printf("success write to efd, write %d bytes(%llu) at %lds %ldus\n", ret, count, tv.tv_sec, tv.tv_usec); } sleep(1); } fail: if (0 != pid) { pthread_join(pid, NULL); pid = 0; } if (efd >= 0) { close(efd); efd = -1; } return ret; }
输出结果如下所示:
success write to efd, write 8 bytes(4) at 1328805612s 21939us
success read from efd, read 8 bytes(4) at 1328805612s 21997us
success write to efd, write 8 bytes(4) at 1328805613s 22247us
success read from efd, read 8 bytes(4) at 1328805613s 22287us
success write to efd, write 8 bytes(4) at 1328805614s 22462us
success read from efd, read 8 bytes(4) at 1328805614s 22503us
success write to efd, write 8 bytes(4) at 1328805615s 22688us
success read from efd, read 8 bytes(4) at 1328805615s 22726us
success write to efd, write 8 bytes(4) at 1328805616s 22973us
success read from efd, read 8 bytes(4) at 1328805616s 23007us
epoll wait timed out
上述例子,首先使用eventfd创建描述符,并在线程里面使用epoll管理这个描述符,当在主线程中write时,线程中的epoll返回,描述符可读。
不难发现,通过eventfd创建的描述符,读/写大小为
sizeof(uint_64)数据,就可以完成两个线程间的唤醒。比如上述例子,由于
epoll_wait()的等待,
pthread_create出来的线程阻塞,在主线程中,通过往eventfd中write数据,使描述符可读,epoll返回,这就达到了唤醒的目的。
2. EventLoop::runInLoop()函数
这个函数的效果就是在loop中执行某个用户回调。这个函数很简单:
void EventLoop::runInLoop(const Functor& cb) { if (isInLoopThread())//如果是在本线程中 { cb(); } else// { queueInLoop(cb); } }
比如说现在线程1的EventLoop对象正在loop中,并且没有可读的描述符,那么它底层的poll调用将一直不停的timeout(用strace调试一个什么都不做的EventLoop可以很清楚的看到)。那么runInLoop()顾名思义,我们要怎么在这个loop中去执行用户回调呢? 基于我们的前提是线程1已经正在loop,因此,通过某些方式,我们从其他线程获得线程1中EventLoop对象的指针,并在该线程中执行其runInLoop方法,此时进入上述函数的eles分支:
void EventLoop::queueInLoop(const Functor& cb) { { MutexLockGuard lock(mutex_); pendingFunctors_.push_back(cb); } // 调用queueInLoop的线程不是当前IO线程则需要唤醒当前IO线程,才能及时执行doPendingFunctors(); // 或者调用queueInLoop的线程是当前IO线程(比如在doPendingFunctors()中执行functors[i]() 时又调用了queueInLoop()) // 并且此时正在调用pending functor,需要唤醒当前IO线程 // 因为在此时doPendingFunctors() 过程中又添加了任务,故循环回去poll的时候需要被唤醒返回,进而继续执行doPendingFunctors() // 只有当前IO线程的事件回调中调用queueInLoop才不需要唤醒 // 即在handleEvent()中调用queueInLoop 不需要唤醒,因为接下来马上就会执行doPendingFunctors(); if (!isInLoopThread() || callingPendingFunctors_) { wakeup(); } }
我们将这些函数放在一个vector中,事宜的时候唤醒正在loop中的线程,再来看看被唤醒后线程做的事情:
void EventLoop::loop() { assert(!looping_); assertInLoopThread(); looping_ = true; quit_ = false; while (!quit_) { activeChannels_.clear(); pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); for (ChannelList::iterator it = activeChannels_.begin(); it != activeChannels_.end(); ++it) { (*it)->handleEvent(); } doPendingFunctors();//执行回调 } LOG_TRACE << "EventLoop " << this << " stop looping"; looping_ = false; }
由于EventFd可读,poll返回,线程1将执行doPendingFunctors,该函数就是将上述vector中的回调执行。
3. 一个示例
#include "EventLoop.h" #include "EventLoopThread.h" #include <stdio.h> void runInThread() { printf("zxzxrunInThread(): pid = %d, tid = %d\n", getpid(), muduo::CurrentThread::tid()); } int main() { printf("main(): pid = %d, tid = %d\n", getpid(), muduo::CurrentThread::tid()); muduo::EventLoopThread loopThread; muduo::EventLoop* loop = loopThread.startLoop(); loop->runInLoop(runInThread); sleep(3); loop->quit(); printf("exit main().\n"); }
这里通过EventLoopThread对象创建线程,并在线程中创建了EventLoop对象,并通过startLoop()将EventLoop对象的指针返回给主线程,并在主线程中调用runInLoop实现了跨线程调用的例子。
这里需要关注的是EventLoopThread的工作原理。
EventLoopThread::EventLoopThread() : loop_(NULL), exiting_(false), thread_(boost::bind(&EventLoopThread::threadFunc, this)),//先绑定线程函数 mutex_(), cond_(mutex_) { }
在构造函数中,先绑定了一个线程函数,我们来看这个函数:
void EventLoopThread::threadFunc() { EventLoop loop; { MutexLockGuard lock(mutex_); loop_ = &loop;//获得EventLoop对象的指针 cond_.notify();//唤醒条件变量 } loop.loop(); //assert(exiting_); }
当线程启动时,将由loop_成员变量获取EventLoop对象的指针,并唤醒条件变量:
EventLoop* EventLoopThread::startLoop() { assert(!thread_.started()); thread_.start();//线程启动 { MutexLockGuard lock(mutex_); while (loop_ == NULL)//条件变量 { cond_.wait(); } } return loop_; }
在startLoop()中启动线程,然后startLoop将wait(),直到threadFunc对其唤醒,之后startLoop将返回loop_,主线程就获得了EventLoop的指针。这么做的原因在于,可以保证在threadFunc中loop_获得loop对象后才唤醒startLoop()对指针进行返回,可以保证loop_的有效性。
4.参考
1.Linux多线程服务端编程使用muduoC++网络库2.http://blog.csdn.net/yusiguyuan/article/details/15026941
3.http://blog.csdn.net/jnu_simba/article/details/14517953
相关文章推荐
- Muduo网络库源码分析(三)线程间使用eventfd通信和EventLoop::runInLoop系列函数
- muduo源码分析---EventLoopThread
- muduo源码分析之EventLoop、Channel、Poller的实现
- Netty学习之旅------再谈线程模型之源码分析NioEventLoopGroup、SingleThreadEventExecutor
- WPA_SUPPLICANT源码分析(1):EVENT LOOP的实现
- Netty源码分析之NioEventLoop(转)
- Libevent源码分析-----与event相关的一些函数和操作
- Netty源码分析之EventLoop相关结构分析
- libevent源码分析--event_add()函数
- 【Netty4.X】Netty源码分析之NioEventLoop(六)
- muduo网络库学习笔记(11):有用的runInLoop()函数
- muduo网络库学习之EventLoop(二):进程(线程)wait/notify 和 EventLoop::runInLoop
- Libevent源码分析-----与event相关的一些函数和操作
- libevent源码分析--event_init()函数
- libevent源码分析(5)--2.1.8--libevent配置信息对象struct event_config的申请和释放函数分析
- uboot下函数run_main_loop分析
- Netty源码学习——EventLoopGroup原理:NioEventLoopGroup分析
- netty4源码分析——NioEventLoop
- muduo::EventLoopThread、EventLoopThreadPool分析
- libevent源码分析(6)--2.1.8--创建和释放libevent句柄event_base的相关函数