[MDEV-8112] Port no slave left behind into 10.2bb-10.2-mdev8112

This patch implements master throttling based on slave lag, aka no slave left behind.
The core feature works as follows:
1) The semi-sync-reply is ammended to also report back SQL-thread position (aka exec position)
2) Transactions are not removed from the "active-transaction-list"
   in the semi-sync-master plugin until atleast one slave has reported
   that it has executed this transaction. the slave lag can then
   be estimated by calculating how long the oldest transaction has been
   lingering in the active-transaction-list.
3) client-threads are forced to wait before commit until slave lag has decreased to acceptable value.

The following variables are introduced on master:
1. rpl_semi_sync_master_max_slave_lag (global)
2. rpl_semi_sync_master_slave_lag_wait_timeout (session)

The following status variables are introduced on master:
1. rpl_semi_sync_master_slave_lag_wait_sessions
2. rpl_semi_sync_master_estimated_slave_lag
3. rpl_semi_sync_master_trx_slave_lag_wait_time
4. rpl_semi_sync_master_trx_slave_lag_wait_num
5. rpl_semi_sync_master_avg_trx_slave_lag_wait_time

The following variables are introduced on slave:
1. rpl_semi_sync_slave_lag_enabled (global)

In addition to this, 2 optimizations that decreases overhead of semi-sync is introduced.
1) the idea of this is that if when a slave should send and transaction,
   it checks if it should be semi-synced, but rather
   than semi-sync:ing each transaction (which is done currently) the code
   will skip semi-syncing transaction if there already is newer transactions
   committed. But, since this can mean that semi-syncing is delayed indefinitely
   a cap is set using 2 new master variables:
   1. rpl_semi_sync_master_max_unacked_event_bytes (global)
   2. rpl_semi_sync_master_max_unacked_event_count (global)
2) rpl_semi_sync_master_group_commit which makes the semi-sync
   plugin only semi-sync the last transaction in a group commit.

1 files changed, 34 insertions, 10 deletions

diff --git a/plugin/semisync/semisync_slave.cc b/plugin/semisync/semisync_slave.cc
index 5f98472d5d7..839e0cce29d 100644
--- a/plugin/semisync/semisync_slave.cc
+++ b/plugin/semisync/semisync_slave.cc
@@ -20,6 +20,7 @@
 char rpl_semi_sync_slave_enabled;
 char rpl_semi_sync_slave_status= 0;
 unsigned long rpl_semi_sync_slave_trace_level;
+char rpl_semi_sync_slave_lag_enabled= 0;
 
 int ReplSemiSyncSlave::initObject()
 {
@@ -42,7 +43,7 @@ int ReplSemiSyncSlave::initObject()
 
 int ReplSemiSyncSlave::slaveReadSyncHeader(const char *header,
                                       unsigned long total_len,
-                                      bool  *need_reply,
+                                      unsigned char *need_reply,
                                       const char **payload,
                                       unsigned long *payload_len)
 {
@@ -52,7 +53,7 @@ int ReplSemiSyncSlave::slaveReadSyncHeader(const char *header,
 
   if ((unsigned char)(header[0]) == kPacketMagicNum)
   {
-    *need_reply  = (header[1] & kPacketFlagSync);
+    *need_reply  = (header[1] & (kPacketFlagSync | kPacketFlagSyncAndReport));
     *payload_len = total_len - 2;
     *payload     = header + 2;
 
@@ -95,16 +96,20 @@ int ReplSemiSyncSlave::slaveStop(Binlog_relay_IO_param *param)
   return 0;
 }
 
-int ReplSemiSyncSlave::slaveReply(MYSQL *mysql,
-                                 const char *binlog_filename,
-                                 my_off_t binlog_filepos)
+int ReplSemiSyncSlave::slaveReply(unsigned char header_byte,
+                                  MYSQL *mysql,
+                                  const char *binlog_filename,
+                                  my_off_t binlog_filepos,
+                                  Master_info * mi)
 {
   const char *kWho = "ReplSemiSyncSlave::slaveReply";
   NET *net= &mysql->net;
-  uchar reply_buffer[REPLY_MAGIC_NUM_LEN
-                     + REPLY_BINLOG_POS_LEN
-                     + REPLY_BINLOG_NAME_LEN];
+  uchar reply_buffer[REPLY_MAGIC_NUM_LEN +
+                     2 * ( REPLY_BINLOG_POS_LEN +
+                           REPLY_BINLOG_NAME_LEN +
+                           /* '\0' */ 1) ];
   int  reply_res, name_len = strlen(binlog_filename);
+  int  msg_len = name_len + REPLY_BINLOG_NAME_OFFSET;
 
   function_enter(kWho);
 
@@ -119,10 +124,29 @@ int ReplSemiSyncSlave::slaveReply(MYSQL *mysql,
     sql_print_information("%s: reply (%s, %lu)", kWho,
                           binlog_filename, (ulong)binlog_filepos);
 
+  if (header_byte & kPacketFlagSyncAndReport)
+  {
+    /**
+     * master requests that we also report back SQL-thread position
+     */
+
+    // where to store sql filename/position
+    char *bufptr = (char*)reply_buffer + msg_len;
+    bufptr[0] = 0; // '\0' terminate previous filename
+    bufptr++;
+
+    my_off_t sql_file_pos;
+    // get file/position and store the filename directly info bufptr+8
+    size_t name_len2 = get_master_log_pos(mi, bufptr + 8, &sql_file_pos);
+    int8store(bufptr, sql_file_pos); // store position
+
+    msg_len += /* '\0' */ 1 + /* position */ 8 + name_len2 + /* '\0' */ 1;
+  }
+
   net_clear(net, 0);
   /* Send the reply. */
-  reply_res = my_net_write(net, reply_buffer,
-                           name_len + REPLY_BINLOG_NAME_OFFSET);
+  reply_res = my_net_write(net, reply_buffer, msg_len);
+
   if (!reply_res)
   {
     reply_res = net_flush(net);