limlog

作一篇文章记录实现，驱动优化迭代。 代码仓库

• 用法
• 实现
  • 后端实现
  • 前端实现
    • 日期时间的处理
    • 线程id的获取
    • 日志行的其他项处理
  • 优化
    • 整形字符串格式化优化
• 测试
  • benchmark
  • 性能分析
• TODO
• Change Log
• 参考

对日志库的 特点期望：

• 正确性，这个是最重要也是最基本的，包括
  • 全部写入.
  • 多个线程间的日志不穿插干扰.
  • 日志线程不能干扰主程序的运行逻辑.
• 易读性
  • 每条日志记录占用一行空间，便于 awk 等工具的时候方便查找, 查看教程.
  • 日志信息包含必要的信息，包括日期时间、线程id、日志等级、日志发生的文件和行数.
• 易用性
  • 在类 printf 和 std::cout 的用法中选择类 std::cout 用法.
  • 根据时间和日志文件大小自动滚动文件.
  • 代码行数不超过 2000 行
• 速度
  • 速度的重要性放在最后，线程增加（不超过CPU数量）的情况下，对速度的影响在30%内，每条日志达到耗时在 1us 左右基本可以满足要求.

日志格式和日志文件格式期望如下：

• 日志文件的格式为：文件名.日期.时间.日志文件滚动的索引.log
  test_log_file.20191224.160354.7.log
• 日志行的格式为 时间（精确到微秒）线程id 日志等级 日志信息 - 文件名:调用函数:行
  20191224 16:04:05.125049 2970 ERROR 5266030 chello, world - LogTest.cpp:thr_1():20

之前看的两个日志库，恰好都叫做 NanoLog:

• Iyengar111/NanoLog, 跨平台、C++11、实现简单（不超过1000行），每个日志行对应一个流对象、速度较快。
• PlatformLab/NanoLog, 斯坦福一个实验室的项目，Linux 平台、C++17、日志行异步写入，每个线程有一个局部的循环字节缓冲区，速度恐怖，是见过最快的日志库了，实现也非常细腻，优化的地方非常多，部分固定的日志信息直接编译期确定下来。

在前一个日志库学到了用户接口宏的定义方式，后一个 nanolog 则是吸收了其后端的设计部分，尤其是 thread_local 的使用。

用法

通过宏定义为用户提供调用接口。

/// Create a logline with log level \p level .
/// Custom macro \p __REL_FILE__ is relative file path as filename.
#ifndef __REL_FILE__
#define __REL_FILE__ __FILE__
#endif
#define LOG(level)                                                             \
    if (limlog::getLogLevel() <= level)                                        \
    limlog::LogLine(level, __REL_FILE__, __FUNCTION__, __LINE__)

#define LOG_TRACE LOG(limlog::LogLevel::TRACE)
#define LOG_DEBUG LOG(limlog::LogLevel::DEBUG)
#define LOG_INFO LOG(limlog::LogLevel::INFO)
#define LOG_WARN LOG(limlog::LogLevel::WARN)
#define LOG_ERROR LOG(limlog::LogLevel::ERROR)
#define LOG_FATAL LOG(limlog::LogLevel::FATAL)

每一条日志都会创建一个 LogLine 对象，这个对象非常小，用完就销毁，几乎没有开销。用法同 std::cout

#include "Log.h"
#include <string>

int main() {
    setLogFile("./test_log_file");
    setLogLevel(limlog::LogLevel::DEBUG);
    setRollSize(64); // 64MB

    std::string str("std::string");
    LOG_DEBUG << 'c' << 65535 << -9223372036854775808 << 3.14159 << "c@string" << str;

    return 0;
}

实现

前后端分离实现

• 后端负责日志信息的落地处理，是整个日志系统的重心。
• 前端负责组织日志信息的格式。

后端实现

采用单例模式，全局只有一个 LimLog 对象，在这个对象构造时也创建一个后台线程，通过无限循环扫描是否有数据可以写入至文件中。通过条件变量控制该循环的终止及 LimLog 对象的销毁。

