Simplify undo log access during InnoDB startup

trx_rseg_mem_restore(): Update the max_trx_id from the undo log pages. trx_sys_init_at_db_start(): Remove; merge with trx_lists_init_at_db_start(). trx_undo_lists_init(): Move to the only calling module, trx0rseg.cc. trx_undo_mem_create_at_db_start(): Declare globally. Return the number of pages.

Author: dr-m

storage/innobase/include/trx0sys.h

@@ -58,9 +58,6 @@ trx_sys_hdr_page(const page_id_t& page_id)
  && page_id.page_no() == TRX_SYS_PAGE_NO);
 }
 
-/** Initialize the transaction system main-memory data structures. */
-void trx_sys_init_at_db_start();
-
 /*****************************************************************//**
 Creates and initializes the transaction system at the database creation. */
 void

storage/innobase/include/trx0undo.h

@@ -210,15 +210,6 @@ trx_undo_truncate_start(
 ulint	hdr_page_no,
 ulint	hdr_offset,
 undo_no_tlimit);
-/********************************************************************//**
-Initializes the undo log lists for a rollback segment memory copy.
-This function is only called when the database is started or a new
-rollback segment created.
-@return the combined size of undo log segments in pages */
-ulint
-trx_undo_lists_init(
-/*================*/
-trx_rseg_t*	rseg);/*!< in: rollback segment memory object */
 /** Assign an undo log for a persistent transaction.
 A new undo log is created or a cached undo log reused.
 @param[in,out]	trx	transaction
@@ -326,6 +317,13 @@ trx_undo_parse_page_header(
 const byte*	end_ptr,
 page_t*	page,
 mtr_t*	mtr);
+/** Read an undo log when starting up the database.
+@param[in,out]	rseg	rollback segment
+@param[in]	id	rollback segment slot
+@param[in]	page_no	undo log segment page number
+@return	size of the undo log in pages */
+ulint
+trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no);
 /************************************************************************
 Frees an undo log memory copy. */
 void

storage/innobase/srv/srv0start.cc

@@ -2173,7 +2173,7 @@ innobase_start_or_create_for_mysql()
 All the remaining rollback segments will be created later,
 after the double write buffer has been created. */
 trx_sys_create_sys_pages();
-trx_sys_init_at_db_start();
+trx_lists_init_at_db_start();
 
 err = dict_create();
 
@@ -2234,7 +2234,7 @@ innobase_start_or_create_for_mysql()
 }
 /* This must precede
 recv_apply_hashed_log_recs(true). */
-trx_sys_init_at_db_start();
+trx_lists_init_at_db_start();
 break;
 case SRV_OPERATION_RESTORE_DELTA:
 case SRV_OPERATION_BACKUP:
@@ -2325,7 +2325,7 @@ innobase_start_or_create_for_mysql()
 }
 
 /* recv_recovery_from_checkpoint_finish needs trx lists which
-are initialized in trx_sys_init_at_db_start(). */
+are initialized in trx_lists_init_at_db_start(). */
 
 recv_recovery_from_checkpoint_finish();
 

storage/innobase/trx/trx0rseg.cc

@@ -168,49 +168,71 @@ trx_rseg_mem_create(ulint id, ulint space, ulint page_no)
 return(rseg);
 }
 
-/** Restore the state of a persistent rollback segment.
-@param[in,out]	rseg	persistent rollback segment
-@param[in,out]	mtr	mini-transaction */
+/** Read the undo log lists.
+@param[in,out]	rseg	rollback segment
+@param[in]	rseg_header	rollback segment header
+@param[in,out]	mtr	mini-transaction
+@return the combined size of undo log segments in pages */
 static
