MDEV-26293 InnoDB: Failing assertion: space->is_ready_to_close() ...

fil_space_t::acquire_low(): Introduce a parameter that specifies
which flags should be avoided. At all times, referenced() must not
be incremented if the STOPPING flag is set. When fil_system.mutex
is not being held by the current thread, the reference must not be
incremented if the CLOSING flag is set (unless NEEDS_FSYNC is set,
in fil_space_t::flush()).

fil_space_t::acquire(): Invoke acquire_low(STOPPING | CLOSING).
In this way, the reference count cannot be incremented after
fil_space_t::try_to_close() invoked fil_space_t::set_closing().

If the CLOSING flag was set, we must retry acquire_low() after
acquiring fil_system.mutex.

fil_space_t::prepare_acquired(): Replaces prepare(true).

fil_space_t::acquire_and_prepare(): Replaces prepare().
This basically retries fil_space_t::acquire() after
acquiring fil_system.mutex.

storage/innobase/fil/fil0crypt.cc

 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2014, 2021, MariaDB Corporation.
+Copyright (c) 2014, 2022, 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
@@ -1456,7 +1456,7 @@ inline bool fil_space_t::acquire_if_not_stopped()
     return true;
   if (UNIV_UNLIKELY(n & STOPPING))
     return false;
-  return UNIV_LIKELY(!(n & CLOSING)) || prepare(true);
+  return UNIV_LIKELY(!(n & CLOSING)) || prepare_acquired();
 }
 
 bool fil_crypt_must_default_encrypt()
@@ -1567,7 +1567,7 @@ inline fil_space_t *fil_space_t::next(fil_space_t *space, bool recheck,
       const uint32_t n= space->acquire_low();
       if (UNIV_LIKELY(!(n & (STOPPING | CLOSING))))
         break;
-      if (!(n & STOPPING) && space->prepare(true))
+      if (!(n & STOPPING) && space->prepare_acquired())
         break;
     }
   }

storage/innobase/fil/fil0fil.cc

@@ -665,11 +665,9 @@ fil_space_extend_must_retry(
 }
 
 /** @return whether the file is usable for io() */
-ATTRIBUTE_COLD bool fil_space_t::prepare(bool have_mutex)
+ATTRIBUTE_COLD bool fil_space_t::prepare_acquired()
 {
   ut_ad(referenced());
-  if (!have_mutex)
-    mutex_enter(&fil_system.mutex);
   ut_ad(mutex_own(&fil_system.mutex));
   fil_node_t *node= UT_LIST_GET_LAST(chain);
   ut_ad(!id || purpose == FIL_TYPE_TEMPORARY ||
@@ -714,8 +712,16 @@ ATTRIBUTE_COLD bool fil_space_t::prepare(bool have_mutex)
 clear:
    n_pending.fetch_and(~CLOSING, std::memory_order_relaxed);
 
-  if (!have_mutex)
-    mutex_exit(&fil_system.mutex);
+  return is_open;
+}
+
+/** @return whether the file is usable for io() */
+ATTRIBUTE_COLD bool fil_space_t::acquire_and_prepare()
+{
+  mutex_enter(&fil_system.mutex);
+  const auto flags= acquire_low() & (STOPPING | CLOSING);
+  const bool is_open= !flags || (flags == CLOSING && prepare_acquired());
+  mutex_exit(&fil_system.mutex);
   return is_open;
 }
 
@@ -1488,13 +1494,13 @@ fil_space_t *fil_space_t::get(ulint id)
   mutex_enter(&fil_system.mutex);
   fil_space_t *space= fil_space_get_by_id(id);
   const uint32_t n= space ? space->acquire_low() : 0;
-  mutex_exit(&fil_system.mutex);
 
   if (n & STOPPING)
     space= nullptr;
-  else if ((n & CLOSING) && !space->prepare())
+  else if ((n & CLOSING) && !space->prepare_acquired())
     space= nullptr;
 
+  mutex_exit(&fil_system.mutex);
   return space;
 }
 

storage/innobase/include/fil0fil.h

 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2021, MariaDB Corporation.
+Copyright (c) 2013, 2022, 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
@@ -507,15 +507,16 @@ struct fil_space_t final
 
 private:
   MY_ATTRIBUTE((warn_unused_result))
-  /** Try to acquire a tablespace reference.
-  @return the old reference count (if STOPPING is set, it was not acquired) */
-  uint32_t acquire_low()
+  /** Try to acquire a tablespace reference (increment referenced()).
+  @param avoid   when these flags are set, nothing will be acquired
+  @return the old reference count */
+  uint32_t acquire_low(uint32_t avoid= STOPPING)
   {
     uint32_t n= 0;
     while (!n_pending.compare_exchange_strong(n, n + 1,
                                               std::memory_order_acquire,
                                               std::memory_order_relaxed) &&
-           !(n & STOPPING));
+           !(n & avoid));
     return n;
   }
 public:
@@ -529,10 +530,8 @@ struct fil_space_t final
   @return whether the file is usable */
   bool acquire()
   {
-    uint32_t n= acquire_low();
-    if (UNIV_LIKELY(!(n & (STOPPING | CLOSING))))
-      return true;
-    return UNIV_LIKELY(!(n & STOPPING)) && prepare();
+    const auto flags= acquire_low(STOPPING | CLOSING) & (STOPPING | CLOSING);
+    return UNIV_LIKELY(!flags) || (flags == CLOSING && acquire_and_prepare());
   }
 
   /** Acquire another tablespace reference for I/O. */
@@ -999,7 +998,9 @@ struct fil_space_t final
 
 private:
   /** @return whether the file is usable for io() */
-  ATTRIBUTE_COLD bool prepare(bool have_mutex= false);
+  ATTRIBUTE_COLD bool prepare_acquired();
+  /** @return whether the file is usable for io() */
+  ATTRIBUTE_COLD bool acquire_and_prepare();
 #endif /*!UNIV_INNOCHECKSUM */
 };