muduo 网络库日志后端

文前导读

muduo 网络库源码剖析系列的第二篇文章，主要着眼于 muduo 网络库中的日志系统后端的设计与实现
为了保证自己对 muduo 的代码有较为深入的理解，我自己写了一个 tmuduo 网络库，用来验证自己对 muduo 源码上的一些想法。
仓库地址为：git@github.com:Phoenix500526/Tmuduo.git，欢迎 fork、start 以及 follow，一起学习。

more >>

展开全文 >>

muduo 网络库日志前端

文前导读

muduo 网络库源码剖析系列的第一篇文章，主要着眼于 muduo 网络库中的日志系统前端的设计与实现
为了保证自己对 muduo 的代码有较为深入的理解，我自己写了一个 tmuduo 网络库，用来验证自己对 muduo 源码上的一些想法。
仓库地址为：git@github.com:Phoenix500526/Tmuduo.git，欢迎 fork、start 以及 follow，一起学习。

more >>

展开全文 >>

googletest测试框架初探

在 CMake 中使用 Google-Test

示例目录的源码树：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
$ tree
.
├── 3rd_party
│   └── google-test
│       ├── CMakeLists.txt  # 用于构建 Google Test 库的 CMake 命令
│       └── CMakeLists.txt.in # 用于下载 Google Test 的辅助脚本
├── CMakeLists.txt
├── Reverse.h
├── Reverse.cpp
├── Palindrome.h
├── Palindrome.cpp
├── unit_tests.cpp  # 基于 Google Test 测试框架的单元测试文件
# 执行命令
$ mkdir build
$ cd build
$ cmake ..
$ make
$ make test

测试通过：

1
2
3
4
5
6
7
8
9
$ make test
Running tests...
Test project /home/phoenix/Project/cmake-examples/05-unit-testing/google-test-download/build
    Start 1: test_all
1/1 Test #1: test_all .........................   Passed    0.01 sec

100% tests passed, 0 tests failed out of 1

Total Test time (real) =   0.01 sec

测试失败:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
$ make test
Running tests...
Test project /home/phoenix/Project/cmake-examples/05-unit-testing/google-test-download/build
    Start 1: test_all
1/1 Test #1: test_all .........................***Failed    0.00 sec

0% tests passed, 1 tests failed out of 1

Total Test time (real) =   0.00 sec

The following tests FAILED:
      1 - test_all (Failed)
Errors while running CTest
Makefile:72: recipe for target 'test' failed
make: ** [test] Error 8

具体的测试信息会保存在 build/Testing/Temporary/LastTest.log 文件

注1：googletest 也可以使用 clang 或 g++ 进行单独编译，编译命令为：

1
$ clang++ UnitTest.cc -I ../google-test/googletest-src/googletest/include/ -L../google-test/googletest-build/googlemock/gtest -lgtest -lgtest_main -lpthread -std=c++11 -Wall

注2：若测试时需要有配置文件，例如一个 ConfigUnittest.cc 会读取一个测试专用的配置文件：Config_unittest.ini 时，可以在对应目录下的 CMakeLists.txt 文件中添加

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# Config_unitest
add_executable(ConfigUnitTest Config_unittest.cc ${CMAKE_SOURCE_DIR}/Config.cc)
target_include_directories(ConfigUnitTest PUBLIC ${CMAKE_SOURCE_DIR})

target_link_libraries(ConfigUnitTest
    GTest::GTest 
    GTest::Main
)
# 将测试目录下的 Config_unittest.ini 文件拷贝至单元测试目录下
# TARGET ConfigUnitTest POST_BUILD 代表在目标 ConfigUnitTest 编译完成后，执行后续命令
# ${CMAKE_CURRENT_SOURCE_DIR}：代表当前源码目录
# $<TARGET_FILE_DIR:ConfigUnitTest>：代表被拷贝文件的目的地(TARGET_FILE_DIR)与 ConfigUnitTest 一致
add_custom_command(TARGET ConfigUnitTest POST_BUILD
    COMMAND ${CMAKE_COMMAND} -E 
    copy ${CMAKE_CURRENT_SOURCE_DIR}/Config_unittest.ini $<TARGET_FILE_DIR:ConfigUnitTest>)

注3：若希望将拷贝形式变成执行 make copy_ini_file 的形式，则可以采用下面的 cmake 命令

1
2
3
4
add_custom_target(copy_ini_file)
add_custom_command(TARGET copy_ini_file 
    COMMAND ${CMAKE_COMMAND} -E 
    copy ${CMAKE_CURRENT_SOURCE_DIR}/Config_unittest.ini $<TARGET_FILE_DIR:ConfigUnitTest>)

3rd_party/google-test/CmakeLists.txt.in 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
cmake_minimum_required(VERSION 3.0)

# NONE 代表该项目是非语言项目(通常语言默认为 C 或 CXX)
project(googletest-download NONE)

# ExternalProject 相当于是一个包，其中包含了许多的定义，例如下面所使用到的 ExternalProject_Add 函数
include(ExternalProject)

# Version bfc0ffc8a698072c794ae7299db9cb6866f4c0bc happens to be master when I set this up.
# To prevent an issue with accidentally installing GTest / GMock with your project you should use a
# commit after 9469fb687d040b60c8749b7617fee4e77c7f6409
# Note: This is after the release of v1.8
ExternalProject_Add(googletest
  URL               https://github.com/google/googletest/archive/bfc0ffc8a698072c794ae7299db9cb6866f4c0bc.tar.gz  # 下载 googletest 的网址
  SOURCE_DIR        "${CMAKE_CURRENT_BINARY_DIR}/googletest-src"  #下载后的源码目录
  BINARY_DIR        "${CMAKE_CURRENT_BINARY_DIR}/googletest-build"  # 编译好后的可执行文件目录
  CONFIGURE_COMMAND ""
  BUILD_COMMAND     ""
  INSTALL_COMMAND   ""
  TEST_COMMAND      ""
)

3rd_party/google-test/CmakeLists.txt 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# Download and unpack googletest at configure time
# See: http://crascit.com/2015/07/25/cmake-gtest/
configure_file(CMakeLists.txt.in googletest-download/CMakeLists.txt)
# Call CMake to download and Google Test (下载 Google Test 源码)
execute_process(COMMAND ${CMAKE_COMMAND} -G "${CMAKE_GENERATOR}" .
  RESULT_VARIABLE result
  WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/googletest-download )
if(result)
  message(FATAL_ERROR "CMake step for googletest failed: ${result}")
endif()
# Build the downloaded google test (编译 Google Test 源码)
execute_process(COMMAND ${CMAKE_COMMAND} --build .
  RESULT_VARIABLE result
  WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/googletest-download )
if(result)
  message(FATAL_ERROR "Build step for googletest failed: ${result}")
endif()

# Prevent overriding the parent project's compiler/linker
# settings on Windows  主要用于设置图形化界面，FORCE代表用用户指定的值覆盖原有的值
set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
# Prevent installation of GTest with your project
set(INSTALL_GTEST OFF CACHE BOOL "" FORCE)
set(INSTALL_GMOCK OFF CACHE BOOL "" FORCE)

# Add googletest directly to our build. This defines
# the gtest and gtest_main targets.
add_subdirectory(${CMAKE_CURRENT_BINARY_DIR}/googletest-src
                 ${CMAKE_CURRENT_BINARY_DIR}/googletest-build)

# Add aliases for GTest and GMock libraries 为 gtest 所用到的库添加别名
if(NOT TARGET GTest::GTest)
    add_library(GTest::GTest ALIAS gtest)
    add_library(GTest::Main ALIAS gtest_main)
endif()

if(NOT TARGET GTest::GMock)
    add_library(GMock::GMock ALIAS gmock)
    add_library(GMock::Main ALIAS gmock_main)
endif()

CMakeLists.txt 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
cmake_minimum_required(VERSION 3.5)

# Set the project name
project (google_test_example)

# Add an library for the example classes
add_library(example_google_test 
    Reverse.cpp
    Palindrome.cpp
)

#############################################
# Unit tests
# 先执行 3rd_party/google-test 下的 CMakeLists.txt 文件
add_subdirectory(3rd_party/google-test)

# enable CTest testing
enable_testing()

# Add a testing executable
add_executable(unit_tests unit_tests.cpp)

target_link_libraries(unit_tests
    example_google_test
    GTest::GTest 
    GTest::Main
)

add_test(test_all unit_tests)

CMakeLists.txt 编写总结：

  1. 最低版本要求 + 项目信息

  2. 将要测试的代码打包成待测库文件

  3. 执行 add_subdirectory 下载并构建 Google Test

  4. 打开测试开关

  5. 添加测试文件 unit_tests.cpp 为可执行目标 unit_tests

  6. 将 unit_tests 和待测库文件，以及 Google Test 的库文件链接到一起

  7. 执行 add_test 运行测试程序

unit_tests.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
#include <string>
#include "Reverse.h"
#include "Palindrome.h"
#include <gtest/gtest.h>

class ReverseTests : public ::testing::Test
{
};

TEST_F(ReverseTests, simple )
{
    std::string toRev = "Hello";
    Reverse rev;
    std::string res = rev.reverse(toRev);
    EXPECT_EQ(res, "olleH" );   
}

TEST_F(ReverseTests, empty )
{
    std::string toRev;
    Reverse rev;
    std::string res = rev.reverse(toRev);
    EXPECT_EQ(res, "" );
}

TEST_F(ReverseTests,  is_palindrome )
{
    std::string pal = "mom";
    Palindrome pally;
    EXPECT_TRUE(pally.isPalindrome(pal));
}

googletest 的基本知识

注意：由于历史原因，googletest 中所提到的部分术语和常规软件工程中的术语不同，以下有一张对应表格

googletest 中的术语 常规软件工程中所使用的术语 备注信息
Test Case Test Suited(测试套件) 由一个或多个测试用例所构成的测试
Test Test Cas(测试用例) 单个的测试代码

断言

  • Fatal Assertion (ASSERT_*)：一旦测试不通过，立即终止测试，并返回测试信息
  • Nonfatal Assertion(EXCEPT_*)：一旦测试不通过，报告测试信息，并接着执行下一个测试
Fatal assertion Nonfatal assertion Verifies
ASSERT_TRUE(condition); EXPECT_TRUE(condition); condition is true
ASSERT_FALSE(condition); EXPECT_FALSE(condition); condition is false
ASSERT_EQ(val1, val2); EXPECT_EQ(val1, val2); val1 == val2
ASSERT_NE(val1, val2); EXPECT_NE(val1, val2); val1 != val2
ASSERT_LT(val1, val2); EXPECT_LT(val1, val2); val1 < val2
ASSERT_LE(val1, val2); EXPECT_LE(val1, val2); val1 <= val2
ASSERT_GT(val1, val2); EXPECT_GT(val1, val2); val1 > val2
ASSERT_GE(val1, val2); EXPECT_GE(val1, val2); val1 >= val2
ASSERT_STREQ(str1,str2); EXPECT_STREQ(str1,str2); the two C strings have the same content
ASSERT_STRNE(str1,str2); EXPECT_STRNE(str1,str2); the two C strings have different contents
ASSERT_STRCASEEQ(str1,str2); EXPECT_STRCASEEQ(str1,str2); the two C strings have the same content, ignoring case
ASSERT_STRCASENE(str1,str2); EXPECT_STRCASENE(str1,str2); the two C strings have different contents, ignoring case

示例代码：

