PS-9719: changing binlog_transaction_dependency_tracking hit segmentation fault

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

Problem
-------
When changing binlog_transaction_dependency_tracking in high load
workload, MySQL can get a segmentation fault.

Analysis
--------
Address sanitizer runs exposed the heap-use-after-free.

READ of size 8 at 0x6030002c3298 thread T52
    #0 _M_hash_code()
    #1 _M_bucket_index()
    ..
    #7 std::unordered_map::insert()
    #8 Writeset_trx_dependency_tracker::get_dependency()
    #9 Transaction_dependency_tracker::get_dependency()
    #10 MYSQL_BIN_LOG::write_transaction()
    #11 binlog_cache_data::flush()
    #12 binlog_cache_mngr::flush()
    #13 MYSQL_BIN_LOG::flush_thread_caches()
    #14 MYSQL_BIN_LOG::process_flush_stage_queue()
    #15 MYSQL_BIN_LOG::ordered_commit()
    #16 MYSQL_BIN_LOG::commit()

freed by thread T49 here:
    #0 operator delete()
    ...
    #7 std::unordered_map::clear()
    #8 Writeset_trx_dependency_tracker::rotate(long)
    #9 Transaction_dependency_tracker::tracking_mode_changed()
    #10 update_binlog_transaction_dependency_tracking
    #11 sys_var::update()

- The Writeset_trx_dependency_tracker uses std::unordered_map for
  storing depdendency information.
- When a transaction is committing, the committing thread inserts the
  dependency information to this map in through get_dependency().
- When the tracking mode is changed, then the map is cleared by
  Writeset_trx_dependency_tracker::rotate(). Note that no lock/mutex is
  taken during the rotation.
- As the rotate() and get_dependency() operations can be concurrently
  called from different threads and there is no mutex protection to
  handle it, it can result in segmentation fault when the get_dependency()
  tries to insert to the already deleted map.

Solution
--------
Use std::shared_ptr with atomic load/store for safer dependency tracker map rotation.

- Replaced direct usage of of map with std::shared_ptr in the
  Writeset_trx_dependency_tracker class.
- Modified the implementation of rotate() to used std::atomic_load and
  std::atomic_store to enable thread-safe reads and rotations.

With the new solution the rotation happens in an atomic manner. So that
transactions calling get_dependency() always use the object returned by
shared_ptr. So, even if rotate() happens in parallel, the memory won't
be freed until all readers are done.

Author: Venkatesh Prasad

sql/rpl_trx_tracking.cc

@@ -245,6 +245,7 @@ void Writeset_trx_dependency_tracker::get_dependency(THD *thd,
       // it did not broke past the capacity already
       !write_set_ctx->was_write_set_limit_reached();
   bool exceeds_capacity = false;
+  auto writeset_history = std::atomic_load(&m_writeset_history);
 
   if (can_use_writesets) {
     /*
@@ -253,7 +254,7 @@ void Writeset_trx_dependency_tracker::get_dependency(THD *thd,
      using its information for current transaction.
     */
     exceeds_capacity =
-        m_writeset_history.size() + writeset->size() > m_opt_max_history_size;
+        writeset_history->size() + writeset->size() > m_opt_max_history_size;
 
     /*
      Compute the greatest sequence_number among all conflicts and add the
@@ -262,15 +263,15 @@ void Writeset_trx_dependency_tracker::get_dependency(THD *thd,
     int64 last_parent = m_writeset_history_start;
     for (std::vector<uint64>::iterator it = writeset->begin();
          it != writeset->end(); ++it) {
-      Writeset_history::iterator hst = m_writeset_history.find(*it);
-      if (hst != m_writeset_history.end()) {
+      Writeset_history::iterator hst = writeset_history->find(*it);
+      if (hst != writeset_history->end()) {
         if (hst->second > last_parent && hst->second < sequence_number)
           last_parent = hst->second;
 
         hst->second = sequence_number;
       } else {
         if (!exceeds_capacity)
-          m_writeset_history.insert(
+          writeset_history->insert(
               std::pair<uint64, int64>(*it, sequence_number));
       }
     }
@@ -292,13 +293,16 @@ void Writeset_trx_dependency_tracker::get_dependency(THD *thd,
 
   if (exceeds_capacity || !can_use_writesets) {
     m_writeset_history_start = sequence_number;
-    m_writeset_history.clear();
+    if (writeset_history) {
+      writeset_history->clear();
+    }
   }
 }
 
 void Writeset_trx_dependency_tracker::rotate(int64 start) {
   m_writeset_history_start = start;
-  m_writeset_history.clear();
+  auto new_map = std::make_shared<Writeset_history>();
+  std::atomic_store(&m_writeset_history, new_map);
 }
 
 /**

sql/rpl_trx_tracking.h

@@ -127,7 +127,10 @@ class Commit_order_trx_dependency_tracker {
 class Writeset_trx_dependency_tracker {
  public:
   Writeset_trx_dependency_tracker(ulong max_history_size)
-      : m_opt_max_history_size(max_history_size), m_writeset_history_start(0) {}
+      : m_opt_max_history_size(max_history_size), m_writeset_history_start(0) {
+    std::atomic_store(&m_writeset_history,
+                      std::make_shared<Writeset_history>());
+  }
 
   /**
     Main function that gets the dependencies using the WRITESET tracker.
@@ -161,7 +164,7 @@ class Writeset_trx_dependency_tracker {
     in the database, using row hashes from the writeset as the index.
   */
   typedef std::unordered_map<uint64, int64> Writeset_history;
-  Writeset_history m_writeset_history;
+  std::shared_ptr<Writeset_history> m_writeset_history;
 };
 
 /**