## TokuDB日志子系統 MySQL重啟后自動加載InnoDB和其他的動態plugin，包括TokuDB。每一plugin在注冊的時候指定init和deinit回調函數。TokuDB的init/deinit函數分別是`tokudb_init_func`和`tokudb_done_func`。 MySQL重啟過程中調用`tokudb_init_func`進行必要的初始化。在`tokudb_init_func`里面，調用`db_env_create`創建一個env實例，進行參數設置和callback設置。`db_env_create`是一個簡單的封裝，最終會調用`toku_env_create`來進行參數設置，callback設置和初始化的。`toku_env_create`初始化工作中一個很重要的事情就是調用`toku_logger_create`初始化TokuDB的日志子系統。 在TokuDB中，日志子系統是由tokulogger數據結構管理的。下面僅列出了主要的數據成員。 ~~~ struct tokulogger { struct mylock input_lock; // 保護lsn和in_buf的mutex toku_mutex_t output_condition_lock; // 保護written_lsn ，fsynced_lsn ，等待out_buf可用的條件變量的mutex toku_cond_t output_condition; // 等待out_buf可用的條件變量 bool output_is_available; // 標志out_buf可用的條件 bool is_open; // 標志logger是否已打開 bool write_log_files; // 標示是否將redo log buffer寫到redo log file bool trim_log_files; // 標志是否要trim redo log file char *directory; // redo log所在目錄 int lg_max; // redo log最大長度，缺省100M LSN lsn; // 下一個可用的lsn struct logbuf inbuf; // 接收redo log entry的buffer LSN written_lsn; // 最后一次寫入的lsn LSN fsynced_lsn; // 最后一次fsync的lsn LSN last_completed_checkpoint_lsn; // 最近一次checkpoint開始時刻的logger的lsn long long next_log_file_number; // 下一個可用的redo log file的序列號 struct logbuf outbuf; // 寫入redo log file的buf int n_in_file; // 當前redo log file存儲日志的字節數 TOKULOGFILEMGR logfilemgr; // log file manager的handle TXN_MANAGER txn_manager; // txn manager的handle }; ~~~ ### Logger初始化 Logger子系統在env->create階段由`toku_logger_create`進行初步的初始化工作。代碼片段如下： ~~~ int toku_logger_create (TOKULOGGER *resultp) { TOKULOGGER CALLOC(result); if (result==0) return get_error_errno(); result->is_open=false; result->write_log_files = true; result->trim_log_files = true; result->directory=0; result->lg_max = 100<<20; // 100MB default // lsn is uninitialized result->inbuf = (struct logbuf) {0, LOGGER_MIN_BUF_SIZE, (char *) toku_xmalloc(LOGGER_MIN_BUF_SIZE), ZERO_LSN}; result->outbuf = (struct logbuf) {0, LOGGER_MIN_BUF_SIZE, (char *) toku_xmalloc(LOGGER_MIN_BUF_SIZE), ZERO_LSN}; // written_lsn is uninitialized // fsynced_lsn is uninitialized result->last_completed_checkpoint_lsn = ZERO_LSN; // next_log_file_number is uninitialized // n_in_file is uninitialized toku_logfilemgr_create(&result->logfilemgr); *resultp=result; ml_init(&result->input_lock); toku_mutex_init(&result->output_condition_lock, NULL); toku_cond_init(&result->output_condition, NULL); result->output_is_available = true; return 0; } ~~~ Logger子系統在env->open階段，調用`toku_logger_open`函數進行進一步的初始化。函數`toku_logger_open`是`toku_logger_open_with_last_xid`的簡單封裝。Env->open最終調用`toku_logger_open_with_last_xid`解析redo log file獲取下一個可用的lsn，下一個可用的redo log file的序列號index并打開相應redo log file。在env->open時，調用`toku_logger_open_with_last_xid`的最后一個參數last_xid為TXNID_NONE，表示由`toku_logger_open_with_last_xid`指定事務子系統初始化時最新的txnid。 解析redo log file的過程在函數`toku_logfilemgr_init`實現，依次解析redo log目錄下的每一個文件名符合特定格式的redo log file，從中讀取最后一個log entry的lsn保存下來。Redo log文件名遵循”log$index.tokulog$version”格式，$index是64位無符號整數表示的redo log file的序列號index，$version是32位無符號整數表示版本信息。 如果最新的redo log file最后一個log entry是LT_shutdown（表示正常關閉不需要進行recovery），那么把對應的txnid記錄在last_xid_if_clean_shutdown變量，作為TokuDB事務子系統初始化時最新的txnid。在解析redo log file的時候，還會用最新的redo log file的最后一個log entry的lsn更新logger的lsn，written_lsn，fsynced_lsn。接著，`toku_logger_find_next_unused_log_file`找到下一個可用的redo log文件的序列號，并創建新的redo log file。每個redo log file最開始的12個字節是固定的，首先是8個字節的magic字符串“tokulogg“，緊接著4個字節是log的版本信息。代碼片段如下： ~~~ int toku_logger_open_with_last_xid(const char *directory /* redo log dir */, TOKULOGGER logger, TXNID last_xid) { if (logger->is_open) return EINVAL; TXNID last_xid_if_clean_shutdown = TXNID_NONE; r = toku_logfilemgr_init(logger->logfilemgr, directory, &last_xid_if_clean_shutdown); if ( r!=0 ) return r; logger->lsn = toku_logfilemgr_get_last_lsn(logger->logfilemgr); logger->written_lsn = logger->lsn; logger->fsynced_lsn = logger->lsn; logger->inbuf.max_lsn_in_buf = logger->lsn; logger->outbuf.max_lsn_in_buf = logger->lsn; r = open_logdir(logger, directory); if (r!=0) return r; long long nexti; r = toku_logger_find_next_unused_log_file(logger->directory, &nexti); if (r!=0) return r; logger->next_log_file_number = nexti; r = open_logfile(logger); if (r!=0) return r; if (last_xid == TXNID_NONE) { last_xid = last_xid_if_clean_shutdown; } toku_txn_manager_set_last_xid_from_logger(logger->txn_manager, last_xid); logger->is_open = true; return 0; } ~~~ 到這里，TokuDB的logger子系統就初始化好了，在處理DDL或者DML或者TokuDB執行checkpoint的時候，都需要先寫rollback（undo）log，redo log。Rollback在之前的月報[MySQL · TokuDB · 事務子系統和 MVCC 實現](http://mysql.taobao.org/monthly/2016/03/01/)?談到過，這里不再贅述。 ### 寫redo log 下面我們一起看一下往redo log新加一條insert的過程。函數`toku_log_enq_insert`的第2，第5，第6，第7，第8參數表示描述一條insert的五元組(lsn, FT, xid, key, value)。代碼片段如下： ~~~ void toku_log_enq_insert (TOKULOGGER logger, LSN *lsnp, int do_fsync, TOKUTXN txn, FILENUM filenum, TXNID_PAIR xid, BYTESTRING key, BYTESTRING value) { if (logger == NULL) { return; } if (txn && !txn->begin_was_logged) { invariant(!txn_declared_read_only(txn)); // 記錄txn begin toku_maybe_log_begin_txn_for_write_operation(txn); } if (!logger->write_log_files) { // logger->write_log_files為FALSE，表示不寫redo，遞增lsn就可以返回了。 ml_lock(&logger->input_lock); logger->lsn.lsn++; if (lsnp) *lsnp=logger->lsn; ml_unlock(&logger->input_lock); return; } const unsigned int buflen= (+4 // log entry的長度，參與crc計算 +1 // log命令，對應insert來說是‘I’ +8 // lsn +toku_logsizeof_FILENUM(filenum) // filenum，表示哪個FT文件 +toku_logsizeof_TXNID_PAIR(xid) // xid，表示txnid +toku_logsizeof_BYTESTRING(key) // key +toku_logsizeof_BYTESTRING(value) // data +8 // crc + len // crc和log entry長度（不參與crc計算） ); struct wbuf wbuf; ml_lock(&logger->input_lock); toku_logger_make_space_in_inbuf(logger, buflen); wbuf_nocrc_init(&wbuf, logger->inbuf.buf+logger->inbuf.n_in_buf, buflen); wbuf_nocrc_int(&wbuf, buflen); wbuf_nocrc_char(&wbuf, 'I'); logger->lsn.lsn++; logger->inbuf.max_lsn_in_buf = logger->lsn; wbuf_nocrc_LSN(&wbuf, logger->lsn); if (lsnp) *lsnp=logger->lsn; wbuf_nocrc_FILENUM(&wbuf, filenum); wbuf_nocrc_TXNID_PAIR(&wbuf, xid); wbuf_nocrc_BYTESTRING(&wbuf, key); wbuf_nocrc_BYTESTRING(&wbuf, value); wbuf_nocrc_int(&wbuf, toku_x1764_memory(wbuf.buf, wbuf.ndone)); wbuf_nocrc_int(&wbuf, buflen); assert(wbuf.ndone==buflen); logger->inbuf.n_in_buf += buflen; toku_logger_maybe_fsync(logger, logger->lsn, do_fsync, true); } ~~~ TokuDB的logger有兩個buffer：inbuf和outbuf。Inbuf表示接收log entry的buffer，而outbuf表示寫到redo log文件的buffer。這兩個buffer是如何切換的呢？當inbuf滿或者inbuf里的free space無法滿足新來的log entry的存儲需求時，需要觸發redo buffer flush過程，即將inbuf日志flush到redo log文件里。這個過程比較耗時，而且很可能inbuf里面還有free space，只是由于當前這個log entry比較大而無法滿足存儲需求，TokuDB實現了output permission機制，使得需要free space的請求等待在output permission的條件變量上，其他client thread上下文的redo log請求可以繼續使用inbuf寫日志。等待上一個flush完成后（即條件變量被signaled），檢查當前inbuf的free space，如果可以滿足這條redo log entry就直接返回，說明別的線程幫我們flush好了。如果free space不夠，需要在當前線程的上下文去做flush，實際上是把inbuf和outbuf互換，然后把outbuf寫到redo log文件中。寫完之后適當調整inbuf的大小使之滿足當前redo log entry請求。最后喚醒等待inbuf提供足夠空間的線程（阻塞在output permission上的線程）。簡而言之，把redo log buffer拆分成inbuf和outbuf，最重要的作用是在redo log flush的時候不會阻塞新的log entry寫入，感興趣的朋友可以看一下函數`toku_logger_maybe_fsync`的實現，這里就不一一展開了。函數`toku_logger_make_space_in_inbuf`的代碼片段如下： ~~~ void toku_logger_make_space_in_inbuf (TOKULOGGER logger, int n_bytes_needed) { if (logger->inbuf.n_in_buf + n_bytes_needed <= LOGGER_MIN_BUF_SIZE) { return; } ml_unlock(&logger->input_lock); LSN fsynced_lsn; // 等待前面的redo log flush完成 grab_output(logger, &fsynced_lsn); ml_lock(&logger->input_lock); if (logger->inbuf.n_in_buf + n_bytes_needed <= LOGGER_MIN_BUF_SIZE) { // 其他線程幫助flush redo log，直接返回。 release_output(logger, fsynced_lsn); return; } if (logger->inbuf.n_in_buf > 0) { // 交換inbuf，outbuf swap_inbuf_outbuf(logger); // 把outbuf里的日志寫回 write_outbuf_to_logfile(logger, &fsynced_lsn); } // 適當調整inbuf大小 if (n_bytes_needed > logger->inbuf.buf_size) { assert(n_bytes_needed < (1<<30)); // redo log entry必須小于1G int new_size = max_int(logger->inbuf.buf_size * 2, n_bytes_needed); assert(new_size < (1<<30)); // inbuf必須小于1G XREALLOC_N(new_size, logger->inbuf.buf); logger->inbuf.buf_size = new_size; } // 喚醒等待flush redo log的線程 release_output(logger, fsynced_lsn); } ~~~ ## TokuDB崩潰恢復過程 ### 判斷是否進行recovery 前面提到MySQL重啟過程中會調用`db_env_create`創建env實例，進行參數設置和callback設置，然后調用env->open來做進一步初始化。同樣env->open也是一個回調函數，它是在`db_env_create`設置的，指向env_open函數。 在env_open里調用validate_env判斷是否需要進行recovery。validate_env函數返回時表明這個env是否是emptyenv (env目錄為空，且不存在rollback文件，不存在數據文件)，是否是newnev (env目錄不存在)，是否是emptyrollback (env目錄存在，rollback文件為空)。 如果滿足條件?!emptyenv && !new_env && is_set(DB_RECOVERY)?就嘗試進行recovery。簡單地說recovery的條件就是env存在，log_dir存在，redo log存在。 判斷是否真正做recovery的函數是`tokuft_needs_recovery`。代碼如下： ~~~ int tokuft_needs_recovery(const char *log_dir, bool ignore_log_empty) { int needs_recovery; int r; TOKULOGCURSOR logcursor = NULL; r = toku_logcursor_create(&logcursor, log_dir); if (r != 0) { needs_recovery = true; goto exit; } struct log_entry *le; le = NULL; r = toku_logcursor_last(logcursor, &le); if (r == 0) { needs_recovery = le->cmd != LT_shutdown; } else { needs_recovery = !(r == DB_NOTFOUND && ignore_log_empty); } exit: if (logcursor) { r = toku_logcursor_destroy(&logcursor); assert(r == 0); } return needs_recovery; } ~~~ `tokuft_needs_recovery`嘗試讀取最后一條redo log entry，如果不是LT_shutdown，就需要真正做recovery。讀取最后一條redo log entry的代碼片段如下： ~~~ int toku_logcursor_last(TOKULOGCURSOR lc, struct log_entry **le) { // 打開最后一個redo log文件 if ( !lc->is_open ) { r = lc_open_logfile(lc, lc->n_logfiles-1); if (r!=0) return r; lc->cur_logfiles_index = lc->n_logfiles-1; } while (1) { // 移到最后一個redo log的文件末尾 r = fseek(lc->cur_fp, 0, SEEK_END); assert(r==0); // 從當前位置（redo log末尾）向前讀一個log entry r = toku_log_fread_backward(lc->cur_fp, &(lc->entry)); if (r==0) // 讀成功 break; if (r>0) { // 讀失敗 toku_log_free_log_entry_resources(&(lc->entry)); // 從當前redo log頭部開始向后scan直到找到第一非法log Entry的位置，并把redo log文件truncate到那個位置。 r = lc_fix_bad_logfile(lc); if ( r != 0 ) { fprintf(stderr, "%.24s TokuFT recovery repair unsuccessful\n", ctime(&tnow)); return DB_BADFORMAT; } // 重新讀redo log entry r = toku_log_fread_backward(lc->cur_fp, &(lc->entry)); if (r==0) // 讀到好的redo log entry break; } // 當前redo log沒有訪問的log entry，切換到上一個redo log文件 r = lc_close_cur_logfile(lc); if (r!=0) return r; if ( lc->cur_logfiles_index == 0 ) return DB_NOTFOUND; lc->cur_logfiles_index--; r = lc_open_logfile(lc, lc->cur_logfiles_index); if (r!=0) return r; } } ~~~ 在讀最后一個log entry的過程中，在讀log entry出錯的情況下（crash的時候把redo log寫壞了）會調用`lc_fix_bad_logfile`嘗試修復redo log文件。修復的過程很簡單：從當前redo log頭部開始向后scan直到找到第一非法log entry的位置，并把redo log文件truncate到那個位置。此時，文件指針也指向文件末尾。極端的情況是，修復完redo log，發現當前redo log中的所有entry都是壞的，那樣需要切換到前面一個redo log文件。 ### Recovery過程 如果需要做recovery，TokuDB會調用do_recovery進行恢復，恢復的時候先做redo log apply，然后進行undo rollback。代碼片段如下： ~~~ static int do_recovery(RECOVER_ENV renv, const char *env_dir, const char *log_dir) { r = toku_logcursor_create(&logcursor, log_dir); assert(r == 0); scan_state_init(&renv->ss); for (unsigned i=0; 1; i++) { // 讀取前一個log entry，第一次讀的是最后一個log entry le = NULL; r = toku_logcursor_prev(logcursor, &le); if (r != 0) { if (r == DB_NOTFOUND) break; rr = DB_RUNRECOVERY; goto errorexit; } // backward階段處理log entry assert(renv->ss.ss == BACKWARD_BETWEEN_CHECKPOINT_BEGIN_END || renv->ss.ss == BACKWARD_NEWER_CHECKPOINT_END); logtype_dispatch_assign(le, toku_recover_backward_, r, renv); if (r != 0) { rr = DB_RUNRECOVERY; goto errorexit; } if (renv->goforward) break; } for (unsigned i=0; 1; i++) { // forward階段處理log entry，首先處理的是checkpoint begin的那個log entry assert(renv->ss.ss == FORWARD_BETWEEN_CHECKPOINT_BEGIN_END || renv->ss.ss == FORWARD_NEWER_CHECKPOINT_END); logtype_dispatch_assign(le, toku_recover_, r, renv); if (r != 0) { rr = DB_RUNRECOVERY; goto errorexit; } // 讀取下一個log entry le = NULL; r = toku_logcursor_next(logcursor, &le); if (r != 0) { if (r == DB_NOTFOUND) break; rr = DB_RUNRECOVERY; goto errorexit; } } // parse redo log結束 assert(renv->ss.ss == FORWARD_NEWER_CHECKPOINT_END); r = toku_logcursor_destroy(&logcursor); assert(r == 0); // 重啟logger toku_logger_restart(renv->logger, lastlsn); // abort所有未提交的事務 recover_abort_all_live_txns(renv); // 在recovery退出前做一個checkpoint r = toku_checkpoint(renv->cp, renv->logger, NULL, NULL, NULL, NULL, RECOVERY_CHECKPOINT); assert(r == 0); return 0; } ~~~ Scan log entry分別兩個階段：backward階段和forward階段。這兩個階段是由scan_state狀態機控制的。在scan開始之前在`scan_state_init`函數中把狀態機ss的初始狀態設置為BACKWARD_NEWER_CHECKPOINT_END。  * Backward階段：從最后一個log entry開始向前讀，直到讀到checkpoint end。對在這個過程中讀到的每一個log entry調用`logtype_dispatch_assign(le, toku_recover_backward_, r, renv)`。在這個階段對于checkpoint以外的操作，toku_recover_backward_前綴的處理函數都是noop。當讀到checkpoint end的log entry時，會把ss狀態設置為BACKWARD_BETWEEN_CHECKPOINT_BEGIN_END，并記錄這個checkpoint的begin_lsn和lsn。然后繼續向前scan直到讀到checkpoint begin的log entry，確保ss中記錄的checkpoint_begin_lsn和log entry的lsn是相等的，然后 把ss的狀態設置為FORWARD_BETWEEN_CHECKPOINT_BEGIN_END，并設置renv->goforward為TRUE。 * Forward階段：對當前的log entry調用`logtype_dispatch_assign(le, toku_recover_, r, renv)`重放redo log。然后向后scan直到讀到checkpoint end，確保ss中記錄的`checkpoint_begin_lsn`和`checkpoint_end_lsn`與log entry里面記錄的`lsn_begin_checkpoint`和lsn是相等的，然后把ss的狀態設置為FORWARD_NEWER_CHECKPOINT_END。這樣，崩潰之前的最后一個checkpoint就回放完成了。下面要做的事情就是，回放committed txn的redo log。代碼片段如下： ~~~ static void scan_state_init(struct scan_state *ss) { ss->ss = BACKWARD_NEWER_CHECKPOINT_END; ss->checkpoint_begin_lsn = ZERO_LSN; ss->checkpoint_end_lsn = ZERO_LSN; ss->checkpoint_num_fassociate = 0; ss->checkpoint_num_xstillopen = 0; ss->last_xid = 0; } static int toku_recover_backward_end_checkpoint (struct logtype_end_checkpoint *l, RECOVER_ENV renv) { switch (renv->ss.ss) { case BACKWARD_NEWER_CHECKPOINT_END: renv->ss.ss = BACKWARD_BETWEEN_CHECKPOINT_BEGIN_END; renv->ss.checkpoint_begin_lsn.lsn = l->lsn_begin_checkpoint.lsn; renv->ss.checkpoint_end_lsn.lsn = l->lsn.lsn; renv->ss.checkpoint_end_timestamp = l->timestamp; return 0; case BACKWARD_BETWEEN_CHECKPOINT_BEGIN_END: abort(); default: break; } abort(); } static int toku_recover_backward_begin_checkpoint (struct logtype_begin_checkpoint *l, RECOVER_ENV renv) { int r; switch (renv->ss.ss) { case BACKWARD_NEWER_CHECKPOINT_END: // incomplete checkpoint, nothing to do r = 0; break; case BACKWARD_BETWEEN_CHECKPOINT_BEGIN_END: assert(l->lsn.lsn == renv->ss.checkpoint_begin_lsn.lsn); renv->ss.ss = FORWARD_BETWEEN_CHECKPOINT_BEGIN_END; renv->ss.checkpoint_begin_timestamp = l->timestamp; renv->goforward = true; r = 0; break; default: abort(); break; } return r; } static int toku_recover_begin_checkpoint (struct logtype_begin_checkpoint *l, RECOVER_ENV renv) { int r; TXN_MANAGER mgr = toku_logger_get_txn_manager(renv->logger); switch (renv->ss.ss) { case FORWARD_BETWEEN_CHECKPOINT_BEGIN_END: assert(l->lsn.lsn == renv->ss.checkpoint_begin_lsn.lsn); invariant(renv->ss.last_xid == TXNID_NONE); renv->ss.last_xid = l->last_xid; toku_txn_manager_set_last_xid_from_recovered_checkpoint(mgr, l->last_xid); r = 0; break; case FORWARD_NEWER_CHECKPOINT_END: assert(l->lsn.lsn > renv->ss.checkpoint_end_lsn.lsn); // Verify last_xid is no older than the previous begin invariant(l->last_xid >= renv->ss.last_xid); // Verify last_xid is no older than the newest txn invariant(l->last_xid >= toku_txn_manager_get_last_xid(mgr)); r = 0; // ignore it (log only has a begin checkpoint) break; default: abort(); break; } return r; } static int toku_recover_end_checkpoint (struct logtype_end_checkpoint *l, RECOVER_ENV renv) { int r; switch (renv->ss.ss) { case FORWARD_BETWEEN_CHECKPOINT_BEGIN_END: assert(l->lsn_begin_checkpoint.lsn == renv->ss.checkpoint_begin_lsn.lsn); assert(l->lsn.lsn == renv->ss.checkpoint_end_lsn.lsn); assert(l->num_fassociate_entries == renv->ss.checkpoint_num_fassociate); assert(l->num_xstillopen_entries == renv->ss.checkpoint_num_xstillopen); renv->ss.ss = FORWARD_NEWER_CHECKPOINT_END; r = 0; break; case FORWARD_NEWER_CHECKPOINT_END: assert(0); return 0; default: assert(0); return 0; } return r; } ~~~ 上面我們是TokuDB recovery的過程。對讀redo log一筆帶過。現在一起看看讀log entry的過程： * 向后讀：從當前位置讀4個字節的長度len1，然后讀1個字節cmd。然后按照不同cmd的定義來讀log entry。 * 向前讀：從當前位置讀nocrc的長度len2，把文件指針向前移動len2個字節。從那個位置向后讀。 * Verify：讀的過程需要計算crc校驗碼。Len1是參與crc計算的，而len2不參與crc計算。計算得到的crc應該與log entry里面記錄的crc相等。而且len1應該等于len2。