介紹

半同步複製binlog dump執行緒需要做的事情。相關結構：

Binlog_transmit_observer transmit_observer = {
  sizeof(Binlog_transmit_observer), // len

  repl_semi_binlog_dump_start,	// start
  repl_semi_binlog_dump_end,	// stop
  repl_semi_reserve_header,	// reserve_header
  repl_semi_before_send_event,	// before_send_event
  repl_semi_after_send_event,	// after_send_event
  repl_semi_reset_master,	// reset
};
 

/**
   Observe and extends the binlog dumping thread.
*/
typedef struct Binlog_transmit_observer {
  uint32 len;
  
  /**
     This callback is called when binlog dumping starts


     @param param Observer common parameter
     @param log_file Binlog file name to transmit from
     @param log_pos Binlog position to transmit from

     @retval 0 Sucess
     @retval 1 Failure
  */
  int (*transmit_start)(Binlog_transmit_param *param,
                        const char *log_file, my_off_t log_pos);

  /**
     This callback is called when binlog dumping stops

     @param param Observer common parameter
     
     @retval 0 Sucess
     @retval 1 Failure
  */
  int (*transmit_stop)(Binlog_transmit_param *param);

  /**
     This callback is called to reserve bytes in packet header for event transmission

     This callback is called when resetting transmit packet header to
     reserve bytes for this observer in packet header.

     The @a header buffer is allocated by the server code, and @a size
     is the size of the header buffer. Each observer can only reserve
     a maximum size of @a size in the header.

     @param param Observer common parameter
     @param header Pointer of the header buffer
     @param size Size of the header buffer
     @param len Header length reserved by this observer

     @retval 0 Sucess
     @retval 1 Failure
  */
  int (*reserve_header)(Binlog_transmit_param *param,
                        unsigned char *header,
                        unsigned long size,
                        unsigned long *len);

  /**
     This callback is called before sending an event packet to slave

     @param param Observer common parameter
     @param packet Binlog event packet to send
     @param len Length of the event packet
     @param log_file Binlog file name of the event packet to send
     @param log_pos Binlog position of the event packet to send

     @retval 0 Sucess
     @retval 1 Failure
  */
  int (*before_send_event)(Binlog_transmit_param *param,
                           unsigned char *packet, unsigned long len,
                           const char *log_file, my_off_t log_pos );

  /**
     This callback is called after an event packet is sent to the
     slave or is skipped.

     @param param             Observer common parameter
     @param event_buf         Binlog event packet buffer sent
     @param len               length of the event packet buffer
     @param skipped_log_file  Binlog file name of the event that
                              was skipped in the master. This is
                              null if the position was not skipped
     @param skipped_log_pos   Binlog position of the event that
                              was skipped in the master. 0 if not
                              skipped
     @retval 0 Sucess
     @retval 1 Failure
   */
  int (*after_send_event)(Binlog_transmit_param *param,
                          const char *event_buf, unsigned long len,
                          const char *skipped_log_file, my_off_t skipped_log_pos);

  /**
     This callback is called after resetting master status

     This is called when executing the command RESET MASTER, and is
     used to reset status variables added by observers.

     @param param Observer common parameter

     @retval 0 Sucess
     @retval 1 Failure
  */
  int (*after_reset_master)(Binlog_transmit_param *param);
} Binlog_transmit_observer;
 

transmit_start

本節主要介紹transmit_start這個函式。

com_binlog_dump->mysql_binlog_send->RUN_HOOK(binlog_transmit, transmit_start,
               (thd, flags, log_ident, pos, &observe_transmission))
每次申請dump請求，都會執行transmit_start這個函式。該函式呼叫repl_semi_binlog_dump_start。

程式碼分析

int repl_semi_binlog_dump_start(Binlog_transmit_param *param,
				 const char *log_file,
				 my_off_t log_pos)
{
  //檢查slave的semi-sync是否開啟
  bool semi_sync_slave= repl_semisync.is_semi_sync_slave();
  
  if (semi_sync_slave)
  {
    /* 至少有一個slave開啟了半同步，將rpl_semi_sync_master_clients++*/
	//是在LOCK_binlog_鎖內進行增加
    repl_semisync.add_slave();//rpl_semi_sync_master_clients++;
    /* Tell server it will observe the transmission.*/
    param->set_observe_flag();

    /*
      Let's assume this semi-sync slave has already received all
      binlog events before the filename and position it requests.
    */
    repl_semisync.reportReplyBinlog(param->server_id, log_file, log_pos);
  }
  else
    param->set_dont_observe_flag();
  
  return 0;
}
 

1、is_semi_sync_slave()->get_user_var_int("rpl_semi_sync_slave", &val, &null_value);
      檢查slave是否是半同步。注：
      slave的before_request_transmit傳送SET @rpl_semi_sync_slave= 1將master的rpl_semi_sync_slave變數置成1

2、在binlog dump執行緒中做transmit_start操作。對於每次DUMP請求，都會首先做這個動作。
     該函式的主要動作：