使用 C++ 11 的关键字 thread_local 为每个线程都创建一个线程局部缓冲区，这样我们的前端的日志写入就可以不用加锁（虽然不用加锁，但是有另外的处理，后说明），每次都写入到各自线程的缓冲区中。然后在上面的循环中扫描这些缓冲区，将其中的日志数据读取至内部的输出缓冲区中，再将其写入到文件中。

每个缓冲区内存只创建一次使用，避免每次使用创建带来不必要的申请消耗。每个线程局部缓冲区的大小为 1M(1 << 20 bytes), LimLog 对象内的输出缓冲区大小为 16M(1 << 24 bytes), 这个大小并不是像看起来那么大就可以高枕无忧了，见 benchmark 性能分析。

后端日志类的成员变量如下，

class LimLog{
    ···
  private:
    LogSink logSink_;
    bool threadSync_; // 前后端同步的标示
    bool threadExit_; // 后台线程标示
    bool outputFull_; // 输出缓冲区满的标示
    LogLevel level_;
    uint32_t bufferSize_;        // 输出缓冲区的大小
    uint32_t sinkCount_;         // 写入文件的次数
    uint32_t perConsumeBytes_;   // 每轮循环读取的字节数
    uint64_t totalConsumeBytes_; // 总读取的字节数
    char *outputBuffer_;         // first internal buffer.
    char *doubleBuffer_;         // second internal buffer, unused.
    std::vector<BlockingBuffer *> threadBuffers_;
    std::thread sinkThread_;
    std::mutex bufferMutex_; // internel buffer mutex.
    std::mutex condMutex_;
    std::condition_variable proceedCond_;  // 后台线程是否继续运行的标条件变量
    std::condition_variable hitEmptyCond_; // 前端缓冲区为空的条件变量
    static thread_local BlockingBuffer *blockingBuffer_; // 线程局部缓冲区
};

对 LimLog 对象的析构和后台线程的退出需要十分的谨慎，这里程序出问题的重灾区，前一个日志库就是这个地方没有处理好，导致有的时候日志还没有写完程序就退出了。为了保证这个逻辑的正确运行，采用了两个条件变量 proceedCond_ 和 hitEmptyCond_。在执行对象的析构函数时，需要等待无可读数据的条件变量 hitEmptyCond_, 保证这个时候线程局部缓冲区的数据都已经读取完成，然后在析构函数中设置后台线程循环退出的条件，并且使用 proceedCond_ 通知后台线程运行。

以上逻辑在析构函数中的逻辑如下：

LimLog::~LimLog() {
    {
        // notify background thread befor the object detoryed.
        std::unique_lock<std::mutex> lock(condMutex_);
        threadSync_ = true;
        proceedCond_.notify_all();
        hitEmptyCond_.wait(lock);
    }

    {
        // stop sink thread.
        std::lock_guard<std::mutex> lock(condMutex_);
        threadExit_ = true;
        proceedCond_.notify_all();
    }
    ···
}

之后就转入到后台线程的循环处理中，以下是所有后台线程处理的逻辑：

// Sink log info to file with async.
void LimLog::sinkThreadFunc() {
    while (!threadExit_) {
        // move front-end data to internal buffer.
        {
            std::lock_guard<std::mutex> lock(bufferMutex_);
            uint32_t bbufferIdx = 0;
            // while (!threadExit_ && !outputFull_ && !threadBuffers_.empty()) {
            while (!threadExit_ && !outputFull_ &&
                   (bbufferIdx < threadBuffers_.size())) {
                BlockingBuffer *bbuffer = threadBuffers_[bbufferIdx];
                uint32_t consumableBytes = bbuffer->consumable(); // 如果这个地方使用 used() 代替，就会出现日志行截断的现象

                if (bufferSize_ - perConsumeBytes_ < consumableBytes) {
                    outputFull_ = true;
                    break;
                }

                if (consumableBytes > 0) {
                    uint32_t n = bbuffer->consume(
                        outputBuffer_ + perConsumeBytes_, consumableBytes);
                    perConsumeBytes_ += n;
                } else {
                    // threadBuffers_.erase(threadBuffers_.begin() +
                    // bbufferIdx);
                }
                bbufferIdx++;
            }
        }

        // not data to sink, go to sleep, 50us.
        if (perConsumeBytes_ == 0) {
            std::unique_lock<std::mutex> lock(condMutex_);

            // if front-end generated sync operation, consume again.
            if (threadSync_) {
                threadSync_ = false;
                continue;
            }

            hitEmptyCond_.notify_one();
            proceedCond_.wait_for(lock, std::chrono::microseconds(50));
        } else {
            logSink_.sink(outputBuffer_, perConsumeBytes_);
            sinkCount_++;
            totalConsumeBytes_ += perConsumeBytes_;
            perConsumeBytes_ = 0;
            outputFull_ = false;
        }
    }
}

