# my_muduo

**Repository Path**: win1010/my_muduo

## Basic Information

- **Project Name**: my_muduo
- **Description**: C++11  手写muduo网络库项目
- **Primary Language**: C++
- **License**: MulanPSL-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2024-05-24
- **Last Updated**: 2024-06-17

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README


## 1. 序
 
 
### 1.1 总述
 
muduo库是基于多Reactor-多线程模型实现的TCP网络编程库，性能良好。如libev作者：“One loop per thread is usually a good model”，muduo库的作者陈硕在其《Linux多线程服务端编程》中也力荐这种“One loop per thread”的IO模型，使我们仅需要关注EventLoop的设计与实现，然后每个线程run一个loop即可。不过由于当时C++11并没有进入实用，在这一书中，作者没有谈及C++11，整个muduo库的实现，也依赖了boost库。

而在项目设计与实现中，按照C++11标准对muduo库中核心部分进行重写，主要涉及了以下模块：Channel、Poller、EventLoop、Thread、EventLoopThread、EventLoopThreadPool、Socket、Acceptor、Buffer、TcpConnection、TcpServer、TimerWheel，下面将进行分述。

 
### 1.2 One loop per thread
 
在多Reactor-多线程模型中，运用one loop per thread的思想，由一个mainReactor负责accept连接，然后把该连接挂载到某个subReactor，多个连接分配到多个线程，充分利用CPU资源。
![image](IMG/yuanli.jpg)
 
## 2. 核心部分
 在手写muduo库项目之中，存在三个核心部分，分别是Channel类、Poller类和EventLoop类，这三大类的组合，实现了reactor用以监听fd并同时处理相应的回调函数。其中Poller和Channel之间通过EventLoop相互通信。
![image](IMG/conn.png)
 
### 2.1 Channel

1. fd_：封装sockfd，两种Channel：listenfd-acceptorChannel，connfd-ConnectionChannel；
1. events_：fd监听的事件类型；
1. revents_：Poller返回的具体监听到的事件。
1. callback：上层设置的各种类型事件回调；
1. tie_：weak_ptr<void>，在事件监听器返回监听结果后，就会调用Channel中的handleEvent()函数。首先会把tie_这个weak_ptr提升为shared_ptr，它会指向当前的TcpConnection对象，即使外面调用了删除析构了其他所有指向该TcpConnection的智能指针，只要没有handleEvent()完，这个TcpConnection都不会被析构释放堆内存。



 
 
### 2.2 Poller/EPollPoller
muduo库提供poll和epoll两种IO多路复用方法来实现事件监听，重写时，通过基类Poller和派生类EPollPoller，支持了Epoll。Poller主要扮演Reactor模型中Demultiplex事件分发器（也可以说是事件监听器）的角色。


1. epollfd_：记录epoll_create返回的句柄
1. channels_：用来记录注册在其上的Channel的unordered_map。


### 2.3 EventLoop
EventLoop扮演Reactor模型中Reactor的角色，是对epoll的封装。EventLoop在epoll_create，注册各个Channel之后，处于epoll_wait阻塞状态，要想唤醒当前的EventLoop去执行新的连接，通过往wakefd上写入一个字符，唤醒当前的EventLoop。（而并非生产者-消费者模型）。

1. 包含了所有的Channel
1. 每一个loop都有一个wakeupFd

 
 
### 2.4 具体方法的部分代码实现
- EventLoop::loop()——开启事件循环

```
// 开启事件循环
void EventLoop::loop()
{
    // ...
    while (!quit_)
    {
        activeChannels_.clear(); // 准备接收发生事件的channel，先清空
        // 这里监听两类fd，一类是用户fd，一类是自己的wakeupfd
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);

        for (Channel *channel : activeChannels_)
        {
            // LOG_DEBUG("发生事件: %d",channel->revents());
            //  EventLoop的poller监听到就绪的channel，上报给EventLoop，然后EVentLoop通知相应的channel处理事件
            channel->handleEvent(pollReturnTime_);
        }

        // ...
    }

    LOG_INFO("EventLoop %p stop looping", this);
    // ...
}
```
- EPollPoller::poll()——开启Poller事件监听，调用了::epoll_wait()

