MDEV-21351: Free processed recv_sys_t::blocks

Release memory as soon as redo log records are processed.

Because the memory allocation and deallocation of parsed redo log
records must be protected by recv_sys.mutex, it is better to avoid
using a std::atomic field for bookkeeping.

buf_page_t::access_time: Keep track of the recv_sys.pages record
allocations. The most significant 16 bits will count allocated
blocks (which were previously counted by buf_page_t::buf_fix_count
in the debug version), and the least significant 16 bits indicate
the number of allocated bytes in the block (which was previously
managed in buf_block_t::modify_clock), which must be a positive
number, up to innodb_page_size. The byte offset 65536 is represented
as the value 0.

recv_recover_page(): Let the caller erase the log.

recv_validate_tablespace(): Acquire recv_sys_t::mutex.

storage/innobase/include/buf0buf.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, 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
@@ -1568,7 +1568,17 @@ class buf_page_t {
 	unsigned	access_time;	/*!< time of first access, or
 					0 if the block was never accessed
 					in the buffer pool. Protected by
-					block mutex */
+					block mutex for buf_page_in_file()
+					blocks.
+
+					For state==BUF_BLOCK_MEMORY
+					blocks, this field can be repurposed
+					for something else.
+
+					When this field counts log records
+					and bytes allocated for recv_sys.pages,
+					the field is protected by
+					recv_sys_t::mutex. */
 # ifdef UNIV_DEBUG
 	ibool		file_page_was_freed;
 					/*!< this is set to TRUE when

storage/innobase/include/log0recv.h

@@ -196,9 +196,9 @@ struct page_recv_t
       tail= recs;
     }
 
-    /** Trim old log records for a page
+    /** Trim old log records for a page.
     @param start_lsn oldest log sequence number to preserve
-    @return whether the entire log was trimmed */
+    @return whether all the log for the page was trimmed */
     inline bool trim(lsn_t start_lsn);
     /** @return the last log snippet */
     const log_rec_t* last() const { return tail; }
@@ -215,11 +215,8 @@ struct page_recv_t
     iterator begin() { return head; }
     iterator end() { return NULL; }
     bool empty() const { ut_ad(!head == !tail); return !head; }
+    /** Clear and free the records; @see recv_sys_t::alloc() */
     inline void clear();
-#ifdef UNIV_DEBUG
-    /** Declare the records as freed; @see recv_sys_t::alloc() */
-    inline void free() const;
-#endif
   } log;
 
   /** Ignore any earlier redo log records for this page. */
@@ -282,32 +279,32 @@ struct recv_sys_t{
 	/** the time when progress was last reported */
 	time_t		progress_time;
 
-	using map = std::map<const page_id_t, page_recv_t,
-			     std::less<const page_id_t>,
-			     ut_allocator
-			     <std::pair<const page_id_t, page_recv_t>>>;
-	/** buffered records waiting to be applied to pages */
-	map pages;
-
-	/** Process a record that indicates that a tablespace is
-	being shrunk in size.
-	@param page_id	first page identifier that is not in the file
-	@param lsn	log sequence number of the shrink operation */
-	inline void trim(const page_id_t page_id, lsn_t lsn);
-
-	/** Undo tablespaces for which truncate has been logged
-	(indexed by id - srv_undo_space_id_start) */
-	struct trunc {
-		/** log sequence number of MLOG_FILE_CREATE2, or 0 if none */
-		lsn_t		lsn;
-		/** truncated size of the tablespace, or 0 if not truncated */
-		unsigned	pages;
-	} truncated_undo_spaces[127];
-
-	recv_dblwr_t	dblwr;
-
-	/** Last added LSN to pages. */
-	lsn_t		last_stored_lsn;
+  using map = std::map<const page_id_t, page_recv_t,
+                       std::less<const page_id_t>,
+                       ut_allocator<std::pair<const page_id_t, page_recv_t>>>;
+  /** buffered records waiting to be applied to pages */
+  map pages;
+
+  /** Process a record that indicates that a tablespace size is being shrunk.
+  @param page_id first page that is not in the file
+  @param lsn     log sequence number of the shrink operation */
+  inline void trim(const page_id_t page_id, lsn_t lsn);
+
+  /** Undo tablespaces for which truncate has been logged
+  (indexed by page_id_t::space() - srv_undo_space_id_start) */
+  struct trunc
+  {
+    /** log sequence number of MLOG_FILE_CREATE2, or 0 if none */
+    lsn_t lsn;
+    /** truncated size of the tablespace, or 0 if not truncated */
+    unsigned pages;
+  } truncated_undo_spaces[127];
+
+  /** The contents of the doublewrite buffer */
+  recv_dblwr_t dblwr;
+
+  /** Last added LSN to pages. */
+  lsn_t last_stored_lsn;
 
 private:
   /** Maximum number of buffer pool blocks to allocate for redo log records */
@@ -374,17 +371,9 @@ struct recv_sys_t{
   @return pointer to len bytes of memory (never NULL) */
   inline byte *alloc(size_t len, bool store_recv= false);
 
-#ifdef UNIV_DEBUG
-private:
-  /** Find the buffer pool block that is storing a redo log record.
-  @param[in] data   pointer to buffer returned by alloc()
-  @return redo list element */
-  inline buf_block_t *find_block(const void *data) const;
-public:
-  /** Declare a redo log record freed from a buffer pool block.
-  @param[in] data   pointer to buffer returned by alloc() */
-  inline void free(const void *data) const;
-#endif
+  /** Free a redo log snippet.
+  @param data buffer returned by alloc() */
+  inline void free(const void *data);
 
   /** @return the free length of the latest alloc() block, in bytes */
   inline size_t get_free_len() const;