muduo

void CountDownLatch::wait() // latchdown之后然后调用wait.等待最后一个线程notifyAll,
// 然后多个线程同时解除锁定就可以同时开始执行了
{
    MutexLockGuard lock(mutex_);
    while (count_ > 0) {
        condition_.wait();
    }
}

void CountDownLatch::countDown() // 每个线程在开始都latchdown,最后一个线程会notifyAll
{
    MutexLockGuard lock(mutex_);
    --count_;
    if (count_ == 0) {
        condition_.notifyAll();
    }
}

static T& instance()
{
  pthread_once(&ponce_, &Singleton::init); // 使用pthread_once来进行构造
  return *value_;
}

// 外部丢入Task放到线程池内部
void ThreadPool::run(const Task& task)
{
  if (threads_.empty()) // 如果没有任何线程
  {
    task();
  }
  else
  {
    MutexLockGuard lock(mutex_);
    queue_.push_back(task);
    cond_.notify();
  }
}
ThreadPool::Task ThreadPool::take()
{
  MutexLockGuard lock(mutex_);
  while (queue_.empty() && running_)
  {
    cond_.wait();
  }
  Task task;
  if(!queue_.empty())
  {
    task = queue_.front();
    queue_.pop_front();
  }
  return task;
}
// 线程回调函数
void ThreadPool::runInThread()
{
  try
  {
    while (running_)
    {
      Task task(take());
      if (task)
      {
        task();
      }
    }
  }
}

void makeSpace(size_t len)
{
    if (writableBytes() + prependableBytes() < len + kCheapPrepend) // 如果当前不能够通过压缩合并的话
    {
        buffer_.resize(writerIndex_+len);
    }
    else
    {
        // move readable data to the front, make space inside buffer
        assert(kCheapPrepend < readerIndex_); // 如果可以压缩的话那么就压缩
        size_t readable = readableBytes();
        std::copy(begin()+readerIndex_,
                  begin()+writerIndex_,
                  begin()+kCheapPrepend);
        readerIndex_ = kCheapPrepend;
        writerIndex_ = readerIndex_ + readable;
        assert(readable == readableBytes());
    }
}

class Channel : boost::noncopyable
{
  public:
    typedef boost::function<void()> EventCallback;
    typedef boost::function<void(Timestamp)> ReadEventCallback;
    void handleEvent(Timestamp receiveTime);
  private:
    EventLoop* loop_; // 属于哪一个Reactor
    const int  fd_; // 关联fd
    int        events_; // 关注事件
    int        revents_; // ready事件
    int        index_; // used by Poller. 在Poller中的编号，实际上没有特别意思

    boost::weak_ptr<void> tie_; // 绑定的对象，这个对于boost::weak_ptr不是很了解，但是这个对于理解框架没有用途
    bool tied_; // 是否绑定了对象上来
    bool eventHandling_; // 当前正在处理event
    ReadEventCallback readCallback_;
    EventCallback writeCallback_; // 定义如何写数据
    EventCallback closeCallback_; // 定义如何关闭连接
    EventCallback errorCallback_; // 定义如果出错的话如何处理
};

void Channel::handleEventWithGuard(Timestamp receiveTime)
{
    eventHandling_ = true;
    if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
    {
        LOG_WARN << "Channel::handle_event() POLLHUP";
        if (closeCallback_) closeCallback_();
    }
    if (revents_ & POLLNVAL)
    {
        LOG_WARN << "Channel::handle_event() POLLNVAL";
    }
    if (revents_ & (POLLERR | POLLNVAL))
    {
        if (errorCallback_) errorCallback_();
    }
    if (revents_ & (POLLIN | POLLPRI | POLLRDHUP))
    {
        if (readCallback_) readCallback_(receiveTime);
    }
    if (revents_ & POLLOUT)
    {
        if (writeCallback_) writeCallback_();
    }
    eventHandling_ = false;
}