```
// 通过epoll_wait监听哪些Channel/fd发生事件
Timestamp EPollPoller::poll(int timeoutMs, ChannelList *activeChannels)
{
    // ...
    int numEvents = epoll_wait(epfd_, &(*events_.begin()), static_cast<int>(events_.size()), timeoutMs);
    // errno是全局的，多线程，多个Poller，执行下面代码时errno可能会变，这里保存当前Poller的错误码
    int savedErrno = errno;
    Timestamp now(Timestamp::now());

    if (numEvents > 0)
    {
        LOG_DEBUG("%d events happened, 属于loop: %p",numEvents,ownerLoop());
        fillActiveChannels(numEvents, activeChannels);
        if (numEvents == events_.size())    //扩容
        {
            events_.resize(events_.size() * 2);
        }
    }
    else if (numEvents == 0)
    {
        LOG_DEBUG("nothing happened, 属于loop: %p",ownerLoop());
    }
    else
    {
        if (savedErrno != EINTR)
        {
            LOG_FATAL("EPollPoller::poll() error: %d: %s", savedErrno, strerror(errno));
        }
    }
    return now;
}
```
- 唤醒机制——通过向eventfd写一个数据

```
// 创建wakeupfd，用来notify唤醒subReactor处理新的Channel
int createEventfd()
{
    int evtfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    // ...
    return evtfd;
}
 
// 用于唤醒loop所在线程: 向wakefd写一个数据
//wakeupFd_在构造函数中通过createsEventFd()函数初始化
// 通过往wakeupfd中写一个数据，而使其读事件发生，从而唤醒当前loop的线程
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8", n);
    }
}
```

## 3. 其他部分
### 3.1 EventLoopThreadPool
EventLoopThreadPool类，可以理解为subLoop池，主要是对EventLoopThread的封装，而EventLoopThread又是对EventLoop（Reactor）和Thread（记录线程的详细信息）的封装。

其中，初始化时，会提供一个baseLoop（mainLoop）来进行基本的事件循环。通过设置numthreads_来创建对应数量的subReactor，每当创建一个线程，就会生成一个EventLoop。

在工作方式上，通过getNextLoop()方法，实现对subReactor的轮询。

```
// ...
class EventLoopThreadPool : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop *)>;

    EventLoopThreadPool(EventLoop *baseloop, const std::string &nameArg);
    ~EventLoopThreadPool();

    void setThreadNum(int numThreads) { numThreads_ = numThreads; }
    void start(const ThreadInitCallback &cb = ThreadInitCallback());
    // 轮询获取subloop。如果没有subloop，baseloop则充当subloop
    EventLoop *getNextLoop();

    // 获取所有的subloop
    std::vector<EventLoop *> getAllLoops();

    bool started() { return started_; }
    const std::string &name() { return name_; }

private:
    EventLoop *baseLoop_;                // 用户最开始创建的loop
    std::string name_;                   // 线程池的名称
    bool started_;                       
    int numThreads_;
    int next_;
    std::vector<std::unique_ptr<EventLoopThread>> threads_;
    std::vector<EventLoop *> loops_;
};

```

### 3.2 Acceptor
Acceptor类，封装的是服务器监听socketfd和相关处理函数。接收新用户连接后，通过轮询来选择subReactor并给它分发连接。

### 3.3 TcpConnection
每个连接进来的客户端，对应一个TcpConnection，封装了一个connfd，一个Channel，各种回调函数（Callback）和读写缓冲区（Buffer）。

state_：记录当前连接状态，一共有四种：kConnected、kConnecting、kDisconnecting、kDisconnected。

整个TcpConnection的工作流程


1. TcpServer通过Acceptor监听用户新连接，用accept拿到connfd
1. TcpConnection设置回调给Channel，Channel注册到Poller
1. Poller监听到事件就通知调用Channel的回调


### 3.4 Buffer

