MDEV-27058: Reduce the size of buf_block_t and buf_page_t

buf_page_t::frame: Moved from buf_block_t::frame.
All 'thin' buf_page_t describing compressed-only ROW_FORMAT=COMPRESSED
pages will have frame=nullptr, while all 'fat' buf_block_t
will have a non-null frame pointing to aligned innodb_page_size bytes.
This eliminates the need for separate states for
BUF_BLOCK_FILE_PAGE and BUF_BLOCK_ZIP_PAGE.

buf_page_t::lock: Moved from buf_block_t::lock. That is, all block
descriptors will have a page latch. The IO_PIN state that was used
for discarding or creating the uncompressed page frame of a
ROW_FORMAT=COMPRESSED block is replaced by a combination of read-fix
and page X-latch.

page_zip_des_t::fix: Replaces state_, buf_fix_count_, io_fix_, status
of buf_page_t with a single std::atomic<uint32_t>. All modifications
will use store(), fetch_add(), fetch_sub(). This space was previously
wasted to alignment on 64-bit systems. We will use the following encoding
that combines a state (partly read-fix or write-fix) and a buffer-fix
count:

buf_page_t::NOT_USED=0 (previously BUF_BLOCK_NOT_USED)
buf_page_t::MEMORY=1 (previously BUF_BLOCK_MEMORY)
buf_page_t::REMOVE_HASH=2 (previously BUF_BLOCK_REMOVE_HASH)
buf_page_t::FREED=3 + fix: pages marked as freed in the file
buf_page_t::UNFIXED=1U<<29 + fix: normal pages
buf_page_t::IBUF_EXIST=2U<<29 + fix: normal pages; may need ibuf merge
buf_page_t::REINIT=3U<<29 + fix: reinitialized pages (skip doublewrite)
buf_page_t::READ_FIX=4U<<29 + fix: read-fixed pages (also X-latched)
buf_page_t::WRITE_FIX=5U<<29 + fix: write-fixed pages (also U-latched)
buf_page_t::WRITE_FIX_IBUF=6U<<29 + fix: write-fixed; may have ibuf
buf_page_t::WRITE_FIX_REINIT=7U<<29 + fix: write-fixed (no doublewrite)

buf_page_t::write_complete(): Change WRITE_FIX or WRITE_FIX_REINIT to
UNFIXED, and WRITE_FIX_IBUF to IBUF_EXIST, before releasing the U-latch.

buf_page_t::read_complete(): Renamed from buf_page_read_complete().
Change READ_FIX to UNFIXED or IBUF_EXIST, before releasing the X-latch.

buf_page_t::can_relocate(): If the page latch is being held or waited for,
or the block is buffer-fixed or io-fixed, return false. (The condition
on the page latch is new.)

Outside buf_page_get_gen(), buf_page_get_low() and buf_page_free(), we
will acquire the page latch before fix(), and unfix() before unlocking.

buf_page_t::flush(): Replaces buf_flush_page(). Optimize the
handling of FREED pages.

buf_pool_t::release_freed_page(): Assume that buf_pool.mutex is held
by the caller.

buf_page_t::is_read_fixed(), buf_page_t::is_write_fixed(): New predicates.

buf_page_get_low(): Ignore guesses that are read-fixed because they
may not yet be registered in buf_pool.page_hash and buf_pool.LRU.

buf_page_optimistic_get(): Acquire latch before buffer-fixing.

buf_page_make_young(): Leave read-fixed blocks alone, because they
might not be registered in buf_pool.LRU yet.

recv_sys_t::recover_deferred(), recv_sys_t::recover_low():
Possibly fix MDEV-26326, by holding a page X-latch instead of
only buffer-fixing the page.

mysql-test/suite/encryption/r/innodb_encrypt_key_rotation_age.result

@@ -67,13 +67,13 @@ test/t2
 test/t3
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
 NAME
-test/t4
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-NAME
 innodb_system
-mysql/innodb_index_stats
 mysql/innodb_table_stats
+mysql/innodb_index_stats
 mysql/transaction_registry
+test/t4
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
 test/t1
 test/t2
 test/t3

mysql-test/suite/encryption/r/innodb_encryption.result

@@ -66,7 +66,7 @@ innodb_encryption_threads	0
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0
 AND NAME NOT LIKE 'innodb_undo%' AND NAME NOT LIKE 'mysql/innodb_%_stats' AND NAME NOT LIKE 'mysql/transaction_registry';
 NAME
-innodb_system
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0
 AND NAME NOT LIKE 'innodb_undo%' AND NAME NOT LIKE 'mysql/innodb_%_stats' AND NAME NOT LIKE 'mysql/transaction_registry';
 NAME
+innodb_system

mysql-test/suite/encryption/t/innodb_encryption.test

