MDEV-21133 Optimize access to InnoDB page header fields

Introduce memcpy_aligned<N>(), memcmp_aligned<N>(), memset_aligned<N>()
and use them for accessing InnoDB page header fields that are known
to be aligned.

MY_ASSUME_ALIGNED(): Wrapper for the GCC/clang __builtin_assume_aligned().
Nothing similar seems to exist in Microsoft Visual Studio, and the
C++20 std::assume_aligned is not available to us yet.

Explicitly specified alignment guarantees allow compilers to generate
faster code on platforms with strict alignment rules, instead of
emitting calls to potentially unaligned memcpy(), memcmp(), or memset().

include/m_string.h

@@ -1,5 +1,6 @@
 /*
    Copyright (c) 2000, 2012, Oracle and/or its affiliates.
+   Copyright (c) 2019, 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
@@ -194,6 +195,33 @@ extern ulonglong strtoull(const char *str, char **ptr, int base);
 
 #if defined(__cplusplus)
 }
+
+# ifdef _MSC_VER
+#  define MY_ASSUME_ALIGNED(x,n) x
+# else
+#  define MY_ASSUME_ALIGNED(x,n) __builtin_assume_aligned(x,n)
+# endif
+
+template <size_t Alignment>
+inline void *memcpy_aligned(void *dest, const void *src, size_t n)
+{
+  static_assert(Alignment && !(Alignment & (Alignment - 1)), "power of 2");
+  return memcpy(MY_ASSUME_ALIGNED(dest, Alignment),
+                MY_ASSUME_ALIGNED(src, Alignment), n);
+}
+template <size_t Alignment>
+inline int memcmp_aligned(const void *s1, const void *s2, size_t n)
+{
+  static_assert(Alignment && !(Alignment & (Alignment - 1)), "power of 2");
+  return memcmp(MY_ASSUME_ALIGNED(s1, Alignment),
+                MY_ASSUME_ALIGNED(s2, Alignment), n);
+}
+template <size_t Alignment>
+inline void *memset_aligned(void *s, int c, size_t n)
+{
+  static_assert(Alignment && !(Alignment & (Alignment - 1)), "power of 2");
+  return memset(MY_ASSUME_ALIGNED(s, Alignment), c, n);
+}
 #endif
 
 #include <mysql/plugin.h>

storage/innobase/btr/btr0btr.cc

@@ -1165,7 +1165,8 @@ btr_create(
 	compile_time_assert(FIL_NULL == 0xffffffff);
 	mlog_memset(block, FIL_PAGE_PREV, 8, 0xff, mtr);
 	if (UNIV_LIKELY_NULL(page_zip)) {
-		memset(page_zip->data + FIL_PAGE_PREV, 0xff, 8);
+		static_assert(FIL_PAGE_PREV % 8 == 0, "alignment");
+		memset_aligned<8>(FIL_PAGE_PREV + page_zip->data, 0xff, 8);
 	}
 
 	/* We reset the free bits for the page in a separate
@@ -1573,10 +1574,13 @@ btr_page_reorganize_low(
 			/* Preserve the PAGE_INSTANT information. */
 			ut_ad(!page_zip);
 			ut_ad(index->is_instant());
