Add tests for disk failure

CMakeLists.txt

@@ -56,6 +56,10 @@ option(ENABLE_JEMALLOC "Use jemalloc (default OFF)")
 option(ENABLE_LUAJIT   "Use LuaJIT (default OFF)")
 option(ENABLE_MIMALLOC "Use mimalloc (default OFF)")
 option(ENABLE_DOCS "Build docs (default OFF)")
+option(ENABLE_DISK_FAILURE_TESTS "Build disk failure tests (enables AIO fault injection, default OFF)" OFF)
+if(ENABLE_DISK_FAILURE_TESTS)
+    add_compile_definitions("AIO_FAULT_INJECTION")
+endif()
 option(ENABLE_AUTEST "Setup autest (default OFF)")
 option(ENABLE_BENCHMARKS "Build benchmarks (default OFF)")
 

configure.ac

@@ -592,6 +592,19 @@ AC_ARG_ENABLE([32bit-build],
 )
 AC_MSG_RESULT([$enable_32bit])
 
+#
+# Check if disk failure handling tests should be built.
+# This means compiling disk failure simulation code into ATS.
+#
+AC_MSG_CHECKING([whether to build disk failure handling tests])
+AC_ARG_ENABLE([disk-failure-tests],
+  [AS_HELP_STRING([--enable-disk-failure-tests],[Build disk failure tests])],
+  [],
+  [enable_disk_failure_tests=no]
+)
+AC_MSG_RESULT([$enable_disk_failure_tests])
+AM_CONDITIONAL([ENABLE_DISK_FAILURE_TESTS], [ test "x${enable_disk_failure_tests}" = "xyes" ])
+
 
 #
 # Installation directories
@@ -1103,6 +1116,12 @@ elif test "x${enable_tsan}" = "xstatic"; then
   TS_ADDTO(AM_CXXFLAGS, [-fsanitize=thread -static-libtsan])
 fi
 
+# Build for disk failure simulation
+if test "x${enable_disk_failure_tests}" = "xyes"; then
+  TS_ADDTO(AM_CFLAGS, [-DAIO_FAULT_INJECTION])
+  TS_ADDTO(AM_CXXFLAGS, [-DAIO_FAULT_INJECTION])
+fi
+
 # Checks for pointer size.
 # TODO: Later this is irrelevant, and we should just bail on 32-bit platforms always
 AC_CHECK_SIZEOF([void*])

iocore/aio/AIO.cc

@@ -32,6 +32,10 @@
 
 #include "P_AIO.h"
 
+#ifdef AIO_FAULT_INJECTION
+#include "AIO_fault_injection.h"
+#endif
+
 #define MAX_DISKS_POSSIBLE 100
 
 // globals
