怎么進行從庫MTS多線程并行回放

今天就跟大家聊聊有關怎么進行從庫MTS多線程并行回放，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

一、綜述

與單SQL線程的回放不同，MTS包含多個工作線程，原有的SQL線程蛻變為協調線程。SQL協調線程同時還承擔了檢查點的工作。我們知道并行回放的方式有兩種，包含LOGICAL_CLOCK和DATABASE，體現在判定哪些事物能夠并行回放的規則不同。實際上源碼對應兩個不同的類：

• Mts_submode_logical_clock

• Mts_submode_database

這里只準備討論基于LOGICAL_CLOCK的并發方式，而不會討論老的基于DATABASE的方式，下面是我設置的參數：

• slave_parallel_type：LOGICAL_CLOCK

• slave_parallel_workers ：4

注意slave_parallel_workers設置的是工作線程的個數，且不包協調線程，因此如果不想使用MTS應該將這個參數設置為0，然后‘stop slave;start slave’才能生效。因為工作線程在啟動的時候已經初始化完畢了。

因為我們知道在5.7中即便不開啟GTID也包含的匿名的GTID Event，它攜帶了last commit和seq number，因此即便關閉GTID也可以使用MTS，但是不建議后面第26節可以找到原因。

在前面我們討論了MySQL層事務提交的流程和基于WRITESET的并行復制方式，我們總共提到了三種生成last commit和seq number的方式：

• ORDER_COMMIT

• WRITESET

• WRITESET_SESSION

它們控制的是生成last commit和seq number的規則。而從庫只要將參數slave_parallel_type設置為LOGICAL_CLOCK，其能否并行的依據就是last commit和seq number。

我們下面的描述還是以一個正常的‘Delete’語句刪除一行數據的Event來描述，那么這個事物Event的順序如下：

同時在此之前我們先來明確一下MySQL中持久化MTS信息的三個場所，因為和傳統的單SQL線程的主從不同，MTS需要存儲更多的信息。注意我們只討論master_info_repository和relay_log_info_repository為TABLE的情況，如下：

• slave_master_info表：由IO線程進行更新，超過sync_master_info設置更新，單位Event個數。

• relay_log_info_repository表：由SQL協調線程執行檢查點的時候進行更新。

• slave_worker_info表：由工作線程每次提交事務的時候更新。

更加詳細的解釋參考第25節，同時會解釋為什么只考慮master_info_repository和relay_log_info_repository為TABLE的原因。

二、協調線程的分發機制

協調線程在Event的分發中主要完成下面兩個工作：

• 判定事務是否可以并行回放。

• 判定事務由哪一個工作線程進行回放。

和單SQL線程執行的流程不同，主要體現在函數apply_event_and_update_pos下面，對于單線程而言會完成Event的應用，而對用MTS而言就是只會完成Event的分發，具體的應用將會由工作線程完成。
這里說一下簡化的流程，具體函數調用參考筆記。下面是一張流程圖（圖19-1，高清原圖包含在文末原圖中）：

三、步驟解析

下面對每一步進行解析如下：

（1）如果是GTID_LOG_EVENT代表事物開始，將本事物加入到GAQ隊列中（下一節會詳細描述GAQ）。可參考函數Log_event::get_slave_worker。

（2）將GTID_LOG_EVENT加入到curr_group_da隊列中暫存。可參考函數Log_event::get_slave_worker。

（3）獲取GTID_LOG_EVENT中的last commit和seq number值。可參考函數Mts_submode_logical_clock::schedule_next_event。

（4）獲取current_lwm值，這個值代表的是所有在GAQ隊列上還沒有提交完成事務中最早的那個事務的前一個已經提交事務的seq number，可能后面的事務已經提交完成了，聽起來可能比較拗口但很重要，如果都提交完成了那么就是取最新提交的事務的seq number，下面的圖表達的就是這個意思，這個圖是源碼中的。這個值的獲取可參考函數Mts_submode_logical_clock::get_lwm_timestamp。

       the last time index containg lwm
               +------+
               | LWM  |
               |  |   |
               V  V   V
GAQ:x  xoooooxxxxxXXXXX...X
             ^   ^
             |   | LWM+1（LWM代表的是檢查點指向的位置）
             |
             + new current_lwm（這里就是current_lwm）
      <---- logical (commit) time ----
