阅读量:0
muduo源码阅读
Timestamp类，时间戳类
时间戳主要是获得时间
成员变量int64_t microSecondsSinceEpoch_ 是时间
now()函数是获得当前时间
toString()函数是把时间转换成人看的string
Timestamp Timestamp::now() { struct timeval tv; gettimeofday(&tv, NULL); int64_t seconds = tv.tv_sec; return Timestamp(seconds * kMicroSecondsPerSecond + tv.tv_usec); //把时间秒换算成毫秒 } string Timestamp::toString() const { char buf[32] = {0}; int64_t seconds = microSecondsSinceEpoch_ / kMicroSecondsPerSecond; int64_t microseconds = microSecondsSinceEpoch_ % kMicroSecondsPerSecond; snprintf(buf, sizeof(buf), "%" PRId64 ".%06" PRId64 "", seconds, microseconds); return buf; } Log日志类
有一个日志等级
enum LogLevel { TRACE, DEBUG, INFO, WARN, ERROR, FATAL, NUM_LOG_LEVELS, }; Channel类
Channel理解为通道，封装了sockfd和其感兴趣的event,如EPOLLIN、EPOLLOUT事件，还绑定了poller返回的具体事件。当拿到文件描述符时，就去拿对应的channel，当文件描述符发生读、写事件，系统就会去channel里拿对应的读、写回调函数。
channel与eventloop打交道，epoll与eventloop打交道
EventLoop* loop_; const int fd_; //因为是文件描述符绑定，所以必须有fd文件描述符 int events_; //注册的事件 int revents_; //epoll或者poll监听到的事件， int index_; // used by Poller. bool logHup_;
四个回调函数，以及四个方法设置回调函数
ReadEventCallback readCallback_; EventCallback writeCallback_; EventCallback closeCallback_; EventCallback errorCallback_; typedef std::function<void()> EventCallback; typedef std::function<void(Timestamp)> ReadEventCallback; void setReadCallback(ReadEventCallback cb) { readCallback_ = std::move(cb); } //std::move是移动，移动完之后，cb内容就消失了 void setWriteCallback(EventCallback cb) { writeCallback_ = std::move(cb); } void setCloseCallback(EventCallback cb) { closeCallback_ = std::move(cb); } void setErrorCallback(EventCallback cb) { errorCallback_ = std::move(cb); } 设置事件
void enableReading() { events_ |= kReadEvent; update(); } //update()类似epollctrl void disableReading() { events_ &= ~kReadEvent; update(); } void enableWriting() { events_ |= kWriteEvent; update(); } void disableWriting() { events_ &= ~kWriteEvent; update(); } void disableAll() { events_ = kNoneEvent; update(); } bool isWriting() const { return events_ & kWriteEvent; } bool isReading() const { return events_ & kReadEvent; } update()
loop_->updateChannel(this); //channel的update是调用eventloop的updateChannel //eventloop的updateChannel是调用poller的updateChannel poller_->updateChannel(channel); //epoller的updateChannel就是调用epollctrl来修改文件描述符的关键 //update(EPOLL_CTL_DEL, channel); ==》 //::epoll_ctl(epollfd_, operation, fd, &event) void EPollPoller::updateChannel(Channel* channel) { Poller::assertInLoopThread(); const int index = channel->index(); LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events() << " index = " << index; if (index == kNew || index == kDeleted) { // a new one, add with EPOLL_CTL_ADD int fd = channel->fd(); if (index == kNew) { assert(channels_.find(fd) == channels_.end()); channels_[fd] = channel; } else // index == kDeleted { assert(channels_.find(fd) != channels_.end()); assert(channels_[fd] == channel); } channel->set_index(kAdded); update(EPOLL_CTL_ADD, channel); } else { // update existing one with EPOLL_CTL_MOD/DEL int fd = channel->fd(); (void)fd; assert(channels_.find(fd) != channels_.end()); assert(channels_[fd] == channel); assert(index == kAdded); if (channel->isNoneEvent()) { update(EPOLL_CTL_DEL, channel); channel->set_index(kDeleted); } else { update(EPOLL_CTL_MOD, channel); } } } void EPollPoller::update(int operation, Channel* channel) { struct epoll_event event; memZero(&event, sizeof event); event.events = channel->events(); event.data.ptr = channel; int fd = channel->fd(); LOG_TRACE << "epoll_ctl op = " << operationToString(operation) << " fd = " << fd << " event = { " << channel->eventsToString() << " }"; if (::epoll_ctl(epollfd_, operation, fd, &event) < 0) { if (operation == EPOLL_CTL_DEL) { LOG_SYSERR << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd; } else { LOG_SYSFATAL << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd; } } } handleEvent()
Channel::handleEvent(Timestamp receiveTime) //eventloop里调用handleEvent //poller监听channel哪些事件发生了变化，然后上报给eventloop //handleEvent就是根据目前fd上发生的事件，然后去调用对应的回调函数
Poller类
poller是基类
typedef std::vector<Channel*> ChannelList; //放channel的 //三个纯虚函数，在底层实现 virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0; virtual void updateChannel(Channel* channel) = 0; virtual void removeChannel(Channel* channel) = 0; //是否包含channel virtual bool hasChannel(Channel* channel) const; // typedef std::map<int, Channel*> ChannelMap; //poller所指向的eventloop EventLoop* ownerLoop_;
epollpoller
Timestamp poll(int timeoutMs, ChannelList* activeChannels) override //在这个函数会调用void fillActiveChannels(int numEvents, ChannelList* activeChannels) const; //fillActiveChannels是填写活跃的连接，将epoll监测到的活跃事件返回给activeChannel里 //成员变量 int epollfd_; //静态成员变量，表示eventList的大小，后面可以扩容 static const int kInitEventListSize = 16; typedef std::vector<struct epoll_event> EventList; //epollfd在构造的时候使用epoll_create1创建 epollfd_(::epoll_create1(EPOLL_CLOEXEC)), //poll函数就是等待事件发生 Timestamp EPollPoller::poll(int timeoutMs, ChannelList* activeChannels) { LOG_TRACE << "fd total count " << channels_.size(); //调用epoll_wait来等待事件发生， int numEvents = ::epoll_wait(epollfd_, &*events_.begin(), static_cast<int>(events_.size()), timeoutMs); int savedErrno = errno; Timestamp now(Timestamp::now()); //如果有事件发生 if (numEvents > 0) { LOG_TRACE << numEvents << " events happened"; fillActiveChannels(numEvents, activeChannels); if (implicit_cast<size_t>(numEvents) == events_.size()) //扩容操作 { events_.resize(events_.size()*2); } } else if (numEvents == 0) { LOG_TRACE << "nothing happened"; } else { // error happens, log uncommon ones if (savedErrno != EINTR) { errno = savedErrno; LOG_SYSERR << "EPollPoller::poll()"; } } return now; } void EPollPoller::fillActiveChannels(int numEvents, ChannelList* activeChannels) const { assert(implicit_cast<size_t>(numEvents) <= events_.size()); for (int i = 0; i < numEvents; ++i) { //把事件取出来，事件和channel是绑定的，因此转化为channel，然后放入到activeChannels里 Channel* channel = static_cast<Channel*>(events_[i].data.ptr); #ifndef NDEBUG int fd = channel->fd(); ChannelMap::const_iterator it = channels_.find(fd); assert(it != channels_.end()); assert(it->second == channel); #endif channel->set_revents(events_[i].events); activeChannels->push_back(channel); } } const int kNew = -1; const int kAdded = 1; const int kDeleted = 2; //上面三个在channel是index void EPollPoller::updateChannel(Channel* channel) { Poller::assertInLoopThread(); const int index = channel->index(); LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events() << " index = " << index; if (index == kNew || index == kDeleted) { // a new one, add with EPOLL_CTL_ADD int fd = channel->fd(); if (index == kNew) { assert(channels_.find(fd) == channels_.end()); channels_[fd] = channel; } else // index == kDeleted { assert(channels_.find(fd) != channels_.end()); assert(channels_[fd] == channel); } channel->set_index(kAdded); update(EPOLL_CTL_ADD, channel); } else { // update existing one with EPOLL_CTL_MOD/DEL int fd = channel->fd(); (void)fd; assert(channels_.find(fd) != channels_.end()); assert(channels_[fd] == channel); assert(index == kAdded); if (channel->isNoneEvent()) { update(EPOLL_CTL_DEL, channel); channel->set_index(kDeleted); } else { update(EPOLL_CTL_MOD, channel); } } } EventLoop类
eventfd是专门用于事件通知的fd
其他线程可以调用eventloop，所以isInLoopThread就是判断是否在本线程里
// 把回调函数放入到队列里 void queueInLoop(Functor cb); //执行回调函数 void runInLoop(Functor cb); //执行上层的回调函数 void EventLoop::doPendingFunctors() //主要的函数loop，在循环里，用poller监测活跃的channel，监测到后，去执行对应的事件，最后调用上层的回调函数 void EventLoop::loop() { assert(!looping_); assertInLoopThread(); looping_ = true; quit_ = false; // FIXME: what if someone calls quit() before loop() ? LOG_TRACE << "EventLoop " << this << " start looping"; while (!quit_) { activeChannels_.clear(); pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); ++iteration_; if (Logger::logLevel() <= Logger::TRACE) { printActiveChannels(); } // TODO sort channel by priority eventHandling_ = true; for (Channel* channel : activeChannels_) { currentActiveChannel_ = channel; currentActiveChannel_->handleEvent(pollReturnTime_); } currentActiveChannel_ = NULL; eventHandling_ = false; doPendingFunctors(); } //这里说明一下，当执行doPendingFunctors();回调时，又加入到5个回调，如果不处理，那么下一次循环中，就会阻塞在poller_->poll //因此queueInLoop里就会wakeup()一下，往wakeupfd写一个事件，然后poller监测到后就不阻塞了 LOG_TRACE << "EventLoop " << this << " stop looping"; looping_ = false; } eventloop都一个threadid，这个threadid是通过currentthread::tid()获得的
wakeupfd是通过createEventfd()创建的
createEventfd()是用linux的eventfd创建，当监听到写事件时，就会唤醒wakeup线程
单reactor和多reactor，
如果是单reactor，上层接受连接、读写，都是在一个eventloop里执行的，因此直接执行回调函数
如果是多reactor，比如上层接受连接，然后下发给子reactor，读写关闭等回调是放到子reactor里，子reactor是通过queueInloop来实现的
void EventLoop::runInLoop(Functor cb) { if (isInLoopThread()) { cb(); } else { queueInLoop(std::move(cb)); } } //将上层的回调函数插入到vector的 pendingFunctors里 void EventLoop::queueInLoop(Functor cb) { { MutexLockGuard lock(mutex_); pendingFunctors_.push_back(std::move(cb)); } //当回调正在执行时，说明上面刚加入的回调是在下一轮被执行，而不是马上执行。因此需要唤醒一下，让下一次loop()里的poller->poll()不再阻塞 if (!isInLoopThread() || callingPendingFunctors_) { wakeup(); } } handleRead是wakeup的回调函数
//构造的时候设置了回调 wakeupChannel_->setReadCallback( std::bind(&EventLoop::handleRead, this)); // we are always reading the wakeupfd wakeupChannel_->enableReading(); void EventLoop::handleRead() { uint64_t one = 1; ssize_t n = sockets::read(wakeupFd_, &one, sizeof one); if (n != sizeof one) { LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8"; } } //唤醒就是向wakeupFd写事件，wakeupchannel就会发生读事件 void EventLoop::wakeup() { uint64_t one = 1; ssize_t n = sockets::write(wakeupFd_, &one, sizeof one); if (n != sizeof one) { LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8"; } } void EventLoop::doPendingFunctors() { std::vector<Functor> functors; //callingPendingFunctors_表示正在执行回调 callingPendingFunctors_ = true; { MutexLockGuard lock(mutex_); functors.swap(pendingFunctors_); //交换的方式减少锁的临界区域范围，提升效率 //如果是先加锁，把所有的回调函数都执行完，然后再解锁。可能会导致死锁，因为在queueInLoop里，会出现在执行回调函数过程中 //还加入回调，加入回调也是加锁，然后就进入死锁。一个线程获得了一个锁，想再次获得该锁，就会死锁。 //因此陈硕通过加锁交换，解锁，再唤醒的方式 } for (const Functor& functor : functors) { functor(); } callingPendingFunctors_ = false; } thread类
成员变量
using ThreadFunc = std::function<void()>; bool started_; bool joined_; std::shared_ptr<std::thread> thread_; pid_t tid_; // 在线程创建时再绑定 ThreadFunc func_; // 线程回调函数 std::string name_; static std::atomic_int numCreated_;
构造函数的时候，先把function放进去，start()的时候，才启动线程开启function
比较重要的是用信号量来保证线程创建的有序性，start()是启动线程
void Thread::start() // 一个Thread对象 记录的就是一个新线程的详细信息 { started_ = true; sem_t sem; sem_init(&sem, false, 0); // false指的是 不设置进程间共享 // 开启线程 thread_ = std::shared_ptr<std::thread>(new std::thread([&]() { tid_ = CurrentThread::tid(); // 获取线程的tid值 sem_post(&sem); func_(); // 开启一个新线程 专门执行该线程函数 })); // 这里必须等待获取上面新创建的线程的tid值 sem_wait(&sem); } EventLoopThread类
成员变量
using ThreadInitCallback = std::function<void(EventLoop *)>; EventLoop *loop_; bool exiting_; Thread thread_; std::mutex mutex_; // 互斥锁 std::condition_variable cond_; // 条件变量 ThreadInitCallback callback_;
EventLoopThread对应的function，这个函数会以回调函数的形式放入到thread_里
在构造函数里，有这样的一条语句。function是放进去了，不过没有执行，这就是回调函数的好处
thread_(std::bind(&EventLoopThread::threadFunc, this), name)
// 下面这个方法是在单独的新线程里运行的 void EventLoopThread::threadFunc() { EventLoop loop; // 创建一个独立的EventLoop对象 和上面的线程是一一对应的 级one loop per thread if (callback_) { callback_(&loop); } { std::unique_lock<std::mutex> lock(mutex_); loop_ = &loop; cond_.notify_one(); } loop.loop(); // 执行EventLoop的loop() 开启了底层的Poller的poll() //只有等eventloop quit()之后才会执行下面的语句，不然会一直阻塞在上面的语句 std::unique_lock<std::mutex> lock(mutex_); loop_ = nullptr; } 然后startloop()才是执行，会去调用thread的start(), thread的start()才是创建真正的线程
EventLoop *EventLoopThread::startLoop() { thread_.start(); // 启用底层线程Thread类对象thread_中通过start()创建的线程 //上面的start创建线程，是Thread类是调用std::thread来创建的，线程真正的创建在背后的内核，所以我们要一直等线程创建好。 EventLoop *loop = nullptr; { std::unique_lock<std::mutex> lock(mutex_); while(loop_ == nullptr) { //一直等待真正的线程创建好 cond_.wait(lock); } //线程创建好之后就可以返回对应的eventloop，说明eventloop的线程创建好了 loop = loop_; } return loop; } 总结：会把回调函数threadFunc先放入到thread_，但是这个时候还没有去创建线程，然后调用startLoop()函数时，再去调用thread的的start()，这个时候回去创建真正的线程。startLoop函数会返回创建好的eventloop指针，但是必须等线程创建好，这个时候就使用锁和条件变量，在回调函数threadFunc里等eventloop创建好后，会使用cond.notify_one();通知startloop线程创建好了*
EventLoopThreadPool
成员变量
EventLoop *baseLoop_; // 用户使用muduo创建的loop 如果线程数为1 那直接使用用户创建的loop 否则创建多EventLoop std::string name_; bool started_; int numThreads_; int next_; // 轮询的下标 std::vector<std::unique_ptr<EventLoopThread>> threads_; //线程vector std::vector<EventLoop *> loops_; //EventLoop的vector
void EventLoopThreadPool::start(const ThreadInitCallback &cb) { started_ = true; for(int i = 0; i < numThreads_; ++i) { char buf[name_.size() + 32]; snprintf(buf, sizeof buf, "%s%d", name_.c_str(), i); EventLoopThread *t = new EventLoopThread(cb, buf); //绑定回调函数cb threads_.push_back(std::unique_ptr<EventLoopThread>(t)); loops_.push_back(t->startLoop()); // 底层创建线程 绑定一个新的EventLoop 并返回该loop的地址 } if(numThreads_ == 0 && cb) // 整个服务端只有一个线程运行baseLoop { cb(baseLoop_); } } 这个baseloop就是一开始程序运行的那个eventloop
// 如果工作在多线程中，baseLoop_(mainLoop)会默认以轮询的方式分配Channel给subLoop EventLoop *EventLoopThreadPool::getNextLoop() { EventLoop *loop = baseLoop_; // 如果只设置一个线程 也就是只有一个mainReactor 无subReactor 那么轮询只有一个线程 getNextLoop()每次都返回当前的baseLoop_ if(!loops_.empty()) // 通过轮询获取下一个处理事件的loop { loop = loops_[next_]; ++next_; if(next_ >= loops_.size()) { next_ = 0; } } return loop; } getAllLoops就是返回所有的eventloop
InetAddress类
主要是封装sockaddr_in，封装地址、端口等等
还有输出地址、端口
Socket类
主要是封装文件描述符socket
提供socket相关的操作函数
bind listen accept
int fd() const { return sockfd_; } void bindAddress(const InetAddress &localaddr); void listen(); int accept(InetAddress *peeraddr); void shutdownWrite(); void setTcpNoDelay(bool on); void setReuseAddr(bool on); void setReusePort(bool on); void setKeepAlive(bool on); Acceptor类
Acceptor类用来接受连接并处理连接
成员变量EventLoop是baseLoop，在TcpServer里会传入baseLoop给Acceptor
using NewConnectionCallback = std::function<void(int sockfd, const InetAddress &)>; EventLoop *loop_; // Acceptor用的就是用户定义的那个baseLoop 也称作mainLoop Socket acceptSocket_; Channel acceptChannel_; NewConnectionCallback NewConnectionCallback_; bool listenning_; int idleFd_; //用于当文件描述符过多时的操作，当文件描述符过多时，就丢到黑洞里
构造函数
Acceptor::Acceptor(EventLoop *loop, const InetAddress &listenAddr, bool reuseport) : loop_(loop) //loop是上层传入的 , acceptSocket_(createNonblocking()) //调用函数来创建socket , acceptChannel_(loop, acceptSocket_.fd()) , listenning_(false) , idleFd_(::open("/dev/null", O_RDONLY | O_CLOEXEC)) { acceptSocket_.setReuseAddr(true); acceptSocket_.setReusePort(true); acceptSocket_.bindAddress(listenAddr); // TcpServer::start() => Acceptor.listen() 如果有新用户连接 要执行一个回调(accept => connfd => 打包成Channel => 唤醒subloop) // baseloop监听到有事件发生 => acceptChannel_(listenfd) => 执行该回调函数 acceptChannel_.setReadCallback( std::bind(&Acceptor::handleRead, this)); //当有连接发生时，eventloop会监测到，然后socketfd的channel会去执行回调函数，这个回调函数就是handleRead } //创建socket，用于构造函数 static int createNonblocking() { int sockfd = ::socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, IPPROTO_TCP); if (sockfd < 0) { LOG_FATAL("%s:%s:%d listen socket create err:%d\n", __FILE__, __FUNCTION__, __LINE__, errno); } return sockfd; } // listenfd有事件发生了，就是有新用户连接了 //该函数功能是响应和接受连接请求 void Acceptor::handleRead() { InetAddress peerAddr; //调用accept int connfd = acceptSocket_.accept(&peerAddr); if (connfd >= 0) { //NewConnectionCallback_上层回调，TcpServer传入的 if (NewConnectionCallback_) { NewConnectionCallback_(connfd, peerAddr); // 轮询找到subLoop 唤醒并分发当前的新客户端的Channel } else { ::close(connfd); } } else { LOG_SYSERR << "in Acceptor::handleRead"; // Read the section named "The special problem of // accept()ing when you can't" in libev's doc. // By Marc Lehmann, author of libev. if (errno == EMFILE) { ::close(idleFd_); idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL); ::close(idleFd_); idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC); } } } 析构函数
Acceptor::~Acceptor() { acceptChannel_.disableAll(); // 把从Poller中感兴趣的事件删除掉 acceptChannel_.remove(); // 调用EventLoop->removeChannel => Poller->removeChannel 把Poller的ChannelMap对应的部分删除 ::close(idleFd_); } 监听
void Acceptor::listen() { listenning_ = true; acceptSocket_.listen(); // listen acceptChannel_.enableReading(); // acceptChannel_注册至Poller !重要 } Buffer类
buffer内部是一个std::vector<char>，
前8个字节预留着用于记录数据长度，readIndex 指向可读取数据的初始位置，writeIndex指向空闲区的起始位置
成员变量
std::vector<char> buffer_; size_t readerIndex_; size_t writerIndex_; static const size_t kCheapPrepend = 8; static const size_t kInitialSize = 1024;
每次读数据，readerIndex_向右移动。每次写数据，writerIndex向右移动。
//返回可读区域 size_t readableBytes() const { return writerIndex_ - readerIndex_; } //返回可写区域 size_t writableBytes() const { return buffer_.size() - writerIndex_; } //返回预留区域 size_t prependableBytes() const { return readerIndex_; } 取数据的操作
//取数据的操作 void retrieve(size_t len) { if (len < readableBytes()) { readerIndex_ += len; // 说明应用只读取了可读缓冲区数据的一部分，就是len长度 还剩下readerIndex+=len到writerIndex_的数据未读 } else // len == readableBytes() { retrieveAll(); } } //当把所有的数据取出来的时候，就把reader指针和writer指针归位，归到开头 void retrieveAll() { readerIndex_ = kCheapPrepend; writerIndex_ = kCheapPrepend; } 下面的函数是把数据取出来，并转化为string类型
// 把onMessage函数上报的Buffer数据 转成string类型的数据返回 std::string retrieveAllAsString() { return retrieveAsString(readableBytes()); } std::string retrieveAsString(size_t len) { std::string result(peek(), len); retrieve(len); // 上面一句把缓冲区中可读的数据已经读取出来 这里肯定要对缓冲区进行复位操作 return result; } 要插入数据，先看一下可写区大小够不够，如果不够，那么就要扩容
// buffer_.size - writerIndex_ void ensureWritableBytes(size_t len) { if (writableBytes() < len) { makeSpace(len); // 扩容 } } 扩容操作，先看一下可写区和预留区的总和是否大于len，如果大于，就说明buffer容量还够，我们只需要把reader指针归位，writer指针往前移动一点距离。如果小于，就说明buffer容量不够，我们需要进行resize扩容
void makeSpace(size_t len) { /** * | kCheapPrepend |xxx| reader | writer | // xxx标示reader中已读的部分 * | kCheapPrepend | reader ｜ len | **/ if (writableBytes() + prependableBytes() < len + kCheapPrepend) // 也就是说 len > xxx + writer的部分 { buffer_.resize(writerIndex_ + len); } else // 这里说明 len <= xxx + writer 把reader搬到从xxx开始 使得xxx后面是一段连续空间 { size_t readable = readableBytes(); // readable = reader的长度 std::copy(begin() + readerIndex_, begin() + writerIndex_, // 把这一部分数据拷贝到begin+kCheapPrepend起始处 begin() + kCheapPrepend); readerIndex_ = kCheapPrepend; writerIndex_ = readerIndex_ + readable; } } 添加数据
// 把[data, data+len]内存上的数据添加到writable缓冲区当中 void append(const char *data, size_t len) { ensureWritableBytes(len); std::copy(data, data+len, beginWrite()); writerIndex_ += len; } 从fd读数据，先开两个缓冲区。内核缓冲区里有数据，应该要尽快一次性读进来。而不是先扩容，再读数据，不然后面内核缓冲区还会继续来数据。
/** * 从fd上读取数据 Poller工作在LT模式 * Buffer缓冲区是有大小的！ 但是从fd上读取数据的时候 却不知道tcp数据的最终大小 * * @description: 从socket读到缓冲区的方法是使用readv先读至buffer_， * Buffer_空间如果不够会读入到栈上65536个字节大小的空间，然后以append的 * 方式追加入buffer_。既考虑了避免系统调用带来开销，又不影响数据的接收。 **/ ssize_t Buffer::readFd(int fd, int *saveErrno) { // 栈额外空间，用于从套接字往出读时，当buffer_暂时不够用时暂存数据，待buffer_重新分配足够空间后，在把数据交换给buffer_。 char extrabuf[65536] = {0}; // 栈上内存空间 65536/1024 = 64KB /* struct iovec { ptr_t iov_base; // iov_base指向的缓冲区存放的是readv所接收的数据或是writev将要发送的数据 size_t iov_len; // iov_len在各种情况下分别确定了接收的最大长度以及实际写入的长度 }; */ // 使用iovec分配两个连续的缓冲区 struct iovec vec[2]; const size_t writable = writableBytes(); // 这是Buffer底层缓冲区剩余的可写空间大小 不一定能完全存储从fd读出的数据 // 第一块缓冲区，指向可写空间 vec[0].iov_base = begin() + writerIndex_; vec[0].iov_len = writable; // 第二块缓冲区，指向栈空间 vec[1].iov_base = extrabuf; vec[1].iov_len = sizeof(extrabuf); // when there is enough space in this buffer, don't read into extrabuf. // when extrabuf is used, we read 128k-1 bytes at most. // 这里之所以说最多128k-1字节，是因为若writable为64k-1，那么需要两个缓冲区 第一个64k-1 第二个64k 所以做多128k-1 // 如果第一个缓冲区>=64k 那就只采用一个缓冲区 而不使用栈空间extrabuf[65536]的内容 const int iovcnt = (writable < sizeof(extrabuf)) ? 2 : 1; const ssize_t n = ::readv(fd, vec, iovcnt); if (n < 0) { *saveErrno = errno; } else if (n <= writable) // Buffer的可写缓冲区已经够存储读出来的数据了 { writerIndex_ += n; } else // extrabuf里面也写入了n-writable长度的数据 { writerIndex_ = buffer_.size(); append(extrabuf, n - writable); // 对buffer_扩容 并将extrabuf存储的另一部分数据追加至buffer_ } return n; } 把数据写到fd上
// inputBuffer_.readFd表示将对端数据读到inputBuffer_中，移动writerIndex_指针 // outputBuffer_.writeFd标示将数据写入到outputBuffer_中，从readerIndex_开始，可以写readableBytes()个字节 ssize_t Buffer::writeFd(int fd, int *saveErrno) { ssize_t n = ::write(fd, peek(), readableBytes()); if (n < 0) { *saveErrno = errno; } return n; } TcpConnection类
TcpServer => Acceptor => 有一个新用户连接，通过accept函数拿到connfd
=> TcpConnection设置回调 => 设置到Channel => Poller => Channel回调
往下层设置的回调
eventloop的doPendingFunctors();解决的上层回调就是这些
void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; } void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; } void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; } void setCloseCallback(const CloseCallback &cb) { closeCallback_ = cb; } void setHighWaterMarkCallback(const HighWaterMarkCallback &cb, size_t highWaterMark) { highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; } 下面这四个是注册到channel中
void handleRead(Timestamp receiveTime); void handleWrite(); void handleClose(); void handleError();
成员变量，socket是与channel绑定在一起
// Socket Channel 这里和Acceptor类似 Acceptor => mainloop TcpConnection => subloop std::unique_ptr<Socket> socket_; std::unique_ptr<Channel> channel_;
const InetAddress localAddr_; const InetAddress peerAddr_; EventLoop *loop_; // 这里是baseloop还是subloop由TcpServer中创建的线程数决定 若为多Reactor 该loop_指向subloop 若为单Reactor 该loop_指向baseloop const std::string name_; std::atomic_int state_; bool reading_;
回调函数
// 这些回调TcpServer也有 用户通过写入TcpServer注册 TcpServer再将注册的回调传递给TcpConnection TcpConnection再将回调注册到Channel中 ConnectionCallback connectionCallback_; // 有新连接时的回调 MessageCallback messageCallback_; // 有读写消息时的回调 WriteCompleteCallback writeCompleteCallback_; // 消息发送完成以后的回调 HighWaterMarkCallback highWaterMarkCallback_; //控制收发速度 CloseCallback closeCallback_; size_t highWaterMark_;
// 数据缓冲区 Buffer inputBuffer_; // 接收数据的缓冲区 Buffer outputBuffer_; // 发送数据的缓冲区 用户send向outputBuffer_发
handleReader
// 将内核缓冲区中的数据读入到inputbuffer中，并调用用户注册的messageCallback // 读是相对服务器而言的 当对端客户端有数据到达 服务器端检测到EPOLLIN 就会触发该fd上的回调 handleRead取读走对端发来的数据 void TcpConnection::handleRead(Timestamp receiveTime) { int savedErrno = 0; //从fd上读数据到inputBuffer ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno); if (n > 0) // 有数据到达 { // 已建立连接的用户有可读事件发生了 调用用户传入的回调操作onMessage shared_from_this就是获取了TcpConnection的智能指针 messageCallback_(shared_from_this(), &inputBuffer_, receiveTime); } else if (n == 0) // 客户端断开 { handleClose(); } else // 出错了 { errno = savedErrno; LOG_ERROR("TcpConnection::handleRead"); handleError(); } } //当内核缓冲区由满变为不满的时候，就会产生写事件 //只要有写事件，就会一直写，这个是重点，所以写完之后，一定要关闭写事件 void TcpConnection::handleWrite() { if (channel_->isWriting()) { int savedErrno = 0; ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno); if (n > 0) { outputBuffer_.retrieve(n); //readableBytes等于0，说明buffer里的所有数据都写进去了 //如果不等于0，说明内核缓冲区里已经满了，写不进去了， if (outputBuffer_.readableBytes() == 0) { //写好了，就要关闭。因为只要有写事件，就会一直写下去。 channel_->disableWriting(); if (writeCompleteCallback_) { // TcpConnection对象在其所在的subloop中 向pendingFunctors_中加入回调 loop_->queueInLoop( std::bind(writeCompleteCallback_, shared_from_this())); } if (state_ == kDisconnecting) { shutdownInLoop(); // 在当前所属的loop中把TcpConnection删除掉 } } } else { LOG_ERROR("TcpConnection::handleWrite"); } } else { LOG_ERROR("TcpConnection fd=%d is down, no more writing", channel_->fd()); } } 下面这个函数并没有用到
void TcpConnection::send(const std::string &buf) { if (state_ == kConnected) { if (loop_->isInLoopThread()) // 这种是对于单个reactor的情况 用户调用conn->send时 loop_即为当前线程 { sendInLoop(buf.c_str(), buf.size()); } else { loop_->runInLoop( std::bind(&TcpConnection::sendInLoop, this, buf.c_str(), buf.size())); } } } /** * 发送数据 应用写的快 而内核发送数据慢 需要把待发送数据写入缓冲区，而且设置了水位回调 **/ void TcpConnection::sendInLoop(const void *data, size_t len) { ssize_t nwrote = 0; size_t remaining = len; bool faultError = false; if (state_ == kDisconnected) // 之前调用过该connection的shutdown 不能再进行发送了 { LOG_ERROR("disconnected, give up writing"); } // 表示channel_第一次开始写数据或者缓冲区没有待发送数据 if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0) { nwrote = ::write(channel_->fd(), data, len); if (nwrote >= 0) { remaining = len - nwrote; if (remaining == 0 && writeCompleteCallback_) { // 既然在这里数据全部发送完成，就不用再给channel设置epollout事件了 loop_->queueInLoop( std::bind(writeCompleteCallback_, shared_from_this())); } } else // nwrote < 0 { nwrote = 0; if (errno != EWOULDBLOCK) // EWOULDBLOCK表示非阻塞情况下没有数据后的正常返回 等同于EAGAIN { LOG_ERROR("TcpConnection::sendInLoop"); if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE RESET { faultError = true; } } } } /** * 说明当前这一次write并没有把数据全部发送出去 剩余的数据需要保存到缓冲区当中 * 然后给channel注册EPOLLOUT事件，Poller发现tcp的发送缓冲区有空间后会通知 * 相应的sock->channel，调用channel对应注册的writeCallback_回调方法， * channel的writeCallback_实际上就是TcpConnection设置的handleWrite回调， * 把发送缓冲区outputBuffer_的内容全部发送完成 **/ if (!faultError && remaining > 0) { // 目前发送缓冲区剩余的待发送的数据的长度 size_t oldLen = outputBuffer_.readableBytes(); if (oldLen + remaining >= highWaterMark_ && oldLen < highWaterMark_ && highWaterMarkCallback_) { loop_->queueInLoop( std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining)); } outputBuffer_.append((char *)data + nwrote, remaining); if (!channel_->isWriting()) { channel_->enableWriting(); // 这里一定要注册channel的写事件 否则poller不会给channel通知epollout } } } // 连接建立 void connectEstablished(); // 连接销毁 void connectDestroyed();
TcpServer
服务器
回调函数
//在callback.h文件里定义的有定义： using ConnectionCallback = std::function<void(const TcpConnectionPtr &)>; using CloseCallback = std::function<void(const TcpConnectionPtr &)>; using WriteCompleteCallback = std::function<void(const TcpConnectionPtr &)>;
这些回调函数是用户定义的，传递给TcpConnection，TcpConnection传递给Channel。因此需要用户写回调函数，然后层层传递下去
ConnectionCallback connectionCallback_; //有新连接时的回调 MessageCallback messageCallback_; // 有读写事件发生时的回调 WriteCompleteCallback writeCompleteCallback_; // 消息发送完成后的回调 void setThreadInitCallback(const ThreadInitCallback &cb) { threadInitCallback_ = cb; } void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; } void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; } void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; } setThreadNum，就是调用threadPool里的setThreadNum
// 设置底层subloop的个数 void TcpServer::setThreadNum(int numThreads) { threadPool_->setThreadNum(numThreads); } // 开启服务器监听 void TcpServer::start() { if (started_++ == 0) // 防止一个TcpServer对象被start多次 { threadPool_->start(threadInitCallback_); // 启动底层的loop线程池 //这个loop是mainloop，用来监听连接 loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get())); } } 成员变量：
using ConnectionMap = std::unordered_map<std::string, TcpConnectionPtr>; EventLoop *loop_; // baseloop 用户自定义的loop const std::string ipPort_; const std::string name_; std::unique_ptr<Acceptor> acceptor_; // 运行在mainloop 任务就是监听新连接事件 std::shared_ptr<EventLoopThreadPool> threadPool_; // one loop per thread std::atomic_int started_; int nextConnId_; ConnectionMap connections_; // 保存所有的连接
构造函数
TcpServer::TcpServer(EventLoop *loop, const InetAddress &listenAddr, const std::string &nameArg, Option option) : loop_(CheckLoopNotNull(loop)) , ipPort_(listenAddr.toIpPort()) , name_(nameArg) //比较重要的两个acceptor和threadPool， , acceptor_(new Acceptor(loop, listenAddr, option == kReusePort)) , threadPool_(new EventLoopThreadPool(loop, name_)) , connectionCallback_() , messageCallback_() , nextConnId_(1) , started_(0) { // 当有新用户连接时，Acceptor类中绑定的acceptChannel_会有读事件发生，执行handleRead()调用TcpServer::newConnection回调 acceptor_->setNewConnectionCallback( std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2)); } TcpServer::newConnection
先从线程池中调用getNextLoop()，拿到一个eventLoop来管理新连接fd对应的channel。
创建新的地址和新的socket，然后创建新的连接TcpConnection。
把TcpConnection放入到对应的map里，并设置对应的函数回调
// 有一个新用户连接，acceptor会执行这个回调操作，负责将mainLoop接收到的请求连接(acceptChannel_会有读事件发生)通过回调轮询分发给subLoop去处理 void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr) { // 轮询算法 选择一个subLoop 来管理connfd对应的channel EventLoop *ioLoop = threadPool_->getNextLoop(); char buf[64] = {0}; snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_); ++nextConnId_; // 这里没有设置为原子类是因为其只在mainloop中执行 不涉及线程安全问题 std::string connName = name_ + buf; LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s\n", name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str()); // 通过sockfd获取其绑定的本机的ip地址和端口信息 sockaddr_in local; ::memset(&local, 0, sizeof(local)); socklen_t addrlen = sizeof(local); if(::getsockname(sockfd, (sockaddr *)&local, &addrlen) < 0) { LOG_ERROR("sockets::getLocalAddr"); } InetAddress localAddr(local); TcpConnectionPtr conn(new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr)); connections_[connName] = conn; // 下面的回调都是用户设置给TcpServer => TcpConnection的，至于Channel绑定的则是TcpConnection设置的四个，handleRead,handleWrite... 这下面的回调用于handlexxx函数中 conn->setConnectionCallback(connectionCallback_); conn->setMessageCallback(messageCallback_); conn->setWriteCompleteCallback(writeCompleteCallback_); // 设置了如何关闭连接的回调 conn->setCloseCallback( std::bind(&TcpServer::removeConnection, this, std::placeholders::_1)); ioLoop->runInLoop( std::bind(&TcpConnection::connectEstablished, conn)); } 