前半部分muduo筆記 日志庫（一），中提到muduo異步日志庫分為兩部分：前端和后端。這部分描述后端。
后端
前端主要實現異步日志中的日志功能，為用戶提供將日志內容轉換為字符串，封裝為一條完整的log消息存放到RAM中；
而實現異步，核心是通過專門的后端線程，與前端線程並發運行，將RAM中的大量日志消息寫到磁盤上。
后端主要包括：AsyncLogging, LogFile, AppendFile，MutexLock。
AsyncLogging 提供后端線程，定時將log緩沖寫到磁盤，維護緩沖及緩沖隊列。
LogFile 提供日志文件滾動功能，寫文件功能。
AppendFile 封裝了OS提供的基礎的寫文件功能。
類圖關系如下：
AsyncLogging類
AsyncLogging 主要職責：提供大緩沖Large Buffer（默認4MB）存放多條日志消息，緩沖隊列BufferVector用於存放多個Large Buffer，為前端線程提供線程安全的寫Large Buffer操作；提供專門的后端線程，用於定時或緩沖隊列非空時，將緩沖隊列中的Large Buffer通過LogFile提供的日志文件操作接口，逐個寫到磁盤上。
數據成員
/**
* Provide async logging function. backend.
* Background thread (just only one) call this module to write log to file.
*/
class AsyncLogging : noncopyable
{
...
private:
typedef muduo::detail::FixedBuffer<muduo::detail::kLargeBuffer> Buffer; // Large Buffer Type
typedef std::vector<std::unique_ptr<Buffer>> BufferVector; // 已滿緩沖隊列類型
typedef BufferVector::value_type BufferPtr;
const int flushInterval_; // 沖刷緩沖數據到文件的超時時間, 默認3秒
std::atomic<bool> running_; // 后端線程loop是否運行標志
const string basename_; // 日志文件基本名稱
const off_t rollSize_; // 日志文件滾動大小
muduo::Thread thread_; // 后端線程
muduo::CountDownLatch latch_; // 門閥, 同步調用線程與新建的后端線程
muduo::MutexLock mutex_; // 互斥鎖, 功能相當於std::mutex
muduo::Condition cond_ GUARDED_BY(mutex_); // 條件變量, 與mutex_配合使用, 等待特定條件滿足
BufferPtr currentBuffer_ GUARDED_BY(mutex_); // 當前緩沖
BufferPtr nextBuffer_ GUARDED_BY(mutex_); // 空閑緩沖
BufferVector buffers_ GUARDED_BY(mutex_); // 已滿緩沖隊列
}
AsyncLogging數據按功能主要分為3部分：1）維護存放log消息的大緩沖Large Buffer；2）后端線程；3）傳遞給其他類對象的參數，如basename_，rollSize_；
LargeBuffer 存放大量log消息
Large Buffer（FixedBuffermuduo::detail::kLargeBuffer）默認大小4MB，用於存儲多條log消息；相對的，還有Small Buffer（FixedBuffermuduo::detail::kSmallBuffer）默認大小4KB，用於存儲一條log消息。
當前端線程通過調用LOG_XXX << "..."時，如何將log消息傳遞給后端呢？
可以通過調用AsyncLogging::append()
void AsyncLogging::append(const char *logline, int len)
{
muduo::MutexLockGuard lock(mutex_);
if (currentBuffer_->avail() > len)
{ // current buffer's free space is enough to fill C string logline[0..len-1]
currentBuffer_->append(logline, static_cast<size_t>(len));
}
else
{ // current buffer's free space is not enough
buffers_.push_back(std::move(currentBuffer_));
if (nextBuffer_)
{
currentBuffer_ = std::move(nextBuffer_);
}
else
{
currentBuffer_.reset(new Buffer); // rarely happens
}
currentBuffer_->append(logline, static_cast<size_t>(len));
cond_.notify();
}
}
append()可能會被多個前端線程調用，因此必須考慮線程安全，可以用mutex_加鎖。
append()基本思路：當前緩沖（currentBuffer_）剩余空間（avail()）足夠存放新log消息大小（len）時，就直接存放到當前緩沖；當前緩沖剩余空間不夠時，說明當前緩沖已滿（或者接近已滿），就將當前緩沖move到已滿緩沖隊列（buffers_），將空閑緩沖move到當前緩沖，再把新log消息存放到當前緩沖中（此時當前緩沖為空，剩余空間肯定夠用），最后喚醒等待中的后端線程。
注意：Large Buffer是通過std::unique_ptr指向的，move操作后，原來的 std::unique_ptr就會值為空。
問題：
1）為什么最后要通過cond_喚醒后端線程？
因為沒有log消息要記錄時，后端線程很可能阻塞等待log消息，當有緩沖滿時，及時喚醒后端將已滿緩沖數據寫到磁盤上，能有效改善新能；否則，短時間內產生大量log消息，可能造成數據堆積，甚至丟失，而后端線程一直休眠（直到3秒超時喚醒）。
2）為什么調用notify()而不是notifyAll()，只喚醒一個線程，而不是喚醒所有線程？
因為一個應用程序通常只有一個日志庫后端，而一個后端通常只有一個后端線程，也只會有一個后端線程在該條件變量上等待，因此喚醒一個線程足以。
后端線程 異步寫數據到log文件
后端線程的創建就是啟動，是在start()中，通過調用Thread::start()完成。門閥latch_目的在於讓調用start()線程等待線程函數啟動完成，而線程函數中調用latch_.countDown()表示啟動完成，當然，前提是latch_計數器初值為1。
// Control background thread
void start()
{
running_ = true;
thread_.start();
latch_.wait();
}
void stop() NO_THREAD_SAFETY_ANALYSIS
{
running_ = false;
cond_.notify();
thread_.join();
}
而AsyncLogging::stop()用於關閉后端線程，通常是在析構函數中，調用AsyncLogging::stop() 停止后端線程。
~AsyncLogging()
{
if (running_)
{
stop();
}
}
后端線程函數threadFunc，會構建1個LogFile對象，用於控制log文件創建、寫日志數據，創建2個空閑緩沖區buffer1、buffer2，和一個待寫緩沖隊列buffersToWrite，分別用於替換當前緩沖currentBuffer_、空閑緩沖nextBuffer_、已滿緩沖隊列buffers_，避免在寫文件過程中，鎖住緩沖和隊列，導致前端無法寫數據到后端緩沖。
threadFunc中，提供了一個loop，基本流程是這樣的：
1）每次當已滿緩沖隊列中有數據時，或者即使沒有數據但3秒超時，就將當前緩沖加入到已滿緩沖隊列（即使當前緩沖沒滿），將buffer1移動給當前緩沖，buffer2移動給空閑緩沖（如果空閑緩沖已移動的話）。
2）然后，再交換已滿緩沖隊列和待寫緩沖隊列，這樣已滿緩沖隊列就為空，待寫緩沖隊列就有數據了。
3）接着，將待寫緩沖隊列的所有緩沖通過LogFile對象，寫入log文件。
4）此時，待寫緩沖隊列中的緩沖，已經全部寫到LogFile指定的文件中（也可能在內核緩沖中），擦除多余緩沖，只用保留兩個，歸還給buffer1和buffer2。
5）此時，待寫緩沖隊列中的緩沖沒有任何用處，直接clear即可。
6）將內核高速緩存中的數據flush到磁盤，防止意外情況造成數據丟失。
后端線程函數threadFunc，會構建1個LogFile對象，用於控制log文件創建、寫日志數據，創建2個空閑緩沖區buffer1、buffer2，和一個待寫緩沖隊列buffersToWrite，分別用於替換當前緩沖currentBuffer_、空閑緩沖nextBuffer_、已滿緩沖隊列buffers_，避免在寫文件過程中，鎖住緩沖和隊列，導致前端無法寫數據到后端緩沖。
threadFunc中，提供了一個loop，基本流程是這樣的：
1）每次當已滿緩沖隊列中有數據時，或者即使沒有數據但3秒超時，就將當前緩沖加入到已滿緩沖隊列（即使當前緩沖沒滿），將buffer1移動給當前緩沖，buffer2移動給空閑緩沖（如果空閑緩沖已移動的話）。
2）然后，再交換已滿緩沖隊列和待寫緩沖隊列，這樣已滿緩沖隊列就為空，待寫緩沖隊列就有數據了。
3）接着，將待寫緩沖隊列的所有緩沖通過LogFile對象，寫入log文件。
4）此時，待寫緩沖隊列中的緩沖，已經全部寫到LogFile指定的文件中（也可能在內核緩沖中），擦除多余緩沖，只用保留兩個，歸還給buffer1和buffer2。
5）此時，待寫緩沖隊列中的緩沖沒有任何用處，直接clear即可。
6）將內核高速緩存中的數據flush到磁盤，防止意外情況造成數據丟失。
void AsyncLogging::threadFunc()
{
assert(running_ == true);
latch_.countDown();
LogFile output(basename_, rollSize_, false); // only called by this thread, so no need to use thread safe
BufferPtr newBuffer1(new Buffer);
BufferPtr newBuffer2(new Buffer);
newBuffer1->bzero();
newBuffer2->bzero();
BufferVector buffersToWrite;
static const int kBuffersToWriteMaxSize = 25;
buffersToWrite.reserve(16); // FIXME: why 16?
while (running_)
{
// ensure empty buffer
assert(newBuffer1 && newBuffer1->length() == 0);
assert(newBuffer2 && newBuffer2->length() == 0);
// ensure buffersToWrite is empty
assert(buffersToWrite.empty());
{ // push buffer to vector buffersToWrite
muduo::MutexLockGuard lock(mutex_);
if (buffers_.empty())
{ // unusual usage!
cond_.waitForSeconds(flushInterval_); // wait condition or timeout
}
// not empty or timeout
buffers_.push_back(std::move(currentBuffer_));
currentBuffer_ = std::move(newBuffer1);
buffersToWrite.swap(buffers_);
if (!nextBuffer_)
{
nextBuffer_ = std::move(newBuffer2);
}
}
// ensure buffersToWrite is not empty
assert(!buffersToWrite.empty());
if (buffersToWrite.size() > kBuffersToWriteMaxSize) // FIXME: why 25? 25x4MB = 100MB, 也就是說, 從上次loop到本次loop已經堆積超過100MB, 就丟棄多余緩沖
{
char buf[256];
snprintf(buf, sizeof(buf), "Dropped log message at %s, %zd larger buffers\n",
Timestamp::now().toFormattedString().c_str(),
buffersToWrite.size() - 2);
fputs(buf, stderr);
output.append(buf, static_cast<int>(strlen(buf)));
buffersToWrite.erase(buffersToWrite.begin() + 2, buffersToWrite.end()); // keep 2 buffer
}
// append buffer content to logfile
for (const auto& buffer : buffersToWrite)
{
// FIXME: use unbuffered stdio FILE? or use ::writev ?
output.append(buffer->data(), buffer->length());
}
if (buffersToWrite.size() > 2)
{
// drop non-bzero-ed buffers, avoid trashing
buffersToWrite.resize(2);
}
// move vector buffersToWrite's last buffer to newBuffer1
if (!newBuffer1)
{
assert(!buffersToWrite.empty());
newBuffer1 = std::move(buffersToWrite.back());
buffersToWrite.pop_back();
newBuffer1->reset(); // reset buffer
}
// move vector buffersToWrite's last buffer to newBuffer2
if (!newBuffer2)
{
assert(!buffersToWrite.empty());
newBuffer2 = std::move(buffersToWrite.back());
buffersToWrite.pop_back();
newBuffer2->reset(); // reset buffer
}
buffersToWrite.clear();
output.flush();
}
output.flush();
}
異常處理：
當已滿緩沖隊列中的數據堆積（默認緩沖數超過25），就會丟棄多余緩沖，只保留最開始2個。
為什么保留2個？個人覺得2個~16個都是可以的，不過，為了有效減輕log導致的負擔，丟棄多余的也未嘗不可。
25的含義：
25個緩沖，每個4MB，共100MB。也就是說，上次處理周期到本次，已經堆積了超過100MB數據待處理。
假設磁盤的寫速度100MB/S，要堆積100MB有2種極端情況：
1）1S內產生200MB數據；
2）25秒內，平均每秒產生104MB數據；
不論哪種情況，都是要超過磁盤的處理速度。而實際應用中，只有產生數據速度不到磁盤寫速度的1/10，應用程序性能才不會受到明顯影響。
[======]
LogFile類
LogFile 主要職責：提供對日志文件的操作，包括滾動日志文件、將log數據寫到當前log文件、flush log數據到當前log文件。
構造函數
LogFile::LogFile(const std::string &basename,
off_t rollSize,
bool threadSafe, // 線程安全控制項, 默認為true. 當只有一個后端AsnycLogging和后端線程時, 該項可置為false
int flushInterval,
int checkEveryN)
: basename_(basename), // 基礎文件名, 用於新log文件命名
rollSize_(rollSize), // 滾動文件大小
flushInterval_(flushInterval), // 沖刷時間限值, 默認3 (秒)
checkEveryN_(checkEveryN), // 寫數據次數限值, 默認1024
count_(0), // 寫數據次數計數, 超過限值checkEveryN_時清除, 然后重新計數
mutex_(threadSafe ? new MutexLock : NULL), // 互斥鎖指針, 根據是否需要線程安全來初始化
startOfPeriod_(0), // 本次寫log周期的起始時間(秒)
lastRoll_(0), // 上次roll日志文件時間(秒)
lastFlush_(0) // 上次flush日志文件時間(秒)
{
assert(basename.find('/') == string::npos); // basename不應該包含'/', 這是路徑分隔符
rollFile();
}
重新啟動時，可能並沒有log文件，因此在構建LogFile對象時，直接調用rollFile()以創建一個全新的日志文件。
滾動日志文件
當日志文件接近指定的滾動限值（rollSize）時，需要換一個新文件寫數據，便於后續歸檔、查看。調用LogFile::rollFile()可以實現文件滾動。
bool LogFile::rollFile()
{
time_t now = 0;
string filename = getLogFileName(basename_, &now);
time_t start = now / kRollPerSeconds_ * kRollPerSeconds_;
if (now > lastRoll_)
{ // to avoid identical roll by roll time
lastRoll_ = now;
lastFlush_ = now;
startOfPeriod_ = start;
// create new log file with new filename
file_.reset(new FileUtil::AppendFile(filename));
return true;
}
return false;
}
滾動日志文件操作的關鍵是：1）取得新log文件名，文件名全局唯一；2）創建並打開一個新log文件，用指向LogFile對象的unique_ptr指針file_表示。
異常處理：
滾動操作會新建一個文件，而為避免頻繁創建新文件，rollFile會確保上次滾動時間到現在如果不到1秒，就不會滾動。
注意：是否滾動日志文件的條件判斷，並不在rollFile，而是在寫數據到log文件的LogFile::append_unlocked()中，因為寫新數據的時候，是判斷當前log文件大小是否足夠大的最合適時機。而rollFile只用專門負責如何滾動log文件即可。
日志文件名
getLogFileName根據調用者提供的基礎名，以及當前時間，得到一個全新的、唯一的log文件名。或許叫nextLogFileName更合適。
string LogFile::getLogFileName(const string &basename, time_t *now) // static
{
string filename;
filename.reserve(basename.size() + 64); // extra 64 bytes for timestamp etc.
filename = basename;
char timebuf[32];
struct tm tmbuf;
*now = time(NULL);
gmtime_r(now, &tmbuf); // FIXME: localtime_r ?
strftime(timebuf, sizeof(timebuf), ".%Y%m%d-%H%M%S.", &tmbuf);
filename += timebuf;
filename += ProcessInfo::hostname();
char pidbuf[32];
snprintf(pidbuf, sizeof(pidbuf), ".%d", ProcessInfo::pid());
filename += pidbuf;
filename += ".log";
return filename;
}
gmtime_r獲取的是gmt時區時間，localtime_r獲取的是本地時間。
新log文件名格式：
basename + now + hostname + pid + ".log"
basename 基礎名, 由用戶指定, 通常可設為應用程序名
now 當前時間, 格式: "%Y%m%d-%H%M%S"
hostname 主機名
pid 進程號, 通常由OS提供, 通過getpid獲取
".log" 固定后綴名, 表明這是一個log文件
各部分之間, 用"."連接
如下面是一個根據basename為"test_log_mt"生成的log文件名：
test_log_mt.20220218-134000.ubuntu.12426.log
寫日志文件操作
LogFile提供了2個接口，用於向當前日志文件file_寫入數據。append本質上是通過append_unlocked完成對日志文件寫操作，但多了線程安全。用戶只需調用第一個接口即可，append會根據線程安全需求，自行判斷是否需要加上；第二個是private接口。
void append(const char *logline, int len);
void append_unlocked(const char *logline, int len);
append_unlocked 會先將log消息寫入file_文件，之后再判斷是否需要滾動日志文件；如果不滾動，就根據append_unlocked的調用次數和時間，確保1）一個log文件超時（默認1天），就創建一個新的；2）flush文件操作，不會頻繁執行（默認間隔3秒）。
void LogFile::append_unlocked(const char *logline, int len)
{
file_->append(logline, len);
if (file_->writtenBytes() > rollSize_)
{ // written bytes to file_ > roll threshold (rollSize_)
rollFile();
}
else
{
++count_;
if (count_ >= checkEveryN_)
{
count_ = 0;
time_t now = ::time(NULL);
time_t thisPeriod_ = now / kRollPerSeconds_ * kRollPerSeconds_;
if (thisPeriod_ != startOfPeriod_)
{ // new period, roll file for log
rollFile();
}
else if (now - lastFlush_ > flushInterval_)
{ // timeout ( flushInterval_ = 3 seconds)
lastFlush_ = now;
file_->flush();
}
}
}
}
append如何根據需要選擇是否線程安全地調用append_unlocked？
可以根據mutex_是否為空。因為構造時，根據用戶傳入的threadSafe實參，決定了mutex_是否為空。
void LogFile::append(const char *logline, int len)
{
if (mutex_)
{
MutexLockGuard lock(*mutex_);
append_unlocked(logline, len);
}
else
{
append_unlocked(logline, len);
}
}
flush日志文件
flush操作往往與write文件操作配套。LogFile::flush實際上是通過AppendFile::flush()，完成對日志文件的沖刷。與LogFile::append()類似，flush也能通過mutex_指針是否為空，自動選擇線程安全版本，還是非線程安全版本。
void LogFile::flush()
{
if (mutex_)
{
MutexLockGuard lock(*mutex_);
file_->flush();
}
else
{
file_->flush();
}
}
[======]
AppendFile類
AppendFile位於FileUtil.h/.cc，封裝了OS提供的，底層的創建/打開文件、寫文件、關閉文件等操作接口，並沒有專門考慮線程安全問題。線程安全由上一層級調用者，如LogFile來保證。
數據結構
AppendFile的數據結構較簡單，
// not thread safe
class AppendFile : public noncopyable
{
public:
explicit AppendFile(StringArg filename);
~AppendFile();
void append(const char* logline, size_t len); // 添加log消息到文件末尾
void flush(); // 沖刷文件到磁盤
off_t writtenBytes() const { return writtenBytes_; } // 返回已寫字節數
private:
size_t write(const char* logline, size_t len); // 寫數據到文件
FILE* fp_; // 文件指針
char buffer_[ReadSmallFile::kBufferSize]; // 文件操作的緩沖區
off_t writtenBytes_; // 已寫字節數
};
RAII方式打開、關閉文件
AppendFile采用RAII方式管理文件資源，構建對象即打開文件，銷毀對象即關閉文件。
AppendFile::AppendFile(StringArg filename)
: fp_(::fopen(filename.c_str(), "ae")), // 'e' for O_CLOEXEC
writtenBytes_(0)
{
assert(fp_);
::setbuffer(fp_, buffer_, sizeof(buffer_)); // change stream fp_'s buffer to buffer_
#if 0
// optimization for predeclaring an access pattern for file data
struct stat statbuf;
fstat(fd_, &statbuf);
::posix_fadvise(fp_, 0, statbuf.st_size, POSIX_FADV_DONTNEED);
#endif
}
AppendFile::~AppendFile()
{
::fclose(fp_);
}
為posix_fadvise(2)指定POSIX_FADV_DONTNEED選項，告訴內核在近期不會訪問文件的指定數據，以便內核對其進行優化。
寫數據到文件
AppendFile有個兩個接口：append和write。其中，append()是供用戶調用的public接口，確保將指定數據附加到文件末尾，實際的寫文件操作是通過write()來完成的；write通過非線程安全的glibc庫函數fwrite_unlocked()來完成寫文件操作，而沒有選擇線程安全的fwrite()，主要是出於性能考慮。
一個后端通常只有一個后端線程，一個LogFile對象，一個AppendFile對象，這樣，也就只會有一個線程寫同一個log文件。
void AppendFile::append(const char *logline, size_t len)
size_t AppendFile::write(const char *logline, size_t len);
append和write實現：
void AppendFile::append(const char *logline, size_t len)
{
size_t written = 0;
/* write len byte to fp_ unless complete writing or error occurs */
while (written != len)
{
size_t remain = len - written;
size_t n = write(logline + written, remain);
if (n != remain)
{
int err = ferror(fp_);
if (err)
{
fprintf(stderr, "AppendFile::append() failed %s\n", strerror_tl(err));
clearerr(fp_); // clear error indicators for fp_
break;
}
}
written += n;
}
writtenBytes_ += written;
}
size_t AppendFile::write(const char *logline, size_t len)
{
// not thread-safe
return ::fwrite_unlocked(logline, 1, len, fp_);
}
可以看出，append是通過一個循環來確保所有數據都寫到磁盤文件上，除非發生錯誤。
[======]
使用異步日志
自此，一個完整的異步日志前端、后端都已完成。但問題在於，應用程序如何使用？
為此，寫一個測試程序，對比為前端Logger設置輸出回調函數前后不同。
#include "muduo/base/AsyncLogging.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Timestamp.h"
#include <stdio.h>
#include <unistd.h>
using namespace muduo;
static const off_t kRollSize = 1*1024*1024;
AsyncLogging* g_asyncLog = NULL;
inline AsyncLogging* getAsyncLog()
{
return g_asyncLog;
}
void test_Logging()
{
LOG_TRACE << "trace";
LOG_DEBUG << "debug";
LOG_INFO << "info";
LOG_WARN << "warn";
LOG_ERROR << "error";
LOG_SYSERR << "sys error";
// 注意不能輕易使用 LOG_FATAL, LOG_SYSFATAL, 會導致程序abort
const int n = 10;
for (int i = 0; i < n; ++i) {
LOG_INFO << "Hello, " << i << " abc...xyz";
}
}
void test_AsyncLogging()
{
const int n = 3*1024*1024;
for (int i = 0; i < n; ++i) {
LOG_INFO << "Hello, " << i << " abc...xyz";
}
}
void asyncLog(const char* msg, int len)
{
AsyncLogging* logging = getAsyncLog();
if (logging)
{
logging->append(msg, len);
}
}
int main(int argc, char* argv[])
{
printf("pid = %d\n", getpid());
AsyncLogging log(::basename(argv[0]), kRollSize);
test_Logging();
sleep(1);
g_asyncLog = &log;
Logger::setOutput(asyncLog); // 為Logger設置輸出回調, 重新配接輸出位置
log.start();
test_Logging();
test_AsyncLogging();
sleep(1);
log.stop();
return 0;
}
可以發現，在調用Logger::setOutput設置輸出回調前，默認輸出位置是stdout（見defaultOutput），而設置了輸出位置為自定義asyncLog后，每當Logger要輸出log消息時，就會通過asyncLog調用預先設置好的g_asyncLog的append()函數，將log消息輸出到AsycnLogging的Large Buffer中。
從這里也可以發現，muduo日志庫AsycnLogging類有個bug：AsycnLogging並沒有同步設置一個flush函數，這樣Logger::flush調用的其實還是默認的flush到stdout，並不能跟AsycnLogging::append()同步。當然，這不是什么難事，直接為其添加一個自定義flush()即可。
[======]
小結
1）AsyncLogging 提供多個Large Buffer緩存多條log消息，前端需要在重新配接輸出位置后，將每條log消息輸出到Large Buffer中。后端線程也是由AsyncLogging 負責維護。
2）LogFile 提供日志文件操作，包括滾動日志文件、寫日志文件。
3）AppendFile 封裝了最底層的的寫文件操作，供LogFile使用。
[======]
參考
https://blog.csdn.net/luotuo44/article/details/19252535
https://docs.microsoft.com/en-us/previous-versions/af6x78h6(v=vs.140)?redirectedfrom=MSDN