@@ -442,9 +446,19 @@ cache_op(AIOCallbackInternal *op)
     while (a->aio_nbytes - res > 0) {
       do {
         if (read) {
+#ifdef AIO_FAULT_INJECTION
+          err = aioFaultInjection.pread(a->aio_fildes, (static_cast<char *>(a->aio_buf)) + res, a->aio_nbytes - res,
+                                        a->aio_offset + res);
+#else
           err = pread(a->aio_fildes, (static_cast<char *>(a->aio_buf)) + res, a->aio_nbytes - res, a->aio_offset + res);
+#endif
         } else {
+#ifdef AIO_FAULT_INJECTION
+          err = aioFaultInjection.pwrite(a->aio_fildes, (static_cast<char *>(a->aio_buf)) + res, a->aio_nbytes - res,
+                                         a->aio_offset + res);
+#else
           err = pwrite(a->aio_fildes, (static_cast<char *>(a->aio_buf)) + res, a->aio_nbytes - res, a->aio_offset + res);
+#endif
         }
       } while ((err < 0) && (errno == EINTR || errno == ENOBUFS || errno == ENOMEM));
       if (err <= 0) {

iocore/aio/AIO_fault_injection.cc

@@ -0,0 +1,122 @@
+/** @file
+
+A mechanism to simulate disk failure by injecting faults in userspace.
+
+@section license License
+
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#include "AIO_fault_injection.h"
+#include "I_Lock.h"
+#include <mutex>
+
+AIOFaultInjection aioFaultInjection;
+
+void
+AIOFaultInjection::_decrement_op_count(int fd)
+{
+  auto it = _state_by_fd.find(fd);
+  if (it != _state_by_fd.end()) {
+    it->second.op_count--;
+  }
+}
+
+AIOFaultInjection::IOFault
+AIOFaultInjection::_op_result(int fd)
+{
+  auto it = _faults_by_fd.find(fd);
+  if (it != _faults_by_fd.end()) {
+    std::size_t op_count = _state_by_fd[fd].op_count++;
+    auto results         = it->second;
+    if (results.find(op_count) != results.end()) {
+      return results[op_count];
+    }
+  }
+
+  return IOFault{0, false};
+}
+
+void
+AIOFaultInjection::inject_fault(const char *path_regex, int op_index, IOFault fault)
+{
+  std::lock_guard<std::mutex> lock{_mutex};
+  _faults_by_regex[path_regex][op_index] = fault;
+}
+
+int
+AIOFaultInjection::open(const char *pathname, int flags, mode_t mode)
+{
+  std::lock_guard<std::mutex> lock{_mutex};
+  std::filesystem::path abspath = std::filesystem::absolute(pathname);
+  int fd                        = ::open(pathname, flags, mode);
+  if (fd >= 0) {
+    for (auto &[re_str, faults] : _faults_by_regex) {
+      std::regex re{re_str};
+      std::cmatch m;
+      if (std::regex_match(abspath.c_str(), m, re)) {
+        _faults_by_fd.insert_or_assign(fd, faults);
+      }
+    }
+  }
+
+  return fd;
+}
+
+ssize_t
+AIOFaultInjection::pread(int fd, void *buf, size_t nbytes, off_t offset)
+{
+  std::lock_guard<std::mutex> lock{_mutex};
+  IOFault result = _op_result(fd);
+  ssize_t ret    = 0;
+  if (result.skip_io) {
+    ink_release_assert(result.err_no != 0);
+  } else {
+    ret = ::pread(fd, buf, nbytes, offset);
+    if (ret < 0 && (errno == EINTR || errno == ENOBUFS || errno == ENOMEM)) {
+      // caller will retry
+      _decrement_op_count(fd);
+    }
+  }
+  if (result.err_no != 0) {
+    errno = result.err_no;
+    ret   = -1;
+  }
+  return ret;
+}
+
+ssize_t
+AIOFaultInjection::pwrite(int fd, const void *buf, size_t n, off_t offset)
+{
+  std::lock_guard<std::mutex> lock{_mutex};
+  IOFault result = _op_result(fd);
+  ssize_t ret    = 0;
+  if (result.skip_io) {
+    ink_release_assert(result.err_no != 0);
+  } else {
+    ret = ::pwrite(fd, buf, n, offset);
+    if (ret < 0 && (errno == EINTR || errno == ENOBUFS || errno == ENOMEM)) {
+      // caller will retry
+      _decrement_op_count(fd);
+    }
+  }
+  if (result.err_no != 0) {
+    errno = result.err_no;
+    ret   = -1;
+  }
+  return ret;
+}

iocore/aio/AIO_fault_injection.h

@@ -0,0 +1,72 @@
+/** @file
+
+  A mechanism to simulate disk failure by injecting faults in userspace.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+ */
+
+#pragma once
+
+#include "I_Continuation.h"
+#include "I_Lock.h"
+#include "tscore/ink_assert.h"
+#include "tscore/ink_mutex.h"
+#include <cerrno>
+#include <fcntl.h>
+#include <filesystem>
+#include <regex>
+#include <unistd.h>
+#include <unordered_map>
+#include <vector>
+
+// We need a way to simulate failures determininstically to test disk
+// initialization
+
+static constexpr auto TAG{"fault"};
+
+class AIOFaultInjection
+{
+  struct IOFault {
+    int err_no;
+    bool skip_io;
+  };
+  using IOFaults = std::unordered_map<int, IOFault>;
+
+  struct IOFaultState {
+    std::size_t op_count = 0;
+  };
+
+  std::unordered_map<std::string, IOFaults> _faults_by_regex;
+  std::unordered_map<int, IOFaults &> _faults_by_fd;
+  std::unordered_map<int, IOFaultState> _state_by_fd;
+
+  void _decrement_op_count(int fd);
+  IOFault _op_result(int fd);
+
+  std::mutex _mutex;
+
+public:
+  void inject_fault(const char *path_regex, int op_index, IOFault fault);
+
+  int open(const char *pathname, int flags, mode_t mode);
+  ssize_t pread(int fd, void *buf, size_t nbytes, off_t offset);
+  ssize_t pwrite(int fd, const void *buf, size_t n, off_t offset);
+};
+
+extern AIOFaultInjection aioFaultInjection;

iocore/aio/CMakeLists.txt

@@ -19,7 +19,8 @@
 add_library(aio STATIC)
 add_library(ts::aio ALIAS aio)
 
-target_sources(aio PRIVATE AIO.cc Inline.cc)
+target_sources(aio PRIVATE AIO.cc Inline.cc AIO_fault_injection.cc)
+
 target_include_directories(aio
     PUBLIC
         ${CMAKE_CURRENT_SOURCE_DIR}

iocore/aio/Makefile.am

@@ -33,7 +33,9 @@ libinkaio_a_SOURCES = \
 	AIO.cc \
 	I_AIO.h \
 	Inline.cc \
-	P_AIO.h
+	P_AIO.h \
+	AIO_fault_injection.h \
+	AIO_fault_injection.cc
 
 test_AIO_LDFLAGS = \
 	@AM_LDFLAGS@ \

iocore/cache/Cache.cc

@@ -37,6 +37,10 @@
 
 #include "tscore/hugepages.h"
 
+#ifdef AIO_FAULT_INJECTION
+#include "AIO_fault_injection.h"
+#endif
+
 #include <atomic>
 
 constexpr ts::VersionNumber CACHE_DB_VERSION(CACHE_DB_MAJOR_VERSION, CACHE_DB_MINOR_VERSION);
@@ -600,11 +604,19 @@ CacheProcessor::start_internal(int flags)
       opts |= O_RDONLY;
     }
 
-    int fd         = open(paths[gndisks], opts, 0644);
+#ifdef AIO_FAULT_INJECTION
+    int fd = aioFaultInjection.open(paths[gndisks], opts, 0644);
+#else
+    int fd = open(paths[gndisks], opts, 0644);
+#endif
     int64_t blocks = sd->blocks;
 
     if (fd < 0 && (opts & O_CREAT)) { // Try without O_DIRECT if this is a file on filesystem, e.g. tmpfs.
+#ifdef AIO_FAULT_INJECTION
+      fd = aioFaultInjection.open(paths[gndisks], DEFAULT_CACHE_OPTIONS | O_CREAT, 0644);
+#else
       fd = open(paths[gndisks], DEFAULT_CACHE_OPTIONS | O_CREAT, 0644);
+#endif
     }
 
     if (fd >= 0) {

iocore/cache/I_CacheDefs.h

@@ -24,6 +24,7 @@
 #pragma once
 
 #include "I_Event.h"
+#include "I_VConnection.h"
 #include "tscore/I_Version.h"
 #include "tscore/CryptoHash.h"