-void
-trx_rseg_mem_restore(trx_rseg_t* rseg, mtr_t* mtr)
+ulint
+trx_undo_lists_init(trx_rseg_t* rseg, const trx_rsegf_t* rseg_header,
+ mtr_t* mtr)
 {
-ulint	len;
-fil_addr_tnode_addr;
-trx_rsegf_t*	rseg_header;
-trx_ulogf_t*	undo_log_hdr;
-ulint	sum_of_undo_sizes;
+ut_ad(srv_force_recovery < SRV_FORCE_NO_UNDO_LOG_SCAN);
 
-rseg_header = trx_rsegf_get_new(rseg->space, rseg->page_no, mtr);
+ulint size = 0;
 
-rseg->max_size = mtr_read_ulint(
-rseg_header + TRX_RSEG_MAX_SIZE, MLOG_4BYTES, mtr);
+for (ulint i = 0; i < TRX_RSEG_N_SLOTS; i++) {
+ulint	page_no = trx_rsegf_get_nth_undo(rseg_header, i);
+if (page_no != FIL_NULL) {
+size += trx_undo_mem_create_at_db_start(
+rseg, i, page_no);
+MONITOR_INC(MONITOR_NUM_UNDO_SLOT_USED);
+}
+}
 
-/* Initialize the undo log lists according to the rseg header */
+return(size);
+}
 