1
2
3
4
5
6
// 可在测试断言后用 << 运算符打印自己的信息
ASSERT_EQ(x.size(), y.size()) << "Vectors x and y are of unequal length";

for (int i = 0; i < x.size(); ++i) {
  EXPECT_EQ(x[i], y[i]) << "Vectors x and y differ at index " << i;
}

注意事项：

  1. 若 val1 和 val2 是用户自定义类型，则必须重载相应的二元运算符以支持对应的比较操作。由于 Google C++ Style 规定禁止重载运算符，因此只能使用如 ASSERT_TRUE( val1 < val2) 这样的形式

  2. 在一般情况下更倾向于使用 ASSERT_EQ(actual, expected) ，而不是 ASSERT_TRUE(actual == expected）。这主要是因为当测试失败时，前者能够返回更多的信息(actual 和 expected 在实际测试中的值)

  3. ASSERT_EQ() 用于判断两个指针是否相等时，只会判断这两个指针变量在内存中的位置是否相同，而并不会比较两个指针所指向的内容是否相同。要比较两个 C 风格的字符串字面值是否相等，应该采用 ASSERT_STREQ()。对于 string 对象则没有这个限制，依然可以使用 ASSERT_EQ

  4. 在判断一个 C 风格的指针是否为空或不为空时，使用 *_EQ(c_str, nullptr) 或者 *_NE(c_str, nullptr)。这是因为 nullptr 是一种类型，而 NULL 不是。而且 NULL 指针和空的 string 对象是不同的。

Test 的创建

一个简单的示例：

1
2
3
4
5
6
7
8
9
10
11
12
// Tests factorial of 0.
TEST(FactorialTest, HandlesZeroInput) {
  EXPECT_EQ(Factorial(0), 1);
}

// Tests factorial of positive numbers.
TEST(FactorialTest, HandlesPositiveInput) {
  EXPECT_EQ(Factorial(1), 1);
  EXPECT_EQ(Factorial(2), 2);
  EXPECT_EQ(Factorial(3), 6);
  EXPECT_EQ(Factorial(8), 40320);
}

在上述的例子中，使用了宏 Test(TestSuitName, TestName) 来定义和命名一个测试函数(无返回值)。TestSuitName 代表测试套件的名称，而 TestName 则代表单个测试的名称。由于测试结果会按照测试套件名称进行分组，因此逻辑上相关的测试最好集中到同一个测试套件当中。另外，不同的测试套件中的同名测试互不影响。TestSuitName 和 TestName 中不能带有 ‘_’ 字符

测试夹具(Test Fixture)【测试夹具可以为多个测试提供相同的数据配置】

一个简单的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
//待测试的源文件
template <typename E>  // E is the element type.
class Queue {
 public:
  Queue();
  void Enqueue(const E& element);
  E* Dequeue();  // Returns NULL if the queue is empty.
  size_t size() const;
  ...
};
//测试夹具类
class QueueTest : public ::testing::Test {
 protected:
  void SetUp() override {
     q1_.Enqueue(1);
     q2_.Enqueue(2);
     q2_.Enqueue(3);
  }
  // 由于没有动态申请内存，因此不需要定义 TearDown 函数
  // void TearDown() override {}

  Queue<int> q0_;
  Queue<int> q1_;
  Queue<int> q2_;
};

//定义测试夹具对象
TEST_F(QueueTest, IsEmptyInitially) {
  EXPECT_EQ(q0_.size(), 0);
}

TEST_F(QueueTest, DequeueWorks) {
  int* n = q0_.Dequeue();
  EXPECT_EQ(n, nullptr);

  n = q1_.Dequeue();
  ASSERT_NE(n, nullptr);
  EXPECT_EQ(*n, 1);
  EXPECT_EQ(q1_.size(), 0);
  delete n;

  n = q2_.Dequeue();
  ASSERT_NE(n, nullptr);
  EXPECT_EQ(*n, 2);
  EXPECT_EQ(q2_.size(), 1);
  delete n;
}

通过例子可以看出，使用测试夹具的基本流程如下：

  1. 从 ::testing::Test 派生一个测试夹具类(通常命名为 类名+Test)
  2. 在夹具类中定义稍后测试所需要使用的对象。注意测试夹具类中的成员均采用 protected 作为访问限定符，以便夹具类的子类也可以访问这些成员。
  3. 必要的情况下，编写构造函数或 SetUp 函数来进行夹具对象的初始化，并编写析构函数或 TearDown 函数来进行回收资源的工作
  4. 若有必要，为测试定义子例程

在执行上述测试代码的过程中，会发生以下情况：

  1. googletest 会创建一个 QueueTest 对象 t1
  2. 对 t1 执行 SetUp() 进行初始化
  3. 在 t1 上运行第一个测试 IsEmptyInitially
  4. 对 t1 执行 TearDown() 函数
  5. 创建一个 QueueTest 对象 t2
  6. 对 t2 对象执行 SetUp() 函数
  7. 在 t2 上运行第二个测试 DequeueWorks
  8. 对 t2 对象执行 TearDown() 函数

什么时候使用构造/析构函数，什么时候又应当使用 SetUp/TearDown 函数 ?

使用构造/析构函数的好处：

  1. 在创建 test fixture 对象的时候，可以将其设置为 const，从而避免意外修改到对象当中的数据
  2. 能够确保 test fixture 对象的派生对象在构造的时候优先构造基类子对象，析构的时候最后析构基类子对象

使用 SetUp / TearDwon 的优点：

  1. C++ 在构造函数中无法使用多态。因此在需要使用多态进行构造的场合，你只能使用 SetUp 和 TearDown 函数

  2. C++ 构造 / 析构函数的函数体中无法使用 ASSERT_* 断言

  3. 如果 TearDown 的操作可能会抛出异常，那么只能使用 TearDown 的方式

在 main 中调用测试

一个简单的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
#include "this/package/foo.h"
#include "gtest/gtest.h"

namespace my {
namespace project {
namespace {

// The fixture for testing class Foo.
class FooTest : public ::testing::Test {
 protected:
  // You can remove any or all of the following functions if their bodies would
  // be empty.

  FooTest() {
     // You can do set-up work for each test here.
  }

  ~FooTest() override {
     // You can do clean-up work that doesn't throw exceptions here.
  }

  // If the constructor and destructor are not enough for setting up
  // and cleaning up each test, you can define the following methods:

  void SetUp() override {
     // Code here will be called immediately after the constructor (right
     // before each test).
  }

  void TearDown() override {
     // Code here will be called immediately after each test (right
     // before the destructor).
  }

  // Class members declared here can be used by all tests in the test suite
  // for Foo.
};

// Tests that the Foo::Bar() method does Abc.
TEST_F(FooTest, MethodBarDoesAbc) {
  const std::string input_filepath = "this/package/testdata/myinputfile.dat";
  const std::string output_filepath = "this/package/testdata/myoutputfile.dat";
  Foo f;
  EXPECT_EQ(f.Bar(input_filepath, output_filepath), 0);
}

// Tests that Foo does Xyz.
TEST_F(FooTest, DoesXyz) {
  // Exercises the Xyz feature of Foo.
}

}  // namespace
}  // namespace project
}  // namespace my

int main(int argc, char **argv) {
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}

在上述的例子当中，googletest 会自动将创建好的测试套件以及测试夹具中的测试注册起来，并通过 RUN_ALL_TESTS()来调用所有的测试。对于 RUN_ALL_TESTS() 有以下两点注意事项：

  • 不要忽略 RUN_ALL_TESTS() 的返回值， 0 代表通过， 1 则代表失败。对于 main 函数而言，如果没有返回 RUN_ALL_TESTS() 的返回值会导致编译错误
  • RUN_ALL_TESTS() 只能调用一次。调用两次或以上的 RUN_ALL_TESTS() 和 googletest 的一些其他特性相冲突，因此并不被支持。
  • 在调用RUN_ALL_TESTS()之前需要调用::testing::InitGoogleTest(&argc, argv);以正确解析命令行参数。这可以使得用户能够通过命令行来控制测试程序的行为

googletest 提供了一个默认的 main 函数，如果你不需要在 main 函数中完成什么定制化的操作，则可以直接将测试程序连接到 gtest_main 当中即可。

展开全文 >>

Leetcode 493. 翻转对(归并排序+二分查找)

问题描述

给定一个数组 nums ，如果 i < j 且 nums[i] > 2*nums[j] 我们就将 (i, j) 称作一个重要翻转对。

你需要返回给定数组中的重要翻转对的数量。

Example

示例 1:

输入: [1,3,2,3,1]

输出: 2

示例 2:

输入: [2,4,3,5,1]

输出: 3

题目链接: 493. 翻转对(难度：困难)

思路

这道题可以采用归并排序 + 二分查找来解决。我们可以先考虑一个最简单的情况，如 [-10, -6]，我们可以先将 [-10, -10] 分割为 L 和 R 两个子数组，其中 L = [-10], R = [-6]。显然，L 和 R 中都只有一个元素，本身就是有序数组，同时翻转对数为0。将 L 和 R 合并排序，由于 L[0] > 2 * R[0], 得到排序后的数组为 [-10, -6], 且翻转对数为 1。

上述过程大体上描绘了代码主体的框架，要求出一个数组的翻转对数，我们需要将其划分为 L 和 R，并求出 L 和 R 中的翻转对数 v1 和 v2，然后对排好序的 L 和 R 进行合并排序，得到翻转对数 v3，则数组本身的翻转对数就是 v1 + v2 + v3。

这道题有两个值得注意的细节：

  • 在对 L 和 R 进行合并排序的过程中，由于 L 和 R 都是有序数组，因此可以使用二分查找来优化
  • 由于可能存在负数，因此可能会出现 L[i] < R[j], 但 L[i] > 2 * R[j] 的情况，对于这种情况，需要从头扫描整个数组，因为 L[i] 前面的元素也可能大于 2 * R[j]。若为正数，则直接扫描 L[i, end) 即可，因为L[i] 前面的元素小于 L[i]，而 L[i] < 2 * R[j]， 所以 L[i] 前面的元素也都不需要考虑了

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Solution {
public:
    using ull = unsigned long long;
    static const int MAX = 50000;
    long long L[MAX / 2 + 2] = {0}, R[MAX / 2 + 2] = {0};
    ull Merge(vector<int>& nums, int left, int mid, int right){
        int L_end = mid - left;
        int R_end = right - mid;
        for(int i = 0;i < L_end;++i)
            L[i] = nums[left + i];
        for(int i = 0;i < R_end;++i)
            R[i] = nums[mid + i];
        L[L_end] = static_cast<long long>(numeric_limits<int>::max()) + 2;
        R[R_end] = static_cast<long long>(numeric_limits<int>::max()) + 2;
        ull cnt = 0;
        int i = 0, j = 0;
        for(int k = left;k < right;++k){
            if(L[i] < R[j]){
                nums[k] = L[i++];
            }else{
                if(R[j] < 0)
                    cnt += L_end - (upper_bound(L, L + L_end, R[j] * 2) - L);
                else
                    cnt += L_end - (upper_bound(L + i, L + L_end, R[j] * 2) - L);
                nums[k] = R[j++];
            }
        }
        return cnt;
    }
    int MergeSort(vector<int>& nums, int left, int right){
        ull v1 = 0, v2 = 0, v3 = 0;
        if(left + 1 < right){
            int mid = (left + right) >> 1;
            v1 = MergeSort(nums, left, mid);
            v2 = MergeSort(nums, mid, right);
            v3 = Merge(nums, left, mid, right);
        }
        return v1 + v2 + v3;
    }
    int reversePairs(vector<int>& nums) {
        return MergeSort(nums, 0, nums.size());
    }
};

运行结果：456ms, 42 MB

相关练习：

剑指 Offer 51. 数组中的逆序对(难度：困难)

315. 计算右侧小于当前元素的个数(难度：困难)

展开全文 >>

Libevent 源码阅读笔记(三)、日志系统的实现

Libevent 如何定义自己的日志输出格式

Libevent 的日志系统中定义了四个日志的输出等级，默认情况下会将日志信息输出到终端。不过 Libevent 允许用户设置自定义的日志输出回调函数以及异常日志输出函数，以提供日志信息的进一步解释和输出重定向。回调函数的指针定义如下：

1
2
3
4
5
6
7
8
9
//event.h
//日志输出等级
#define EVENT_LOG_DEBUG 0   
#define EVENT_LOG_MSG   1
#define EVENT_LOG_WARN  2
#define EVENT_LOG_ERR   3
//用户自定义的日志输出回调函数，其中 severity 代表日志输出等级
typedef void (*event_log_cb)(int severity, const char *msg);
typedef void (*event_fatal_cb)(int err);

Libevent 默认将这两个回调函数置为 NULL，用户可以通过函数event_set_log_callback和event_set_fatal_callback进行设置

1
2
3
4
5
6
7
8
9
10
11
//log.c
static event_log_cb log_fn = NULL;
void event_set_log_callback(event_log_cb cb)
{
    log_fn = cb;
}
static event_fatal_cb fatal_fn = NULL;
void event_set_fatal_callback(event_fatal_cb cb)
{
    fatal_fn = cb;
}

若用户没有自定义日志输出函数，那么 Libevent 会根据 severity 的值调用相应的默认日志输出函数，并向 stderr 输出日志信息。具体的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
//log.c
static void event_log(int severity, const char *msg)
{
    if (log_fn)
        log_fn(severity, msg);
    else {
        const char *severity_str;
        switch (severity) {
        case EVENT_LOG_DEBUG:
            severity_str = "debug";
            break;
        case EVENT_LOG_MSG:
            severity_str = "msg";
            break;
        case EVENT_LOG_WARN:
            severity_str = "warn";
            break;
        case EVENT_LOG_ERR:
            severity_str = "err";
            break;
        default:
            severity_str = "???";
            break;
        }
        //(void)的作用是减少编译器因为“未被使用的返回值”而产生的警告信息。
        (void)fprintf(stderr, "[%s] %s\n", severity_str, msg);
    }
}

static void event_exit(int errcode)
{
    if (fatal_fn) {
        fatal_fn(errcode);
        exit(errcode); /* should never be reached */
    } else if (errcode == EVENT_ERR_ABORT_)
        abort();
    else
        exit(errcode);
}

日志系统的底层函数

看完了 Libevent 的日志系统留给用户的两个接口定义后，我们可以来看看 Libevent 默认的日志输出函数。在源码当中，日志系统相关的文件主要有两个: log.c 和 log-internal.h。我们先来看 log-internal.h 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// log-internal.h 
#ifdef __GNUC__
#define EV_CHECK_FMT(a,b) __attribute__((format(printf, a, b)))
#define EV_NORETURN __attribute__((noreturn))
#else
#define EV_CHECK_FMT(a,b)
#define EV_NORETURN
#endif

//EVENT_DEBUG_LOGGING_ENABLED 为调试模式的开关
#ifdef EVENT_DEBUG_LOGGING_ENABLED
EVENT2_CORE_EXPORT_SYMBOL extern ev_uint32_t event_debug_logging_mask_;
#define event_debug_get_logging_mask_() (event_debug_logging_mask_)
#else
#define event_debug_get_logging_mask_() (0)
#endif

//EVENT_ERR_ABORT_ 代表终止程序时的错误码
#define EVENT_ERR_ABORT_ ((int)0xdeaddead)

/* EVENT2_EXPORT_SYMBOL是定义在 include/event2/visibility.h 下的一个宏
 * define EVENT2_EXPORT_SYMBOL __attribute__ ((visibility("default")))
 */
EVENT2_EXPORT_SYMBOL
void event_err(int eval, const char *fmt, ...) EV_CHECK_FMT(2,3) EV_NORETURN;
EVENT2_EXPORT_SYMBOL
void event_warn(const char *fmt, ...) EV_CHECK_FMT(1,2);
EVENT2_EXPORT_SYMBOL
void event_sock_err(int eval, evutil_socket_t sock, const char *fmt, ...) EV_CHECK_FMT(3,4) EV_NORETURN;
EVENT2_EXPORT_SYMBOL
void event_sock_warn(evutil_socket_t sock, const char *fmt, ...) EV_CHECK_FMT(2,3);
EVENT2_EXPORT_SYMBOL
void event_errx(int eval, const char *fmt, ...) EV_CHECK_FMT(2,3) EV_NORETURN;
EVENT2_EXPORT_SYMBOL
void event_warnx(const char *fmt, ...) EV_CHECK_FMT(1,2);
EVENT2_EXPORT_SYMBOL
void event_msgx(const char *fmt, ...) EV_CHECK_FMT(1,2);
EVENT2_EXPORT_SYMBOL
void event_debugx_(const char *fmt, ...) EV_CHECK_FMT(1,2);

EVENT2_EXPORT_SYMBOL
void event_logv_(int severity, const char *errstr, const char *fmt, va_list ap)
    EV_CHECK_FMT(3,0);

在 log-internal.h 中，我们可以看到 Libevent 中的日志系统其实运用到了编译器属性 attribute 的一些技巧。

attribute 是 GNU C 下的一种机制，可以用来设置函数属性、变量属性以及类型属性，它的语法格式如下

1
__attribute__((attribute-list))

其中 attribute-list 是用逗号分隔开的列表。在 Libevent 的日志系统中，主要应用到的有以下三种 attribute 属性：

1. attribute((format(archetype, string-index, first-to-check)))

该属性可以为函数添加上类似于 print、scanf、strftime 以及 strfmon 风格的格式化字符串，从而可以帮助我们完成对格式化字符串的类型检查。其中，archetype 代表参数的格式化风格，string-index 代表格式化字符串位于形参列表中的哪个位置，first-to-check 代表可变参数列表在形参列表中的哪个位置。结合一个 event_warn 的例子：

void event_warn(const char *fmt, …) attribute((format(printf, 1, 2)));

attribute((format(printf, 1, 2))) 代表按照 printf 的风格检查格式化字符串，其中参数列表的第 1 个参数代表格式化字符串，第 2 个参数代表可变参数列表。

2. attribute((noreturn))

noreturn 是一个函数属性，用来通知编译器该函数在执行后不会将控制权返回给它的调用者。以函数 event_err为例，当程序使用该日志函数时，说明程序已然产生了严重的错误，不能再继续执行下去，除了抛出异常，或者调用如exit()或abort()函数终止程序以外，可以利用 noreturn 来处理。当函数执行 event_err 时，主程序会将控制权交给 event_err 函数，当 event_err 执行完毕后，控制流将不会返回主程序，而是停留在此处。

3. attribute ((visibility(“default”)))

visibility属性可以将从动态库中导出可见符号，default 代表可见，而 hidden 代表隐藏。新版本的 Linux 内核限制了动态库的符号可见性，因为这样做可以提高程序的模块性，同时动态库装载和识别的符号越少，程序启动和运行的速度也就越快。Libevent 中被 EVENT2_EXPORT_SYMBOL 修饰的函数符号都将会被导出，因此使用该动态库程序都可以使用到这些函数。

对于event_err，event_sock_err，event_debugx_等函数，它们的实现大同小异，区别仅是部分参数的不同，以下选取部分函数进行讨论

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
void event_err(int eval, const char *fmt, ...)
{
    va_list ap;
    va_start(ap, fmt);
    event_logv_(EVENT_LOG_ERR, strerror(errno), fmt, ap);
    va_end(ap);
    event_exit(eval);
}

void event_sock_err(int eval, evutil_socket_t sock, const char *fmt, ...)
{
    va_list ap;
    int err = evutil_socket_geterror(sock);

    va_start(ap, fmt);
    event_logv_(EVENT_LOG_ERR, evutil_socket_error_to_string(err), fmt, ap);
    va_end(ap);
    event_exit(eval);
}

void event_debugx_(const char *fmt, ...)
{
    va_list ap;

    va_start(ap, fmt);
    event_logv_(EVENT_LOG_DEBUG, NULL, fmt, ap);
    va_end(ap);
}

void event_logv_(int severity, const char *errstr, const char *fmt, va_list ap)
{
    char buf[1024];
    size_t len;
    //当 event_debug_get_logging_mask_() == 0 时关闭调试日志的输出
    if (severity == EVENT_LOG_DEBUG && !event_debug_get_logging_mask_())
        return;

    if (fmt != NULL)
        //将日志信息按照格式字符串进行格式化，并存入 buf 中
        evutil_vsnprintf(buf, sizeof(buf), fmt, ap);
    else
        buf[0] = '\0';
    //若有额外的错误信息，则将其一起放入 buf 
    if (errstr) {
        len = strlen(buf);
        if (len < sizeof(buf) - 3) {
            evutil_snprintf(buf + len, sizeof(buf) - len, ": %s", errstr);
        }
    }

    event_log(severity, buf);
}

上述的日志函数大体的结构是相同的，都采用了可变参数列表的形式实现，最终通过event_logv_以及event_log两个函数将日志输出。

一些语法糖

在 log-internal.h 当中还定义了一个 event_debug 的宏，可以用于在调试期间打印出调试信息。利用这个宏来代替event_debugx_函数，一方面可以使得调试语义更加清晰，另一方面也可以非常方便地实现调试模式的开关。它的定义如下：

1
2
3
4
5
6
7
8
9
#ifdef EVENT_DEBUG_LOGGING_ENABLED
#define event_debug(x) do {         \
    if (event_debug_get_logging_mask_()) {  \
        event_debugx_ x;        \
    }                   \
    } while (0)