@@ -90,6 +90,9 @@ AND NAME NOT LIKE 'innodb_undo%' AND NAME NOT LIKE 'mysql/innodb_%_stats' AND NA
 
 SHOW VARIABLES LIKE 'innodb_encrypt%';
 
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+--source include/wait_condition.inc
+
 --sorted_result
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0
 AND NAME NOT LIKE 'innodb_undo%' AND NAME NOT LIKE 'mysql/innodb_%_stats' AND NAME NOT LIKE 'mysql/transaction_registry';

mysql-test/suite/innodb_i_s/innodb_buffer_page.result

@@ -18,7 +18,7 @@ INNODB_BUFFER_PAGE	CREATE TEMPORARY TABLE `INNODB_BUFFER_PAGE` (
   `DATA_SIZE` bigint(21) unsigned NOT NULL DEFAULT 0,
   `COMPRESSED_SIZE` bigint(21) unsigned NOT NULL DEFAULT 0,
   `PAGE_STATE` enum('NOT_USED','MEMORY','REMOVE_HASH','FILE_PAGE') NOT NULL,
-  `IO_FIX` enum('IO_NONE','IO_READ','IO_WRITE','IO_PIN') NOT NULL,
+  `IO_FIX` enum('IO_NONE','IO_READ','IO_WRITE') NOT NULL,
   `IS_OLD` int(1) NOT NULL DEFAULT 0,
   `FREE_PAGE_CLOCK` bigint(21) unsigned NOT NULL DEFAULT 0
 ) ENGINE=MEMORY DEFAULT CHARSET=utf8mb3

mysql-test/suite/innodb_i_s/innodb_buffer_page_lru.result

@@ -18,7 +18,7 @@ INNODB_BUFFER_PAGE_LRU	CREATE TEMPORARY TABLE `INNODB_BUFFER_PAGE_LRU` (
   `DATA_SIZE` bigint(21) unsigned NOT NULL DEFAULT 0,
   `COMPRESSED_SIZE` bigint(21) unsigned NOT NULL DEFAULT 0,
   `COMPRESSED` int(1) NOT NULL DEFAULT 0,
-  `IO_FIX` enum('IO_NONE','IO_READ','IO_WRITE','IO_PIN') NOT NULL,
+  `IO_FIX` enum('IO_NONE','IO_READ','IO_WRITE') NOT NULL,
   `IS_OLD` int(1) DEFAULT NULL,
   `FREE_PAGE_CLOCK` bigint(21) unsigned NOT NULL DEFAULT 0
 ) ENGINE=MEMORY DEFAULT CHARSET=utf8mb3

storage/innobase/CMakeLists.txt

@@ -95,7 +95,6 @@ SET(INNOBASE_SOURCES
 	include/buf0dblwr.h
 	include/buf0dump.h
 	include/buf0flu.h
-	include/buf0flu.ic
 	include/buf0lru.h
 	include/buf0rea.h
 	include/buf0types.h

storage/innobase/btr/btr0bulk.cc

@@ -494,8 +494,8 @@ inline void PageBulk::finishPage()
 
 inline bool PageBulk::needs_finish() const
 {
-  ut_ad(page_align(m_cur_rec) == m_block->frame);
-  ut_ad(m_page == m_block->frame);
+  ut_ad(page_align(m_cur_rec) == m_block->page.frame);
+  ut_ad(m_page == m_block->page.frame);
   if (!m_page[PAGE_HEADER + PAGE_DIRECTION_B])
     return true;
   ulint heap_no, n_heap= page_header_get_field(m_page, PAGE_N_HEAP);
@@ -839,7 +839,7 @@ PageBulk::release()
 	finish();
 
 	/* We fix the block because we will re-pin it soon. */
-	buf_block_buf_fix_inc(m_block);
+	m_block->page.fix();
 
 	/* No other threads can modify this block. */
 	m_modify_clock = buf_block_get_modify_clock(m_block);
@@ -873,11 +873,11 @@ PageBulk::latch()
 		ut_ad(m_block != NULL);
 	}
 
-	buf_block_buf_fix_dec(m_block);
+	ut_d(const auto buf_fix_count =) m_block->page.unfix();
 
-	ut_ad(m_block->page.buf_fix_count());
-
-	ut_ad(m_cur_rec > m_page && m_cur_rec < m_heap_top);
+	ut_ad(buf_fix_count);
+	ut_ad(m_cur_rec > m_page);
+	ut_ad(m_cur_rec < m_heap_top);
 
 	return (m_err);
 }
@@ -1208,7 +1208,7 @@ BtrBulk::finish(dberr_t	err)
 		last_block = btr_block_get(*m_index, last_page_no, RW_X_LATCH,
 					   false, &mtr);
 		first_rec = page_rec_get_next(
-			page_get_infimum_rec(last_block->frame));
+			page_get_infimum_rec(last_block->page.frame));
 		ut_ad(page_rec_is_user_rec(first_rec));
 
 		/* Copy last page to root page. */

storage/innobase/btr/btr0defragment.cc