-sum_of_undo_sizes = trx_undo_lists_init(rseg);
+/** Restore the state of a persistent rollback segment.
+@param[in,out]	rseg	persistent rollback segment
+@param[in,out]	max_trx_id	maximum observed transaction identifier
+@param[in,out]	mtr	mini-transaction */
+static
+void
+trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
+{
+const trx_rsegf_t*	rseg_header = trx_rsegf_get_new(
+rseg->space, rseg->page_no, mtr);
+rseg->max_size = mach_read_from_4(rseg_header + TRX_RSEG_MAX_SIZE);
 
-rseg->curr_size = mtr_read_ulint(
-rseg_header + TRX_RSEG_HISTORY_SIZE, MLOG_4BYTES, mtr)
-+ 1 + sum_of_undo_sizes;
+/* Initialize the undo log lists according to the rseg header */
 
-len = flst_get_len(rseg_header + TRX_RSEG_HISTORY);
+rseg->curr_size = mach_read_from_4(rseg_header + TRX_RSEG_HISTORY_SIZE)
++ 1 + trx_undo_lists_init(rseg, rseg_header, mtr);
 
-if (len > 0) {
+if (ulint len = flst_get_len(rseg_header + TRX_RSEG_HISTORY)) {
 my_atomic_addlint(&trx_sys.rseg_history_len, len);
 
-node_addr = trx_purge_get_log_from_hist(
+fil_addr_tnode_addr = trx_purge_get_log_from_hist(
 flst_get_last(rseg_header + TRX_RSEG_HISTORY, mtr));
 
 rseg->last_page_no = node_addr.page;
 rseg->last_offset = node_addr.boffset;
 
-undo_log_hdr = trx_undo_page_get(
+const trx_ulogf_t*undo_log_hdr = trx_undo_page_get(
 page_id_t(rseg->space, node_addr.page), mtr)
 + node_addr.boffset;
 
-rseg->last_trx_no = mach_read_from_8(
-undo_log_hdr + TRX_UNDO_TRX_NO);
+trx_id_t id = mach_read_from_8(undo_log_hdr + TRX_UNDO_TRX_ID);
+if (id > max_trx_id) {
+max_trx_id = id;
+}
+id = mach_read_from_8(undo_log_hdr + TRX_UNDO_TRX_NO);
+rseg->last_trx_no = id;
+if (id > max_trx_id) {
+max_trx_id = id;
+}
 unsigned purge = mach_read_from_2(
 undo_log_hdr + TRX_UNDO_NEEDS_PURGE);
 ut_ad(purge <= 1);
@@ -233,11 +255,30 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, mtr_t* mtr)
 void
 trx_rseg_array_init()
 {
-mtr_tmtr;
+trx_id_t max_trx_id = 0;
 
 for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS; rseg_id++) {
+mtr_t mtr;
 mtr.start();
 if (const buf_block_t* sys = trx_sysf_get(&mtr, false)) {
+if (rseg_id == 0) {
+/* VERY important: after the database
+is started, max_trx_id value is
+divisible by TRX_SYS_TRX_ID_WRITE_MARGIN,
+and the first call of
+trx_sys.get_new_trx_id() will invoke
+flush_max_trx_id()! Thus trx id values
+will not overlap when the database is
+repeatedly started! */
+
+max_trx_id = 2 * TRX_SYS_TRX_ID_WRITE_MARGIN
++ ut_uint64_align_up(
+mach_read_from_8(
+TRX_SYS
++ TRX_SYS_TRX_ID_STORE
++ sys->frame),
+TRX_SYS_TRX_ID_WRITE_MARGIN);
+}
 const uint32_tpage_no = trx_sysf_rseg_get_page_no(
 sys, rseg_id);
 if (page_no != FIL_NULL) {
@@ -249,12 +290,14 @@ trx_rseg_array_init()
 ut_ad(rseg->id == rseg_id);
 ut_ad(!trx_sys.rseg_array[rseg_id]);
 trx_sys.rseg_array[rseg_id] = rseg;
-trx_rseg_mem_restore(rseg, &mtr);
+trx_rseg_mem_restore(rseg, max_trx_id, &mtr);
 }
 }
 
 mtr.commit();
 }
+
+trx_sys.init_max_trx_id(max_trx_id);
 }
 
 /** Create a persistent rollback segment.

storage/innobase/trx/trx0sys.cc

@@ -415,39 +415,6 @@ trx_sysf_create(
 ut_a(page_no == FSP_FIRST_RSEG_PAGE_NO);
 }
 
-/** Initialize the transaction system main-memory data structures. */
-void
-trx_sys_init_at_db_start()
-{
-/* VERY important: after the database is started, max_trx_id value is
-divisible by TRX_SYS_TRX_ID_WRITE_MARGIN, and the 'if' in
-trx_sys.get_new_trx_id will evaluate to TRUE when the function
-is first time called, and the value for trx id will be written
-to the disk-based header! Thus trx id values will not overlap when
-the database is repeatedly started! */
-
-mtr_tmtr;
-mtr.start();
-
-buf_block_t* block = trx_sysf_get(&mtr);
-
-trx_id_t max_trx_id = block
-? 2 * TRX_SYS_TRX_ID_WRITE_MARGIN
-+ ut_uint64_align_up(mach_read_from_8(TRX_SYS
- + TRX_SYS_TRX_ID_STORE
- + block->frame),
- TRX_SYS_TRX_ID_WRITE_MARGIN)
-: 0;
-trx_sys.init_max_trx_id(max_trx_id);
-
-mtr.commit();
-
-trx_dummy_sess = sess_open();
-
-trx_lists_init_at_db_start();
-trx_sys.mvcc.clone_oldest_view(&purge_sys->view);
-}
-
 /** Create the instance */
 void
 trx_sys_t::create()

storage/innobase/trx/trx0trx.cc

@@ -895,22 +895,23 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
 void
 trx_lists_init_at_db_start()
 {
-uint64_trows_to_undo	= 0;
 ut_a(srv_is_being_started);
 ut_ad(!srv_was_started);
 ut_ad(!purge_sys);
-
-purge_sys = UT_NEW_NOKEY(purge_sys_t());
+ut_ad(!trx_dummy_sess);
 
 if (srv_force_recovery >= SRV_FORCE_NO_UNDO_LOG_SCAN) {
 return;
 }
 
+trx_dummy_sess = sess_open();
+purge_sys = UT_NEW_NOKEY(purge_sys_t());
 trx_rseg_array_init();
 
 /* Look from the rollback segments if there exist undo logs for
 transactions. */
-const ib_time_t start_time = ut_time();
+const ib_time_tstart_time	= ut_time();
+uint64_trows_to_undo	= 0;
 
 for (ulint i = 0; i < TRX_SYS_N_RSEGS; ++i) {
 trx_undo_t*	undo;
@@ -979,6 +980,7 @@ trx_lists_init_at_db_start()
 }
 
 std::sort(trx_sys.rw_trx_ids.begin(), trx_sys.rw_trx_ids.end());
+trx_sys.mvcc.clone_oldest_view(&purge_sys->view);
 }
 
 /** Assign a persistent rollback segment in a round-robin fashion,