#else
#define event_debug(x) ((void)0)
#endif

总结

  1. Libevent 的日志系统为用户提供了两个接口event_log_cb和event_fatal_cb，以便用户实现自定义的日志处理功能，在需要的时候可以使用event_set_log_callback和event_set_fatal_callback进行设置
  2. 如果用户没有定义自己的日志处理回调函数，则采用 Libevent 默认的日志处理系统，它将日志输出等级定义为四个，并会将日志输出到终端。
  3. 在日志系统的开发过程当中，利用好 GNU C 的一些机制，如 attribute 等，可以极大简化一些繁琐的检查过程，不仅有利于调试，帮助找到一些难以发现的错误，还有助于编译器进行优化。

展开全文 >>

Libevent 源码阅读笔记(二)、从一个简单例子开始(下)

阅读本文之前请确保你看过上一篇文章：Libevent 源码阅读笔记(一)、从一个简单例子开始(上)

在上一篇文章中，我们提到了关于 Libevent 使用的一个基本逻辑：

  1. 调用event_init()创建 event_base 对象，一个 event_base 对象相当于一个 Reactor 实例
  2. 创建具体的事件处理器，其中 evsignal_new()创建的是信号事件处理器和evtimer_new()创建的是定时事件处理器
  3. 调用event_add()将事件处理器插入到注册事件队列中，并将其对应的事件插入到事件多路分发器中，相当于 Reactor 中的register_handler方法
  4. 调用event_base_dispatch()执行事件循环
  5. 循环结束后，调用*_free()释放系统资源