@@ -476,7 +476,7 @@ btr_defragment_n_pages(
 	/* It doesn't make sense to call this function with n_pages = 1. */
 	ut_ad(n_pages > 1);
 
-	if (!page_is_leaf(block->frame)) {
+	if (!page_is_leaf(block->page.frame)) {
 		return NULL;
 	}
 

storage/innobase/btr/btr0pcur.cc

@@ -112,8 +112,9 @@ btr_pcur_store_position(
 	page_cursor = btr_pcur_get_page_cur(cursor);
 
 	rec = page_cur_get_rec(page_cursor);
-	offs = rec - block->frame;
-	ut_ad(block->page.id().page_no() == page_get_page_no(block->frame));
+	offs = rec - block->page.frame;
+	ut_ad(block->page.id().page_no()
+	      == page_get_page_no(block->page.frame));
 	ut_ad(block->page.buf_fix_count());
 	/* For spatial index, when we do positioning on parent
 	buffer if necessary, it might not hold latches, but the
@@ -126,13 +127,13 @@ btr_pcur_store_position(
 
 	cursor->old_stored = true;
 
-	if (page_is_empty(block->frame)) {
+	if (page_is_empty(block->page.frame)) {
 		/* It must be an empty index tree; NOTE that in this case
 		we do not store the modify_clock, but always do a search
 		if we restore the cursor position */
 
-		ut_a(!page_has_siblings(block->frame));
-		ut_ad(page_is_leaf(block->frame));
+		ut_a(!page_has_siblings(block->page.frame));
+		ut_ad(page_is_leaf(block->page.frame));
 		ut_ad(block->page.id().page_no() == index->page);
 
 		if (page_rec_is_supremum_low(offs)) {
@@ -159,9 +160,9 @@ btr_pcur_store_position(
 #endif
 			ut_ad(index->is_instant()
 			      || block->page.id().page_no() != index->page);
-			ut_ad(page_get_n_recs(block->frame) == 1);
-			ut_ad(page_is_leaf(block->frame));
-			ut_ad(!page_has_prev(block->frame));
+			ut_ad(page_get_n_recs(block->page.frame) == 1);
+			ut_ad(page_is_leaf(block->page.frame));
+			ut_ad(!page_has_prev(block->page.frame));
 			cursor->rel_pos = BTR_PCUR_AFTER_LAST_IN_TREE;
 			return;
 		}
@@ -171,7 +172,7 @@ btr_pcur_store_position(
 		rec = page_rec_get_next(rec);
 
 		if (rec_is_metadata(rec, *index)) {
-			ut_ad(!page_has_prev(block->frame));
+			ut_ad(!page_has_prev(block->page.frame));
 			rec = page_rec_get_next(rec);
 			if (page_rec_is_supremum(rec)) {
 				goto before_first;

storage/innobase/buf/buf0block_hint.cc

@@ -43,16 +43,15 @@ void Block_hint::buffer_fix_block_if_still_valid()
   different page, and that slice of buf_pool.page_hash could be protected
   by another hash_lock that we are not holding.)
 
-  Finally, assuming that we have correct hash bucket latched, we must
-  validate m_block->state() to ensure that the block is not being freed. */
+  Finally, we must ensure that the block is not being freed. */
   if (m_block)
   {
     auto &cell= buf_pool.page_hash.cell_get(m_page_id.fold());
     transactional_shared_lock_guard<page_hash_latch> g
       {buf_pool.page_hash.lock_get(cell)};
     if (buf_pool.is_uncompressed(m_block) && m_page_id == m_block->page.id() &&
-        m_block->page.state() == BUF_BLOCK_FILE_PAGE)
-      m_block->fix();
+        m_block->page.frame && m_block->page.in_file())
+      m_block->page.fix();
     else
       clear();
   }

storage/innobase/buf/buf0buddy.cc

@@ -354,11 +354,11 @@ buf_buddy_block_free(void* buf)
 	ut_a(!ut_align_offset(buf, srv_page_size));
 
 	HASH_SEARCH(hash, &buf_pool.zip_hash, fold, buf_page_t*, bpage,
-		    ut_ad(bpage->state() == BUF_BLOCK_MEMORY
+		    ut_ad(bpage->state() == buf_page_t::MEMORY
 			  && bpage->in_zip_hash),
-		    ((buf_block_t*) bpage)->frame == buf);
+		    bpage->frame == buf);
 	ut_a(bpage);
-	ut_a(bpage->state() == BUF_BLOCK_MEMORY);
+	ut_a(bpage->state() == buf_page_t::MEMORY);
 	ut_ad(bpage->in_zip_hash);
 	ut_d(bpage->in_zip_hash = false);
 	HASH_DELETE(buf_page_t, hash, &buf_pool.zip_hash, fold, bpage);
@@ -383,10 +383,10 @@ buf_buddy_block_register(
 	buf_block_t*	block)	/*!< in: buffer frame to allocate */
 {
 	const ulint	fold = BUF_POOL_ZIP_FOLD(block);
-	ut_ad(block->page.state() == BUF_BLOCK_MEMORY);
+	ut_ad(block->page.state() == buf_page_t::MEMORY);
 
-	ut_a(block->frame);
-	ut_a(!ut_align_offset(block->frame, srv_page_size));
+	ut_a(block->page.frame);
+	ut_a(!ut_align_offset(block->page.frame, srv_page_size));
 
 	ut_ad(!block->page.in_zip_hash);
 	ut_d(block->page.in_zip_hash = true);
@@ -462,8 +462,8 @@ byte *buf_buddy_alloc_low(ulint i, bool *lru)
 alloc_big:
 	buf_buddy_block_register(block);
 
-	block = (buf_block_t*) buf_buddy_alloc_from(
-		block->frame, i, BUF_BUDDY_SIZES);
+	block = reinterpret_cast<buf_block_t*>(
+		buf_buddy_alloc_from(block->page.frame, i, BUF_BUDDY_SIZES));
 
 func_exit:
 	buf_pool.buddy_stat[i].used++;
@@ -694,7 +694,7 @@ buf_buddy_realloc(void* buf, ulint size)
 
 		block = reinterpret_cast<buf_block_t*>(
 			buf_buddy_alloc_from(
-				block->frame, i, BUF_BUDDY_SIZES));
+				block->page.frame, i, BUF_BUDDY_SIZES));
 	}
 
 	buf_pool.buddy_stat[i].used++;

storage/innobase/buf/buf0dblwr.cc

@@ -86,11 +86,12 @@ bool buf_dblwr_t::create()
   buf_block_t *trx_sys_block= buf_dblwr_trx_sys_get(&mtr);
 
   if (mach_read_from_4(TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_MAGIC +
-                       trx_sys_block->frame) == TRX_SYS_DOUBLEWRITE_MAGIC_N)
+                       trx_sys_block->page.frame) ==
+      TRX_SYS_DOUBLEWRITE_MAGIC_N)
   {
     /* The doublewrite buffer has already been created: just read in
     some numbers */
-    init(TRX_SYS_DOUBLEWRITE + trx_sys_block->frame);
+    init(TRX_SYS_DOUBLEWRITE + trx_sys_block->page.frame);
     mtr.commit();
     return true;
   }
@@ -121,7 +122,7 @@ bool buf_dblwr_t::create()
   }
 
   byte *fseg_header= TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_FSEG +