class Poller : boost::noncopyable
{
  public:
    typedef std::vector<Channel*> ChannelList;
    /// Polls the I/O events.
    /// Must be called in the loop thread.
    virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;  // 进行poll操作，活跃事件放在activeChannels里面
    /// Changes the interested I/O events.
    /// Must be called in the loop thread.
    virtual void updateChannel(Channel* channel) = 0; // 更新channel
    /// Remove the channel, when it destructs.
    /// Must be called in the loop thread.
    virtual void removeChannel(Channel* channel) = 0; // 删除channel
  private:
    EventLoop* ownerLoop_;
};

class EventLoop : boost::noncopyable
{
  public:
    typedef boost::function<void()> Functor;
    typedef boost::function<void()> TimerCallback;
    void loop();
    void quit();
    /// Runs callback immediately in the loop thread.
    /// It wakes up the loop, and run the cb.
    /// If in the same loop thread, cb is run within the function.
    /// Safe to call from other threads.
    void runInLoop(const Functor& cb);
    /// Queues callback in the loop thread.
    /// Runs after finish pooling.
    /// Safe to call from other threads.
    void queueInLoop(const Functor& cb);

    /// Runs callback at 'time'.
    /// Safe to call from other threads.
    ///
    TimerId runAt(const Timestamp& time, const TimerCallback& cb);
    ///
    /// Runs callback after @c delay seconds.
    /// Safe to call from other threads.
    ///
    TimerId runAfter(double delay, const TimerCallback& cb);
    ///
    /// Runs callback every @c interval seconds.
    /// Safe to call from other threads.
    ///
    TimerId runEvery(double interval, const TimerCallback& cb);
    ///
    /// Cancels the timer.
    /// Safe to call from other threads.
    ///
    // void cancel(TimerId timerId);
    // internal usage
    void wakeup();
    void updateChannel(Channel* channel);
    void removeChannel(Channel* channel);
    bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
  private:
    typedef std::vector<Channel*> ChannelList;
    bool looping_; /* atomic */
    bool quit_; /* atomic */
    bool eventHandling_; /* atomic */
    bool callingPendingFunctors_; /* atomic */
    const pid_t threadId_;
    Timestamp pollReturnTime_;
    boost::scoped_ptr<Poller> poller_;
    boost::scoped_ptr<TimerQueue> timerQueue_;
    int wakeupFd_;
    // unlike in TimerQueue, which is an internal class,
    // we don't expose Channel to client.
    boost::scoped_ptr<Channel> wakeupChannel_;
    ChannelList activeChannels_;
    MutexLock mutex_;
    std::vector<Functor> pendingFunctors_; // @BuardedBy mutex_
};

TimerId TimerQueue::addTimer(const TimerCallback& cb,
                             Timestamp when,
                             double interval)
{
  Timer* timer = new Timer(cb, when, interval); // 首先创建一个Timer对象，然后将cb放在里面。内部有一个run函数，调用的就是cb
  loop_->runInLoop(
      boost::bind(&TimerQueue::scheduleInLoop, this, timer)); // 然后将这个timer丢到eventLoop里面去执行
  return TimerId(timer, timer->sequence());
}
void TimerQueue::scheduleInLoop(Timer* timer)
{
  loop_->assertInLoopThread();
  bool earliestChanged = insert(timer); // 将timer插入到内部的链表里面去，按照超时时间顺序插入，并且判断这个插入是否会影响最早时间

  if (earliestChanged)
  {
    resetTimerfd(timerfd_, timer->expiration()); // 如果影响的话，那么要修改这个timerfd超时时间。
  }
}

void TimerQueue::handleRead()
{
  loop_->assertInLoopThread();
  Timestamp now(Timestamp::now());
  readTimerfd(timerfd_, now);
  std::vector<Entry> expired = getExpired(now); // 我们可以知道有哪些计时器超时
  // safe to callback outside critical section
  for (std::vector<Entry>::iterator it = expired.begin();
      it != expired.end(); ++it)
  {
    it->second->run(); // 对于这些超时的Timer,执行run()函数，对应也就是我们一开始注册的回调函数。
  }
  reset(expired, now);
}