在这篇文章中，我们将继续分析剩余的后 3 个步骤

将事件处理器添加到注册事件队列中

要了解如何将事件处理器插入到注册事件队列中，就需要先知道 Libevent 是如何将句柄和注册事件之间的联系定义出来。其相关的数据结构主要定义在 event-internal.h 和 evmap.c 文件中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
//event-internal.h
//若定义了 EVMAP_USE_HT 则将 event_io_map 定义为哈希表。该哈希表存储了 event_map_entry 对象和 I/O 事件队列之间的映射关系
#ifdef EVMAP_USE_HT
#define HT_NO_CACHE_HASH_VALUES
#include "ht-internal.h"
struct event_map_entry;
HT_HEAD(event_io_map, event_map_entry);
#else
#define event_io_map event_signal_map
#endif

/* Used to map signal numbers to a list of events.  If EVMAP_USE_HT is not
   defined, this structure is also used as event_io_map, which maps fds to a
   list of events.
*/
struct event_signal_map {
    void **entries; //用于存放 evmap_io 或 evmap_signal 数组
    /* The number of entries available in entries */
    int nentries;   //entries 的大小
};

// evmap.c 文件
/** An entry for an evmap_io list: notes all the events that want to read or
    write on a given fd, and the number of each.
  */
struct evmap_io {
    struct event_dlist events;  //I/O 事件队列
    ev_uint16_t nread;      //记录事件队列中的读事件数量
    ev_uint16_t nwrite;     //记录事件队列中的写事件数量
    ev_uint16_t nclose;     //记录事件队列中的关闭事件数量
};

/* An entry for an evmap_signal list: notes all the events that want to know
   when a signal triggers. */
struct evmap_signal {
    struct event_dlist events;  //信号事件队列
};

/* On some platforms, fds start at 0 and increment by 1 as they are
   allocated, and old numbers get used.  For these platforms, we
   implement io maps just like signal maps: as an array of pointers to
   struct evmap_io.  But on other platforms (windows), sockets are not
   0-indexed, not necessarily consecutive, and not necessarily reused.
   There, we use a hashtable to implement evmap_io.
*/
#ifdef EVMAP_USE_HT
//哈希表中的节点类型：以 fd 为键， ent 为值
struct event_map_entry {
    HT_ENTRY(event_map_entry) map_node;
    evutil_socket_t fd;
    union { /* This is a union in case we need to make more things that can
               be in the hashtable. */
        struct evmap_io evmap_io;
    } ent;
};

创建好的 event 事件可以利用 event_add()插入相应的注册事件队列中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
//向注册事件队列中添加事件处理器
int event_add(struct event *ev, const struct timeval *tv)
{
    int res;
    if (EVUTIL_FAILURE_CHECK(!ev->ev_base)) {
        event_warnx("%s: event has no event_base set.", __func__);
        return -1;
    }
    EVBASE_ACQUIRE_LOCK(ev->ev_base, th_base_lock);
    res = event_add_nolock_(ev, tv, 0);
    EVBASE_RELEASE_LOCK(ev->ev_base, th_base_lock);
    return (res);
}
int event_add_nolock_(struct event *ev, const struct timeval *tv,int tv_is_absolute)
{
    struct event_base *base = ev->ev_base;
    int res = 0;
    int notify = 0;

    EVENT_BASE_ASSERT_LOCKED(base);
    event_debug_assert_is_setup_(ev);

    event_debug((
         "event_add: event: %p (fd "EV_SOCK_FMT"), %s%s%s%scall %p",
         ev,
         EV_SOCK_ARG(ev->ev_fd),
         ev->ev_events & EV_READ ? "EV_READ " : " ",
         ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
         ev->ev_events & EV_CLOSED ? "EV_CLOSED " : " ",
         tv ? "EV_TIMEOUT " : " ",
         ev->ev_callback));

    EVUTIL_ASSERT(!(ev->ev_flags & ~EVLIST_ALL));

    if (ev->ev_flags & EVLIST_FINALIZING) {
        /* XXXX debug */
        return (-1);
    }

    /* 如果添加的事件处理器是一个尚未被插入到通用定时器队列或时间堆内的定时器，则为该定时器在时间堆上预留一个位置*/
    if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)) {
        if (min_heap_reserve_(&base->timeheap,
            1 + min_heap_size_(&base->timeheap)) == -1)
            return (-1);  /* ENOMEM == errno */
    }

    /* 如果当前调用者不是主线程，且被添加的是信号事件处理器，而主线程也正在调用该信号事件的回调函数，则当前调用者必须等待主线程完成调用，以避免导致竞态的发生 */
#ifndef EVENT__DISABLE_THREAD_SUPPORT
    if (base->current_event == event_to_event_callback(ev) &&
        (ev->ev_events & EV_SIGNAL)
        && !EVBASE_IN_THREAD(base)) {
        ++base->current_event_waiters;
        EVTHREAD_COND_WAIT(base->current_event_cond, base->th_base_lock);
    }
#endif
    
    if ((ev->ev_events & (EV_READ|EV_WRITE|EV_CLOSED|EV_SIGNAL)) &&
        !(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE|EVLIST_ACTIVE_LATER))) {
        if (ev->ev_events & (EV_READ|EV_WRITE|EV_CLOSED))
            //添加 IO 事件和 IO 事件处理器的映射关系
            res = evmap_io_add_(base, ev->ev_fd, ev);
        else if (ev->ev_events & EV_SIGNAL)
            //添加信号事件和信号事件处理器的映射关系
            res = evmap_signal_add_(base, (int)ev->ev_fd, ev);
        if (res != -1)
            //将事件处理器插入到注册事件队列
            event_queue_insert_inserted(base, ev);
        if (res == 1) {
            //通知主线程，事件多路分发器中添加了新的事件
            notify = 1;
            res = 0;
        }
    }

    /* 在前一个事件添加成功，且该事件是一个定时器的话，将其添加至时间堆或通用定时器队列当中。*/
    if (res != -1 && tv != NULL) {
        struct timeval now;
        int common_timeout;
#ifdef USE_REINSERT_TIMEOUT
        int was_common;
        int old_timeout_idx;
#endif

        /* 对于永久性事件处理器，如果其超时时间不是绝对时间，则将该事件的超时时间保存到 ev->ev_io_timeout 中*/
        if (ev->ev_closure == EV_CLOSURE_EVENT_PERSIST && !tv_is_absolute)
            ev->ev_io_timeout = *tv;

#ifndef USE_REINSERT_TIMEOUT
        //若该事件处理器已经被插入到通用定时器队列或时间堆当中，则先对其进行删除
        if (ev->ev_flags & EVLIST_TIMEOUT) {
            event_queue_remove_timeout(base, ev);
        }
#endif

        /* Check if it is active due to a timeout.  Rescheduling
         * this timeout before the callback can be executed
         * removes it from the active list. */
        //如果待添加的事件处理器因超时而被激活，则要从活动事件队列当中删除它以便重新调度
        if ((ev->ev_flags & EVLIST_ACTIVE) &&
            (ev->ev_res & EV_TIMEOUT)) {
            if (ev->ev_events & EV_SIGNAL) {
                //若事件处理器的回调函数正在执行当中，则设置 ev_ncalls 的值为 0，就可以退出回调函数的执行循环
                if (ev->ev_ncalls && ev->ev_pncalls) {
                    /* Abort loop */
                    *ev->ev_pncalls = 0;
                }
            }

            event_queue_remove_active(base, event_to_event_callback(ev));
        }

        gettime(base, &now);

        common_timeout = is_common_timeout(tv, base);   //用于判断应该将定时器插入到通用定时器队列还是时间堆当中
#ifdef USE_REINSERT_TIMEOUT
        was_common = is_common_timeout(&ev->ev_timeout, base);
        old_timeout_idx = COMMON_TIMEOUT_IDX(&ev->ev_timeout);
#endif

        if (tv_is_absolute) {
            ev->ev_timeout = *tv;
        } else if (common_timeout) {
            struct timeval tmp = *tv;
            tmp.tv_usec &= MICROSECONDS_MASK;
            evutil_timeradd(&now, &tmp, &ev->ev_timeout);
            ev->ev_timeout.tv_usec |=
                (tv->tv_usec & ~MICROSECONDS_MASK);
        } else {
            //加上当前的系统时间以获得定时器超时的绝对时间
            evutil_timeradd(&now, tv, &ev->ev_timeout);
        }

        event_debug((
             "event_add: event %p, timeout in %d seconds %d useconds, call %p",
             ev, (int)tv->tv_sec, (int)tv->tv_usec, ev->ev_callback));
//插入定时器
#ifdef USE_REINSERT_TIMEOUT
        //根据 common_timeout 决定插入的位置是通用定时器队列还是时间堆
        event_queue_reinsert_timeout(base, ev, was_common, common_timeout, old_timeout_idx);
#else
        event_queue_insert_timeout(base, ev);
#endif

        if (common_timeout) {
            struct common_timeout_list *ctl =
                get_common_timeout_list(base, &ev->ev_timeout);
            //若被插入的定时器是通用定时器队列的首元素，则调用 common_timeout_schedule 将其转移到时间堆单中，这样可以使得时间堆和定时器链表的定时器得到相同的处理。
            if (ev == TAILQ_FIRST(&ctl->events)) {
                common_timeout_schedule(ctl, &now, ev);
            }
        } else {
            struct event* top = NULL;
            /* See if the earliest timeout is now earlier than it
             * was before: if so, we will need to tell the main
             * thread to wake up earlier than it would otherwise.
             * We double check the timeout of the top element to
             * handle time distortions due to system suspension.
             */
            //判断当前的定时器是否位于最小堆的堆顶，如果是的话，通知主线程以便更新 tick 的周期
            if (min_heap_elt_is_top_(ev))
                notify = 1;
            //若定时器不位于堆顶，则取出堆顶的定时器，判断是否小于当前时间，如果是，则通知主线程
            else if ((top = min_heap_top_(&base->timeheap)) != NULL &&
                     evutil_timercmp(&top->ev_timeout, &now, <))
                notify = 1;
            //猜测：time distortions：在一般情况下，主线程每次 tick 都对应了时间堆堆顶的定时器，因此不会出现堆顶定时器的时间小于当前时间的情况。但考虑到主线程可能因为阻塞无法及时发出 tick 信号，因此如果检测到堆顶定时器的时间小于当前时间，那说明发生了 time distortions，需要通知主线程，让它尽快 tick
        }
    }

    /* 若有必要，唤醒主线程 */
    if (res != -1 && notify && EVBASE_NEED_NOTIFY(base))
        evthread_notify_base(base);

    event_debug_note_add_(ev);

    return (res);
}

可以看到，在整个 event_add_nolock_函数中，主要的工作是为不同的事件提供对应的操作，主要分为 3 类操作，

  • 调用 evmap_io_add_ 将 I/O 事件添加到事件多路分发器中，并将对应的事件处理器添加到 I/O 事件队列当中，同时建立起 I/O 事件和 I/O 事件处理器之间的映射，最后将 I/O 事件处理器插入到注册事件队列中
  • 调用 evmap_signal_add_ 将信号事件添加到事件多路分发器中，并将对应的事件处理器添加到信号事件队列当中，同时建立起信号事件和信号事件处理器之间的映射，最后将信号事件处理器插入到注册事件队列中
  • 调用 event_queue_insert_inserted修改 event_base 的事件处理器总数
  • 根据是否允许将定时器重新插入到通用定时器队列或时间堆中，调用event_queue_reinsert_timeout和event_queue_insert_timeout来将定时器插入到通用定时器队列或时间堆中