-    trx_sys_block->frame;
+    trx_sys_block->page.frame;
   for (uint32_t prev_page_no= 0, i= 0, extent_size= FSP_EXTENT_SIZE;
        i < 2 * size + extent_size / 2; i++)
   {
@@ -149,12 +150,12 @@ bool buf_dblwr_t::create()
     tablespace, then the page has not been written to in
     doublewrite. */
 
-    ut_ad(new_block->lock.not_recursive());
+    ut_ad(new_block->page.lock.not_recursive());
     const page_id_t id= new_block->page.id();
     /* We only do this in the debug build, to ensure that the check in
     buf_flush_init_for_writing() will see a valid page type. The
     flushes of new_block are actually unnecessary here.  */
-    ut_d(mtr.write<2>(*new_block, FIL_PAGE_TYPE + new_block->frame,
+    ut_d(mtr.write<2>(*new_block, FIL_PAGE_TYPE + new_block->page.frame,
                       FIL_PAGE_TYPE_SYS));
 
     if (i == size / 2)
@@ -162,10 +163,10 @@ bool buf_dblwr_t::create()
       ut_a(id.page_no() == size);
       mtr.write<4>(*trx_sys_block,
                    TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_BLOCK1 +
-                   trx_sys_block->frame, id.page_no());
+                   trx_sys_block->page.frame, id.page_no());
       mtr.write<4>(*trx_sys_block,
                    TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_REPEAT +
-                   TRX_SYS_DOUBLEWRITE_BLOCK1 + trx_sys_block->frame,
+                   TRX_SYS_DOUBLEWRITE_BLOCK1 + trx_sys_block->page.frame,
                    id.page_no());
     }
     else if (i == size / 2 + size)
@@ -173,10 +174,10 @@ bool buf_dblwr_t::create()
       ut_a(id.page_no() == 2 * size);
       mtr.write<4>(*trx_sys_block,
                    TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_BLOCK2 +
-                   trx_sys_block->frame, id.page_no());
+                   trx_sys_block->page.frame, id.page_no());
       mtr.write<4>(*trx_sys_block,
                    TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_REPEAT +
-                   TRX_SYS_DOUBLEWRITE_BLOCK2 + trx_sys_block->frame,
+                   TRX_SYS_DOUBLEWRITE_BLOCK2 + trx_sys_block->page.frame,
                    id.page_no());
     }
     else if (i > size / 2)
@@ -193,7 +194,7 @@ bool buf_dblwr_t::create()
       mtr.start();
       trx_sys_block= buf_dblwr_trx_sys_get(&mtr);
       fseg_header= TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_FSEG +
-        trx_sys_block->frame;
+        trx_sys_block->page.frame;
     }
 
     prev_page_no= id.page_no();
