opt(log,trace):1.10.15, 在AsyncSink中添加写文件时长的监控；优化trace::Sink写文件效率

Author: hevake

modules/log/async_sink.cpp

@@ -22,6 +22,7 @@
 #include <cstring>
 #include <algorithm>
 #include <iostream>
+#include <chrono>
 
 constexpr uint32_t LOG_MAX_LEN = (100 << 10);   //! 限定单条日志最大长度
 
@@ -59,6 +60,8 @@ void AsyncSink::onLogFrontEnd(const LogContent *content)
 
 void AsyncSink::onLogBackEndReadPipe(const void *data_ptr, size_t data_size)
 {
+    auto start_ts = std::chrono::steady_clock::now();
+
     buffer_.append(data_ptr, data_size);
 
     bool is_need_flush = false;
@@ -81,6 +84,10 @@ void AsyncSink::onLogBackEndReadPipe(const void *data_ptr, size_t data_size)
 
     if (is_need_flush)
         flushLog();
+
+    auto time_cost = std::chrono::steady_clock::now() - start_ts;
+    if (time_cost > std::chrono::milliseconds(500))
+        std::cerr << timestamp_str_ << " NOTICE: log sink cost > 500 ms, " << time_cost.count() / 1000 << " us" << std::endl;
 }
 
 void AsyncSink::onLogBackEnd(const LogContent &content)

modules/trace/sink.cpp

@@ -25,6 +25,7 @@
 #include <cstring>
 #include <vector>
 #include <sstream>
+#include <chrono>
 
 #include <sys/syscall.h>
 #include <tbox/base/defines.h>