-			memcpy(FIL_PAGE_TYPE + page,
-			       FIL_PAGE_TYPE + temp_page, 2);
-			memcpy(PAGE_HEADER + PAGE_INSTANT + page,
-			       PAGE_HEADER + PAGE_INSTANT + temp_page, 2);
+			static_assert(!(FIL_PAGE_TYPE % 2), "alignment");
+			memcpy_aligned<2>(FIL_PAGE_TYPE + page,
+					  FIL_PAGE_TYPE + temp_page, 2);
+			static_assert(!((PAGE_HEADER+PAGE_INSTANT) % 2), "");
+			memcpy_aligned<2>(PAGE_HEADER + PAGE_INSTANT + page,
+					  PAGE_HEADER + PAGE_INSTANT
+					  + temp_page, 2);
 			if (!index->table->instant) {
 			} else if (page_is_comp(page)) {
 				memcpy(PAGE_NEW_INFIMUM + page,
@@ -1947,7 +1951,8 @@ btr_root_raise_and_insert(
 	compile_time_assert(FIL_NULL == 0xffffffff);
 	mlog_memset(new_block, FIL_PAGE_PREV, 8, 0xff, mtr);
 	if (UNIV_LIKELY_NULL(new_page_zip)) {
-		memset(new_page_zip->data + FIL_PAGE_PREV, 0xff, 8);
+		static_assert(FIL_PAGE_PREV % 8 == 0, "alignment");
+		memset_aligned<8>(new_page_zip->data + FIL_PAGE_PREV, 0xff, 8);
 	}
 
 	/* Copy the records from root to the new page one by one. */
@@ -3225,8 +3230,10 @@ void btr_level_list_remove(const buf_block_t& block, const dict_index_t& index,
 			= buf_block_get_frame(prev_block);
 #ifdef UNIV_BTR_DEBUG
 		ut_a(page_is_comp(prev_page) == page_is_comp(page));
-		ut_a(!memcmp(prev_page + FIL_PAGE_NEXT, page + FIL_PAGE_OFFSET,
-			     4));
+		static_assert(FIL_PAGE_NEXT % 4 == 0, "alignment");
+		static_assert(FIL_PAGE_OFFSET % 4 == 0, "alignment");
+		ut_a(!memcmp_aligned<4>(prev_page + FIL_PAGE_NEXT,
+					page + FIL_PAGE_OFFSET, 4));
 #endif /* UNIV_BTR_DEBUG */
 
 		btr_page_set_next(prev_page,
@@ -3242,8 +3249,10 @@ void btr_level_list_remove(const buf_block_t& block, const dict_index_t& index,
 			= buf_block_get_frame(next_block);
 #ifdef UNIV_BTR_DEBUG
 		ut_a(page_is_comp(next_page) == page_is_comp(page));
-		ut_a(!memcmp(next_page + FIL_PAGE_PREV, page + FIL_PAGE_OFFSET,
-			     4));
+		static_assert(FIL_PAGE_PREV % 4 == 0, "alignment");
+		static_assert(FIL_PAGE_OFFSET % 4 == 0, "alignment");
+		ut_a(!memcmp_aligned<4>(next_page + FIL_PAGE_PREV,
+					page + FIL_PAGE_OFFSET, 4));
 #endif /* UNIV_BTR_DEBUG */
 
 		btr_page_set_prev(next_page,
@@ -3818,11 +3827,12 @@ btr_compress(
 			invoked by page_copy_rec_list_end() below,
 			requires that FIL_PAGE_PREV be FIL_NULL.
 			Clear the field, but prepare to restore it. */
+			static_assert(FIL_PAGE_PREV % 8 == 0, "alignment");
 #ifdef UNIV_BTR_DEBUG
 			memcpy(fil_page_prev, merge_page + FIL_PAGE_PREV, 4);
 #endif /* UNIV_BTR_DEBUG */
 			compile_time_assert(FIL_NULL == 0xffffffffU);
-			memset(merge_page + FIL_PAGE_PREV, 0xff, 4);
+			memset_aligned<4>(merge_page + FIL_PAGE_PREV, 0xff, 4);
 		}
 
 		orig_succ = page_copy_rec_list_end(merge_block, block,

storage/innobase/btr/btr0bulk.cc

@@ -95,7 +95,8 @@ PageBulk::init()
 		if (new_page_zip) {
 			page_create_zip(new_block, m_index, m_level, 0,
 					&m_mtr);
-			memset(FIL_PAGE_PREV + new_page, 0xff, 8);
+			static_assert(FIL_PAGE_PREV % 8 == 0, "alignment");
+			memset_aligned<8>(FIL_PAGE_PREV + new_page, 0xff, 8);
 			page_zip_write_header(new_page_zip,
 					      FIL_PAGE_PREV + new_page,
 					      8, &m_mtr);

storage/innobase/buf/buf0buf.cc

@@ -416,9 +416,10 @@ static bool buf_tmp_page_decrypt(byte* tmp_frame, byte* src_frame)
 		return false;
 	}
 
-	memcpy(tmp_frame + srv_page_size - FIL_PAGE_FCRC32_CHECKSUM,
-	       src_frame + srv_page_size - FIL_PAGE_FCRC32_CHECKSUM,
-	       FIL_PAGE_FCRC32_CHECKSUM);
+	static_assert(FIL_PAGE_FCRC32_CHECKSUM == 4, "alignment");
+	memcpy_aligned<4>(tmp_frame + srv_page_size - FIL_PAGE_FCRC32_CHECKSUM,
+			  src_frame + srv_page_size - FIL_PAGE_FCRC32_CHECKSUM,
+			  FIL_PAGE_FCRC32_CHECKSUM);
 
 	memcpy(src_frame, tmp_frame, srv_page_size);
 	srv_stats.pages_decrypted.inc();
@@ -959,12 +960,16 @@ buf_page_is_corrupted(
 				      size - FIL_PAGE_FCRC32_CHECKSUM)) {
 			return true;
 		}
+		static_assert(FIL_PAGE_FCRC32_KEY_VERSION == 0, "alignment");
+		static_assert(FIL_PAGE_LSN % 4 == 0, "alignment");
+		static_assert(FIL_PAGE_FCRC32_END_LSN % 4 == 0, "alignment");
 		if (!compressed
 		    && !mach_read_from_4(FIL_PAGE_FCRC32_KEY_VERSION
 					 + read_buf)
-		    && memcmp(read_buf + (FIL_PAGE_LSN + 4),
-			      end - (FIL_PAGE_FCRC32_END_LSN
-				     - FIL_PAGE_FCRC32_CHECKSUM), 4)) {
+		    && memcmp_aligned<4>(read_buf + (FIL_PAGE_LSN + 4),
+					 end - (FIL_PAGE_FCRC32_END_LSN
+						- FIL_PAGE_FCRC32_CHECKSUM),
+					 4)) {
 			return true;
 		}
 
@@ -998,10 +1003,13 @@ buf_page_is_corrupted(
 		return(false);
 	}
 
-	if (!zip_size && memcmp(read_buf + FIL_PAGE_LSN + 4,
-				read_buf + srv_page_size
-				- FIL_PAGE_END_LSN_OLD_CHKSUM + 4, 4)) {
+	static_assert(FIL_PAGE_LSN % 4 == 0, "alignment");
+	static_assert(FIL_PAGE_END_LSN_OLD_CHKSUM % 4 == 0, "alignment");
 
+	if (!zip_size
+	    && memcmp_aligned<4>(read_buf + FIL_PAGE_LSN + 4,
+				 read_buf + srv_page_size
+				 - FIL_PAGE_END_LSN_OLD_CHKSUM + 4, 4)) {
 		/* Stored log sequence numbers at the start and the end
 		of page do not match */
 
@@ -1029,7 +1037,7 @@ buf_page_is_corrupted(
 	checksum_field2 = mach_read_from_4(
 		read_buf + srv_page_size - FIL_PAGE_END_LSN_OLD_CHKSUM);
 
-	compile_time_assert(!(FIL_PAGE_LSN % 8));
+	static_assert(FIL_PAGE_LSN % 8 == 0, "alignment");
 
 	/* A page filled with NUL bytes is considered not corrupted.
 	The FIL_PAGE_FILE_FLUSH_LSN field may be written nonzero for
@@ -2176,8 +2184,12 @@ buf_page_realloc(
 		ut_ad(new_block->page.in_page_hash);
 
 		buf_block_modify_clock_inc(block);
-		memset(block->frame + FIL_PAGE_OFFSET, 0xff, 4);
-		memset(block->frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID, 0xff, 4);
+		static_assert(FIL_PAGE_OFFSET % 4 == 0, "alignment");
+		memset_aligned<4>(block->frame + FIL_PAGE_OFFSET, 0xff, 4);
+		static_assert(FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID % 4 == 2,
+			      "not perfect alignment");
+		memset_aligned<2>(block->frame
+				  + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID, 0xff, 4);
 		UNIV_MEM_INVALID(block->frame, srv_page_size);
 		buf_block_set_state(block, BUF_BLOCK_REMOVE_HASH);
 		block->page.id
@@ -5499,8 +5511,9 @@ buf_page_create(
 
 	frame = block->frame;
 
-	memset(frame + FIL_PAGE_PREV, 0xff, 4);
-	memset(frame + FIL_PAGE_NEXT, 0xff, 4);
+	static_assert(FIL_PAGE_PREV % 8 == 0, "alignment");
+	static_assert(FIL_PAGE_PREV + 4 == FIL_PAGE_NEXT, "adjacent");
+	memset_aligned<8>(frame + FIL_PAGE_PREV, 0xff, 8);
 	mach_write_to_2(frame + FIL_PAGE_TYPE, FIL_PAGE_TYPE_ALLOCATED);
 
 	/* FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION is only used on the
@@ -5510,7 +5523,8 @@ buf_page_create(
 	(3) key_version on encrypted pages (not page 0:0) */
 
 	memset(frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0, 8);
-	memset(frame + FIL_PAGE_LSN, 0, 8);
+	static_assert(FIL_PAGE_LSN % 8 == 0, "alignment");
+	memset_aligned<8>(frame + FIL_PAGE_LSN, 0, 8);
 
 #if defined UNIV_DEBUG || defined UNIV_BUF_DEBUG
 	ut_a(++buf_dbg_counter % 5771 || buf_validate());
@@ -5692,25 +5706,21 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 
 /** Check if the encrypted page is corrupted for the full crc32 format.
 @param[in]	space_id	page belongs to space id
-@param[in]	dst_frame	page
+@param[in]	d		page
 @param[in]	is_compressed	compressed page
 @return true if page is corrupted or false if it isn't */
-static bool buf_page_full_crc32_is_corrupted(
-	ulint		space_id,
-	const byte*	dst_frame,
-	bool		is_compressed)
+static bool buf_page_full_crc32_is_corrupted(ulint space_id, const byte* d,
+                                             bool is_compressed)
 {
-	if (!is_compressed
-	    && memcmp(dst_frame + FIL_PAGE_LSN + 4,
-		      dst_frame + srv_page_size - FIL_PAGE_FCRC32_END_LSN, 4)) {
-		return true;
-	}
+  if (space_id != mach_read_from_4(d + FIL_PAGE_SPACE_ID))
+    return true;
 
-	if (space_id != mach_read_from_4(dst_frame + FIL_PAGE_SPACE_ID)) {
-		return true;
-	}
+  static_assert(FIL_PAGE_LSN % 4 == 0, "alignment");
+  static_assert(FIL_PAGE_FCRC32_END_LSN % 4 == 0, "alignment");
 
-	return false;
+  return !is_compressed &&
+    memcmp_aligned<4>(FIL_PAGE_LSN + 4 + d,
+                      d + srv_page_size - FIL_PAGE_FCRC32_END_LSN, 4);
 }
 
 /** Check if page is maybe compressed, encrypted or both when we encounter

storage/innobase/buf/buf0dblwr.cc

@@ -501,8 +501,7 @@ buf_dblwr_init_or_load_pages(
 
 				return(err);
 			}
-
-		} else if (memcmp(field_ref_zero, page + FIL_PAGE_LSN, 8)) {
+		} else if (mach_read_from_8(page + FIL_PAGE_LSN)) {
 			/* Each valid page header must contain
 			a nonzero FIL_PAGE_LSN field. */
 			recv_dblwr.add(page);
@@ -789,29 +788,21 @@ buf_dblwr_update(
 
 #ifdef UNIV_DEBUG
 /** Check the LSN values on the page.
-@param[in]	page	page to check
-@param[in]	s	tablespace */
+@param[in] page  page to check
+@param[in] s     tablespace */
 static void buf_dblwr_check_page_lsn(const page_t* page, const fil_space_t& s)
 {
-	/* Ignore page compressed or encrypted pages */
-	if (s.is_compressed()
-	    || buf_page_get_key_version(page, s.flags)) {
-		return;
-	}
-
-	const unsigned lsn1 = mach_read_from_4(page + FIL_PAGE_LSN + 4),
-		lsn2 = mach_read_from_4(page + srv_page_size
-					- (s.full_crc32()
-					   ? FIL_PAGE_FCRC32_END_LSN
-					   : FIL_PAGE_END_LSN_OLD_CHKSUM - 4));
-	if (UNIV_UNLIKELY(lsn1 != lsn2)) {
-		ib::error() << "The page to be written to "
-			    << s.chain.start->name <<
-			" seems corrupt!"
-			" The low 4 bytes of LSN fields do not match"
-			" (" << lsn1 << " != " << lsn2 << ")!"
-			" Noticed in the buffer pool.";
-	}
+  /* Ignore page compressed or encrypted pages */
+  if (s.is_compressed() || buf_page_get_key_version(page, s.flags))
+    return;
+  const byte* lsn_start= FIL_PAGE_LSN + 4 + page;
+  const byte* lsn_end= page +
+    srv_page_size - (s.full_crc32()
+    ? FIL_PAGE_FCRC32_END_LSN
+    : FIL_PAGE_END_LSN_OLD_CHKSUM - 4);
+  static_assert(FIL_PAGE_FCRC32_END_LSN % 4 == 0, "alignment");
+  static_assert(FIL_PAGE_LSN % 4 == 0, "alignment");
+  ut_ad(!memcmp_aligned<4>(lsn_start, lsn_end, 4));
 }
 
 static void buf_dblwr_check_page_lsn(const buf_page_t& b, const byte* page)

storage/innobase/buf/buf0flu.cc

@@ -831,11 +831,17 @@ buf_flush_init_for_writing(
 	}
 
 	if (use_full_checksum) {
-		memcpy(page + srv_page_size - FIL_PAGE_FCRC32_END_LSN,
-		       FIL_PAGE_LSN + 4 + page, 4);
+		static_assert(FIL_PAGE_FCRC32_END_LSN % 4 == 0, "aligned");
+		static_assert(FIL_PAGE_LSN % 4 == 0, "aligned");
+		memcpy_aligned<4>(page + srv_page_size
+				  - FIL_PAGE_FCRC32_END_LSN,
+				  FIL_PAGE_LSN + 4 + page, 4);
 	} else {
-		memcpy(page + srv_page_size - FIL_PAGE_END_LSN_OLD_CHKSUM,
-		       FIL_PAGE_LSN + page, 8);
+		static_assert(FIL_PAGE_END_LSN_OLD_CHKSUM % 8 == 0, "aligned");
+		static_assert(FIL_PAGE_LSN % 8 == 0, "aligned");
+		memcpy_aligned<8>(page + srv_page_size
+				  - FIL_PAGE_END_LSN_OLD_CHKSUM,
+				  FIL_PAGE_LSN + page, 8);
 	}
 
 	if (block && !use_full_checksum && srv_page_size == 16384) {
@@ -1030,16 +1036,21 @@ static byte* buf_page_encrypt(fil_space_t* space, buf_page_t* bpage, byte* s)
   if (!encrypted && !page_compressed)
   {
     /* No need to encrypt or compress. Clear key-version & crypt-checksum. */
+    static_assert(FIL_PAGE_FCRC32_KEY_VERSION % 4 == 0, "alignment");
+    static_assert(FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION % 4 == 2,
+                  "not perfect alignment");
     if (full_crc32)
-      memset(s + FIL_PAGE_FCRC32_KEY_VERSION, 0, 4);
+      memset_aligned<4>(s + FIL_PAGE_FCRC32_KEY_VERSION, 0, 4);
     else
-      memset(s + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0, 8);
+      memset_aligned<2>(s + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0, 8);
     return s;
   }
 
+  static_assert(FIL_PAGE_FCRC32_END_LSN % 4 == 0, "alignment");
+  static_assert(FIL_PAGE_LSN % 8 == 0, "alignment");
   if (full_crc32)
-    memcpy(s + srv_page_size - FIL_PAGE_FCRC32_END_LSN,
-           FIL_PAGE_LSN + 4 + s, 4);
+    memcpy_aligned<4>(s + srv_page_size - FIL_PAGE_FCRC32_END_LSN,
+                      FIL_PAGE_LSN + 4 + s, 4);
 
   ut_ad(!bpage->zip_size() || !page_compressed);
   buf_pool_t *buf_pool= buf_pool_from_bpage(bpage);
@@ -1101,7 +1112,7 @@ static byte* buf_page_encrypt(fil_space_t* space, buf_page_t* bpage, byte* s)
 
     if (full_crc32)
     {
-      compile_time_assert(FIL_PAGE_FCRC32_CHECKSUM == 4);
+      static_assert(FIL_PAGE_FCRC32_CHECKSUM == 4, "alignment");
       mach_write_to_4(tmp + len - 4, ut_crc32(tmp, len - 4));
       ut_ad(!buf_page_is_corrupted(true, tmp, space->flags));
     }