@@ -201,15 +202,16 @@ bool buf_dblwr_t::create()
 
   mtr.write<4>(*trx_sys_block,
                TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_MAGIC +
-               trx_sys_block->frame, TRX_SYS_DOUBLEWRITE_MAGIC_N);
+               trx_sys_block->page.frame, TRX_SYS_DOUBLEWRITE_MAGIC_N);
   mtr.write<4>(*trx_sys_block,
                TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_MAGIC +
-               TRX_SYS_DOUBLEWRITE_REPEAT + trx_sys_block->frame,
+               TRX_SYS_DOUBLEWRITE_REPEAT + trx_sys_block->page.frame,
                TRX_SYS_DOUBLEWRITE_MAGIC_N);
 
   mtr.write<4>(*trx_sys_block,
                TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_SPACE_ID_STORED +
-               trx_sys_block->frame, TRX_SYS_DOUBLEWRITE_SPACE_ID_STORED_N);
+               trx_sys_block->page.frame,
+               TRX_SYS_DOUBLEWRITE_SPACE_ID_STORED_N);
   mtr.commit();
 
   /* Flush the modified pages to disk and make a checkpoint */
@@ -519,8 +521,9 @@ static void buf_dblwr_check_page_lsn(const buf_page_t &b, const byte *page)
 /** Check the LSN values on the page with which this block is associated. */
 static void buf_dblwr_check_block(const buf_page_t *bpage)
 {
-  ut_ad(bpage->state() == BUF_BLOCK_FILE_PAGE);
-  const page_t *page= reinterpret_cast<const buf_block_t*>(bpage)->frame;
+  ut_ad(bpage->in_file());
+  const page_t *page= bpage->frame;
+  ut_ad(page);
 
   switch (fil_page_get_type(page)) {
   case FIL_PAGE_INDEX:
@@ -619,9 +622,7 @@ static void *get_frame(const IORequest &request)
   if (request.slot)
     return request.slot->out_buf;
   const buf_page_t *bpage= request.bpage;
-  return bpage->zip.data
-    ? bpage->zip.data
-    : reinterpret_cast<const buf_block_t*>(bpage)->frame;
+  return bpage->zip.data ? bpage->zip.data : bpage->frame;
 }
 
 void buf_dblwr_t::flush_buffered_writes_completed(const IORequest &request)
@@ -673,7 +674,6 @@ void buf_dblwr_t::flush_buffered_writes_completed(const IORequest &request)
     }
     else
     {
-      ut_ad(bpage->state() == BUF_BLOCK_FILE_PAGE);
       ut_ad(!bpage->zip_size());
       ut_d(buf_dblwr_check_page_lsn(*bpage, static_cast<const byte*>(frame)));
     }

storage/innobase/buf/buf0dump.cc