接下来，对于注册事件，Libevent 通过evmap_io_add_、 evmap_signal_add_ 以及 event_queue_insert_inserted进行将它们插入到对应的事件队列当中，其中evmap_io_add_、 evmap_signal_add_ 被定义在 evmap.c 文件中，而event_queue_insert_inserted被定义在 event.c 文件中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
// evmap.c 文件
int evmap_io_add_(struct event_base *base, evutil_socket_t fd, struct event *ev)
{
    //获得 event_base 的后端 I/O 复用机制实例
    const struct eventop *evsel = base->evsel;
    //获得 event_base 中文件描述符与 I/O 事件队列的映射表(哈希表或数组)，event_io_map 由宏 HT_HEAD 定义
    struct event_io_map *io = &base->io;
    //ctx代表 fd 所对应的 I/O 事件队列
    struct evmap_io *ctx = NULL;
    int nread, nwrite, nclose, retval = 0;
    short res = 0, old = 0;
    struct event *old_ev;

    EVUTIL_ASSERT(fd == ev->ev_fd);

    if (fd < 0)
        return 0;

#ifndef EVMAP_USE_HT
    if (fd >= io->nentries) {
        //当 fd 超过了映射表 io 的最大容量时，扩充哈希表
        if (evmap_make_space(io, fd, sizeof(struct evmap_io *)) == -1)
            return (-1);
    }
#endif
    //在映射表 io 中获取 ctx 和 fd 之间的映射关系，若没有映射关系，则建立对应的映射关系
    GET_IO_SLOT_AND_CTOR(ctx, io, fd, evmap_io, evmap_io_init,
                         evsel->fdinfo_len);

    nread = ctx->nread;
    nwrite = ctx->nwrite;
    nclose = ctx->nclose;

    if (nread)
        old |= EV_READ;
    if (nwrite)
        old |= EV_WRITE;
    if (nclose)
        old |= EV_CLOSED;
    //根据 ev 的事件类型，增加 I/O 事件队列中的相应的事件数量
    if (ev->ev_events & EV_READ) {
        if (++nread == 1)
            res |= EV_READ;
    }
    if (ev->ev_events & EV_WRITE) {
        if (++nwrite == 1)
            res |= EV_WRITE;
    }
    if (ev->ev_events & EV_CLOSED) {
        if (++nclose == 1)
            res |= EV_CLOSED;
    }
    if (EVUTIL_UNLIKELY(nread > 0xffff || nwrite > 0xffff || nclose > 0xffff)) {
        event_warnx("Too many events reading or writing on fd %d",
            (int)fd);
        return -1;
    }
    if (EVENT_DEBUG_MODE_IS_ON() &&
        (old_ev = LIST_FIRST(&ctx->events)) &&
        (old_ev->ev_events&EV_ET) != (ev->ev_events&EV_ET)) {
        event_warnx("Tried to mix edge-triggered and non-edge-triggered"
            " events on fd %d", (int)fd);
        return -1;
    }

    if (res) {
        void *extra = ((char*)ctx) + sizeof(struct evmap_io);
        /* XXX(niels): we cannot mix edge-triggered and
         * level-triggered, we should probably assert on
         * this. */
        //在事件多路分发器中注册事件。add 函数是事件多路分发器的接口。
        if (evsel->add(base, ev->ev_fd,
            old, (ev->ev_events & EV_ET) | res, extra) == -1)
            return (-1);
        retval = 1;
    }

    ctx->nread = (ev_uint16_t) nread;
    ctx->nwrite = (ev_uint16_t) nwrite;
    ctx->nclose = (ev_uint16_t) nclose;
    //将 ev 插入到 I/O 事件队列 ctx 的头部
    LIST_INSERT_HEAD(&ctx->events, ev, ev_io_next);

    return (retval);
}

int evmap_signal_add_(struct event_base *base, int sig, struct event *ev)
{
    const struct eventop *evsel = base->evsigsel;
    //获得 event_base 中信号值与信号事件队列的映射表(哈希表或数组)，event_signal_map 由宏 HT_HEAD 定义
    struct event_signal_map *map = &base->sigmap;
    struct evmap_signal *ctx = NULL;

    if (sig < 0 || sig >= NSIG)
        return (-1);
    //若信号值超出了 event_signal_map 的索引范围，则对 event_signal_map 进行扩容
    if (sig >= map->nentries) {
        if (evmap_make_space(
            map, sig, sizeof(struct evmap_signal *)) == -1)
            return (-1);
    }
    //在映射表 signal 中获取 ctx 和信号值之间的映射关系，若没有映射关系，则建立对应的映射关系
    GET_SIGNAL_SLOT_AND_CTOR(ctx, map, sig, evmap_signal, evmap_signal_init,
        base->evsigsel->fdinfo_len);
    //若 ctx 中的信号事件队列不为空，则调用 I/O 复用机制 evsel 中的 add 函数，对信号事件进行注册
    if (LIST_EMPTY(&ctx->events)) {
        if (evsel->add(base, ev->ev_fd, 0, EV_SIGNAL, NULL)
            == -1)
            return (-1);
    }
    //插入信号事件处理器
    LIST_INSERT_HEAD(&ctx->events, ev, ev_signal_next);

    return (1);
}

//event.c
static void event_queue_insert_inserted(struct event_base *base, struct event *ev)
{
    EVENT_BASE_ASSERT_LOCKED(base);

    if (EVUTIL_FAILURE_CHECK(ev->ev_flags & EVLIST_INSERTED)) {
        event_errx(1, "%s: %p(fd "EV_SOCK_FMT") already inserted", __func__,
            ev, EV_SOCK_ARG(ev->ev_fd));
        return;
    }
    //更新 base 当中 ev->ev_flags 类型的计数器
    INCR_EVENT_COUNT(base, ev->ev_flags);
    //标记事件 ev 已经被插入到注册事件队列中
    ev->ev_flags |= EVLIST_INSERTED;
}

对于超时事件，Libevent 则调用了event_queue_insert_timeout来将定时器插入到通用定时器队列或时间堆。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
static void event_queue_insert_timeout(struct event_base *base, struct event *ev)
{
    EVENT_BASE_ASSERT_LOCKED(base);

    if (EVUTIL_FAILURE_CHECK(ev->ev_flags & EVLIST_TIMEOUT)) {
        event_errx(1, "%s: %p(fd "EV_SOCK_FMT") already on timeout", __func__,
            ev, EV_SOCK_ARG(ev->ev_fd));
        return;
    }
    //更新 base 中的超时事件计数器
    INCR_EVENT_COUNT(base, ev->ev_flags);
    //标记以将这个超时事件添加到通用定时器链表或时间堆当中
    ev->ev_flags |= EVLIST_TIMEOUT;
    //判断该定时器的类型是否为通用定时器，若是则插入到通用定时器链表中，否则插入到时间堆当中
    if (is_common_timeout(&ev->ev_timeout, base)) {
        struct common_timeout_list *ctl =
            get_common_timeout_list(base, &ev->ev_timeout);
        insert_common_timeout_inorder(ctl, ev);
    } else {
        min_heap_push_(&base->timeheap, ev);
    }
}

这里需要指出，早期的 Libevent 使用了时间堆来设计定时器。由于在某些场合下，升序定时器链表的效率会更高，因此新版本的 Libevent 同时支持通用定时器链表和时间堆的操作。

Libevent 的核心动力 —— 事件循环

正如前面提到，Libevent 是基于事件驱动的，整个框架库的运转核心在于事件驱动：Reactor 在接收到相应事件后，通过 event_base_dispatch() 的方法，将事件分派给对应的事件处理器去处理。event_base_dispatch()定义在 event.c 文件中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
int event_base_dispatch(struct event_base *event_base)
{
    return (event_base_loop(event_base, 0));
}
int event_base_loop(struct event_base *base, int flags)
{
    const struct eventop *evsel = base->evsel;
    struct timeval tv;
    struct timeval *tv_p;
    int res, done, retval = 0;

    /* Grab the lock.  We will release it inside evsel.dispatch, and again
     * as we invoke user callbacks. */
    EVBASE_ACQUIRE_LOCK(base, th_base_lock);
    //一个 event_base 仅允许执行一个事件循环
    if (base->running_loop) {
        event_warnx("%s: reentrant invocation.  Only one event_base_loop"
            " can run on each event_base at once.", __func__);
        EVBASE_RELEASE_LOCK(base, th_base_lock);
        return -1;
    }

    base->running_loop = 1;

    clear_time_cache(base);
    //设置信号事件的 event_base 实例
    if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
        evsig_set_base_(base);

    done = 0;

#ifndef EVENT__DISABLE_THREAD_SUPPORT
    //获取当前线程 ID
    base->th_owner_id = EVTHREAD_GET_ID();
#endif

    base->event_gotterm = base->event_break = 0;

    while (!done) {
        base->event_continue = 0;
        base->n_deferreds_queued = 0;

        /* Terminate the loop if we have been asked to */
        //这几个值代表了事件执行后应当对其执行什么操作，具体可看上篇文章中 event_base 的注释代码
        if (base->event_gotterm) {
            break;
        }

        if (base->event_break) {
            break;
        }

        tv_p = &tv;
        if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {
            //获取时间堆的堆顶元素的超时值，并设为此次事件的超时值
            timeout_next(base, &tv_p);
        } else {
            //若有就绪事件尚未处理，则将I/O复用系统调用的超时时间设置为0，这样系统调用会直接返回，程序也就可以立即处理就绪事件了
            evutil_timerclear(&tv);
        }

        //如果 event_base 中没有注册任何事件，则直接退出事件循环
        if (0==(flags&EVLOOP_NO_EXIT_ON_EMPTY) &&
            !event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
            event_debug(("%s: no events registered.", __func__));
            retval = 1;
            goto done;
        }

        event_queue_make_later_events_active(base);

        clear_time_cache(base);
        //调用多路事件分发器的 dispatch 方法分发事件，将就绪事件插入到活动事件队列中
        res = evsel->dispatch(base, tv_p);

        if (res == -1) {
            event_debug(("%s: dispatch returned unsuccessfully.",
                __func__));
            retval = -1;
            goto done;
        }
        //更新时间缓冲为当前的系统时间
        update_time_cache(base);
        //检查时间堆上的到期事件并处理
        timeout_process(base);

        if (N_ACTIVE_CALLBACKS(base)) {
            //调用 event_process_active 处理活动事件
            int n = event_process_active(base);
            if ((flags & EVLOOP_ONCE)
                && N_ACTIVE_CALLBACKS(base) == 0
                && n != 0)
                done = 1;
        } else if (flags & EVLOOP_NONBLOCK)
            done = 1;
    }
    event_debug(("%s: asked to terminate loop.", __func__));

done:
    //循环结束时的工作：清空时间缓存，并设置停止循环标志位
    clear_time_cache(base);
    base->running_loop = 0;

    EVBASE_RELEASE_LOCK(base, th_base_lock);

    return (retval);
}

从代码中可以看出，event_base_dispatch()调用了event_base_loop()，而event_base_loop()的关键是

调用了 event_base 中的evsel->dispatch()，该方法封装了 I/O 复用系统调用。至于如何根据编程环境的不同，自动选择合适的 I/O 复用技术，可以看上一篇文章中关于 eventop 的介绍

