Parallel catchups

core/consensus/grandpa/impl/grandpa_impl.cpp

@@ -428,34 +428,34 @@ namespace kagome::consensus::grandpa {
       // Check if needed to catch-up peer, then do that
       if (msg.round_number
           >= current_round_->roundNumber() + kCatchUpThreshold) {
-        // Do catch-up only when another one is not in progress
-        if (not pending_catchup_request_.has_value()) {
+        std::lock_guard _{peer_id_catcup_mutex_};
+        // Do catch-up only when another one is not in progress for this peer
+        if (not pending_catchup_requests_.contains(peer_id)) {
           environment_->onCatchUpRequested(
               peer_id, msg.voter_set_id, msg.round_number - 1);
-          if (pending_catchup_request_.has_value()) {
-            SL_WARN(logger_,
-                    "Catch up request pending, but another one has done");
-          }
-          pending_catchup_request_.emplace(
+          pending_catchup_requests_.emplace(
               peer_id,
               network::CatchUpRequest{.round_number = msg.round_number - 1,
                                       .voter_set_id = msg.voter_set_id});
-          catchup_request_timer_handle_ = scheduler_->scheduleWithHandle(
-              [wp{weak_from_this()}] {
-                auto self = wp.lock();
-                if (not self) {
-                  return;
-                }
-                if (self->pending_catchup_request_.has_value()) {
-                  const auto &peer_id =
-                      std::get<0>(self->pending_catchup_request_.value());
-                  self->reputation_repository_->change(
-                      peer_id,
-                      network::reputation::cost::CATCH_UP_REQUEST_TIMEOUT);
-                  self->pending_catchup_request_.reset();
-                }
-              },
-              toMilliseconds(kCatchupRequestTimeout));
+          catchup_request_timer_handles_.emplace(
+              peer_id,
+              scheduler_->scheduleWithHandle(
+                  [wp{weak_from_this()}, peer_id] {
+                    auto self = wp.lock();
+                    if (not self) {
+                      return;
+                    }
+                    std::lock_guard _{self->peer_id_catcup_mutex_};
+                    self->catchup_request_timer_handles_.erase(peer_id);
+                    if (auto it = self->pending_catchup_requests_.find(peer_id);
+                        it != self->pending_catchup_requests_.end()) {
+                      self->reputation_repository_->change(
+                          peer_id,
+                          network::reputation::cost::CATCH_UP_REQUEST_TIMEOUT);
+                      self->pending_catchup_requests_.erase(it);
+                    }
+                  },
+                  toMilliseconds(kCatchupRequestTimeout)));
         }
       }
       return;
@@ -609,8 +609,10 @@ namespace kagome::consensus::grandpa {
 
     bool need_cleanup_when_exiting_scope = false;
 
+    std::lock_guard _{peer_id_catcup_mutex_};
+    auto it = pending_catchup_requests_.find(peer_id);
     if (allow_missing_blocks) {
-      if (not pending_catchup_request_.has_value()) {
+      if (it == pending_catchup_requests_.end()) {
         SL_DEBUG(logger_,
                  "Catch-up request to round #{} received from {}, "
                  "but catch-up request is not pending or timed out",
@@ -621,8 +623,7 @@ namespace kagome::consensus::grandpa {
         return;
       }
 
-      const auto &[remote_peer_id, catchup_request] =
-          pending_catchup_request_.value();
+      const auto &[remote_peer_id, catchup_request] = *it;
 
       if (peer_id != remote_peer_id) {
         SL_DEBUG(logger_,
@@ -679,8 +680,8 @@ namespace kagome::consensus::grandpa {
 
     ::libp2p::common::FinalAction cleanup([&] {
       if (need_cleanup_when_exiting_scope) {
-        catchup_request_timer_handle_.reset();
-        pending_catchup_request_.reset();
+        catchup_request_timer_handles_.erase(peer_id);
+        pending_catchup_requests_.erase(peer_id);
       }
     });
 
@@ -1409,7 +1410,7 @@ namespace kagome::consensus::grandpa {
       auto *index = vote.is<Prevote>()   ? &votes.prevote_idx
                   : vote.is<Precommit>() ? &votes.precommit_idx
                                          : nullptr;
-      if (index and not *index) {
+      if (index and not*index) {
         *index = votes.seen.size();
       }
     }

core/consensus/grandpa/impl/grandpa_impl.hpp

@@ -275,10 +275,11 @@ namespace kagome::consensus::grandpa {
     std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
 
     std::shared_ptr<VotingRound> current_round_;
-    std::optional<
-        const std::tuple<libp2p::peer::PeerId, network::CatchUpRequest>>
-        pending_catchup_request_;
-    libp2p::basic::Scheduler::Handle catchup_request_timer_handle_;
+    std::mutex peer_id_catcup_mutex_;
+    std::unordered_map<libp2p::peer::PeerId, network::CatchUpRequest>
+        pending_catchup_requests_;
+    std::unordered_map<libp2p::peer::PeerId, libp2p::basic::Scheduler::Handle>
+        catchup_request_timer_handles_;
     libp2p::basic::Scheduler::Handle fallback_timer_handle_;
 
     std::vector<WaitingBlock> waiting_blocks_;