@@ -328,16 +328,15 @@ buf_dump(
 	for (bpage = UT_LIST_GET_FIRST(buf_pool.LRU), j = 0;
 	     bpage != NULL && j < n_pages;
 	     bpage = UT_LIST_GET_NEXT(LRU, bpage)) {
-
-		ut_a(bpage->in_file());
-		const page_id_t id(bpage->id());
-
-		if (id.space() == SRV_TMP_SPACE_ID) {
-			/* Ignore the innodb_temporary tablespace. */
+		const auto status = bpage->state();
+		if (status < buf_page_t::UNFIXED) {
+			ut_a(status >= buf_page_t::FREED);
 			continue;
 		}
+		const page_id_t id{bpage->id()};
 
-		if (bpage->status == buf_page_t::FREED) {
+		if (id.space() == SRV_TMP_SPACE_ID) {
+			/* Ignore the innodb_temporary tablespace. */
 			continue;
 		}
 

storage/innobase/buf/buf0rea.cc

@@ -59,12 +59,9 @@ inline void buf_pool_t::watch_remove(buf_page_t *watch,
   ut_ad(page_hash.lock_get(chain).is_write_locked());
   ut_a(watch_is_sentinel(*watch));
   if (watch->buf_fix_count())
-  {
     page_hash.remove(chain, watch);
-    watch->set_buf_fix_count(0);
-  }
   ut_ad(!watch->in_page_hash);
-  watch->set_state(BUF_BLOCK_NOT_USED);
+  watch->set_state(buf_page_t::NOT_USED);
   watch->id_= page_id_t(~0ULL);
 }
 
@@ -109,10 +106,10 @@ static buf_page_t* buf_page_init_for_read(ulint mode, const page_id_t page_id,
   if (!zip_size || unzip || recv_recovery_is_on())
   {
     block= buf_LRU_get_free_block(false);
-    block->initialise(page_id, zip_size);
+    block->initialise(page_id, zip_size, buf_page_t::READ_FIX);
     /* x_unlock() will be invoked
-    in buf_page_read_complete() by the io-handler thread. */
-    block->lock.x_lock(true);
+    in buf_page_t::read_complete() by the io-handler thread. */
+    block->page.lock.x_lock(true);
   }
 
   buf_pool_t::hash_chain &chain= buf_pool.page_hash.cell_get(page_id.fold());
@@ -125,7 +122,8 @@ static buf_page_t* buf_page_init_for_read(ulint mode, const page_id_t page_id,
     /* The page is already in the buffer pool. */
     if (block)
     {
-      block->lock.x_unlock(true);
+      block->page.lock.x_unlock(true);
+      ut_d(block->page.set_state(buf_page_t::MEMORY));
       buf_LRU_block_free_non_file_page(block);
     }
     goto func_exit;
@@ -143,14 +141,13 @@ static buf_page_t* buf_page_init_for_read(ulint mode, const page_id_t page_id,
       if (hash_page)
       {
         /* Preserve the reference count. */
-        auto buf_fix_count= hash_page->buf_fix_count();
-        ut_a(buf_fix_count > 0);
-        block->page.add_buf_fix_count(buf_fix_count);
+        uint32_t buf_fix_count= hash_page->state();
+        ut_a(buf_fix_count >= buf_page_t::UNFIXED);
+        ut_a(buf_fix_count < buf_page_t::READ_FIX);
         buf_pool.watch_remove(hash_page, chain);
+        block->page.fix(buf_fix_count - buf_page_t::UNFIXED);
       }
 
-      block->page.set_io_fix(BUF_IO_READ);
-      block->page.set_state(BUF_BLOCK_FILE_PAGE);
       buf_pool.page_hash.append(chain, &block->page);
     }
 
@@ -198,13 +195,14 @@ static buf_page_t* buf_page_init_for_read(ulint mode, const page_id_t page_id,
       }
     }
 
-    bpage= buf_page_alloc_descriptor();
+    bpage= static_cast<buf_page_t*>(ut_zalloc_nokey(sizeof *bpage));
 
     page_zip_des_init(&bpage->zip);
     page_zip_set_size(&bpage->zip, zip_size);
     bpage->zip.data = (page_zip_t*) data;
 
-    bpage->init(BUF_BLOCK_ZIP_PAGE, page_id);
+    bpage->init(buf_page_t::READ_FIX, page_id);
+    bpage->lock.x_lock(true);
 
     {
       transactional_lock_guard<page_hash_latch> g
@@ -215,12 +213,14 @@ static buf_page_t* buf_page_init_for_read(ulint mode, const page_id_t page_id,
         /* Preserve the reference count. It can be 0 if
         buf_pool_t::watch_unset() is executing concurrently,
         waiting for buf_pool.mutex, which we are holding. */
-        bpage->add_buf_fix_count(hash_page->buf_fix_count());
+        uint32_t buf_fix_count= hash_page->state();
+        ut_a(buf_fix_count >= buf_page_t::UNFIXED);
+        ut_a(buf_fix_count < buf_page_t::READ_FIX);
+        bpage->fix(buf_fix_count - buf_page_t::UNFIXED);
         buf_pool.watch_remove(hash_page, chain);
       }
 
       buf_pool.page_hash.append(chain, bpage);
-      bpage->set_io_fix(BUF_IO_READ);
     }
 
     /* The block must be put to the LRU list, to the old blocks.
@@ -315,16 +315,7 @@ buf_read_page_low(
 		 "read page " << page_id << " zip_size=" << zip_size
 		 << " unzip=" << unzip << ',' << (sync ? "sync" : "async"));
 
-	void*	dst;
-
-	if (zip_size) {
-		dst = bpage->zip.data;
-	} else {
-		ut_a(bpage->state() == BUF_BLOCK_FILE_PAGE);
-
-		dst = ((buf_block_t*) bpage)->frame;
-	}
-
+	void* dst = zip_size ? bpage->zip.data : bpage->frame;
 	const ulint len = zip_size ? zip_size : srv_page_size;
 
 	auto fio = space->io(IORequest(sync
@@ -347,7 +338,7 @@ buf_read_page_low(
 		thd_wait_end(NULL);
 
 		/* The i/o was already completed in space->io() */
-		*err = buf_page_read_complete(bpage, *fio.node);
+		*err = bpage->read_complete(*fio.node);
 		space->release();
 
 		if (*err != DB_SUCCESS) {
@@ -628,19 +619,7 @@ buf_read_ahead_linear(const page_id_t page_id, ulint zip_size, bool ibuf)
       on the page, we do not acquire an s-latch on the page, this is to
       prevent deadlocks. The hash_lock is only protecting the
       buf_pool.page_hash for page i, not the bpage contents itself. */
-      const byte *f;
-      switch (UNIV_EXPECT(bpage->state(), BUF_BLOCK_FILE_PAGE)) {
-      case BUF_BLOCK_FILE_PAGE:
-        f= reinterpret_cast<const buf_block_t*>(bpage)->frame;
-        break;
-      case BUF_BLOCK_ZIP_PAGE:
-        f= bpage->zip.data;
-        break;
-      default:
-        ut_ad("invalid state" == 0);
-        goto fail;
-      }
-
+      const byte *f= bpage->frame ? bpage->frame : bpage->zip.data;
       uint32_t prev= mach_read_from_4(my_assume_aligned<4>(f + FIL_PAGE_PREV));
       uint32_t next= mach_read_from_4(my_assume_aligned<4>(f + FIL_PAGE_NEXT));
       if (prev == FIL_NULL || next == FIL_NULL)

storage/innobase/dict/dict0boot.cc

 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2020, MariaDB Corporation.
+Copyright (c) 2016, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -55,29 +55,29 @@ dict_hdr_get_new_id(
 
 	if (table_id) {
 		id = mach_read_from_8(DICT_HDR + DICT_HDR_TABLE_ID
-				      + dict_hdr->frame);
+				      + dict_hdr->page.frame);
 		id++;
 		mtr.write<8>(*dict_hdr, DICT_HDR + DICT_HDR_TABLE_ID
-			     + dict_hdr->frame, id);
+			     + dict_hdr->page.frame, id);
 		*table_id = id;
 	}
 
 	if (index_id) {
 		id = mach_read_from_8(DICT_HDR + DICT_HDR_INDEX_ID
-				      + dict_hdr->frame);
+				      + dict_hdr->page.frame);
 		id++;
 		mtr.write<8>(*dict_hdr, DICT_HDR + DICT_HDR_INDEX_ID
-			     + dict_hdr->frame, id);
+			     + dict_hdr->page.frame, id);
 		*index_id = id;
 	}
 
 	if (space_id) {
 		*space_id = mach_read_from_4(DICT_HDR + DICT_HDR_MAX_SPACE_ID
-					     + dict_hdr->frame);
+					     + dict_hdr->page.frame);
 		if (fil_assign_new_space_id(space_id)) {
 			mtr.write<4>(*dict_hdr,
 				     DICT_HDR + DICT_HDR_MAX_SPACE_ID
-				     + dict_hdr->frame, *space_id);
+				     + dict_hdr->page.frame, *space_id);
 		}
 	}
 
@@ -90,7 +90,7 @@ void dict_hdr_flush_row_id(row_id_t id)
   mtr_t mtr;
   mtr.start();
   buf_block_t* d= dict_hdr_get(&mtr);
-  byte *row_id= DICT_HDR + DICT_HDR_ROW_ID + d->frame;
+  byte *row_id= DICT_HDR + DICT_HDR_ROW_ID + d->page.frame;
   if (mach_read_from_8(row_id) < id)
     mtr.write<8>(*d, row_id, id);
   mtr.commit();
@@ -119,17 +119,18 @@ static bool dict_hdr_create()
 
 	/* Start counting row, table, index, and tree ids from
 	DICT_HDR_FIRST_ID */
-	mtr.write<8>(*d, DICT_HDR + DICT_HDR_ROW_ID + d->frame,
+	mtr.write<8>(*d, DICT_HDR + DICT_HDR_ROW_ID + d->page.frame,
 		     DICT_HDR_FIRST_ID);
-	mtr.write<8>(*d, DICT_HDR + DICT_HDR_TABLE_ID + d->frame,
+	mtr.write<8>(*d, DICT_HDR + DICT_HDR_TABLE_ID + d->page.frame,
 		     DICT_HDR_FIRST_ID);
-	mtr.write<8>(*d, DICT_HDR + DICT_HDR_INDEX_ID + d->frame,
+	mtr.write<8>(*d, DICT_HDR + DICT_HDR_INDEX_ID + d->page.frame,
 		     DICT_HDR_FIRST_ID);
 
-	ut_ad(!mach_read_from_4(DICT_HDR + DICT_HDR_MAX_SPACE_ID + d->frame));
+	ut_ad(!mach_read_from_4(DICT_HDR + DICT_HDR_MAX_SPACE_ID
+				+ d->page.frame));
 
 	/* Obsolete, but we must initialize it anyway. */
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_MIX_ID_LOW + d->frame,
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_MIX_ID_LOW + d->page.frame,
 		     DICT_HDR_FIRST_ID);
 
 	/* Create the B-tree roots for the clustered indexes of the basic
@@ -145,7 +146,8 @@ static bool dict_hdr_create()
 		goto func_exit;
 	}
 
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_TABLES + d->frame, root_page_no);
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_TABLES + d->page.frame,
+		     root_page_no);
 	/*--------------------------*/
 	root_page_no = btr_create(DICT_UNIQUE,
 				  fil_system.sys_space, DICT_TABLE_IDS_ID,
@@ -154,7 +156,7 @@ static bool dict_hdr_create()
 		goto failed;
 	}
 
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_TABLE_IDS + d->frame,
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_TABLE_IDS + d->page.frame,
 		     root_page_no);
 	/*--------------------------*/
 	root_page_no = btr_create(DICT_CLUSTERED | DICT_UNIQUE,
@@ -164,7 +166,7 @@ static bool dict_hdr_create()
 		goto failed;
 	}
 
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_COLUMNS + d->frame,
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_COLUMNS + d->page.frame,
 		     root_page_no);
 	/*--------------------------*/
 	root_page_no = btr_create(DICT_CLUSTERED | DICT_UNIQUE,
@@ -174,7 +176,8 @@ static bool dict_hdr_create()
 		goto failed;
 	}
 
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_INDEXES + d->frame, root_page_no);
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_INDEXES + d->page.frame,
+		     root_page_no);
 	/*--------------------------*/
 	root_page_no = btr_create(DICT_CLUSTERED | DICT_UNIQUE,
 				  fil_system.sys_space, DICT_FIELDS_ID,
@@ -183,7 +186,8 @@ static bool dict_hdr_create()
 		goto failed;
 	}
 