当扫描发现无数据可取时，这个时候我们再循环读取一下，防止析构函数在扫描数据和判断 threadSync_ 执行的时间差之内又有日志产生，在又一轮循环后，确保所有的数据都已经读取，通知析构函数继续执行，然后退出循环，该后台线程函数退出，线程销毁。 wait_for 是节约系统资源，出现无数据时当前线程就直接休眠 50us 避免不必要的循环消耗.

以上是对后台线程的处理，对数据的处理其实还是比较简单的，后端提供了一个写入数据的接口，该接口将前端产生的数据统一写入至线程局部缓冲区中。

void LimLog::produce(const char *data, size_t n) {
    if (!blockingBuffer_) {
        std::lock_guard<std::mutex> lock(bufferMutex_);
        blockingBuffer_ =
            static_cast<BlockingBuffer *>(malloc(sizeof(BlockingBuffer)));
        threadBuffers_.push_back(blockingBuffer_);  // 添加到后端的缓冲区数组中
    }

    blockingBuffer_->produce(data, static_cast<uint32_t>(n)); // 将数据添加到线程局部缓冲区中
}

后台写入线程这个时候，遍历缓冲数组，读取数据至内部的输出缓冲区中，当出现可读取的数据大于缓冲区时，直接写入文件，剩余的数据下一路再读取；当无可读取的数据时，判断是对象析构了还只是简单的没有数据读取，并且休眠 50us；有数据可读则写入至文件，重置相关数据。

线程局部缓冲区为一个阻塞环形生产者消费者队列，设计为阻塞态的原因为缓冲区的大小足够大了，如果改为非阻塞态，当缓冲区的满了的时候重新分配内存，还要管理一次内存的释放操作。虽然线程内的操作不需要使用锁来同步，在后台线程的循环范围内，每次都要获取数据的可读大小，这里可能就涉及多个线程对数据的访问了。最初的版本测试在没有同步并且开启优化的情况下，有一定概率在 consume() 操作陷入死循环，原因是该线程的 cache 的变量未及时得到更新。现在使用内存屏障来解决这个问题，避免编译器优化。

对于后台线程读取各线程内的数据有一个数据截断的问题，如果使用 BlockingBuffer::unsed()获取已读数据长度，可能会出现日志行只被读取了一半，然后立刻被另外的一个线程的日志截断的情况，这不是我们期望的。为了解决这个问题，这里又提供了一个接口 BlockingBuffer::incConsumablePos() 来移动可以读取的位置，该接口每次递增的长度为一条日志行的长度，在一日志行数据写入到局部缓冲区后调用该接口来移动可以读取到的位置。而又提供接口 BlockingBuffer::consumable() 来获取可以读取的长度，这样就避免了一条日志行被截断的情况。

另外关于线程内无锁的处理，使用 fence 指令来保持缓存一致性，因为局部线程内缓冲区中的 BlockingBuffer::consumablePos_ 和 BlockingBuffer::consumePos_ 会由于 BlockingBuffer::consumable() 暴露给其他线程。如果不使用同步措施，在多核的情况下，一个core cache 中的数据发生改变时，其他core感知不到变化，仍然对老的数据做处理。在这里一个 core 中的 producePos_ 等于 consumePos_ 的时候获取不到新的数据变化，就会陷入无限循环。加上 fence 指令，保证 fence 前操作先于在 fence 后的操作完成，且变化被fence之后的操作感知。

