add timer_task_thread_

include/pika_binlog.h

@@ -13,6 +13,7 @@
 #include "pstd/include/pstd_status.h"
 #include "pstd/include/noncopyable.h"
 #include "include/pika_define.h"
+#include "net/src/dispatch_thread.h"
 
 std::string NewFileName(const std::string& name, uint32_t current);
 
@@ -78,6 +79,7 @@ class Binlog : public pstd::noncopyable {
   }
 
   void Close();
+  void FlushBufferedFile();
 
  private:
   pstd::Status Put(const char* item, int len);
@@ -110,6 +112,8 @@ class Binlog : public pstd::noncopyable {
   std::string filename_;
 
   std::atomic<bool> binlog_io_error_;
+
+  net::TimerTaskThread timer_task_thread_;
 };
 
 #endif

src/pika_binlog.cc

@@ -120,15 +120,29 @@ Binlog::Binlog(std::string  binlog_path, const int file_size)
     uint64_t filesize = queue_->Filesize();
     DLOG(INFO) << "Binlog: filesize is " << filesize;
   }
-
   InitLogFile();
+
+  timer_task_thread_.AddTimerTask("flush_binlog_task", 500, true,
+                                [this] { this->FlushBufferedFile(); });
+  timer_task_thread_.StartThread();
 }
 
 Binlog::~Binlog() {
   std::lock_guard l(mutex_);
+  timer_task_thread_.StopThread();
   Close();
 }
 
+void Binlog::FlushBufferedFile() {
+  std::lock_guard l(mutex_);
+  if (!opened_.load()) {
+    return;
+  }
+  if (queue_) {
+    queue_->Flush();
+  }
+}
+
 void Binlog::Close() {
   if (!opened_.load()) {
     return;

src/pstd/src/env.cc

@@ -1,15 +1,11 @@
 #include "pstd/include/env.h"
 
-#include <dirent.h>
 #include <fcntl.h>
 #include <sys/mman.h>
 #include <sys/resource.h>
-#include <sys/stat.h>
-#include <sys/time.h>
 #include <cassert>
 
 #include <cstdio>
-#include <fstream>
 #include <sstream>
 #include <utility>
 #include <thread>