A lightweight and efficient multi-threaded c++stream style asynchronous log (ii)

preface

This article follows the previous article: it shows you how each of the above logs can be asynchronously imported into a local file.

I’ll start with a brief introduction to the LogFile class and then go into the double buffering mechanism in AsyncLogging.

The overall log module structure diagram,

LogFile class

LogFile the LogFile class manages log files. RollFile () : rolls a new LogFile when the log exceeds the size of m_rollSize. When used with scroll log, name the log file with scroll time suffixed. M_mutex: lock file when used with append() data. Append () : stick to log. Flush () : flush buffer.

LogFile has a AppendFIle class that is ultimately used to manipulate local files. Append () : fwrite() is called to write to the local file. Flush () : flush the buffer. WrittenBytes () : gets the number of written sections.

AsyncLogging class

AsyncLogging Logs Asynchronously write logs. Before introducing its interface, let’s describe its working logic.

AsyncLogging has the following types of caches: m_currentBuffer: indicates the cache that is currently receiving logs from another thread append. M_nextBuffer: Points to the cache used to replace m_currentBuffer when m_currentBuffer is full.

BackupBuffer1: backup cache backupBuffer2: backup cache buffersToWrite: and m_buffers are exchanged after swap() and append() to the pointer container of LogFile.

AsyncLogging uses a double-buffering mechanism with two cache containers: m_buffers and buffersToWrite are used alternately. A and B. A is used to receive incoming logs from the other thread Append (). B is used to write the currently accepted cache to the log file. When B is done, clean() B, swap A, B, and so on.

Advantages: The newly created log does not have to wait for disk operation, and avoids triggering the log thread for each new log. Instead, multiple logs are sent to the log thread to write a large buffer to the file. Equivalent to batch processing, reduce thread wake up frequency, reduce overhead. In addition, in order to write the log message to the file in A timely manner, the log performs the above write operation every three seconds, even if there is no push in Buffer A.

AsyncLogging uses a larger LogBuffer to hold logs sent from one Logger to another.

Mutex: Used to control multithreaded writes.

Condition: Waits for data in the buffer.

Thread: Uses a single Thread to handle the swapping of caches and the writing of logs.

AsyncLogging implementation

A simple implementation of AsyncLogging is shown below. There are actually several spare caches that I haven’t added here to make it easier to understand the program; The standby cache is mainly to reduce the overhead caused by repeated new operations,

#ifndef _ASYNC_LOGGING_HH #define _ASYNC_LOGGING_HH #include "MutexLock.hh" #include "Thread.hh" #include "LogStream.hh"  #include "ptr_vector.hh" #include "Condition.hh" #include <string> class AsyncLogging { public: AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval = 3); ~AsyncLogging(); void start(){ m_isRunning = true; m_thread.start(); } void stop(){ m_isRunning = false; m_cond.notify(); } void append(const char *logline, int len); private: AsyncLogging(const AsyncLogging&); AsyncLogging& operator=(const AsyncLogging&); void threadRoutine(); typedef LogBuffer<kLargeBuffer> Buffer; typedef oneself::ptr_vector<Buffer> BufferVector; typedef oneself::auto_ptr<Buffer> BufferPtr; const int m_flushInterval; bool m_isRunning; off_t m_rollSize; std::string m_filePath; Thread m_thread; MutexLock m_mutex; Condition m_cond; BufferPtr m_currentBuffer; BufferVector m_buffers; }; #endif //AsyncLogging.cpp #include "AsyncLogging.hh" #include "LogFile.hh" #include <assert.h> #include <stdio.h> AsyncLogging::AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval) :m_filePath(filePath), m_rollSize(2048), m_flushInterval(flushInterval), m_isRunning(false), m_thread(std::bind(&AsyncLogging::threadRoutine, this)), m_mutex(), m_cond(m_mutex), m_currentBuffer(new Buffer), m_buffers() { } AsyncLogging::~AsyncLogging(){ if(m_isRunning) stop(); } void AsyncLogging::append(const char* logline, int len){ MutexLockGuard lock(m_mutex); if(m_currentBuffer->avail() > len){ m_currentBuffer->append(logline, len); } else{ m_buffers.push_back(m_currentBuffer.release()); m_currentBuffer.reset(new Buffer); m_currentBuffer->append(logline, len); m_cond.notify(); } } void AsyncLogging::threadRoutine(){ assert(m_isRunning == true); LogFile output(m_filePath, m_rollSize, false); BufferVector buffersToWrite; buffersToWrite.reserve(8); while(m_isRunning){ assert(buffersToWrite.empty()); { MutexLockGuard lock(m_mutex); if(m_buffers.empty()){ m_cond.waitForSeconds(m_flushInterval); } m_buffers.push_back(m_currentBuffer.release()); m_currentBuffer.reset(new Buffer); m_buffers.swap(buffersToWrite); } assert(! buffersToWrite.empty()); for(size_t i = 0; i < buffersToWrite.size(); ++i){ output.append(buffersToWrite[i]->data(), buffersToWrite[i]->length()); } buffersToWrite.clear(); output.flush(); } output.flush(); }Copy the code

