BUG#35403452 PAR Dump Enhancements: File Buffering and Worker Session Pool

When dumping using PAR, the workers split the data dump operation in 2 parts:

- Buffering data
- Transferring data to Bucket

This patch replaces the memory buffering with file buffering. The dump files will
first be dumped in local disk to later be transferred to the target bucket.
The location of the files will in the system default temporary folder.

Additionally, a new session pool is introduced on the dumper to enable the
workers to release the database session as soon as the buffering step is complete.
This way, the worker thread can continue with the data transfer while another
worker can pick the next dump operation.

Also the number of workers when dumping with PAR is internally doubled, so there
can be up to 'threads'*2 workers doing dump/transfer operations while there are
at most 'threads' workers doing dump operations.

Notes on System Temporary Directory
-----------------------------------

The following order of precedence is used to determine the value of the temporary
directory:

POSIX: TMPDIR, TMP, TEMP, TEMPDIR. If none is defined uses /tmp.

Windows: TMP, TEMP, USERPROFILE. If none defined, uses the Windows directory.

Change-Id: I99c534c6b24857ceb7544a14f847cf5a4c7bb9c9

Author: rennox

modules/util/dump/ddl_dumper_options.cc

@@ -65,7 +65,7 @@ const shcore::Option_pack_def<Ddl_dumper_options>
           .include<Dump_options>()
           .optional("chunking", &Ddl_dumper_options::m_split)
           .optional("bytesPerChunk", &Ddl_dumper_options::set_bytes_per_chunk)
-          .optional("threads", &Ddl_dumper_options::m_threads)
+          .optional("threads", &Ddl_dumper_options::set_threads)
           .optional("triggers", &Ddl_dumper_options::m_dump_triggers)
           .optional("tzUtc", &Ddl_dumper_options::m_timezone_utc)
           .optional("ddlOnly", &Ddl_dumper_options::m_ddl_only)
@@ -174,6 +174,13 @@ void Ddl_dumper_options::set_dry_run(bool dry_run) {
   set_dry_run_mode(dry_run ? Dry_run::DONT_WRITE_ANY_FILES : Dry_run::DISABLED);
 }
 