@@ -173,6 +174,14 @@ void Sink::commitRecord(const char *name, const char *module, uint32_t line, uin
 
 void Sink::onBackendRecvData(const void *data, size_t size)
 {
+    auto start_ts = std::chrono::steady_clock::now();
+
+    if (!checkAndCreateRecordFile() ||
+        !checkAndWriteNames() ||
+        !checkAndWriteModules() ||
+        !checkAndWriteThreads())
+        return;
+
     buffer_.append(data, size);
     std::vector<uint8_t> write_cache;
 
@@ -200,6 +209,7 @@ void Sink::onBackendRecvData(const void *data, size_t size)
     }
 
     if (!write_cache.empty()) {
+
         auto wsize = ::write(curr_record_fd_, write_cache.data(), write_cache.size());
         if (wsize != static_cast<ssize_t>(write_cache.size())) {
             LogErrno(errno, "write record file '%s' fail", curr_record_filename_.c_str());
@@ -210,16 +220,17 @@ void Sink::onBackendRecvData(const void *data, size_t size)
         if (total_write_size_ >= record_file_max_size_)
             CHECK_CLOSE_RESET_FD(curr_record_fd_);
     }
+
+    auto time_cost = std::chrono::steady_clock::now() - start_ts;
+    if (time_cost > std::chrono::milliseconds(100))
+        LogNotice("trace sink cost > 100 ms, %lu us", time_cost.count() / 1000);
 }
 
 void Sink::onBackendRecvRecord(const RecordHeader &record, const char *name, const char *module, std::vector<uint8_t> &write_cache)
 {
     if (!isFilterPassed(module))
         return;
 
-    if (!checkAndCreateRecordFile())
-        return;
-
     auto thread_index = allocThreadIndex(record.thread_id);
     auto name_index = allocNameIndex(name, record.line);
     auto module_index = allocModuleIndex(module);
@@ -285,7 +296,7 @@ bool Sink::checkAndCreateRecordFile()
     return true;
 }
 
-Sink::Index Sink::allocNameIndex(const std::string &name, uint32_t line)
+bool Sink::checkAndWriteNames()
 {
     //! 如果文件不存在了，则重写所有的名称列表
     if (!util::fs::IsFileExist(name_list_filename_)) {
@@ -297,24 +308,15 @@ Sink::Index Sink::allocNameIndex(const std::string &name, uint32_t line)
         for (auto &content: name_vec)
             oss << content << ENDLINE;
 
-        util::fs::WriteStringToTextFile(name_list_filename_, oss.str(), is_file_sync_enabled_);
+        return util::fs::WriteStringToTextFile(name_list_filename_, oss.str(), is_file_sync_enabled_);
     }
 
-    std::string content = name + " at L" + std::to_string(line);
-    auto iter = name_to_index_map_.find(content);
-    if (iter != name_to_index_map_.end())
-        return iter->second;
-
-    auto new_index = next_name_index_++;
-    name_to_index_map_[content] = new_index;
-
-    util::fs::AppendStringToTextFile(name_list_filename_, content + ENDLINE, is_file_sync_enabled_);
-    return new_index;
+    return true;
 }
 
-Sink::Index Sink::allocModuleIndex(const std::string &module)
+bool Sink::checkAndWriteModules()
 {
-    //! 如果文件不存在了，则重写所有的名称列表
+    //! 如果文件不存在了，则重写所有的模块列表
     if (!util::fs::IsFileExist(module_list_filename_)) {
         std::vector<std::string> module_vec(module_to_index_map_.size());
         for (auto &item : module_to_index_map_)
@@ -324,21 +326,13 @@ Sink::Index Sink::allocModuleIndex(const std::string &module)
         for (auto &module : module_vec)
             oss << module << ENDLINE;
 
-        util::fs::WriteStringToTextFile(module_list_filename_, oss.str(), is_file_sync_enabled_);
+        return util::fs::WriteStringToTextFile(module_list_filename_, oss.str(), is_file_sync_enabled_);
     }
 
-    auto iter = module_to_index_map_.find(module);
-    if (iter != module_to_index_map_.end())
-        return iter->second;
-
-    auto new_index = next_module_index_++;
-    module_to_index_map_[module] = new_index;
-
-    util::fs::AppendStringToTextFile(module_list_filename_, module + ENDLINE, is_file_sync_enabled_);
-    return new_index;
+    return true;
 }
 
-Sink::Index Sink::allocThreadIndex(long thread_id)
+bool Sink::checkAndWriteThreads()
 {
     //! 如果文件不存在了，则重写所有的线程列表
     if (!util::fs::IsFileExist(thread_list_filename_)) {
@@ -350,9 +344,41 @@ Sink::Index Sink::allocThreadIndex(long thread_id)
         for (auto thread_id : thread_vec)
             oss << thread_id << ENDLINE;
 
-        util::fs::WriteStringToTextFile(thread_list_filename_, oss.str(), is_file_sync_enabled_);
+        return util::fs::WriteStringToTextFile(thread_list_filename_, oss.str(), is_file_sync_enabled_);
     }
 
+    return true;
+}
+
+Sink::Index Sink::allocNameIndex(const std::string &name, uint32_t line)
+{
+    std::string content = name + " at L" + std::to_string(line);
+    auto iter = name_to_index_map_.find(content);
+    if (iter != name_to_index_map_.end())
+        return iter->second;
+
+    auto new_index = next_name_index_++;
+    name_to_index_map_[content] = new_index;
+
+    util::fs::AppendStringToTextFile(name_list_filename_, content + ENDLINE, is_file_sync_enabled_);
+    return new_index;
+}
+
+Sink::Index Sink::allocModuleIndex(const std::string &module)
+{
+    auto iter = module_to_index_map_.find(module);
+    if (iter != module_to_index_map_.end())
+        return iter->second;
+
+    auto new_index = next_module_index_++;
+    module_to_index_map_[module] = new_index;
+
+    util::fs::AppendStringToTextFile(module_list_filename_, module + ENDLINE, is_file_sync_enabled_);
+    return new_index;
+}
+
+Sink::Index Sink::allocThreadIndex(long thread_id)
+{
     auto iter = thread_to_index_map_.find(thread_id);
     if (iter != thread_to_index_map_.end())
         return iter->second;

modules/trace/sink.h

@@ -113,6 +113,9 @@ class Sink {
     void onBackendRecvRecord(const RecordHeader &record, const char *name, const char *module,
                              std::vector<uint8_t> &write_cache);
 
+    bool checkAndWriteNames();
+    bool checkAndWriteModules();
+    bool checkAndWriteThreads();
     bool checkAndCreateRecordFile();
 
     bool isFilterPassed(const std::string &module) const;

version.mk

@@ -21,4 +21,4 @@
 # TBOX版本号
 TBOX_VERSION_MAJOR := 1
 TBOX_VERSION_MINOR := 10
-TBOX_VERSION_REVISION := 14
+TBOX_VERSION_REVISION := 15