libhv网络库源码剖析

此文归纳整理和libhv源码分析相关的文章，献给感兴趣钻研libhv源码的同学。如有其他同学有写过不错的libhv源码分析文章，可以联系我加上。文章目录: 事件循环和IO多路复用机制介绍、事件的定义以及数据结构介绍、事件的优先级机制、IO事件、定时器事件、空闲事件、自定义事件、心跳和保活机制、TCP如何处理粘包与分包、回显、聊天、代理三种经典服务实现详解、多线程/多进程服务端编程、网络编程十宗罪、

ithewei

14492人浏览 · 2022-03-05 17:46:53

ithewei · 2022-03-05 17:46:53 发布

此文归纳整理和libhv源码分析相关的文章，献给感兴趣钻研libhv源码的同学。
如有其他同学有写过不错的libhv源码分析文章，欢迎联系我加上。

libhv是一个比libevent、libev、libuv更易用的跨平台c/c++国产网络库，用来开发TCP/UDP/SSL/HTTP/WebSocket/MQTT客户端/服务端。

项目地址：https://github.com/ithewei/libhv.git
码云镜像：https://gitee.com/libhv/libhv.git
QQ技术交流群：739352073

文章目录

事件循环和IO多路复用机制介绍

https://hewei.blog.csdn.net/article/details/113724474

事件循环是libevent、libev、libuv、libhv这类网络库里最核心的概念，即在事件循环里处理IO读写事件、定时器事件、自定义事件等各种事件；

IO多路复用即在一个IO线程监听多个fd，如最早期的select、后来的poll，linux的epoll、windows的iocp、bsd的kqueue、solaris的port等，都属于IO多路复用机制。非阻塞NIO搭配IO多路复用机制就是高并发的钥匙。

libhv下的event模块正是封装了多种平台的IO多路复用机制，提供了统一的事件接口，是libhv的核心模块。

事件的定义以及数据结构介绍

libhv中的事件包括IO事件、timer定时器事件、idle空闲事件、自定义事件。
所有事件都是继承自公共的基类hevent_t，便于放入事件队列中统一调度。

事件结构体hevent_t定义在事件循环模块对外头文件 hloop.h 里：

struct hevent_s {
    hloop_t*            loop;           // 事件所属的事件循环
    hevent_type_e       event_type;     // 事件类型
    uint64_t            event_id;       // 事件ID
    hevent_cb           cb;             // 事件回调
    void*               userdata;       // 事件用户数据
    void*               privdata;       // 事件私有数据
    struct hevent_s*    pending_next;   // 下一个事件（用于实现事件队列）
    int                 priority;       // 事件优先级
};

事件循环结构体hloop_t定义在内部 hevent.h 头文件里：

struct hloop_s {
    uint32_t    flags;			// 事件循环flags
    hloop_status_e status; 		// 状态：running、stop、pause
    uint64_t    start_ms;       // 开始时间（现实时间realtime）
    uint64_t    start_hrtime;   // 开始时间（线性时间，不受系统时间调整影响）
    uint64_t    end_hrtime;   	// 结束时间
    uint64_t    cur_hrtime;		// 当前时间
    uint64_t    loop_cnt;		// 循环计数
    long        pid;			// 进程ID
    long        tid;			// 线程ID
    void*       userdata;		// 用户数据
//private:
    // events
    uint32_t                    intern_nevents;	// 内部激活事件数
    uint32_t                    nactives;	// 激活事件数
    uint32_t                    npendings;	// 行将发生事件数
    // pendings: with priority as array.index
    hevent_t*                   pendings[HEVENT_PRIORITY_SIZE]; // 事件优先级队列
    // idles
    struct list_head            idles;	// 空闲事件链表
    uint32_t                    nidles;	// 空闲事件数
    // timers
    struct heap                 timers;  // 定时器事件堆
    uint32_t                    ntimers; // 定时器事件数
    // ios: with fd as array.index
    struct io_array             ios; 	// IO事件数组
    uint32_t                    nios;	// IO事件数
    // one loop per thread, so one readbuf per loop is OK.
    hbuf_t                      readbuf;	// 读缓存
    void*                       iowatcher;	// IO事件监视器
    // custom_events
    int                         eventfds[2]; 	// 事件FD，用于唤醒事件循环
    event_queue                 custom_events;  // 自定义事件队列
    hmutex_t                    custom_events_mutex; // 互斥锁，用于自定义事件队列的线程安全性
};