释放资源

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
static int event_base_free_queues_(struct event_base *base, int run_finalizers)
{
    int deleted = 0, i;
    //释放掉活动事件队列中的剩余事件
    for (i = 0; i < base->nactivequeues; ++i) {
        struct event_callback *evcb, *next;
        for (evcb = TAILQ_FIRST(&base->activequeues[i]); evcb; ) {
            next = TAILQ_NEXT(evcb, evcb_active_next);
            deleted += event_base_cancel_single_callback_(base, evcb, run_finalizers);
            evcb = next;
        }
    }
    //释放延后活动事件队列中的事件
    {
        struct event_callback *evcb;
        while ((evcb = TAILQ_FIRST(&base->active_later_queue))) {
            deleted += event_base_cancel_single_callback_(base, evcb, run_finalizers);
        }
    }

    return deleted;
}
static int event_base_cancel_single_callback_(struct event_base *base,
    struct event_callback *evcb,
    int run_finalizers)
{
    int result = 0;
    //将已初始化的事件处理器释放掉
    if (evcb->evcb_flags & EVLIST_INIT) {
        struct event *ev = event_callback_to_event(evcb);
        if (!(ev->ev_flags & EVLIST_INTERNAL)) {
            event_del_(ev, EVENT_DEL_EVEN_IF_FINALIZING);
            result = 1;
        }
    } else {
        EVBASE_ACQUIRE_LOCK(base, th_base_lock);
        event_callback_cancel_nolock_(base, evcb, 1);
        EVBASE_RELEASE_LOCK(base, th_base_lock);
        result = 1;
    }

    if (run_finalizers && (evcb->evcb_flags & EVLIST_FINALIZING)) {
        switch (evcb->evcb_closure) {
        //对于类型为 EV_CLOSURE_EVENT_FINALIZE 和 EV_CLOSURE_EVENT_FINALIZE_FREE 的事件，调用它们相应的回调函数，结束后释放掉它们的资源
        case EV_CLOSURE_EVENT_FINALIZE:
        case EV_CLOSURE_EVENT_FINALIZE_FREE: {
            struct event *ev = event_callback_to_event(evcb);
            ev->ev_evcallback.evcb_cb_union.evcb_evfinalize(ev, ev->ev_arg);
            if (evcb->evcb_closure == EV_CLOSURE_EVENT_FINALIZE_FREE)
                mm_free(ev);
            break;
        }
        case EV_CLOSURE_CB_FINALIZE:
            //执行 EV_CLOSURE_CB_FINALIZE 类型事件的回调函数，它们的回调函数当中包含了事件终止时所应当执行的操作
            evcb->evcb_cb_union.evcb_cbfinalize(evcb, evcb->evcb_arg);
            break;
        default:
            break;
        }
    }
    return result;
}

参考资料

《Linux 高性能服务器编程》 —— 游双著

《libevent-book》 —— Libevent 上的官方文档

展开全文 >>

Libevent 源码阅读笔记(一)、从一个简单例子开始(上)

什么是 Libevent

Libevent 是一款基于 Reactor 模式实现，由事件驱动的高性能开源 I/O 框架库，它提供一组 API 并允许开发者为事件注册回调函数。当事件发生时，Libevent 实例会执行相应的回调函数。使用 Libevent 来实现网络服务器的事件循环检查框架，可以使开发者更加集中于程序的逻辑之上，减少了工作量，而且 Libevent 在稳定性，性能及可移植性方面有着出色的表现。基于 Libevent 框架库实现的著名软件有：高性能分布式的内存缓存软件 memcached 和 Google 浏览器的 Chromium 的 Linux 版本。

一个基于 Reactor 模式实现的 I/O 框架库大致上应当包含如下几个组件：

I/O 框架库组件

  • 句柄(Handle)：一个句柄通常代表了一个事件源，在 Linux 下， I/O 事件对应的句柄是文件描述符(fd)，而信号事件对应的句柄是信号值
  • 事件多路分发器(EventDemultiplexer)：将系统支持的各种 I/O 复用的系统调用封装成统一的接口，称为事件多路分发器。一般包含了事件注册register_event()，事件删除remove_event()和事件等待demultiplex()。其中demultiplex()是等待事件的核心函数，其内部调用的是select、poll、epoll_wait 等函数
  • 事件处理器和具体事件处理器(EventHandler 和 ConcreteEventHandler)：事件处理器是一个接口，其中包含了一个或多个handle_event()方法。用户需要继承这一接口，并实现自己的事件处理器。
  • Reactor：整个 I/O 框架库的核心，主要提供了以下几个方法：
    • handle_events()：该方法执行事件循环，不断地等待事件，然后依次处理所有就绪事件对应的事件处理器
    • register_handler()：调用事件多路分发器的 register_event() 方法来向事件多路分发器中注册一个事件
    • remove_handler()：调用事件多路分发器的 remove_event() 方法来删除事件多路分发器中的一个事件

I/O 框架库的工作时序

Libevent 的源码链接：git clone https://github.com/libevent/libevent.git

Libevent 官网: http://libevent.org/

编译及安装方法(在 libevent 的源码目录下)：

./configure

make

make verify # (optional)

sudo make install

一个简单的 libevent 示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
//helloworld.cpp
#include <sys/signal.h>
#include <event.h>

void signal_cb(int fd, short event, void* argc){
    event_base* base = (event_base*)argc;
    timeval delay = {2,0};
    printf("Caught an interrupt signal; exiting cleanly in two seconds ...\n");
    event_base_loopexit(base, &delay);
}

void timeout_cb(int fd, short event, void* argc){
    printf("timeout\n");
}

int main(void){
    event_base* base = event_init();
    event* signal_event = evsignal_new(base, SIGINT,signal_cb, base);
    event_add(signal_event, nullptr);
    timeval tv = {1,0};
    event* timeout_event = evtimer_new(base, timeout_cb, nullptr);
    event_add(timeout_event,&tv);
    event_base_dispatch(base);
    event_free(timeout_event);
    event_free(signal_event);
    event_base_free(base);
    return 0;
}

编译方法：

g++ helloworld.cpp -levent_core -std=c++11

在上述的 Demo 中，包含了使用 Libevent 库的一个基本逻辑：

  1. 调用event_init()创建 event_base 对象，一个 event_base 对象相当于一个 Reactor 实例
  2. 创建具体的事件处理器，其中 evsignal_new()创建的是信号事件处理器和evtimer_new()创建的是定时事件处理器
  3. 调用event_add()将事件处理器插入到注册事件队列中，并将其对应的事件插入到事件多路分发器中，相当于 Reactor 中的register_handler方法
  4. 调用event_base_dispatch()执行事件循环
  5. 循环结束后，调用*_free()释放系统资源

接下来我们便按照上述 5 个步骤进行源码分析，出于篇幅考虑，会将这 5 个步骤拆分成 2 篇文章分别叙述，本篇只讨论前 2 个步骤。

event_init 函数的实现与相关的数据结构

在了解 event_int()函数做了何种操作之前，我们需要了解 event_base、eventop 这两个数据结构，它们都定义在 libevent-2.1.12-stable/event-internal.h 中。其中 event_base 构成了 Reactor 的角色， 而 eventop 则构成了 EventDemultiplexer 角色

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
/** Structure to define the backend of a given event_base. */
struct eventop {
    //后端 I/O 复用技术的名称
    const char *name;
    
    //初始化函数，返回的指针将会通过 event_init 函数保存到 event_base.evbase 当中，失败返回 nullptr
    void *(*init)(struct event_base *);
    
    //注册事件：events 代表要新添加的事件，而 old 代表 fd 在添加新事件之前的状态。
    int (*add)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
    
    //删除事件，参数与 add 函数类似，但是 events 包含了我们想要删除的事件
    int (*del)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);

    //分派事件：是实现事件循环的核心函数，它会判断被添加的事件是否处于激活状态，并通过 event_io_active 或者其他的方式来为每一个激活事件执行 event_active 函数
    int (*dispatch)(struct event_base *, struct timeval *);
    //从 event_base 中释放相应的资源
    void (*dealloc)(struct event_base *);

    //当 fork 之后是否需要重新初始化
    int need_reinit;
    /*event_method_feature 由EV_FEATURE_ET、EV_FEATURE_O1、EV_FEATURE_FDS 和 EV_FEATURE_EARLY_CLOSE 组成的枚举类型。
    EV_FEATURE_ET：支持边沿触发事件
    EV_FEATURE_O1：要求事件检测算法的事件复杂度O(1)，【select与poll均不在此列】
    EV_FEATURE_FDS：除了监听socket上的事件，还能监听其他类型文件描述符上的事件
    EV_FEATURE_EARLY_CLOSE：允许使用 EV_CLOSED 事件来直接检测关闭的链接，而无需读取全部的数据
    */
    enum event_method_feature features;
    
    //为每个拥有激活事件的 fd 提供额外的内存长度信息
    size_t fdinfo_len;
};

struct event_base {
    //代表所选择的 I/O 复用机制
    const struct eventop *evsel;
    // I/O 复用机制时所存储的数据，由 evsel 的 init 函数来初始化
    void *evbase;

    //事件变化队列，如果一个 fd 上注册的事件被多次修改，则可以使用缓冲来避免重复的系统调用(比如 epoll_ctl)，仅支持时间复杂度为 O(1) 的I/O复用技术
    struct event_changelist changelist;
    
    //代表信号的后端处理机制
    const struct eventop *evsigsel;
    //信号处理器使用的数据结构，其中封装了一个由 socketpair 创建的管道，被运用于信号处理函数和事件多路分发器之间的通信。(统一事件源)
    struct evsig_info sig;
    
    //被添加到该 event_base 的虚拟事件总数、虚拟事件最大值、事件总数、事件总数的上限，激活事件总数以及激活事件总数的上限
    int virtual_event_count;
    int virtual_event_count_max;
    int event_count;
    int event_count_max;
    int event_count_active;
    int event_count_active_max;
    
    //是否执行完活动时间队列上的剩余任务之后就退出事件循环
    int event_gotterm;
    //不管是否还有剩余事件，强制退出事件循环
    int event_break;
    //是否应该启动一个新的事件循环
    int event_continue;
    
    //目前处理的活动事件队列的优先级
    int event_running_priority;
    
    //循环是否已启动
    int running_loop;
    
    /** Set to the number of deferred_cbs we've made 'active' in the
     * loop.  This is a hack to prevent starvation; it would be smarter
     * to just use event_config_set_max_dispatch_interval's max_callbacks
     * feature */
    //用于设置在循环中被设置为激活状态的延迟回调函数的数量
    int n_deferreds_queued;
    
    //活动事件队列数组，索引值越小的队列，优先级越高
    struct evcallback_list *activequeues;
    //活动事件队列数组的大小
    int nactivequeues;
    /** A list of event_callbacks that should become active the next time
     * we process events, but not this time. */
    //稍后活动队列：队列当中存放的是下次应当被激活的事件
    struct evcallback_list active_later_queue;
    
    /* common timeout logic */
    //管理通用定时器队列的一些成员
    struct common_timeout_list **common_timeout_queues;
    //通用定时器队列中的定时器个数
    int n_common_timeouts;
    //通用定时器队列中的总容量
    int n_common_timeouts_allocated;
    
    //fd 与 I/O 事件之间的映射表
    struct event_io_map io;
    
    //信号值和信号事件之间的映射表
    struct event_signal_map sigmap;
    
    //时间堆
    struct min_heap timeheap;
    
    /** Stored timeval: used to avoid calling gettimeofday/clock_gettime
     * too often. */
    struct timeval tv_cache;
    
