PS-10245 feature: Implement receiving binlog events in GTID mode (part 4) (#85)

https://perconadev.atlassian.net/browse/PS-10245

Implemented more sophisticated event data buffering and flushing at the
'binstv::storage' level. We now accumulate data from several events in the
'binstv::storage'internal memory buffer (16K initially but can grow) and perform
actual writing only at transaction boundaries. This helps a lot with making sure that
after a crash we can continue operations from a position that corresponds to a
committed transaction. This is vitally important for GTID-based replication.
'binstv::storage::write_event()' method extended with an extra parameter that indicates
whether the event data being written corresponds to the end of a transaction.

Reworked 'binsrv::basic_storage_engine' interface: removed 'flush_stream()' method
and all 'write_data_to_stream()' method overloads are now expected to immediately
flush data. 'binsrv::filesystem_storage_backend' and 'binsrv::s3_storage_backend'
concrete implementations updated to accommodate these changes.

We also added "rdbuf()->pubsetbuf(nullptr, 0U)" calls before `std::fstream` read/writes
inside 'binsrv::filesystem_storage_backend' to disable internal buffering.

Fixed an issue with unnecessary data uploading upon stream closing / object
destruction inside 'binsrv::s3_storage_backend'.

'binlog_streaming.binsrv' MTR test case adapted for new buffering behavior.

Fixed problem with the 'checkpointing_both' combination in the
'binlog_streaming.checkpointing' MTR test case when checkpointing parameters were not set properly.

Author: percona-ysorokin

mtr/binlog_streaming/t/binsrv.test

@@ -42,6 +42,7 @@ INSERT INTO t1 VALUES(DEFAULT);
 --let $binsrv_read_timeout = 60
 --let $binsrv_idle_time = 10
 --let $binsrv_verify_checksum = $extracted_init_connect_variable_value
+--let $binsrv_checkpoint_size = 1
 --source ../include/set_up_binsrv_environment.inc
 
 --echo

mtr/binlog_streaming/t/checkpointing.test

@@ -70,7 +70,7 @@ if ($extracted_init_connect_variable_value == 'interval')
 {
   --let $binsrv_checkpoint_interval = 5s
 }
-if ($extracted_init_connect_variable_value == 'interval')
+if ($extracted_init_connect_variable_value == 'both')
 {
   --let $binsrv_checkpoint_size = 2M
   --let $binsrv_checkpoint_interval = 5s

src/app.cpp

@@ -357,6 +357,7 @@ void log_span_dump(binsrv::basic_logger &logger,
 
 void process_binlog_event(const binsrv::event::event &current_event,
                           util::const_byte_span portion,
+                          const binsrv::event::reader_context &context,
                           binsrv::storage &storage, bool &skip_open_binlog) {
   const auto &current_common_header = current_event.get_common_header();
   const auto code = current_common_header.get_type_code();
@@ -399,7 +400,7 @@ void process_binlog_event(const binsrv::event::event &current_event,
     }
   }
   if (!is_artificial && !is_pseudo) {
-    storage.write_event(portion);
+    storage.write_event(portion, context.is_at_transaction_boundary());
   }
   if ((code == binsrv::event::code_type::rotate && !is_artificial) ||
       code == binsrv::event::code_type::stop) {
@@ -515,7 +516,8 @@ void receive_binlog_events(
               boost::lexical_cast<std::string>(context.get_transaction_gtid()));
     }
 
-    process_binlog_event(current_event, portion, storage, skip_open_binlog);
+    process_binlog_event(current_event, portion, context, storage,
+                         skip_open_binlog);
   }
   if (termination_flag.test()) {
     logger.log(binsrv::log_severity::info,
@@ -531,6 +533,8 @@ void receive_binlog_events(
     util::exception_location().raise<std::logic_error>(
         "fetch operation did not reach EOF reading binlog events");
   }
+  // TODO: here (upon timing out) we also need to flush internal buffers in
+  //       the storage
   logger.log(binsrv::log_severity::info,
              "timed out waiting for events and disconnected");
 }

src/binsrv/basic_storage_backend.cpp

@@ -58,14 +58,6 @@ void basic_storage_backend::write_data_to_stream(util::const_byte_span data) {
   do_write_data_to_stream(data);
 }
 
-void basic_storage_backend::flush_stream() {
-  if (!stream_open_) {
-    util::exception_location().raise<std::logic_error>(
-        "cannot flush the stream as it has not been opened");
-  }
-  do_flush_stream();
-}
-
 void basic_storage_backend::close_stream() {
   if (!stream_open_) {
     util::exception_location().raise<std::logic_error>(

src/binsrv/basic_storage_backend.hpp

@@ -43,7 +43,6 @@ class basic_storage_backend {
   void open_stream(std::string_view name,
                    storage_backend_open_stream_mode mode);
   void write_data_to_stream(util::const_byte_span data);
-  void flush_stream();
   void close_stream();
 
   [[nodiscard]] std::string get_description() const;
@@ -59,7 +58,6 @@ class basic_storage_backend {
   virtual void do_open_stream(std::string_view name,
                               storage_backend_open_stream_mode mode) = 0;
   virtual void do_write_data_to_stream(util::const_byte_span data) = 0;
-  virtual void do_flush_stream() = 0;
   virtual void do_close_stream() = 0;
 
   [[nodiscard]] virtual std::string do_get_description() const = 0;

src/binsrv/event/reader_context.hpp

@@ -58,8 +58,9 @@ class [[nodiscard]] reader_context {
     return transaction_gtid_;
   }
   [[nodiscard]] bool is_at_transaction_boundary() const noexcept {
-    return state_ == state_type::any_other_expected &&
-           current_transaction_length_ == expected_transaction_length_;
+    return (state_ == state_type::any_other_expected &&
+            current_transaction_length_ == expected_transaction_length_) ||
+           (state_ == state_type::rotate_artificial_expected);
   }
 
 private:

src/binsrv/filesystem_storage_backend.cpp

@@ -40,6 +40,9 @@ filesystem_storage_backend::filesystem_storage_backend(
     : root_path_{}, ofs_{} {
   // TODO: switch to utf8 file names
 
+  // setting "unbuffered mode" as we will be using our own buffer
+  ofs_.rdbuf()->pubsetbuf(nullptr, 0U);
+
   const auto &backend_uri = config.get<"uri">();
 
   const auto uri_parse_result{boost::urls::parse_absolute_uri(backend_uri)};
@@ -109,8 +112,9 @@ filesystem_storage_backend::do_get_object(std::string_view name) {
   const auto object_path{get_object_path(name)};
 
   // opening in binary mode
-  std::ifstream object_ifs{object_path,
-                           std::ios_base::in | std::ios_base::binary};
+  std::ifstream object_ifs{};
+  object_ifs.rdbuf()->pubsetbuf(nullptr, 0U);
+  object_ifs.open(object_path, std::ios_base::in | std::ios_base::binary);
   if (!object_ifs.is_open()) {
     util::exception_location().raise<std::runtime_error>(
         "cannot open underlying object file");
@@ -134,9 +138,10 @@ void filesystem_storage_backend::do_put_object(std::string_view name,
                                                util::const_byte_span content) {
   const auto object_path = get_object_path(name);
   // opening in binary mode with truncating
-  std::ofstream object_ofs{object_path, std::ios_base::out |
-                                            std::ios_base::binary |
-                                            std::ios_base::trunc};
+  std::ofstream object_ofs{};
+  object_ofs.rdbuf()->pubsetbuf(nullptr, 0U);
+  object_ofs.open(object_path, std::ios_base::out | std::ios_base::binary |
+                                   std::ios_base::trunc);
   if (!object_ofs.is_open()) {
     util::exception_location().raise<std::runtime_error>(
         "cannot open underlying object file for writing");
@@ -174,10 +179,6 @@ void filesystem_storage_backend::do_write_data_to_stream(
     util::exception_location().raise<std::runtime_error>(
         "cannot write data to the underlying stream file");
   }
-}
-
-void filesystem_storage_backend::do_flush_stream() {
-  assert(ofs_.is_open());
   if (!ofs_.flush()) {
     util::exception_location().raise<std::runtime_error>(
         "cannot flush the underlying stream file");

src/binsrv/filesystem_storage_backend.hpp

@@ -51,7 +51,6 @@ class [[nodiscard]] filesystem_storage_backend final
   void do_open_stream(std::string_view name,
                       storage_backend_open_stream_mode mode) override;
   void do_write_data_to_stream(util::const_byte_span data) override;
-  void do_flush_stream() override;
   void do_close_stream() override;
 
   [[nodiscard]] std::string do_get_description() const override;

src/binsrv/s3_storage_backend.cpp

@@ -683,10 +683,9 @@ void s3_storage_backend::do_write_data_to_stream(util::const_byte_span data) {
     util::exception_location().raise<std::runtime_error>(
         "cannot flush the temporary file for S3 object");
   }
+  upload_tmp_stream_internal();
 }
 
-void s3_storage_backend::do_flush_stream() { upload_tmp_stream_internal(); }
-
 void s3_storage_backend::do_close_stream() {
   assert(tmp_fstream_.is_open());
   close_stream_internal();
@@ -743,7 +742,6 @@ void s3_storage_backend::upload_tmp_stream_internal() {
 }
 
 void s3_storage_backend::close_stream_internal() {
-  upload_tmp_stream_internal();
   tmp_fstream_.close();
   // we allow std::filesystem::remove() here to fail - worst case scenario
   // we will have a temorary file not removed

src/binsrv/s3_storage_backend.hpp

@@ -76,7 +76,6 @@ class [[nodiscard]] s3_storage_backend final : public basic_storage_backend {
   void do_open_stream(std::string_view name,
                       storage_backend_open_stream_mode mode) override;
   void do_write_data_to_stream(util::const_byte_span data) override;
-  void do_flush_stream() override;
   void do_close_stream() override;
 
   [[nodiscard]] std::string do_get_description() const override;