IO事件使用了数组，使用fd作为数组索引，便于随机访问；
数组实现见 array.h，源码分析文章暂缺；
空闲事件使用了链表，添加删除复杂度为O(1)，遍历访问；
链表实现见 list.h，源码分析文章见：https://blog.csdn.net/qu1993/article/details/110731150
定时器事件使用了大小堆，查询最小的超时时间复杂度为O(1)，添加删除复杂度O(lgN)
大小堆实现见 heap.h，源码分析文章见：https://blog.csdn.net/qu1993/article/details/110855013
自定义事件使用了队列，具有先进先出的特性；
队列实现见 queue.h，源码分析文章暂缺；

TCP如何处理粘包与分包

libhv提供了设置拆包规则接口，c接口见hio_set_unpack，c++接口见SocketChannel::setUnpack，支持固定包长、分隔符、头部长度字段三种常见的拆包方式，调用该接口设置拆包规则后，内部会根据拆包规则处理粘包与分包，保证回调上来的是完整的一包数据，大大节省了上层处理粘包与分包的成本，该接口具体定义如下：

typedef enum {
    UNPACK_BY_FIXED_LENGTH  = 1,    // 根据固定长度拆包
    UNPACK_BY_DELIMITER     = 2,    // 根据分隔符拆包，如常见的“\r\n”
    UNPACK_BY_LENGTH_FIELD  = 3,    // 根据头部长度字段拆包
} unpack_mode_e;

#define DEFAULT_PACKAGE_MAX_LENGTH  (1 << 21)   // 2M

// UNPACK_BY_DELIMITER
#define PACKAGE_MAX_DELIMITER_BYTES 8

// UNPACK_BY_LENGTH_FIELD
typedef enum {
    ENCODE_BY_VARINT        = 1,                // varint编码
    ENCODE_BY_LITTEL_ENDIAN = LITTLE_ENDIAN,    // 小端编码
    ENCODE_BY_BIG_ENDIAN    = BIG_ENDIAN,       // 大端编码
} unpack_coding_e;

typedef struct unpack_setting_s {
    unpack_mode_e   mode; // 拆包模式
    unsigned int    package_max_length; // 最大包长度限制
    // UNPACK_BY_FIXED_LENGTH
    unsigned int    fixed_length; // 固定包长度
    // UNPACK_BY_DELIMITER
    unsigned char   delimiter[PACKAGE_MAX_DELIMITER_BYTES]; // 分隔符
    unsigned short  delimiter_bytes; // 分隔符长度
    // UNPACK_BY_LENGTH_FIELD
    unsigned short  body_offset; // body偏移量（即头部长度）real_body_offset = body_offset + varint_bytes - length_field_bytes
    unsigned short  length_field_offset; // 头部长度字段偏移量
    unsigned short  length_field_bytes; // 头部长度字段所占字节数
    unpack_coding_e length_field_coding; // 头部长度字段编码方式，支持varint、大小端三种编码方式，通常使用大端字节序（即网络字节序）
#ifdef __cplusplus
    unpack_setting_s() {
        // Recommended setting:
        // head = flags:1byte + length:4bytes = 5bytes
        mode = UNPACK_BY_LENGTH_FIELD;
        package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
        fixed_length = 0;
        delimiter_bytes = 0;
        body_offset = 5;
        length_field_offset = 1;
        length_field_bytes = 4;
        length_field_coding = ENCODE_BY_BIG_ENDIAN;
    }
#endif
} unpack_setting_t;

HV_EXPORT void hio_set_unpack(hio_t* io, unpack_setting_t* setting);

以ftp为例（分隔符方式）可以这样设置：

unpack_setting_t ftp_unpack_setting;
memset(&ftp_unpack_setting, 0, sizeof(unpack_setting_t));
ftp_unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
ftp_unpack_setting.mode = UNPACK_BY_DELIMITER;
ftp_unpack_setting.delimiter[0] = '\r';
ftp_unpack_setting.delimiter[1] = '\n';
ftp_unpack_setting.delimiter_bytes = 2;

以mqtt为例（头部长度字段方式）可以这样设置：

unpack_setting_t mqtt_unpack_setting = {
    .mode = UNPACK_BY_LENGTH_FIELD,
    .package_max_length = DEFAULT_PACKAGE_MAX_LENGTH,
    .body_offset = 2,
    .length_field_offset = 1,
    .length_field_bytes = 1,
    .length_field_coding = ENCODE_BY_VARINT,
};