uint32_t BlockingBuffer::used() const {
    std::atomic_thread_fence(std::memory_order_acquire);
    return producePos_ - consumePos_;
}

void BlockingBuffer::produce(const char *from, uint32_t n) {
    while (unused() < n)
        /* blocking */;
    ···
}

前端实现

一条日志信息在内存中的布局如下：
+------+-----------+-------+------+------+----------+------+
| time | thread id | level | logs | file | function | line |
+------+-----------+-------+------+------+----------+------+

20191224 16:04:05.125044 2970 ERROR  5266025chello, world - LogTest.cpp:thr_1():20
20191224 16:04:05.125044 2970 ERROR  5266026chello, world - LogTest.cpp:thr_1():20
20191224 16:04:05.125045 2970 ERROR  5266027chello, world - LogTest.cpp:thr_1():20

前端实现虽然简单，但是优化的空间很大，使用 perf 做性能分析，性能热点集中在参数格式化为字符串和时间的处理上面。

我们用 LogLine 来表示一个日志行，这个类非常简单，重载了多个参数类型的操作符 <<, 及保存一些日志行相关信息，包括 文件名称，函数，行和最重要的写入字节数。
LogLine 不包含缓冲区，直接调用后端提供的接口 LimLog::produce() 将数据写入到局部缓冲区内。

在 limlog 中，多次使用到了 thread_local 关键字，在前端实现部分也是如此。

time 的处理

在 Linux 中使用 gettimeofday(2) 来获取当前的时间戳，其他平台使用 std::chrono 来获取时间戳，用 localtime() 和 strftime() 获取本地的格式化时间 %Y%m%d %H:%M:%S, 合并微秒 timestamp % 1000000 组成 time.

这里利用 thread_local 做一个优化，格式化时间的函数调用次数。定义一个线程局部变量来存储时间戳对应的秒数 t_second 和 字符数组来存储格式化时间 t_datetime[24]，当秒数未发生改变时，直接取线程局部字符数组而不用再调用格式化函数。

在 x86-64 体系下，gettimeofday(2) 在 vdso 机制下已经不是系统调用了，经测试发现直接调用 gettimeofday(2) 比 std::chrono 的高精度时钟快 15% 左右，虽然不是系统调用但是耗时还是大头，一个调用大概需要 600ns 左右的时间。

thread id 的获取

在 Linux 下，使用 gettid() 而不是 pthread_self() 来获取线程id。
Windows 现在也支持了，我们要获取当前线程id需要调用 std::this_thread::get_id()，但是返回的是 thread::id 对象，不是期望的整形数字，stl 实现如下(thread文件)：

    /// get_id
    inline thread::id
    get_id() noexcept
    {
#ifdef __GLIBC__
      if (!__gthread_active_p())
    return thread::id(1);
#endif
      return thread::id(__gthread_self());
    }

使用 pthread_self() 构造一个 id 对象返回，观察这个id对象是有一个 _M_thread 的私有变量的，这个值是 pthread_self()返回得来的。虽然不能直接获取 _M_thread 变量，但是 thread::id 重载了 << 操作符，根据这个特点借助 std::stringstream 就可以取出这个变量了，实现如下（Log.cpp）:

pid_t tid;

std::stringstream ss;
ss << std::this_thread::get_id();
ss >> tid;

这里继续使用 thread_local 来对线程id优化，每个线程的id调用一次线程id获取函数。创建一个线程局部线程id变量，判断线程id存在时，就不再调用 gettid() 了。

日志行的其他项

日志等级、文件、函数和行数是简单的字符串写入操作了。

优化

使用一些先进的算法来实现加速处理。

整形转换至字符串优化

实现参考了 叶劲锋 大佬的 itoa-benchmark

老的算法是使用 std::to_string(vsnprintf) 来进行字符串格式化的操作，效率一般，这里 使用查表的方式加速字符串转换处理，极大加快了处理的速度。

测试