Buffer缓冲区通过vector来实现，空间不足时，通过vector类的成员函数resize()即可实现扩容。在空间的设计上，主要分为如下图三个区域：
![image](IMG/image.png)




### 3.5 TcpServer
在TcpServer类中，有一个Acceptor，一个EventLoopThreadPool，一些回调函数，一个记录所有连接的unordered_map<string, TcpConnectionPtr>。

```
// 对外服务器编程需要使用的类
class TcpServer : noncopyable
{
public:
    using ThreadInitCallbcak = std::function<void(EventLoop *)>;
    using Functor = std::function<void()>;

    enum Option
    {
        kNoReusePort,
        kReusePort
    };

    TcpServer(EventLoop *loop,
              const InetAddress &listenAddr,
              const std::string &nameArg,
              Option option = kNoReusePort);
    ~TcpServer();

    // 设置subloop的个数
    void setThreadNum(int numThreads);

    void setThreadInitCallback(const ThreadInitCallbcak &cb) { threadInitCallbcak_ = cb; }
    void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; }
    void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; }
    void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; }

    // 设置超时销毁时间函数
    void EnableInactiveRelease(int timeout)
    {
        timeout_ = timeout;
        enableInactiveRelease_ = true;
    }

    // 开启baseloop的监听
    void start();

    //添加定时任务的接口
    void runAfter(const Functor& task, int delay);

private:
    void newConnection(int sockfd, const InetAddress &peerAddr);
    // 移除connections_中的存储的连接，因为他的类型是shared_ptr，会计数
    void removeConnection(const TcpConnectionPtr &conn);
    void removeConnectionInLoop(const TcpConnectionPtr &conn);

    void runAfterInLoop(const Functor& task, int delay);

    using ConnectionMap = std::unordered_map<std::string, TcpConnectionPtr>;

    EventLoop *loop_; // baseloop/mainloop
    const std::string ipPort_;
    const std::string name_;
    std::unique_ptr<Acceptor> acceptor_;
    std::shared_ptr<EventLoopThreadPool> threadPool_;

    ConnectionCallback connectionCallback_;       // 有新连接时的回调
    MessageCallback messageCallback_;             // 有读写消息时的回调
    WriteCompleteCallback writeCompleteCallback_; // 消息发送完成后的回调
    ThreadInitCallbcak threadInitCallbcak_;       // subloop线程初始化的函数
    std::atomic<int> started_;
    /**
     * 连接的非活跃销毁这个定时任务的key值是TcpConnection对象的名称，即 name_-ipport#nextConnId_
     * 我们收动添加的定时任务的key值是to_string(taskId_), 且由baseLoop执行
     */
    int nextConnId_;            // 连接序号
    int taskId_;                //手动添加的定时任务的序号
    ConnectionMap connevtions_; // 保存TcpConnectionPtr与连接名称的键值对

    int timeout_;                   // 非活跃链接的统计时间
    int enableInactiveRelease_;  //标识是否启动非活跃连接超时销毁
};
```
start()：启动EventLoopThreadPool，调用acceptor_的listen()方法，监听客户端的连接套接字。

newConnection()：该方法被注册到了acceptor_中，当acceptor_监听到新用户连接时会执行该回调，轮询选择一个subReactor；根据连接成功的sockfd，创建一个连接对象并加入到TcpServer的存储连接信息的connections_中；给这个连接设置回调；然后在mainLoop执行connectEstablished()；

上面提到的关闭连接的回调函数，真实的调用过程：TcpConnection::setCloseCallBack() --> TcpServer::removeConnection() --> TcpServer::removeConnectionInLoop() --> TcpConnection::connectionDestroyed()
### 3.6 TimerWheel
TimerWheel类实现了一个基于时间轮算法的定时器，这种定时器适用于处理大量的定时任务，同时能够高效地添加、取消和触发任务。