EventLoop* EventLoopThread::startLoop(){
    thread_.start();
    {
        MutexLockGuard lock(mutex_);
        while (loop_ == NULL)
        {
            cond_.wait();
        }
    }
    return loop_;
}

void EventLoopThread::threadFunc(){
    EventLoop loop;
    {
        MutexLockGuard lock(mutex_);
        loop_ = &loop;
        cond_.notify();
    }
    loop.loop();
}

typedef boost::function<void (int sockfd,const InetAddress&)> NewConnectionCallback;

typedef boost::function<void (int sockfd)> NewConnectionCallback;

class TcpConnection : boost::noncopyable,
                      public boost::enable_shared_from_this<TcpConnection>
{
  public:
    /// Constructs a TcpConnection with a connected sockfd
    ///
    /// User should not create this object.
    TcpConnection(EventLoop* loop, // 建立连接需要一个Reactor
                  const string& name, // 连接名称
                  int sockfd, // 连接fd.
                  const InetAddress& localAddr, // 连接的address.
                  const InetAddress& peerAddr);
    // called when TcpServer accepts a new connection
    void connectEstablished();   // should be called only once
    // called when TcpServer has removed me from its map
    void connectDestroyed();  // should be called only once
  private:
    enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };
    void sendInLoop(const void* message, size_t len); // 发送消息
    void setState(StateE s) { state_ = s; }

    EventLoop* loop_;
    string name_;
    StateE state_;  // FIXME: use atomic variable
    // we don't expose those classes to client.
    boost::scoped_ptr<Socket> socket_; // socket.
    boost::scoped_ptr<Channel> channel_; // 连接channel
    InetAddress localAddr_;
    InetAddress peerAddr_;
    ConnectionCallback connectionCallback_; // 连接回调，这个触发包括在连接建立和断开都会触发
    MessageCallback messageCallback_; // 有数据可读的回调
    WriteCompleteCallback writeCompleteCallback_; // 写完毕的回调
    CloseCallback closeCallback_; // 连接关闭回调
    Buffer inputBuffer_; // 数据读取buffer.
    Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.
    boost::any context_; // 上下文环境
    // FIXME: creationTime_, lastReceiveTime_
    //        bytesReceived_, bytesSent_
};

TcpClient::TcpClient(EventLoop* loop,
                     const InetAddress& serverAddr,
                     const string& name)
        : loop_(CHECK_NOTNULL(loop)),
          connector_(new Connector(loop, serverAddr)),
          name_(name),
          connectionCallback_(defaultConnectionCallback),
          messageCallback_(defaultMessageCallback),
          retry_(false),
          connect_(true),
          nextConnId_(1)
{
    connector_->setNewConnectionCallback(
        boost::bind(&TcpClient::newConnection, this, _1));
    // FIXME setConnectFailedCallback
}

TcpServer::TcpServer(EventLoop* loop,
                     const InetAddress& listenAddr,
                     const string& nameArg)
  : loop_(CHECK_NOTNULL(loop)),
    hostport_(listenAddr.toHostPort()),
    name_(nameArg),
    acceptor_(new Acceptor(loop, listenAddr)),
    threadPool_(new EventLoopThreadPool(loop)),
    connectionCallback_(defaultConnectionCallback),
    messageCallback_(defaultMessageCallback),
    started_(false),
    nextConnId_(1)
{
  acceptor_->setNewConnectionCallback(
      boost::bind(&TcpServer::newConnection, this, _1, _2));
}

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
    loop_->assertInLoopThread();
    EventLoop* ioLoop = threadPool_->getNextLoop();
    char buf[32];
    snprintf(buf, sizeof buf, ":%s#%d", hostport_.c_str(), nextConnId_);
    ++nextConnId_;
    string connName = name_ + buf;
    // FIXME poll with zero timeout to double confirm the new connection
    TcpConnectionPtr conn(
        new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));
    connections_[connName] = conn;
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);
    conn->setCloseCallback(
        boost::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
    ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
}