    struct evutil_monotonic_timer monotonic_timer;
    
    /** Difference between internal time (maybe from clock_gettime) and
     * gettimeofday. */
    struct timeval tv_clock_diff;
    /** Second in which we last updated tv_clock_diff, in monotonic time. */
    time_t last_updated_clock_diff;

//多线程支持
#ifndef EVENT__DISABLE_THREAD_SUPPORT
    //当前运行该 event_base 的事件循环的线程
    unsigned long th_owner_id;
    //event_base 的独占锁
    void *th_base_lock;

    //条件变量，用于唤醒正在等待某个事件处理完毕的线程
    void *current_event_cond;
    
    //正在等待 current_event_cond 的线程数
    int current_event_waiters;
#endif
    //代表当前正在执行的是哪个事件处理器的回调函数
    struct event_callback *current_event;

#ifdef _WIN32
    /** IOCP support structure, if IOCP is enabled. */
    struct event_iocp_port *iocp;
#endif

    /** Flags that this base was configured with */
    enum event_base_config_flag flags;
    
    struct timeval max_dispatch_time;
    int max_dispatch_callbacks;
    int limit_callbacks_after_prio;
    
    //为唤醒主线程提供了相关的变量与方法
    int is_notify_pending;
    evutil_socket_t th_notify_fd[2];
    /** An event used by some th_notify functions to wake up the main
     * thread. */
    struct event th_notify;
    /** A function used to wake up the main thread from another thread. */
    int (*th_notify_fn)(struct event_base *base);

    //伪随机数种子，以在不同的 socket 中实现公平的随机调度
    struct evutil_weakrand_state weakrand_seed;
    
    /** List of event_onces that have not yet fired. */
    LIST_HEAD(once_event_list, event_once) once_events;
};

正如先前所提到的那样，Libevent 向用户提供了统一的 I/O 复用接口，屏蔽了底层系统的之间的差异，而这一起都要通过定义在 event.c 文件中的 eventops 数组来实现。在不同的系统下有不同的优先级，对于 Linux 系统而言，默认使用的 I/O 复用技术是 epoll。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
//优先级
#ifdef EVENT__HAVE_WORKING_KQUEUE
#include "kqueue-internal.h"
#endif

#ifdef EVENT__HAVE_EVENT_PORTS
extern const struct eventop evportops;
#endif
#ifdef EVENT__HAVE_SELECT
extern const struct eventop selectops;
#endif
#ifdef EVENT__HAVE_POLL
extern const struct eventop pollops;
#endif
#ifdef EVENT__HAVE_EPOLL
extern const struct eventop epollops;
#endif
#ifdef EVENT__HAVE_WORKING_KQUEUE
extern const struct eventop kqops;
#endif
#ifdef EVENT__HAVE_DEVPOLL
extern const struct eventop devpollops;
#endif
#ifdef _WIN32
extern const struct eventop win32ops;
#endif

/* Array of backends in order of preference. */
static const struct eventop *eventops[] = {
#ifdef EVENT__HAVE_EVENT_PORTS
    &evportops,
#endif
#ifdef EVENT__HAVE_WORKING_KQUEUE
    &kqops,
#endif
#ifdef EVENT__HAVE_EPOLL
    &epollops,
#endif
#ifdef EVENT__HAVE_DEVPOLL
    &devpollops,
#endif
#ifdef EVENT__HAVE_POLL
    &pollops,
#endif
#ifdef EVENT__HAVE_SELECT
    &selectops,
#endif
#ifdef _WIN32
    &win32ops,
#endif
    NULL
};

