Add source to Binlog::sync.

7 files changed, 29 insertions, 24 deletions

diff --git a/tddb/td/db/BinlogKeyValue.h b/tddb/td/db/BinlogKeyValue.h
index 20f802df4..a43c1ba8f 100644
--- a/tddb/td/db/BinlogKeyValue.h
+++ b/tddb/td/db/BinlogKeyValue.h
@@ -213,8 +213,8 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
     return it->second.first;
   }
 
-  void force_sync(Promise<> &&promise) final {
-    binlog_->force_sync(std::move(promise));
+  void force_sync(Promise<> &&promise, const char *source) final {
+    binlog_->force_sync(std::move(promise), source);
   }
 
   void lazy_sync(Promise<> &&promise) {
@@ -288,14 +288,14 @@ inline void BinlogKeyValue<Binlog>::add_event(uint64 seq_no, BufferSlice &&event
 }
 
 template <>
-inline void BinlogKeyValue<Binlog>::force_sync(Promise<> &&promise) {
-  binlog_->sync();
+inline void BinlogKeyValue<Binlog>::force_sync(Promise<> &&promise, const char *source) {
+  binlog_->sync(source);
   promise.set_value(Unit());
 }
 
 template <>
 inline void BinlogKeyValue<Binlog>::lazy_sync(Promise<> &&promise) {
-  force_sync(std::move(promise));
+  force_sync(std::move(promise), "lazy_sync");
 }
 
 }  // namespace td
diff --git a/tddb/td/db/KeyValueSyncInterface.h b/tddb/td/db/KeyValueSyncInterface.h
index d4e43ffef..2e68e3f7f 100644
--- a/tddb/td/db/KeyValueSyncInterface.h
+++ b/tddb/td/db/KeyValueSyncInterface.h
@@ -48,7 +48,7 @@ class KeyValueSyncInterface {
 
   virtual void erase_by_prefix(Slice prefix) = 0;
 
-  virtual void force_sync(Promise<> &&promise) = 0;
+  virtual void force_sync(Promise<> &&promise, const char *source) = 0;
 
   virtual void close(Promise<> promise) = 0;
 };
diff --git a/tddb/td/db/binlog/Binlog.cpp b/tddb/td/db/binlog/Binlog.cpp
index a6221b8d7..76ca502a9 100644
--- a/tddb/td/db/binlog/Binlog.cpp
+++ b/tddb/td/db/binlog/Binlog.cpp
@@ -286,9 +286,9 @@ Status Binlog::close(bool need_sync) {
     return Status::OK();
   }
   if (need_sync) {
-    sync();
+    sync("close");
   } else {
-    flush();
+    flush("close");
   }
 
   fd_.lock(FileFd::LockFlags::Unlock, path_, 1).ensure();
@@ -373,7 +373,7 @@ void Binlog::do_event(BinlogEvent &&event) {
         LOG(INFO) << "Load: init encryption";
       } else {
         CHECK(state_ == State::Reindex);
-        flush();
+        flush("do_event");
         update_write_encryption();
         //LOG(INFO) << format::cond(state_ == State::Run, "Run", "Reindex") << ": init encryption";
       }
@@ -404,19 +404,21 @@ void Binlog::do_event(BinlogEvent &&event) {
   fd_size_ += event_size;
 }
 
-void Binlog::sync() {
-  flush();
+void Binlog::sync(const char *source) {
+  flush(source);
   if (need_sync_) {
+    LOG(INFO) << "Sync binlog from " << source;
     auto status = fd_.sync();
     LOG_IF(FATAL, status.is_error()) << "Failed to sync binlog: " << status;
     need_sync_ = false;
   }
 }
 
-void Binlog::flush() {
+void Binlog::flush(const char *source) {
   if (state_ == State::Load) {
     return;
   }
+  LOG(DEBUG) << "Flush binlog from " << source;
   flush_events_buffer(true);
   // NB: encryption happens during flush
   if (byte_flow_flag_) {
@@ -448,7 +450,7 @@ void Binlog::lazy_flush() {
   buffer_reader_.sync_with_writer();
   auto size = buffer_reader_.size() + events_buffer_size;
   if (size > (1 << 14)) {
-    flush();
+    flush("lazy_flush");
   } else if (size > 0 && need_flush_since_ == 0) {
     need_flush_since_ = Time::now_cached();
   }
@@ -660,7 +662,7 @@ void Binlog::do_reindex() {
     do_event(std::move(event));  // NB: no move is actually happens
   });
   {
-    flush();
+    flush("do_reindex");
     if (start_size != 0) {  // must sync creation of the file if it is non-empty
       auto status = fd_.sync_barrier();
       LOG_IF(FATAL, status.is_error()) << "Failed to sync binlog: " << status;
diff --git a/tddb/td/db/binlog/Binlog.h b/tddb/td/db/binlog/Binlog.h
index 14cb25517..82b8276a2 100644
--- a/tddb/td/db/binlog/Binlog.h
+++ b/tddb/td/db/binlog/Binlog.h
@@ -109,8 +109,8 @@ class Binlog {
   }
 
   void add_event(BinlogEvent &&event);
-  void sync();
-  void flush();
+  void sync(const char *source);
+  void flush(const char *source);
   void lazy_flush();
   double need_flush_since() const {
     return need_flush_since_;
diff --git a/tddb/td/db/binlog/BinlogInterface.h b/tddb/td/db/binlog/BinlogInterface.h
index 37c6cdc88..c6d6b9664 100644
--- a/tddb/td/db/binlog/BinlogInterface.h
+++ b/tddb/td/db/binlog/BinlogInterface.h
@@ -74,7 +74,7 @@ class BinlogInterface {
     return seq_no;
   }
 
-  virtual void force_sync(Promise<> promise) = 0;
+  virtual void force_sync(Promise<> promise, const char *source) = 0;
   virtual void force_flush() = 0;
   virtual void change_key(DbKey db_key, Promise<> promise) = 0;
 
diff --git a/tddb/td/db/binlog/ConcurrentBinlog.cpp b/tddb/td/db/binlog/ConcurrentBinlog.cpp
index 7ae0df269..329e08d30 100644
--- a/tddb/td/db/binlog/ConcurrentBinlog.cpp
+++ b/tddb/td/db/binlog/ConcurrentBinlog.cpp
@@ -61,7 +61,8 @@ class BinlogActor final : public Actor {
     try_flush();
   }
 
-  void force_sync(Promise<> &&promise) {
+  void force_sync(Promise<> &&promise, const char *source) {
+    LOG(INFO) << "Force binlog sync from " << source;
     auto seq_no = processor_.max_unfinished_seq_no();
     if (processor_.max_finished_seq_no() == seq_no) {
       do_immediate_sync(std::move(promise));
@@ -72,7 +73,7 @@ class BinlogActor final : public Actor {
 
   void force_flush() {
     // TODO: use same logic as in force_sync
-    binlog_->flush();
+    binlog_->flush("force_flush");
     flush_flag_ = false;
   }
 
@@ -115,7 +116,7 @@ class BinlogActor final : public Actor {
     auto need_flush_since = binlog_->need_flush_since();
     auto now = Time::now_cached();
     if (now > need_flush_since + FLUSH_TIMEOUT - 1e-9) {
-      binlog_->flush();
+      binlog_->flush("try_flush");
     } else {
       if (!force_sync_flag_) {
         flush_flag_ = true;
@@ -161,7 +162,7 @@ class BinlogActor final : public Actor {
     flush_flag_ = false;
     wakeup_at_ = 0;
     if (need_sync) {
-      binlog_->sync();
+      binlog_->sync("timeout_expired");
       // LOG(ERROR) << "BINLOG SYNC";
       set_promises(sync_promises_);
     } else if (need_flush) {
@@ -205,12 +206,14 @@ void ConcurrentBinlog::add_raw_event_impl(uint64 event_id, BufferSlice &&raw_eve
   send_closure(binlog_actor_, &detail::BinlogActor::add_raw_event, event_id, std::move(raw_event), std::move(promise),
                info);
 }
-void ConcurrentBinlog::force_sync(Promise<> promise) {
-  send_closure(binlog_actor_, &detail::BinlogActor::force_sync, std::move(promise));
+void ConcurrentBinlog::force_sync(Promise<> promise, const char *source) {
+  send_closure(binlog_actor_, &detail::BinlogActor::force_sync, std::move(promise), source);
 }
+
 void ConcurrentBinlog::force_flush() {
   send_closure(binlog_actor_, &detail::BinlogActor::force_flush);
 }
+
 void ConcurrentBinlog::change_key(DbKey db_key, Promise<> promise) {
   send_closure(binlog_actor_, &detail::BinlogActor::change_key, std::move(db_key), std::move(promise));
 }
diff --git a/tddb/td/db/binlog/ConcurrentBinlog.h b/tddb/td/db/binlog/ConcurrentBinlog.h
index 464ddb22d..321e2ebd5 100644
--- a/tddb/td/db/binlog/ConcurrentBinlog.h
+++ b/tddb/td/db/binlog/ConcurrentBinlog.h
@@ -41,7 +41,7 @@ class ConcurrentBinlog final : public BinlogInterface {
   ConcurrentBinlog &operator=(ConcurrentBinlog &&) = delete;
   ~ConcurrentBinlog() final;
 
-  void force_sync(Promise<> promise) final;
+  void force_sync(Promise<> promise, const char *source) final;
   void force_flush() final;
   void change_key(DbKey db_key, Promise<> promise) final;