```
// ...
class TimerWheel
{
public:
    using TaskFunc = std::function<void()>;

    TimerWheel(EventLoop* loop);
    ~TimerWheel();

    //添加定时任务
    void timerAdd(std::string name,uint32_t timeout, const TaskFunc& cb);
    //刷新或延迟定时任务
    void timerRefresh(std::string name);
    //取消定时任务
    void timerCancel(std::string name);
    bool hasTimer(std::string name);

private:
    //删除对应任务
    void removeTimer(std::string name);
    
    static int creatTimerfd();
    int readTimerfd();

    //这个函数每秒钟被执行一次，相当于秒针向后走了一步
    void runTimerTask();
    void onTime();

    void timerAddInLoop(std::string name,uint32_t timeout, const TaskFunc& cb);
    void timerRefreshInLoop(std::string name);
    void timerCancelInLoop(std::string name);

private:
    EventLoop* loop_;
    //定时器描述符
    int timerfd_;

    int tick_;      //当前秒针，走到哪就释放哪，即执行定时任务
    int capacity_;  //最大延时时间

    //管理定时器任务的生命周期
    using weakTask=std::weak_ptr<TimerTask>;
    using ptrTask=std::shared_ptr<TimerTask>;

    std::vector<std::vector<ptrTask>> wheel_;           //轮子, 二维数组，可能过5秒后同时执行多个任务
    std::unordered_map<std::string, weakTask> timers_;  //TimerTask对象的名称 和 TimerTask的映射
    std::unique_ptr<Channel> timerChannel_;             //存储timerfd_的channel
};
```


## 4. 工作流程
### 4.1 安装
下载仓库后，执行sudo ./autobuild.sh，运行编译和安装脚本，生成的动态库和相关头文件会自动添加到系统路径。
### 4.2 测试代码
通过取消下面代码中部分语句的注释，使用telnet完成4项测试。
```
#include <myMuduo/TcpServer.h>
#include <myMuduo/Logger.h>

#include <string>
#include <functional>
#include <iostream>

class EchoServer
{
public:
    EchoServer(EventLoop *loop, const InetAddress &addr, const std::string &name)
        : loop_(loop)
        , server_(loop, addr, name)
    {
        // 注册回调函数
        server_.setConnectionCallback(std::bind(&EchoServer::onConnection, this, std::placeholders::_1));

        server_.setMessageCallback(std::bind(&EchoServer::onMessage, this,
                                             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        // 设置线程数量
        server_.setThreadNum(3);
        server_.EnableInactiveRelease(10);
    }

    void start()
    {
        server_.start();
        // server_.runAfter(std::bind(&EchoServer::print, this), 15);  // 添加定时任务测试
    }

    void print()
    {
        std::cout<<"-------------------定时任务---------------"<<std::endl;
    }

private:
    // 连接回调
    void onConnection(const TcpConnectionPtr &conn)
    {
        if (conn->connected())
        {
            LOG_INFO("Connection UP : %s", conn->peerAddress().toIpPort().c_str());
        }
        else
        {
            LOG_INFO("Connection DOWN : %s", conn->peerAddress().toIpPort().c_str());
        }
    }

    // 有读写消息时的回调
    void onMessage(const TcpConnectionPtr &conn, Buffer *buf, Timestamp recivetime)
    {
        std::string msg = buf->retrieveAllAsString();
        conn->send(msg);
        // conn->shutdown();   // 回显测试
    }

    EventLoop *loop_;
    TcpServer server_;
};

int main()
{
    EventLoop loop; //mainloop
    InetAddress localAddr(6666);    //本地地址，ip地址缺省值为"127.0.0.1"
    EchoServer server(&loop, localAddr, "EchoServer-01");
    server.start();     
    loop.loop();    //启动mainloop的poller，监听新连接

    return 0;
}
```

1. 回显测试。
![image](IMG/huixian.png)
1. 长连接测试，超时时间为10秒。
客户端通过发送数据刷新活跃度，长连接测试正常。
![输入图片说明](IMG/chang.png)
1. 超时连接测试，超时时间为10秒。
客户端连接后不再做处理，服务器关闭超时连接。
![image](IMG/chaoshi.png)
1. 添加定时任务测试。
成功执行添加的15秒后的定时任务
![image](IMG/dingshi.png)