Do snapshot fsync on UV threadpool (#7035)

Author: eddyashton

src/host/test/ledger.cpp

@@ -8,6 +8,7 @@
 #include "ds/files.h"
 #include "ds/serialized.h"
 #include "kv/serialised_entry_format.h"
+#define TEST_MODE_EXECUTE_SYNC_INLINE
 #include "snapshots/snapshot_manager.h"
 
 #define DOCTEST_CONFIG_IMPLEMENT

src/snapshots/snapshot_manager.h

@@ -81,6 +81,62 @@ namespace snapshots
       return snapshot;
     }
 
+#define THROW_ON_ERROR(x, name) \
+  do \
+  { \
+    auto rc = x; \
+    if (rc == -1) \
+    { \
+      throw std::runtime_error(fmt::format( \
+        "Error ({}) writing snapshot {} in " #x, strerror(errno), name)); \
+    } \
+  } while (0)
+
+    struct AsyncSnapshotSyncAndRename
+    {
+      // Inputs, populated at construction
+      const std::filesystem::path dir;
+      const std::string tmp_file_name;
+      const int snapshot_fd;
+
+      // Outputs, populated by callback
+      std::string committed_file_name = {};
+    };
+
+    static void on_snapshot_sync_and_rename(uv_work_t* req)
+    {
+      auto data = static_cast<AsyncSnapshotSyncAndRename*>(req->data);
+
+      {
+        asynchost::TimeBoundLogger log_if_slow(
+          fmt::format("Committing snapshot - fsync({})", data->tmp_file_name));
+        fsync(data->snapshot_fd);
+      }
+
+      close(data->snapshot_fd);
+
+      // e.g. snapshot_100_105.committed
+      data->committed_file_name =
+        fmt::format("{}{}", data->tmp_file_name, snapshot_committed_suffix);
+      const auto full_committed_path = data->dir / data->committed_file_name;
+
+      const auto full_tmp_path = data->dir / data->tmp_file_name;
+      files::rename(full_tmp_path, full_committed_path);
+    }
+
+    static void on_snapshot_sync_and_rename_complete(uv_work_t* req, int status)
+    {
+      auto data = static_cast<AsyncSnapshotSyncAndRename*>(req->data);
+
+      LOG_INFO_FMT(
+        "Renamed temporary snapshot {} to {}",
+        data->tmp_file_name,
+        data->committed_file_name);
+
+      delete data;
+      delete req;
+    }
+
     void commit_snapshot(
       ::consensus::Index snapshot_idx,
       const uint8_t* receipt_data,
@@ -130,42 +186,40 @@ namespace snapshots
             {
               const auto& snapshot = it->second.snapshot;
 
-#define THROW_ON_ERROR(x) \
-  do \
-  { \
-    auto rc = x; \
-    if (rc == -1) \
-    { \
-      throw std::runtime_error(fmt::format( \
-        "Error ({}) writing snapshot {} in " #x, errno, file_name)); \
-    } \
-  } while (0)
-
               THROW_ON_ERROR(
-                write(snapshot_fd, snapshot->data(), snapshot->size()));
-              THROW_ON_ERROR(write(snapshot_fd, receipt_data, receipt_size));
-
-              THROW_ON_ERROR(fsync(snapshot_fd));
-              THROW_ON_ERROR(close(snapshot_fd));
-
-#undef THROW_ON_ERROR
+                write(snapshot_fd, snapshot->data(), snapshot->size()),
+                file_name);
+              THROW_ON_ERROR(
+                write(snapshot_fd, receipt_data, receipt_size), file_name);
 
               LOG_INFO_FMT(
-                "New snapshot file written to {} [{} bytes]",
+                "New snapshot file written to {} [{} bytes] (unsynced)",
                 file_name,
                 snapshot->size() + receipt_size);
 
-              // e.g. snapshot_100_105.committed
-              const auto committed_file_name =
-                fmt::format("{}{}", file_name, snapshot_committed_suffix);
-              const auto full_committed_path =
-                snapshot_dir / committed_file_name;
+              // Call fsync and rename on a worker-thread via uv async, as they
+              // may be slow
+              uv_work_t* work_handle = new uv_work_t;
 
-              files::rename(full_snapshot_path, full_committed_path);
-              LOG_INFO_FMT(
-                "Renamed temporary snapshot {} to committed {}",
-                file_name,
-                committed_file_name);
+              {
+                auto* data = new AsyncSnapshotSyncAndRename{
+                  .dir = snapshot_dir,
+                  .tmp_file_name = file_name,
+                  .snapshot_fd = snapshot_fd};
+
+                work_handle->data = data;
+              }
+
+#ifdef TEST_MODE_EXECUTE_SYNC_INLINE
+              on_snapshot_sync_and_rename(work_handle);
+              on_snapshot_sync_and_rename_complete(work_handle, 0);
+#else
+              uv_queue_work(
+                uv_default_loop(),
+                work_handle,
+                &on_snapshot_sync_and_rename,
+                &on_snapshot_sync_and_rename_complete);
+#endif
             }
 
             pending_snapshots.erase(it);
@@ -184,6 +238,7 @@ namespace snapshots
           e.what());
       }
     }
+#undef THROW_ON_ERROR
 
     std::optional<std::pair<fs::path, fs::path>>
     find_latest_committed_snapshot()