[v23.1.x] Removed lock contention when writing Raft snapshots

src/v/raft/consensus.cc

@@ -1244,7 +1244,10 @@ ss::future<> consensus::do_start() {
                 return _offset_translator.start(
                   must_reset, std::move(bootstrap));
             })
-            .then([this] { return hydrate_snapshot(); })
+            .then([this] {
+                return _snapshot_lock.with(
+                  [this] { return hydrate_snapshot(); });
+            })
             .then([this] {
                 vlog(
                   _ctxlog.debug,
@@ -1910,7 +1913,9 @@ consensus::install_snapshot(install_snapshot_request&& r) {
     return with_gate(_bg, [this, r = std::move(r)]() mutable {
         return _op_lock
           .with([this, r = std::move(r)]() mutable {
-              return do_install_snapshot(std::move(r));
+              return _snapshot_lock.with([this, r = std::move(r)]() mutable {
+                  return do_install_snapshot(std::move(r));
+              });
           })
           .handle_exception_type([this](const ss::broken_semaphore&) {
               return install_snapshot_reply{.term = _term, .success = false};
@@ -2111,7 +2116,7 @@ ss::future<install_snapshot_reply> consensus::finish_snapshot(
 
 ss::future<> consensus::write_snapshot(write_snapshot_cfg cfg) {
     model::offset last_included_index = cfg.last_included_index;
-    bool updated = co_await _op_lock
+    bool updated = co_await _snapshot_lock
                      .with([this, cfg = std::move(cfg)]() mutable {
                          // do nothing, we already have snapshot for this offset
                          // MUST be checked under the _op_lock
@@ -2144,16 +2149,18 @@ ss::future<> consensus::write_snapshot(write_snapshot_cfg cfg) {
     co_await _log.truncate_prefix(storage::truncate_prefix_config(
       model::next_offset(last_included_index), _scheduling.default_iopc));
 
-    co_await _op_lock.with([this, last_included_index] {
-        return _configuration_manager.prefix_truncate(last_included_index)
-          .then([this, last_included_index] {
-              return _offset_translator.prefix_truncate(last_included_index);
-          })
-          .then([this, last_included_index] {
-              // when log was prefix truncate flushed offset should be
-              // equal to at least last snapshot index
-              _flushed_offset = std::max(last_included_index, _flushed_offset);
-          });
+    /*
+     * We do not need to keep an oplock when updating the flushed offset here as
+     * it is only moved forward so there is no risk of moving the flushed offset
+     * back, also there is no risk incorrectly marking dirty batches flushed as
+     * the offset will be at most equal to log start offset
+     */
+    _flushed_offset = std::max(last_included_index, _flushed_offset);
+
+    co_await _snapshot_lock.with([this, last_included_index]() mutable {
+        return ss::when_all_succeed(
+          _configuration_manager.prefix_truncate(last_included_index),
+          _offset_translator.prefix_truncate(last_included_index));
     });
 }
 

src/v/raft/consensus.h

@@ -684,6 +684,12 @@ class consensus {
     /// all raft operations must happen exclusively since the common case
     /// is for the operation to touch the disk
     mutex _op_lock;
+    /// since snapshot state is orthogonal to raft state when writing snapshot
+    /// it is enough to grab the snapshot mutex, there is no need to keep
+    /// oplock, if the two locks are expected to be acquired at the same time
+    /// the snapshot lock should always be an internal (taken after the
+    /// _op_lock)
+    mutex _snapshot_lock;
     /// used for notifying when commits happened to log
     event_manager _event_manager;
     probe _probe;