+void Ddl_dumper_options::set_threads(uint64_t threads) {
+  m_threads = threads;
+
+  // By default, m_worker_threads is equal to m_threads
+  m_worker_threads = threads;
+}
+
 const Object_storage_options *Ddl_dumper_options::object_storage_options()
     const {
   if (m_dump_manifest_options) {
@@ -201,6 +208,9 @@ void Ddl_dumper_options::set_output_url(http://github.com/const%20std::string%20&url) {
     } else {
       if (par.type() == mysqlshdk::oci::PAR_type::PREFIX) {
         set_storage_config(dump::common::get_par_config(par));
+
+        // For dumps with PAR prefix, doubles the number of worker threads
+        m_worker_threads = m_threads * 2;
       } else {
         throw std::invalid_argument("The given URL is not a prefix PAR.");
       }

modules/util/dump/ddl_dumper_options.h

@@ -56,6 +56,8 @@ class Ddl_dumper_options : public Dump_options {
 
   std::size_t threads() const override { return m_threads; }
 
+  std::size_t worker_threads() const override { return m_worker_threads; }
+
   bool is_export_only() const override { return false; }
 
   bool use_single_file() const override { return false; }
@@ -96,6 +98,7 @@ class Ddl_dumper_options : public Dump_options {
   void set_ocimds(bool value);
   void set_compatibility_options(const std::vector<std::string> &options);
   void set_dry_run(bool dry_run);
+  void set_threads(uint64_t threads);
   const Object_storage_options *object_storage_options() const;
 
   Dump_manifest_options m_dump_manifest_options;
@@ -106,8 +109,15 @@ class Ddl_dumper_options : public Dump_options {
 
   bool m_split = true;
   uint64_t m_bytes_per_chunk;
+
+  // Number of threads requested by the user (or default)
+  // At most this number of database connections will be used in the dump
+  // operation.
   uint64_t m_threads = 4;
 
+  // Internal number of threads (to be doubled in the case of prefix PAR dumps)
+  uint64_t m_worker_threads = 4;
+
   bool m_dump_triggers = true;
   bool m_timezone_utc = true;
   bool m_ddl_only = false;

modules/util/dump/dump_options.h

@@ -148,6 +148,8 @@ class Dump_options {
 
   virtual std::size_t threads() const = 0;
 
+  virtual std::size_t worker_threads() const { return threads(); }
+
   virtual bool is_export_only() const = 0;
 
   virtual bool use_single_file() const = 0;

modules/util/dump/dumper.cc

@@ -25,7 +25,6 @@
 
 #include <algorithm>
 #include <chrono>
-#include <condition_variable>
 #include <cstdint>
 #include <functional>
 #include <iterator>
@@ -57,6 +56,7 @@
 #include "mysqlshdk/libs/utils/rate_limit.h"
 #include "mysqlshdk/libs/utils/std.h"
 #include "mysqlshdk/libs/utils/strformat.h"
+#include "mysqlshdk/libs/utils/utils_file.h"
 #include "mysqlshdk/libs/utils/utils_general.h"
 #include "mysqlshdk/libs/utils/utils_lexing.h"
 #include "mysqlshdk/libs/utils/utils_mysql_parsing.h"
@@ -345,31 +345,6 @@ bool check_if_transactions_are_ddl_safe(
 
 }  // namespace
 
-class Dumper::Synchronize_workers final {
- public:
-  Synchronize_workers() = default;
-  ~Synchronize_workers() = default;
-
-  void wait_for(const uint16_t count) {
-    std::unique_lock<std::mutex> lock(m_mutex);
-    m_condition.wait(lock, [this, count]() { return m_count >= count; });
-    m_count -= count;
-  }
-
-  void notify() {
-    {
-      std::lock_guard<std::mutex> lock(m_mutex);
-      ++m_count;
-    }
-    m_condition.notify_one();
-  }
-
- private:
-  std::mutex m_mutex;
-  std::condition_variable m_condition;
-  uint16_t m_count = 0;
-};
-
 class Dumper::Dump_writer_controller {
  public:
   using Create_file = std::function<std::unique_ptr<mysqlshdk::storage::IFile>(
@@ -705,6 +680,12 @@ class Dumper::Table_worker final {
 
   ~Table_worker() = default;
 
+  void release_session() {
+    if (m_session) {
+      m_dumper->session_pool().push(std::move(m_session));
+    }
+  }
+
   void run() {
     std::string context;
 
@@ -714,14 +695,6 @@ class Dumper::Table_worker final {
                                    {"id", std::to_string(m_id)}}));
 
       mysqlsh::Mysql_thread mysql_thread;
-      shcore::on_leave_scope close_session([this]() {
-        if (m_session) {
-          m_session->close();
-        }
-      });
-
-      open_session();
-
       m_rate_limit =
           mysqlshdk::utils::Rate_limit(m_dumper->m_options.max_rate());
 
@@ -738,14 +711,20 @@ class Dumper::Table_worker final {
 
         context = std::move(work.info);
 
+        m_session = m_dumper->session_pool().pop();
+        shcore::on_leave_scope session_releaser(
+            [this]() { release_session(); });
+
+        if (m_dumper->m_worker_interrupt) {
+          return;
+        }
+
         work.task(this);
 
         if (m_dumper->m_worker_interrupt) {
           return;
         }
       }
-
-      m_dumper->assert_transaction_is_open(m_session);
     } catch (const mysqlshdk::db::Error &e) {
       handle_exception(context, e.format().c_str());
     } catch (const shcore::Error &e) {
@@ -765,18 +744,6 @@ class Dumper::Table_worker final {
  private:
   friend class Dumper;
 
-  void open_session() {
-    // notify dumper that the session has been established
-    shcore::on_leave_scope notify_dumper(
-        [this]() { m_dumper->m_worker_synchronization->notify(); });
-
-    m_session =
-        establish_session(m_dumper->session()->get_connection_options(), false);
-
-    m_dumper->start_transaction(m_session);
-    m_dumper->on_init_thread_session(m_session);
-  }
-
   inline std::shared_ptr<mysqlshdk::db::IResult> query(
       const std::string &sql) const {
     return Dumper::query(m_session, sql);
@@ -875,6 +842,8 @@ class Dumper::Table_worker final {
       throw;
     }
 
+    release_session();
+
     controller->finish_writing();
 
     duration.finish();
@@ -1579,6 +1548,7 @@ class Dumper::Table_worker final {
 
   void handle_exception(const std::string &context, const char *msg) {
     m_dumper->m_worker_exceptions[m_id] = std::current_exception();
+    m_dumper->m_worker_exception_thrown = true;
     current_console()->print_error(
         m_log_id + (context.empty() ? "" : "Error while " + context + ": ") +
         msg);
@@ -1828,10 +1798,6 @@ Dumper::Dumper(const Dump_options &options)
       mysqlshdk::storage::get_extension(m_options.compression());
 }
 
-// needs to be defined here due to Dumper::Synchronize_workers being
-// incomplete type
-Dumper::~Dumper() = default;
-
 void Dumper::run() {
   if (m_options.is_dry_run()) {
     current_console()->print_info(
@@ -1934,15 +1900,15 @@ void Dumper::do_run() {
         return;
       }
 
+      create_worker_sessions();
+
       create_worker_threads();
 
       m_current_stage->finish();
 
       // initialize cache while threads are starting up
       initialize_instance_cache();
 
-      wait_for_workers();
-
       if (m_options.consistent_dump() && !m_worker_interrupt) {
         current_console()->print_info("All transactions have been started");
         lock_instance();
@@ -1956,6 +1922,20 @@ void Dumper::do_run() {
       }
     }
 
+    shcore::on_leave_scope cleanup([this]() {
+      // Ensures all the worker sessions get closed
+      size_t worker_sessions = m_options.threads();
+      while (worker_sessions) {
+        auto session = m_session_pool.pop();
+
+        if (!m_worker_exception_thrown) {
+          assert_transaction_is_open(session);
+        }
+        session->close();
+        --worker_sessions;
+      }
+    });
+
     create_schema_tasks();
 
     validate_privileges();
@@ -2803,23 +2783,30 @@ void Dumper::close_output_directory() {
   m_output_dir.reset();
 }
 
+void Dumper::create_worker_sessions() {
+  for (std::size_t i = 0; i < m_options.threads(); ++i) {
+    auto worker_session =
+        establish_session(session()->get_connection_options(), false);
+
+    start_transaction(worker_session);
+    on_init_thread_session(worker_session);
+
+    m_session_pool.push(std::move(worker_session));
+  }
+}
+
 void Dumper::create_worker_threads() {
   m_worker_exceptions.clear();
-  m_worker_exceptions.resize(m_options.threads());
-  m_worker_synchronization = std::make_unique<Synchronize_workers>();
+  m_worker_exceptions.resize(m_options.worker_threads());
 
-  for (std::size_t i = 0; i < m_options.threads(); ++i) {
+  for (std::size_t i = 0; i < m_options.worker_threads(); ++i) {
     auto t = mysqlsh::spawn_scoped_thread(
         &Table_worker::run,
         Table_worker{i, this, Table_worker::Exception_strategy::ABORT});
     m_workers.emplace_back(std::move(t));
   }
 }
 
-void Dumper::wait_for_workers() {
-  m_worker_synchronization->wait_for(m_workers.size());
-}
-
 void Dumper::maybe_push_shutdown_tasks() {
   if (0 == m_chunking_tasks &&
       m_main_thread_finished_producing_chunking_tasks) {

modules/util/dump/dumper.h

@@ -25,9 +25,11 @@
 #define MODULES_UTIL_DUMP_DUMPER_H_
 
 #include <atomic>
+#include <condition_variable>
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <queue>
 #include <string>
 #include <string_view>
 #include <thread>
@@ -65,7 +67,7 @@ class Dumper {
   Dumper &operator=(const Dumper &) = delete;
   Dumper &operator=(Dumper &&) = delete;
 
-  virtual ~Dumper();
+  virtual ~Dumper() = default;
 
   void run();
 
@@ -161,8 +163,6 @@ class Dumper {
     std::function<void(Table_worker *)> task;
   };
 
-  class Synchronize_workers;
-
   class Dump_info;
 
   class Memory_dumper;
@@ -228,9 +228,9 @@ class Dumper {
 
   void close_output_directory();
 
-  void create_worker_threads();
+  void create_worker_sessions();
 
-  void wait_for_workers();
+  void create_worker_threads();
 
   void maybe_push_shutdown_tasks();
 
@@ -370,6 +370,11 @@ class Dumper {
 
   void throw_if_cannot_dump_users() const;
 
+  shcore::Synchronized_queue<std::shared_ptr<mysqlshdk::db::ISession>>
+      &session_pool() {
+    return m_session_pool;
+  }
+
   // session
   std::shared_ptr<mysqlshdk::db::ISession> m_session;
   std::vector<std::shared_ptr<mysqlshdk::db::ISession>> m_lock_sessions;
@@ -435,10 +440,10 @@ class Dumper {
   // threads
   std::vector<std::thread> m_workers;
   std::vector<std::exception_ptr> m_worker_exceptions;
+  std::atomic<bool> m_worker_exception_thrown = false;
   shcore::Synchronized_queue<Task_info> m_worker_tasks;
   std::atomic<uint64_t> m_chunking_tasks;
   std::atomic<bool> m_main_thread_finished_producing_chunking_tasks;
-  std::unique_ptr<Synchronize_workers> m_worker_synchronization;
   std::function<std::unique_ptr<Dump_writer>()> m_writer_creator;
   volatile bool m_worker_interrupt = false;
 
@@ -448,6 +453,9 @@ class Dumper {
   mutable Progress_thread m_progress_thread;
   mutable Progress_thread::Stage *m_current_stage = nullptr;
   Progress_thread::Stage *m_data_dump_stage = nullptr;
+
+  shcore::Synchronized_queue<std::shared_ptr<mysqlshdk::db::ISession>>
+      m_session_pool;
 };
 
 }  // namespace dump