Fix inuse_relaylog atomics memory ordering

The inuse_relaylog is refcounted. The SQL driver thread first increments
queued_count until everything in the relaylog has been queued, then sets
completed=true. The worker threads increment dequeued_count, so that once
completed==true and queued_count==dequeued_count we know the inuse_relaylog
object can be safely de-allocated.

The accounting is done with atomic counters; however everything with
std::memory_order_relaxed ordering. This is not correct on weakly ordered
CPU architectures.

When we set completed==true, it most be done with release semantics so that
we know that all increments to queued_count will be visible to other
threads; this way another thread will not see completed==true with a too low
queued_count value. Similarly, when we increment dequeued_count it must be
done with release semantics so that previous accesses to inuse_relaylog will
have completed before the dequeue is visible to other threads. And the check
for completed==true must be done with matching acquire semantics.

Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>

sql/log.cc

@@ -5841,7 +5841,15 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included)
   while (ir)
   {
     inuse_relaylog *next= ir->next;
-    if (!ir->completed || ir->dequeued_count < ir->queued_count)
+    /*
+      Acquire semantics on the load on ir->completed, so we are sure that
+      all writes to ir->queued_count are visible. And on the ir->dequeued_count
+      to match release on its increment.
+    */
+    int64 loc_completed= ir->completed.load(std::memory_order_acquire);
+    if (!loc_completed ||
+        ir->dequeued_count.load(std::memory_order_acquire) <
+        ir->queued_count.load(std::memory_order_relaxed))
     {
       included= false;
       break;

sql/rpl_parallel.cc

@@ -2057,7 +2057,14 @@ rpl_parallel_thread::inuse_relaylog_refcount_update()
   inuse_relaylog *ir= accumulated_ir_last;
   if (ir)
   {
-    ir->dequeued_count+= accumulated_ir_count;
+    /*
+      Dequeue with release semantics, so we are sure none of our previous
+      accesses to the inuse_relaylog will be re-ordered past this increment
+      of dequeued_count (and thus potentially past the freeing of the
+      inuse_relaylog object).
+    */
+    ir->dequeued_count.fetch_add(accumulated_ir_count,
+                                 std::memory_order_release);
     accumulated_ir_count= 0;
     accumulated_ir_last= NULL;
   }
@@ -3083,8 +3090,9 @@ bool Relay_log_info::are_sql_threads_caught_up()
     Therefore, this place needs to check if it can now be `true`.
   */
   if (!worker_threads_caught_up && ( // No need to re-check if already `true`.
-    !last_inuse_relaylog || // `nullptr` case
-      last_inuse_relaylog->queued_count == last_inuse_relaylog->dequeued_count
+        !last_inuse_relaylog || // `nullptr` case
+        last_inuse_relaylog->queued_count.load(std::memory_order_relaxed) ==
+        last_inuse_relaylog->dequeued_count.load(std::memory_order_relaxed)
   ))
     worker_threads_caught_up= true; // Refresh
   return worker_threads_caught_up;
@@ -3132,7 +3140,7 @@ rpl_parallel_entry::queue_master_restart(rpl_group_info *rgi,
   qev->typ= rpl_parallel_thread::queued_event::QUEUED_MASTER_RESTART;
   qev->entry_for_queued= this;
   qev->ir= rli->last_inuse_relaylog;
-  ++qev->ir->queued_count;
+  qev->ir->queued_count.fetch_add(1, std::memory_order_relaxed);
   thr->enqueue(qev);
   mysql_cond_signal(&thr->COND_rpl_thread);
   mysql_mutex_unlock(&thr->LOCK_rpl_thread);
@@ -3658,7 +3666,7 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
     Queue the event for processing.
   */
   qev->ir= rli->last_inuse_relaylog;
-  ++qev->ir->queued_count;
+  qev->ir->queued_count.fetch_add(1, std::memory_order_relaxed);
   cur_thread->enqueue(qev);
   unlock_or_exit_cond(rli->sql_driver_thd, &cur_thread->LOCK_rpl_thread,
                       &did_enter_cond, &old_stage);

sql/rpl_rli.cc

@@ -1460,7 +1460,7 @@ Relay_log_info::alloc_inuse_relaylog(const char *name)
     inuse_relaylog_list= ir;
   else
   {
-    last_inuse_relaylog->completed= true;
+    last_inuse_relaylog->completed.store(true, std::memory_order_release);
     last_inuse_relaylog->next= ir;
   }
   last_inuse_relaylog= ir;

sql/rpl_rli.h

@@ -17,6 +17,8 @@
 #ifndef RPL_RLI_H
 #define RPL_RLI_H
 
+#include <atomic>
+
 #include "rpl_relay_log_info_file.h"
 #include "rpl_tblmap.h"
 #include "rpl_reporting.h"
@@ -677,11 +679,11 @@ struct inuse_relaylog {
   rpl_gtid *relay_log_state;
   uint32 relay_log_state_count;
   /* Number of events in this relay log queued for worker threads. */
-  Atomic_counter<int64> queued_count;
+  std::atomic<int64> queued_count;
   /* Number of events completed by worker threads. */
-  Atomic_counter<int64> dequeued_count;
+  std::atomic<int64> dequeued_count;
   /* Set when all events have been read from a relaylog. */
-  bool completed;
+  std::atomic<bool> completed;
   char name[FN_REFLEN];
 
   inuse_relaylog(Relay_log_info *rli_arg, rpl_gtid *relay_log_state_arg,

sql/slave.cc

@@ -7585,7 +7585,8 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size)
       DBUG_ASSERT(rli->cur_log_fd >= 0);
       mysql_file_close(rli->cur_log_fd, MYF(MY_WME));
       rli->cur_log_fd = -1;
-      rli->last_inuse_relaylog->completed= true;
+      rli->last_inuse_relaylog->completed.store(true,
+                                                std::memory_order_release);
       rli->relay_log.description_event_for_sql_thread->reset_crypto();
 
       if (relay_log_purge)