int ReplSemiSyncMaster::reportReplyBinlog(uint32 server_id,
                                          const char *log_file_name,
                                          my_off_t log_file_pos,
                                          bool skipped_event)
{
  const char *kWho = "ReplSemiSyncMaster::reportReplyBinlog";
  int   cmp;
  bool  can_release_threads = false;
  bool  need_copy_send_pos = true;
  //判斷master半同步是否開啟：master_enabled_，這個變數和state_關係？
  if (!(getMasterEnabled()))
    return 0;

  function_enter(kWho);
  //LOCK_binlog_鎖
  lock();

  /* 鎖內再次判斷是否已經開啟半同步 */
  if (!getMasterEnabled())
    goto l_end;
  //state_如果半同步由於超時暫時關閉，嘗試重新開啟
  if (!is_on())
    try_switch_on(server_id, log_file_name, log_file_pos);

  //第一個slave請求dump，需要複製響應位置，並將reply_file_name_inited_置1
  //否則，比較當前的位置，如果比已響應的最大位置小，則不需要調整已響應最大位置
  //如果大，則需要調整已響應最大位置
  if (reply_file_name_inited_)
  {
    cmp = ActiveTranx::compare(log_file_name, log_file_pos,
                               reply_file_name_, reply_file_pos_);
    if (cmp < 0)
    {
      /* If the position is behind, do not copy it. */
      need_copy_send_pos = false;
    }
  }

  if (need_copy_send_pos)
  {
    strncpy(reply_file_name_, log_file_name, sizeof(reply_file_name_) - 1);
    reply_file_name_[sizeof(reply_file_name_) - 1]= '\0';
    reply_file_pos_ = log_file_pos;
    reply_file_name_inited_ = true;
  }
  //等待slave回覆ACK的事務有多個
  if (rpl_semi_sync_master_wait_sessions > 0)
  {
    //比較接收到的slave的ACK，reply的值比當前wait_file_name的大
    cmp = ActiveTranx::compare(reply_file_name_, reply_file_pos_,
                               wait_file_name_, wait_file_pos_);
    if (cmp >= 0)
    {
	  //可以釋放等待slave的ACK的事務了
	  //這意味著新的等待事務，需要重新設定等待位點資訊（一組?）
      can_release_threads = true;
      wait_file_name_inited_ = false;
    }
  }

 l_end:

  if (can_release_threads)
  {
    active_tranxs_->signal_waiting_sessions_up_to(reply_file_name_, reply_file_pos_);
  }
  unlock();
  return function_exit(kWho, 0);
}

int ReplSemiSyncMaster::try_switch_on(int server_id,
				      const char *log_file_name,
				      my_off_t log_file_pos)
{
  const char *kWho = "ReplSemiSyncMaster::try_switch_on";
  bool semi_sync_on = false;

  function_enter(kWho);

  //如果當前傳送event的位置比最大提交的事務binlog位置還要大，slave已經跟上master，即
  //slave接收的event已經跟上master，將半同步重新開啟。
  //如果commit_file_name_inited_為FALSE，表示沒有事務，可以立即開啟半同步
  if (commit_file_name_inited_)
  {
    int cmp = ActiveTranx::compare(log_file_name, log_file_pos,
                                   commit_file_name_, commit_file_pos_);
    semi_sync_on = (cmp >= 0);
  }
  else
  {
    semi_sync_on = true;
  }

  if (semi_sync_on)
  {
    /* 將半同步開啟 */
    state_ = true;
  }
  return function_exit(kWho, 0);
}

int ActiveTranx::signal_waiting_sessions_up_to(const char *log_file_name,
                                               my_off_t log_file_pos)
{
  const char *kWho = "ActiveTranx::signal_waiting_sessions_up_to";
  function_enter(kWho);
  //掃描所有正在等待的事務。如果其位點小，表示slave已經收到其binlog，可以廣播到使用者執行緒喚醒。
  TranxNode* entry= trx_front_;
  int cmp= ActiveTranx::compare(entry->log_name_, entry->log_pos_, log_file_name, log_file_pos) ;
  while (entry && cmp <= 0)
  {
    mysql_cond_broadcast(&entry->cond);
    entry= entry->next_;
    if (entry)
      cmp= ActiveTranx::compare(entry->log_name_, entry->log_pos_, log_file_name, log_file_pos) ;
  }
  return function_exit(kWho, (entry != NULL));
}

總結

1、如果slave開啟半同步，那麼會向master註冊：repl_semisync.is_semi_sync_slave->rpl_semi_sync_slave會被置成TRUE。所以首先獲取該變數，判斷是否至少有一個slave開啟slave。

2、至少有一個slave開啟半同步下：LOCK_binlog_鎖內增加rpl_semi_sync_master_clients值（增加連線的備庫計數）

3、reportReplyBinlog：

    1）判斷master是否開啟半同步。通過master_enabled_變數判斷。（注，和state_什麼關係?）

    2）master開啟半同步下。加LOCK_binlog_鎖，在這個鎖內再次判斷master是否開啟半同步，如果沒有開啟則退出。

    3）通過state_判斷半同步是否正常，如果半同步由於超時關閉，嘗試重新開啟：

           如果當前傳送的event位置比最大提交的事務binlog位置還要大，表示slave已經跟上master，即slave已經接收到master所有binlog，此時將半同步重新開啟。

           如果commit_file_name_inited_為FALSE，表示沒有事務，可以立即開啟半同步。

    4）第一個開啟slave，請求dump，需要儲存複製響應位置reply_file_name_並將reply_file_name_inited_設為TRUE

    5）否則，比較當前reply_file_name_位置。如果比reply_file_name_小，則需要將ACK的binlog位置儲存下來，需要調整最大響應位置。

    6）等待slave回覆ACK的事務有多個：

         比較接收slave的ACK，reply值比當前wait_file_name_還要打，則需要釋放等待slave ACK的事務。

    7）通過遍歷ActiveTranx連結串列，比較所有等待事務的binlog位置，如果等待事務的binlog的位點小，表示slave已經收到其binlog，可以broadcast，將等待事務的執行緒喚醒。