TDD 是一种比较好的个人开发习惯，完整的测试用例有助于提高程序的正确性。

• Timestamp, 对时间戳的测试，这里是日期的测试，见 test_timestamp().
• BlockingBuffer, 对循环生产者消费者队列的测试，见 test_blocking_buffer(), 对队列的相关属性（大小，生产消费索引，可消费索引，已/未使用空间）进行测试；队列的相关索引都是自然增长的，到了 UINT32_MAX 自然的归零，所以这里也测试了索引不超过 BlockingBuffer::size() 和超过这个大小的情况。
• NumToString, 目前只有整形转字符串，对 INT*_MAX, INT*_MIN, UINT*_MAX 进行测试，目前程序刻意留出了三个错误的用例，见 test_itoa().

benchmark

即使速度在日志库中不是最重要的，但是速度肯定是要测的。这里使用的测试用例比较多，也是测试的一个难点了，因为磁盘的的容量比较有限。测试的机器为 i7 9700K，普通硬盘，OS 为 Linux 4.4(WSL).

测试用例 如下：

注: - 表示一个该日志包含的元素，-4 表示4个，-+4096 表示除了写入一个该元素外，再追加 4096 字节长度的该元素。99(596/6) 则表示该用例包括对六种日志的测试，每种日志分别打印 4 次相同元素，六种日志平均长度为 99.

单位：微秒/条。

由于是在WSL下跑的测试用例，这个性能比我老i3还慢，在vs2019下编译，速度大概一条80字节的速度大概在0.55us左右，后面再在原生linux上和用vs编译一下进行压测。测试的用例数没有继续调高，是因为在单个用例在 100w 的时候，8线程压测的情况下生成的日志就有 100G 以上了，这里可以优化一下，写一个脚本单批次运行这些用例。

观察这个表格，用例1是比较简单的写入了，100w 单个用例条数在8线程的情况下明显耗时升高，CPU是8核心，程序此时是 9 个线程满负荷的运行了，猜测这个情况是后端日志明显变多，这个时候后台线程不可能进入睡眠状态了，只能前端生产线程进行切换了，切换有开销，且另外一个线程此时得不到处理器。

<spand id="bench&#95;analysis"> 性能分析 </spand>

在数据量小的时候，增加线程对速度的影响并不大，一旦增加单条日志的长度，速度线性降低，产生这个现象的原因是：
后端是一个消费线程对应多个生产线程，消费线程读取数据的能力固定（磁盘写入的速度180M/s），当生产线程数量增加时，均摊到各生产线程的数据读取延迟增加，而前端写入是阻塞模式的，线程局部缓冲区大小只有 1M(1<<20 bytes)， 在单条日志数据长度很大的时候，线程局部缓冲区很容易达到饱并且而发生阻塞，进而整个日志系统的速度下降，就产生了耗时线性增加的情况。

在有些日志库中提供了 非阻塞 写入方式，如果达到磁盘的写入瓶颈，就丢弃最老的日志数据，见 spdlog 的处理。

TODO

• 速度
  • 对 LogLine::operator<<() 浮点数转字符串做优化
  • 日志行的时间戳获取优化，这里的时间占去60%左右的开销

Change Log

1. 跨平台实现.
   1. windows平台线程id获取. commit.
   2. 使用 std::atomic_thread_fence 代替gcc内联汇编 fence 指令. commit
2. 优化整形数字转换到字符串的算法. commit.
3. 增加测试用例和性能压测. commit.
4. 增加每天日志文件滚动的功能. commit

参考

1. Iyengar111/NanoLog, Low Latency C++11 Logging Library.
2. PlatformLab/NanoLog, Nanolog is an extremely performant nanosecond scale logging system for C++ that exposes a simple printf-like API.
3. kfifo, 内核环形生产者消费者队列.
4. memory barries, 内核的内存屏障.
5. itoa-benchmark, 几种对整形数字转换成字符串的实现和性能比较，limlog 选择的是相对简单的查表算法。
6. spdlog, Fast C++ logging library. 易用性和性能非常好的日志库，工业使用值得推荐。