event_init()定义在 event.c 中，它的主要作用是执行 event_base_new_with_config(NULL)来创建一个 event_base 实例并将其保存在全局静态变量 current_base 当中。其中 ``event_base_new_with_config(NULL)`的参数为 NULL 代表根据程序的环境设置相应的 event_base 配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
struct event_base *event_init(void)
{
    struct event_base *base = event_base_new_with_config(NULL);
    if (base == NULL) {
        event_errx(1, "%s: Unable to construct event_base", __func__);
        return NULL;
    }
    //current_base 是一个用于保存全局 event_base 的静态指针
    current_base = base;
    return (base);
}
struct event_base *event_base_new_with_config(const struct event_config *cfg)
{
    int i;
    struct event_base *base;
    int should_check_environment;

#ifndef EVENT__DISABLE_DEBUG_MODE
    event_debug_mode_too_late = 1;
#endif

    if ((base = mm_calloc(1, sizeof(struct event_base))) == NULL) {
        event_warn("%s: calloc", __func__);
        return NULL;
    }

    if (cfg)
        base->flags = cfg->flags;

    should_check_environment =
        !(cfg && (cfg->flags & EVENT_BASE_FLAG_IGNORE_ENV));

    {
        struct timeval tmp;
        int precise_time =
            cfg && (cfg->flags & EVENT_BASE_FLAG_PRECISE_TIMER);
        int flags;
        if (should_check_environment && !precise_time) {
            precise_time = evutil_getenv_("EVENT_PRECISE_TIMER") != NULL;
            if (precise_time) {
                base->flags |= EVENT_BASE_FLAG_PRECISE_TIMER;
            }
        }
        flags = precise_time ? EV_MONOT_PRECISE : 0;
        evutil_configure_monotonic_time_(&base->monotonic_timer, flags);

        gettime(base, &tmp);
    }
    //初始化时间堆
    min_heap_ctor_(&base->timeheap);
    //初始化 base 的 sig 成员中的 Unix 域套接字，为统一事件源的实现做准备
    base->sig.ev_signal_pair[0] = -1;
    base->sig.ev_signal_pair[1] = -1;
    base->th_notify_fd[0] = -1;
    base->th_notify_fd[1] = -1;
    //初始化 event_base 当中的延后活动队列
    TAILQ_INIT(&base->active_later_queue);

    evmap_io_initmap_(&base->io);
    evmap_signal_initmap_(&base->sigmap);
    event_changelist_init_(&base->changelist);

    base->evbase = NULL;

    if (cfg) {
        memcpy(&base->max_dispatch_time,
            &cfg->max_dispatch_interval, sizeof(struct timeval));
        base->limit_callbacks_after_prio =
            cfg->limit_callbacks_after_prio;
    } else {
        base->max_dispatch_time.tv_sec = -1;
        base->limit_callbacks_after_prio = 1;
    }
    if (cfg && cfg->max_dispatch_callbacks >= 0) {
        base->max_dispatch_callbacks = cfg->max_dispatch_callbacks;
    } else {
        base->max_dispatch_callbacks = INT_MAX;
    }
    if (base->max_dispatch_callbacks == INT_MAX &&
        base->max_dispatch_time.tv_sec == -1)
        base->limit_callbacks_after_prio = INT_MAX;
    //遍历 eventops 数组，找到所应当使用的 I/O 复用技术，并将其保存到 base->evsel 中
    for (i = 0; eventops[i] && !base->evbase; i++) {
        if (cfg != NULL) {
            /* determine if this backend should be avoided */
            if (event_config_is_avoided_method(cfg,
                eventops[i]->name))
                continue;
            if ((eventops[i]->features & cfg->require_features)
                != cfg->require_features)
                continue;
        }

        /* also obey the environment variables */
        if (should_check_environment &&
            event_is_method_disabled(eventops[i]->name))
            continue;

        base->evsel = eventops[i];
        //利用多态调用相关的 init 函数进行初始化。例如 Linux 系统下会调用 epoll.c 中的 epoll_init 函数。该函数会创建一个 epoll 对象，成功返回该对象的指针，失败返回 nullptr
        base->evbase = base->evsel->init(base);
    }
    //获取 I/O 复用技术失败，输出警告信息并释放 base 对象
    if (base->evbase == NULL) {
        event_warnx("%s: no event mechanism available",
            __func__);
        base->evsel = NULL;
        event_base_free(base);
        return NULL;
    }
    //如果定义了 EVENT_SHOW_METHOD，则会输出当前后端所选择的 I/O 复用技术的名称，linux下会输出 “epoll”
    if (evutil_getenv_("EVENT_SHOW_METHOD"))
        event_msgx("libevent using: %s", base->evsel->name);

    /* allocate a single active event queue */
    if (event_base_priority_init(base, 1) < 0) {
        event_base_free(base);
        return NULL;
    }

    /* prepare for threading */

#if !defined(EVENT__DISABLE_THREAD_SUPPORT) && !defined(EVENT__DISABLE_DEBUG_MODE)
    event_debug_created_threadable_ctx_ = 1;
#endif

#ifndef EVENT__DISABLE_THREAD_SUPPORT
    if (EVTHREAD_LOCKING_ENABLED() &&
        (!cfg || !(cfg->flags & EVENT_BASE_FLAG_NOLOCK))) {
        int r;
        EVTHREAD_ALLOC_LOCK(base->th_base_lock, 0);
        EVTHREAD_ALLOC_COND(base->current_event_cond);
        r = evthread_make_base_notifiable(base);
        if (r<0) {
            event_warnx("%s: Unable to make base notifiable.", __func__);
            event_base_free(base);
            return NULL;
        }
    }
#endif

#ifdef _WIN32
    if (cfg && (cfg->flags & EVENT_BASE_FLAG_STARTUP_IOCP))
        event_base_start_iocp_(base, cfg->n_cpus_hint);
#endif

    return (base);
}

创建具体的事件处理器

在 Linux 服务器编程中，需要考虑的事件通常有三种：I/O 事件、信号以及定时器事件，而 Libevent 中采用了统一事件源的方式，将这三种事件利用 Unix 域套接字和 I/O 复用的系统调用统一转换成为 I/O 事件进行处理。统一对于信号和定时器，通过socketpair()创建一对 Unix 域套接字，这对套接字起到了一个全双工管道的作用，当信号或是超时事件发生时，通过管道将信号或超时事件发送至 Reactor 进行处理。

要了解 Libevent 的事件类型，我们需要了解位于 libevent-2.1.12-stable\include\event2\event_struct.h 中的 event 结构体和 event_callback 结构体。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
//TAILQ_ENTRY：尾队列中的节点类型
#define TAILQ_ENTRY(type)                       \
struct {                                \
    struct type *tqe_next;  /* next element */          \
    struct type **tqe_prev; /* address of previous next element */  \
}

#define EVLIST_TIMEOUT      0x01    //事件处理器从属于通用定时器队列或时间堆
#define EVLIST_INSERTED     0x02    //事件处理器从属于注册事件队列
#define EVLIST_SIGNAL       0x04    //没有使用
#define EVLIST_ACTIVE       0x08    //事件处理器从属于活动事件队列
#define EVLIST_INTERNAL     0x10    //内部使用
#define EVLIST_ACTIVE_LATER 0x20    
#define EVLIST_FINALIZING   0x40
#define EVLIST_INIT     0x80    //事件处理器已被初始化

struct event_callback {
    TAILQ_ENTRY(event_callback) evcb_active_next;
    short evcb_flags;   //事件标志，可选值定义在上侧，如 EVLIST_TIMEOUT 等
    ev_uint8_t evcb_pri;    //事件处理器优先级，值越小优先级越高
    ev_uint8_t evcb_closure;//代表event_base执行事件处理器的回调函数时的行为
    /* evcb_closure的可选值定义在 event-internal.h 中
       #define EV_CLOSURE_EVENT 0  //默认行为
       #define EV_CLOSURE_EVENT_SIGNAL 1    //将回调函数执行 ev.ev_signal.ev_ncalls 次
       #define EV_CLOSURE_EVENT_PERSIST 2   //执行完回调函数后，再将事件处理器加入到注册事件队列中
       #define EV_CLOSURE_CB_SELF 3         //以 evcb_cb_union.evcb_selfcb 函数作为回调函数
       #define EV_CLOSURE_CB_FINALIZE 4     //以 evcb_cb_union.evcb_cbfinalize 函数作为回调函数
       #define EV_CLOSURE_EVENT_FINALIZE 5  //以 evcb_cb_union.evcb_evfinalize 函数作为回调函数
       #define EV_CLOSURE_EVENT_FINALIZE_FREE 6 //与 EV_CLOSURE_EVENT_FINALIZE 相同，但执行结束后会释放 ev 所占有的资源
     */
    /* allows us to adopt for different types of events */
        union {
        void (*evcb_callback)(evutil_socket_t, short, void *);
        void (*evcb_selfcb)(struct event_callback *, void *);
        void (*evcb_evfinalize)(struct event *, void *);
        void (*evcb_cbfinalize)(struct event_callback *, void *);
    } evcb_cb_union;
    void *evcb_arg; //回调函数参数
};

//event 结构体定义
struct event {
    struct event_callback ev_evcallback;

    /* for managing timeouts */
    union {
        TAILQ_ENTRY(event) ev_next_with_common_timeout; //指出该定时器在通用定时器队列中的位置
        int min_heap_idx;   //定时器在最小堆中的id
    } ev_timeout_pos;   //ev_timeout_pos仅用于定时事件处理器，而一个定时器是否为通用定时器，则由 event.c 中的 is_common_timeout 来判断
    evutil_socket_t ev_fd;  //对于I/O事件处理器而言，它是文件描述符；对于信号处理器而言，它是信号值

    struct event_base *ev_base; //该事件处理器所从属的 event_base 实例
    union {
        /* used for io events */
        struct {
            LIST_ENTRY (event)  ev_io_next //和下一节点相连。该尾队列称之为 I/O 事件队列;  
            struct timeval ev_timeout;  //对定时器有效，代表定时器的超时值
        } ev_io;    //I/O事件队列节点(所有具有相同文件描述符值的I/O事件处理器通过ev_io_next串联而成的尾队列)

        /* used by signal events */
        struct {
            LIST_ENTRY (event) ev_signal_next;  
            short ev_ncalls;    //指定信号发生时，Reactor需要执行多少次对应信号处理器中的回调函数
            /* Allows deletes in callback */
            short *ev_pncalls;  //要么为 nullptr，要么指向ev_ncalls
        } ev_signal;    //信号事件队列节点(所有具有相同信号值的信号事件处理器通过ev_signal_next串联而成的尾队列)
    } ev_;  

    short ev_events;    //代表事件类型，它所能支持的类型定义在 include/event2/event.h
    /*
    #define EV_TIMEOUT  0x01    定时事件
    #define EV_READ     0x02    可读事件
    #define EV_WRITE    0x04    可写事件
    #define EV_SIGNAL   0x08    信号事件
    #define EV_PERSIST  0x10    永久事件
    #define EV_ET       0x20    边缘触发事件
    #define EV_FINALIZE     0x40    在一个线程中以非阻塞形式调用 event_del() 而不管它的回调函数在其他的线程中是否已完成
    #define EV_CLOSED   0x80    连接关闭事件，可直接检测到连接是否关闭而无需读出所有数据
    */
    short ev_res;       //记录了当前激活事件的类型
    struct timeval ev_timeout;
};

通过上述对 event 和 event_callback 的定义，我们对它们所起到的作用有了一定的了解。首先对于 event_callback 而言，它相当于一个 ConcreteEventHadler，其中主要的两个成员是 evcb_active_next，它指向了活动事件链表中的下一个节点，以及联合体成员 evcb_cb_union，它代表了某一个特定的回调函数。而对于 event 结构体而言，它相当于一个 EventHadler，其中主要的成员有具体的事件处理器 ev_evcallback、超时事件ev_timeout_pos、注册事件 ev_（其中包含了 I/O 事件 ev_io 和信号事件 ev_signal)，以及句柄 ev_fd。

一旦了解了事件以及相应的处理器的定义后，我们就需要去看一看如何创建这两个数据结构，并将它们插入到注册事件列表中。event 的创建可以通过定义在 event.c 中的 event_new()来创建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//evsignal_new 和 evtimer_new 都是定义在 include/event2/event.h 中的宏,都是调用了 event.c 中的 event_new() 来实现
#define evsignal_new(b, x, cb, arg)             \
    event_new((b), (x), EV_SIGNAL|EV_PERSIST, (cb), (arg))
#define evtimer_new(b, cb, arg)     event_new((b), -1, 0, (cb), (arg))

struct event* event_new(struct event_base *base, evutil_socket_t fd, short events, void (*cb)(evutil_socket_t, short, void *), void *arg)
{
    struct event *ev;
    //分配内存空间
    ev = mm_malloc(sizeof(struct event));
    if (ev == NULL)
        return (NULL);
    //为 ev 进行相关的赋值操作
    if (event_assign(ev, base, fd, events, cb, arg) < 0) {
        mm_free(ev);
        return (NULL);
    }
    return (ev);
}

参考资料：

《Linux 高性能服务器编程》 —— 游双著

《libevent-book》 —— Libevent 上的官方文档

下篇：Libevent 源码阅读笔记(二)、从一个简单例子开始(下)

展开全文 >>

Leetcode 337. 打家劫舍 III（树形 dp）

问题描述

在上次打劫完一条街道之后和一圈房屋后，小偷又发现了一个新的可行窃的地区。这个地区只有一个入口，我们称之为“根”。 除了“根”之外，每栋房子有且只有一个“父“房子与之相连。一番侦察之后，聪明的小偷意识到“这个地方的所有房屋的排列类似于一棵二叉树”。 如果两个直接相连的房子在同一天晚上被打劫，房屋将自动报警。

计算在不触动警报的情况下，小偷一晚能够盗取的最高金额。

Example

示例 1：

输入: [3,2,3,null,3,null,1]

输出: 7

解释: 小偷一晚能够盗取的最高金额 = 3 + 3 + 1 = 7.

示例 2：

输入: [3,4,5,1,3,null,1]

输出: 9

解释: 小偷一晚能够盗取的最高金额 = 4 + 5 = 9.

题目链接： 337. 打家劫舍 III (难度:中等)

思路：

从题目上看，这显然是一个树形 dp 问题，同样还是离不开动规三要素

1、dp 定义：设 dp[root] 表示从 root 开始抢劫所能得到的最大价值

2、状态转移：

  • 若不抢 root ，则 dp[root] = dp[root->left] + dp[root->right]
  • 若抢 root，则 root 的左右子树均不能抢， dp[root] = dp[root->left->left] + dp[root->left->right] + dp[root->right->left] + dp[root->right->right] + root->val;

3、初态和终态

初态：dp[nullptr] = 0

终态（答案）：dp[root]

由于最终答案为 dp[root]，所以采用后序遍历的方式进行访问

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
/**
 * Definition for a binary tree node.
 * struct TreeNode {
 *     int val;
 *     TreeNode *left;
 *     TreeNode *right;
 *     TreeNode(int x) : val(x), left(NULL), right(NULL) {}
 * };
 */
class Solution {
private:
    unordered_map<TreeNode*, int> dp;
public:
    void helper(TreeNode* root){
        if(root == nullptr){
            if(!dp.count(root)){
                dp[root] = 0;
            }
            return;
        }
        helper(root->left);
        helper(root->right);
        // 不抢的情况
        int not_rob = dp[root->left] + dp[root->right];
        // 抢的情况
        int rob = root->val;
        if(root->left){
            rob += dp[root->left->left] + dp[root->left->right];
        }
        if(root->right){
            rob += dp[root->right->left] + dp[root->right->right];
        }
        dp[root] = max(rob, not_rob);
    }
    int rob(TreeNode* root) {
        helper(root);
        return dp[root];
    }
};

执行结果： 40 ms, 25.9 MB

相关练习题：

展开全文 >>

Leetcode 213. 打家劫舍 II(动态规划，递推)

问题描述：

你是一个专业的小偷，计划偷窃沿街的房屋，每间房内都藏有一定的现金。这个地方所有的房屋都围成一圈，这意味着第一个房屋和最后一个房屋是紧挨着的。同时，相邻的房屋装有相互连通的防盗系统，如果两间相邻的房屋在同一晚上被小偷闯入，系统会自动报警。

给定一个代表每个房屋存放金额的非负整数数组，计算你在不触动警报装置的情况下，能够偷窃到的最高金额。

Example

示例 1：

输入: [2,3,2]

输出: 3

解释: 你不能先偷窃 1 号房屋（金额 = 2），然后偷窃 3 号房屋（金额 = 2）, 因为他们是相邻的。

示例 2：

输入: [1,2,3,1]

输出: 4

解释: 你可以先偷窃 1 号房屋（金额 = 1），然后偷窃 3 号房屋（金额 = 3）。偷窃到的最高金额 = 1 + 3 = 4 。

题目链接：213. 打家劫舍 II (难度:中等)

思路

对于首尾的情况，一共可以分为4种

  • 首尾都不偷；
  • 偷首不偷尾；
  • 偷尾不偷首；
  • 首尾都偷；

显然由题目的描述可知，第 4 种情况会触发报警系统。因此我们需要排除第 4 种情况。另外，第 1 种情况，其实已经包含在第 2 种和第 3 种情况当中。因此我们只需要考虑第 2 种和第 3 种情况即可，可以采用两次dp，一次从0偷到n-1，一次从1偷到n，返回两次dp中的最大值即可。

动态规划三要素

  1. dp定义：

    dp[i] = x：从第 i 间房开始偷，小偷所能得到的最大金额

  2. 状态与状态转移方程：

    ①、第 i 间房不偷，则去第 i + 1 间房看看，dp[i] = dp[i+1]

    ②、第 i 间房偷，则去第 i + 2 间房看看，dp[i] = dp[i + 2] + nums[i];

    方程：dp[i] = max(dp[i + 1], dp[i + 2] + nums[i]);

  3. 初态与终态

    初态：dp[n] = 0

    终态：dp1[0], dp2[1]

    其中，dp1[i]表示从0偷到n-1号房，dp2[i]表示从1偷到n号房

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Solution {
public:
    int rob(vector<int>& nums) {
        if(nums.empty()) return 0;
        int n = nums.size();
        if(n == 1) return nums[0];
        // 此处采用递推实现，dp1_i_1 表示 dp1[i-1]，dp1_i_2 表示 dp1[i-2]，以此类推
        int dp1_i = 0, dp1_i_1 = 0, dp1_i_2 = 0;
        int dp2_i = 0, dp2_i_1 = 0, dp2_i_2 = 0;
        for(int i = n - 1; i >= 1; --i){
            dp1_i = max(dp1_i_1, dp1_i_2 + nums[i]);
            dp1_i_2 = dp1_i_1;
            dp1_i_1 = dp1_i;
        }
        for(int j = n - 2;j >= 0;--j){
            dp2_i = max(dp2_i_1,dp2_i_2 + nums[j]);
            dp2_i_2 = dp2_i_1;
            dp2_i_1 = dp2_i;
        }
        return dp1_i > dp2_i ? dp1_i : dp2_i;
    }
};

执行结果： 0 ms, 7.9 MB

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
C++, 后端开发, lua语言
开源爱好者、Linux

热爱阅读，健身，旅游及美食

联系QQ:519788069