-	mtr.write<4>(*d, DICT_HDR + DICT_HDR_FIELDS + d->frame, root_page_no);
+	mtr.write<4>(*d, DICT_HDR + DICT_HDR_FIELDS + d->page.frame,
+		     root_page_no);
 func_exit:
 	mtr.commit();
 	return fail;
@@ -228,7 +232,7 @@ dict_boot(void)
 	dict_sys.lock(SRW_LOCK_CALL);
 
 	/* Get the dictionary header */
-	const byte* dict_hdr = &dict_hdr_get(&mtr)->frame[DICT_HDR];
+	const byte* dict_hdr = &dict_hdr_get(&mtr)->page.frame[DICT_HDR];
 
 	/* Because we only write new row ids to disk-based data structure
 	(dictionary header) when it is divisible by

storage/innobase/dict/dict0defrag_bg.cc

@@ -314,11 +314,11 @@ btr_get_size_and_reserved(
 	mtr->x_lock_space(index->table->space);
 
 	ulint n = fseg_n_reserved_pages(*root, PAGE_HEADER + PAGE_BTR_SEG_LEAF
-					+ root->frame, used, mtr);
+					+ root->page.frame, used, mtr);
 	if (flag == BTR_TOTAL_SIZE) {
 		n += fseg_n_reserved_pages(*root,
 					   PAGE_HEADER + PAGE_BTR_SEG_TOP
-					   + root->frame, &dummy, mtr);
+					   + root->page.frame, &dummy, mtr);
 		*used += dummy;
 	}
 

storage/innobase/dict/dict0load.cc

@@ -936,7 +936,8 @@ void dict_check_tablespaces_and_store_max_id()
 	mtr.start();
 	ulint max_space_id = mach_read_from_4(DICT_HDR_MAX_SPACE_ID
 					      + DICT_HDR
-					      + dict_hdr_get(&mtr)->frame);
+					      + dict_hdr_get(&mtr)
+					      ->page.frame);
 	mtr.commit();
 
 	fil_set_max_space_id_if_bigger(max_space_id);
@@ -2373,15 +2374,15 @@ static dict_table_t *dict_load_table_one(const span<const char> &name,
 				page_id, table->space->zip_size(),
 				RW_S_LATCH, &mtr);
 			const bool corrupted = !block
-				|| page_get_space_id(block->frame)
+				|| page_get_space_id(block->page.frame)
 				!= page_id.space()
-				|| page_get_page_no(block->frame)
+				|| page_get_page_no(block->page.frame)
 				!= page_id.page_no()
 				|| (mach_read_from_2(FIL_PAGE_TYPE
-						    + block->frame)
+						    + block->page.frame)
 				    != FIL_PAGE_INDEX
 				    && mach_read_from_2(FIL_PAGE_TYPE
-							+ block->frame)
+							+ block->page.frame)
 				    != FIL_PAGE_TYPE_INSTANT);
 			mtr.commit();
 			if (corrupted) {

storage/innobase/handler/ha_innodb.cc

@@ -17515,8 +17515,10 @@ innodb_make_page_dirty(THD*, st_mysql_sys_var*, void*, const void* save)
 	if (block != NULL) {
 		ib::info() << "Dirtying page: " << block->page.id();
 		mtr.write<1,mtr_t::FORCED>(*block,
-					   block->frame + FIL_PAGE_SPACE_ID,
-					   block->frame[FIL_PAGE_SPACE_ID]);
+					   block->page.frame
+					   + FIL_PAGE_SPACE_ID,
+					   block->page.frame
+					   [FIL_PAGE_SPACE_ID]);
 	}
 	mtr.commit();
 	log_write_up_to(mtr.commit_lsn(), true);
@@ -17970,7 +17972,8 @@ static bool innodb_buffer_pool_evict_uncompressed()
 	for (buf_block_t* block = UT_LIST_GET_LAST(buf_pool.unzip_LRU);
 	     block != NULL; ) {
 		buf_block_t*	prev_block = UT_LIST_GET_PREV(unzip_LRU, block);
-		ut_ad(block->page.state() == BUF_BLOCK_FILE_PAGE);
+		ut_ad(block->page.in_file());
+		ut_ad(block->page.belongs_to_unzip_LRU());
 		ut_ad(block->in_unzip_LRU_list);
 		ut_ad(block->page.in_LRU_list);