Add backup cache

Add an alternate buffer to optimize the above program, which performs new operations in two places. When it’s time to write m_currentBuffer to a local file, it will remove its current contents, so you need to create a new cache for m_currentBuffer.

So we prepare a m_nextBuffer to be the standby cache of m_currentBuffer. Add two backupBuffers to m_nextBuffer as backup buffers. When the log volume becomes too large, consider adding the cache dynamically with the new operation.

#ifndef _ASYNC_LOGGING_HH #define _ASYNC_LOGGING_HH #include "MutexLock.hh" #include "Thread.hh" #include "LogStream.hh"  #include "ptr_vector.hh" #include "Condition.hh" #include <memory> #include <string> class AsyncLogging { public: AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval = 3); ~AsyncLogging(); void start(){ m_isRunning = true; m_thread.start(); } void stop(){ m_isRunning = false; m_cond.notify(); } void append(const char *logline, int len); private: AsyncLogging(const AsyncLogging&); AsyncLogging& operator=(const AsyncLogging&); void threadRoutine(); typedef LogBuffer<kLargeBuffer> Buffer; typedef myself::ptr_vector<Buffer> BufferVector; typedef std::unique_ptr<Buffer> BufferPtr; const int m_flushInterval; bool m_isRunning; off_t m_rollSize; std::string m_filePath; Thread m_thread; MutexLock m_mutex; Condition m_cond; BufferPtr m_currentBuffer; BufferPtr m_nextBuffer; BufferVector m_buffers; }; #endif //AsynvLogging.cpp #include "AsyncLogging.hh" #include "LogFile.hh" #include <assert.h> #include <stdio.h> AsyncLogging::AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval) :m_filePath(filePath), m_rollSize(rollSize), m_flushInterval(flushInterval), m_isRunning(false), m_thread(std::bind(&AsyncLogging::threadRoutine, this)), m_mutex(), m_cond(m_mutex), m_currentBuffer(new Buffer), m_nextBuffer(new Buffer), m_buffers() { } AsyncLogging::~AsyncLogging(){ if(m_isRunning) stop(); } void AsyncLogging::append(const char* logline, int len){ MutexLockGuard lock(m_mutex); if(m_currentBuffer->avail() > len){ m_currentBuffer->append(logline, len); } else{ m_buffers.push_back(m_currentBuffer.release()); if(m_nextBuffer){ m_currentBuffer = std::move(m_nextBuffer); } else{ m_currentBuffer.reset(new Buffer); } m_currentBuffer->append(logline, len); m_cond.notify(); } } void AsyncLogging::threadRoutine(){ assert(m_isRunning == true); LogFile output(m_filePath, m_rollSize, false); BufferPtr backupBuffer1(new Buffer); BufferPtr backupBuffer2(new Buffer); BufferVector buffersToWrite; buffersToWrite.reserve(8); while(m_isRunning){ assert(buffersToWrite.empty()); { MutexLockGuard lock(m_mutex); if(m_buffers.empty()){ m_cond.waitForSeconds(m_flushInterval); } m_buffers.push_back(m_currentBuffer.release()); m_currentBuffer = std::move(backupBuffer1); m_buffers.swap(buffersToWrite); if(! m_nextBuffer) m_nextBuffer = std::move(backupBuffer2); } assert(! buffersToWrite.empty()); for(size_t i = 0; i < buffersToWrite.size(); ++i){ output.append(buffersToWrite[i]->data(), buffersToWrite[i]->length()); } if(buffersToWrite.size() > 2) { // drop non-bzero-ed buffers, avoid trashing buffersToWrite.resize(2); } if(! backupBuffer1) { assert(! buffersToWrite.empty()); backupBuffer1 = std::move(buffersToWrite.pop_back()); backupBuffer1->reset(); } if(! backupBuffer2) { assert(! buffersToWrite.empty()); backupBuffer2 = std::move(buffersToWrite.pop_back()); backupBuffer2->reset(); } buffersToWrite.clear(); output.flush(); } output.flush(); }Copy the code

conclusion

This article focuses on the implementation of the AsyncLogging class in Muduo, which includes a double-caching mechanism. The LogFile and AppendFIle classes are basic operations on log files and local files, respectively. If you are interested, you can look at the source code of Muduo. This article will not be written further. If you want the full source code, you can leave a comment.

Alumiana

www.cnblogs.com/ailumiyana/

Alumiana

A lightweight and efficient multi-threaded c++stream style asynchronous log (ii)