here `x' stands for committed, `X' for committed and discarded from
the running range of the queue, `o' for not committed.

我們可以先不看LWM部分，對于檢查點的LWM后面在討論。seq number從右向左遞增，在GAQ中實際上有三種值：

• X：已經做了檢查點，在GAQ中出隊的事物。

• x：已經提交完成的事物。

• o：沒有提交完成的事物。

我們可以看到我們需要獲取的current_lwm并不是最新一次提交事物的seq number的值，而是最早未提交事物的前一個已經提交事物的seq number。這一點很重要，因為理解后就會知道大事務是如何影響MTS的并行回放的，同時中間的5個‘o’實際上就是所謂的‘gap’，關于‘gap’下一節還會詳細描述。

（5）將GTID_LOG_EVENT中的last commit和當前current_lwm進行比較。可以參考函數Mts_submode_logical_clock::schedule_next_event。下面是大概的規則：

• 如果last commit小于等于current_lwm表示可以進行并行回放，繼續。

• 如果last commit大于current_lwm則表示不能進行并行回放。這個時候協調線程就需要等待了，直到小于等于的條件成立。成立后協調線程會被工作線程喚醒。等待期間狀態被置為“Waiting for dependent transaction to commit”。

源碼處也比較簡單如下：

    longlong lwm_estimate= estimate_lwm_timestamp(); 
//這個值 只有在 出現 下面等待的時候 才會設置 min_waited_timestamp ，
//設置了min_waited_timestamp才會更新lwm_estimate
    if (!clock_leq(last_committed, lwm_estimate) && 
//  @return   true  when a "<=" b,false otherwise  last_committed<=lwm_estimate
        rli->gaq->assigned_group_index != rli->gaq->entry) 
    {
      if (wait_for_last_committed_trx(rli, last_committed, lwm_estimate)) 
//等待上一次 組提交的完成 Waiting for dependent transaction to commit

（6）如果是QUERY_EVENT則加入到curr_group_da隊列中暫存。

（7）如果是MAP_EVENT進行工作線程的分配。參考函數Mts_submode_logical_clock::get_least_occupied_worker，分配工作線程如下：

• 如果有空閑的工作線程則分配完成，繼續。

• 如果沒有空閑的工作線程則等待空閑的工作線程。這種情況下狀態會置為“Waiting for slave workers to process their queues”。

下面是分配的標準，其實也很簡單：

  for (Slave_worker **it= rli->workers.begin(); it != rli->workers.end(); ++it)
  {
    Slave_worker *w_i= *it;
    if (w_i->jobs.len == 0)
//任務隊列為0表示本Worker線程空閑可以分配
      return w_i;
  }
  return 0;

（8）將GTID_LOG_EVENT和QUERY_EVENT分配給工作線程。可參考append_item_to_jobs函數。

前面工作線程已經分配了，這里就可以開始將Event分配給這個工作線程了。分配的時候需要檢查工作線程的任務隊列是否已滿，如果滿了需要等待，狀態置為“Waiting for Slave Worker queue”。因為分配的單位是Event，對于一個事務而言可能包含很多Event，如果工作線程應用的速度趕不上協調線程入隊的速度，可能導致任務隊列的積壓，因此任務隊列被占滿是可能的。任務隊列的大小為16384如下：

mts_slave_worker_queue_len_max= 16384;

下面是入隊的部分代碼：

  while (worker->running_status == Slave_worker::RUNNING && !thd->killed &&
         (ret= en_queue(&worker->jobs, job_item)) == -1)
//如果已經滿了
  {
    thd->ENTER_COND(&worker->jobs_cond, &worker->jobs_lock,
                    &stage_slave_waiting_worker_queue, &old_stage);
//標記等待狀態
    worker->jobs.overfill= TRUE;
    worker->jobs.waited_overfill++;
    rli->mts_wq_overfill_cnt++; //標記隊列滿的次數
    mysql_cond_wait(&worker->jobs_cond, &worker->jobs_lock);
//等待喚醒
    mysql_mutex_unlock(&worker->jobs_lock);
    thd->EXIT_COND(&old_stage);
    mysql_mutex_lock(&worker->jobs_lock);
  }

（9）MAP_EVENT分配給工作線程，同上。
（10）DELETE_EVENT分配給工作線程，同上。
（11）XID_EVENT分配給工作線程，但是這里還需要額外的處理，主要處理一些和檢查點相關的信息，這里關注一點如下：

ptr_group->checkpoint_log_name= my_strdup(key_memory_log_event, 
rli->get_group_master_log_name(), MYF(MY_WME));
ptr_group->checkpoint_log_pos= rli->get_group_master_log_pos();
ptr_group->checkpoint_relay_log_name=my_strdup(key_memory_log_event, 
rli->get_group_relay_log_name(), MYF(MY_WME));
ptr_group->checkpoint_relay_log_pos= rli->get_group_relay_log_pos();
ptr_group->ts= common_header->when.tv_sec + (time_t) exec_time; 
//Seconds_behind_master related .checkpoint
//的時候會將這個值再次傳遞 mts_checkpoint_routine()      
ptr_group->checkpoint_seqno= rli->checkpoint_seqno;
//獲取seqno 這個值會在chkpt后減去偏移量

如果檢查點處于這個事務上，那么這些信息會出現在表 slave_worker_info中，并且會出現在show slave status中。也就是說，show slave status中很多信息是來自MTS的檢查點。下一節將詳細描述檢查點。

（12）如果上面Event的分配過程大于2分鐘（120秒），可能會出現一個日志如下：

這個截圖也是一個朋友問的問題。實際上這個日志可以算一個警告。實際上對應的源碼為：

sql_print_information("Multi-threaded slave statistics%s: "
                "seconds elapsed = %lu; "
                "events assigned = %llu; "
                "worker queues filled over overrun level = %lu; "
                "waited due a Worker queue full = %lu; "
                "waited due the total size = %lu; "
                "waited at clock conflicts = %llu "
                "waited (count) when Workers occupied = %lu "
                "waited when Workers occupied = %llu",
                rli->get_for_channel_str(),
                static_cast<unsigned long>
                (my_now - rli->mts_last_online_stat),
//消耗總時間 單位秒
                rli->mts_events_assigned,
//總的event分配的個數
                rli->mts_wq_overrun_cnt,
// worker線程分配隊列大于 90%的次數 當前硬編碼  14746
                rli->mts_wq_overfill_cnt,    
//由于work 分配隊列已滿造成的等待次數 當前硬編碼 16384
                rli->wq_size_waits_cnt, 
//大Event的個數 一般不會存在
                rli->mts_total_wait_overlap,
//由于上一組并行有大事物沒有提交導致不能分配worker線程的等待時間 單位納秒
                rli->mts_wq_no_underrun_cnt, 
//work線程由于沒有空閑的而等待的次數
                rli->mts_total_wait_worker_avail);
//work線程由于沒有空閑的而等待的時間   單位納秒

因為經常看到朋友問這里詳細說明一下它們的含義，從前面的分析中我們一共看到三個等待點：

• “Waiting for dependent transaction to commit”

由于協調線程判定本事務由于last commit大于current_lwm因此不能并行回放，協調線程處于等待，大事務會加劇這種情況。

• “Waiting for slave workers to process their queues”

由于沒有空閑的工作線程，協調線程會等待。這種情況說明理論上的并行度是理想的，但是可能是參數slave_parallel_workers設置不夠。當然設置工作線程的個數應該和服務器的配置和負載相結合考慮，因為第29節我們會看到線程是CPU調度最小的單位。

• “Waiting for Slave Worker queue”

由于工作線程的任務隊列已滿，協調線程會等待。這種情況前面說過是由于一個事務包含了過多的Event并且工作線程應用Event的速度趕不上協調線程分配Event的速度，導致了積壓并且超過了16384個Event。

另外實際上還有一種等待如下：
“Waiting for Slave Workers to free pending events”：由所謂的‘big event’造成的，什么是‘big event’呢，源碼中描述為：event size is greater than slave_pending_jobs_size_max but less than slave_max_allowed_packet。我個人認為出現的可能性不大，因此沒做過多考慮。可以在函數append_item_to_jobs中找到答案。

我們下面對應日志中的輸出進行詳細解釋，如下：

我們可以看到這個日志還是記錄很全的，基本覆蓋了前面我們討論的全部可能性。那么我們再看看案例中的日志，waited at clock conflicts=91895169800 大約91秒。120秒鐘大約91秒都因為不能并行回放而造成的等待，很明顯應該考慮是否有大事物的存在。

四、并行回放判定的列子

下面是我主庫使用WRITESET方式生成的一段binary log片段，我們主要觀察lastcommit和seq number，通過分析來熟悉這種過程。

我們根據剛才說的并行判斷規則，即：

• 如果last commit小于等于current_lwm表示可以進行并行回放，繼續。

• 如果last commit大于current_lwm則表示不能進行并行回放，需要等待。

具體解析如下：

（last commit：22  seq number：23）這個事務會在（last commit：21 seq number：22）事務執行完成后執行因為（last commit：22<= seq number：22），后面的事務直到（last_commit：22  seq number：30），實際上都可以并行執行，我們先假設他們都執行完成了。我們繼續觀察隨后的三個事務如下：

• last_committed：29       sequence_number：31

• last_committed：30       sequence_number：32

• last_committed：27       sequence_number：33

我們注意到到這是基于WRITESET的并行復制下明顯的特征。 last commit可能比上一個事務更小，這就是我們前面說的根據Writeset的歷史MAP信息計算出來的。因此還是根據上面的規則它們三個是可以并行執行的。因為很明顯：

• last_committed：29  <= current_lwm：30

• last_committed：30  <= current_lwm：30

• last_committed：27  <= current_lwm：30

但是如果（last commit：22  seq number：30）這個事務之前有一個大事務沒有執行完成的話，那么current_lwm的取值將不會是30。比如（last commit：22  seq number：27）這個事務是大事務那么current_lwm將會標記為26，上面的三個事務將會被堵塞，并且分配（last commit：29  seq number：31）的時候就已經堵塞了，原因如下：

• last_committed：29  > current_lwm：26

• last_committed：30  > current_lwm：26

• last_committed：27  > current_lwm：26

我們再考慮一下基于WRITESET的并行復制下（last commit：27 seq number：33）這個事務，因為在我們并行規則下last commit越小獲得并發的可能性越高。因此基于WRITESET的并行復制確實提高了從庫回放的并行度，但正如第16節所講主庫會有一定的開銷。

看完上述內容，你們對怎么進行從庫MTS多線程并行回放有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。