Merge remote-tracking branch 'tiram-origin/parallel_block_submits' in…

…to tn11-compiled

kaspad/src/daemon.rs

@@ -441,7 +441,7 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
         mining_manager,
         flow_context,
         index_service.as_ref().map(|x| x.utxoindex().unwrap()),
-        config,
+        config.clone(),
         core.clone(),
         processing_counters,
         wrpc_borsh_counters.clone(),
@@ -451,7 +451,7 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
         grpc_tower_counters.clone(),
     ));
     let grpc_service =
-        Arc::new(GrpcService::new(grpc_server_addr, rpc_core_service.clone(), args.rpc_max_clients, grpc_tower_counters));
+        Arc::new(GrpcService::new(grpc_server_addr, config, rpc_core_service.clone(), args.rpc_max_clients, grpc_tower_counters));
 
     // Create an async runtime and register the top-level async services
     let async_runtime = Arc::new(AsyncRuntime::new(args.async_threads));

protocol/flows/src/flow_context.rs

@@ -381,12 +381,8 @@ impl FlowContext {
         // Broadcast as soon as the block has been validated and inserted into the DAG
         self.hub.broadcast(make_message!(Payload::InvRelayBlock, InvRelayBlockMessage { hash: Some(hash.into()) })).await;
 
-        let ctx = self.clone();
-        let consensus = consensus.clone();
-        tokio::spawn(async move {
-            ctx.on_new_block(&consensus, block, virtual_state_task).await;
-            ctx.log_block_acceptance(hash, BlockSource::Submit);
-        });
+        self.on_new_block(consensus, block, virtual_state_task).await;
+        self.log_block_acceptance(hash, BlockSource::Submit);
 
         Ok(())
     }

rpc/core/src/model/message.rs

@@ -33,13 +33,15 @@ impl SubmitBlockRequest {
 pub enum SubmitBlockRejectReason {
     BlockInvalid = 1,
     IsInIBD = 2,
+    RouteIsFull = 3,
 }
 impl SubmitBlockRejectReason {
     fn as_str(&self) -> &'static str {
         // see app\appmessage\rpc_submit_block.go, line 35
         match self {
             SubmitBlockRejectReason::BlockInvalid => "Block is invalid",
             SubmitBlockRejectReason::IsInIBD => "Node is in IBD",
+            SubmitBlockRejectReason::RouteIsFull => "Route is full",
         }
     }
 }

rpc/grpc/core/proto/rpc.proto

@@ -143,6 +143,7 @@ message SubmitBlockResponseMessage{
     NONE = 0;
     BLOCK_INVALID = 1;
     IS_IN_IBD = 2;
+    ROUTE_IS_FULL = 3;
   }
   RejectReason rejectReason = 1;
   RPCError error = 1000;

rpc/grpc/core/src/convert/message.rs

@@ -109,6 +109,7 @@ from!(item: &kaspa_rpc_core::SubmitBlockReport, RejectReason, {
         kaspa_rpc_core::SubmitBlockReport::Success => RejectReason::None,
         kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::BlockInvalid) => RejectReason::BlockInvalid,
         kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::IsInIBD) => RejectReason::IsInIbd,
+        kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::RouteIsFull) => RejectReason::RouteIsFull,
     }
 });
 
@@ -451,6 +452,7 @@ from!(item: RejectReason, kaspa_rpc_core::SubmitBlockReport, {
         RejectReason::None => kaspa_rpc_core::SubmitBlockReport::Success,
         RejectReason::BlockInvalid => kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::BlockInvalid),
         RejectReason::IsInIbd => kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::IsInIBD),
+        RejectReason::RouteIsFull => kaspa_rpc_core::SubmitBlockReport::Reject(kaspa_rpc_core::SubmitBlockRejectReason::RouteIsFull),
     }
 });
 

rpc/grpc/server/Cargo.toml

@@ -8,14 +8,15 @@ include.workspace = true
 license.workspace = true
 
 [dependencies]
+kaspa-consensus-core.workspace = true
 kaspa-core.workspace = true
 kaspa-grpc-core.workspace = true
 kaspa-notify.workspace = true
 kaspa-rpc-core.workspace = true
 kaspa-rpc-macros.workspace = true
 kaspa-rpc-service.workspace = true
-kaspa-utils.workspace = true
 kaspa-utils-tower.workspace = true
+kaspa-utils.workspace = true
 
 async-channel.workspace = true
 async-stream.workspace = true

rpc/grpc/server/src/adaptor.rs

@@ -33,13 +33,14 @@ impl Adaptor {
 
     pub fn server(
         serve_address: NetAddress,
+        network_bps: u64,
         manager: Manager,
         core_service: DynRpcService,
         core_notifier: Arc<Notifier<Notification, ChannelConnection>>,
         counters: Arc<TowerConnectionCounters>,
     ) -> Arc<Self> {
         let (manager_sender, manager_receiver) = mpsc_channel(Self::manager_channel_size());
-        let connection_handler = ConnectionHandler::new(manager_sender, core_service.clone(), core_notifier, counters);
+        let connection_handler = ConnectionHandler::new(network_bps, manager_sender, core_service.clone(), core_notifier, counters);
         let server_termination = connection_handler.serve(serve_address);
         let adaptor = Arc::new(Adaptor::new(Some(server_termination), connection_handler, manager, serve_address));
         adaptor.manager.clone().start_event_loop(manager_receiver);

rpc/grpc/server/src/connection.rs

@@ -2,8 +2,10 @@ use crate::{
     connection_handler::ServerContext,
     error::{GrpcServerError, GrpcServerResult},
     manager::ManagerEvent,
-    request_handler::{factory::Factory, interface::Interface},
+    request_handler::{factory::Factory, interface::Interface, method::RoutingPolicy},
 };
+use async_channel::{bounded, Receiver as MpmcReceiver, Sender as MpmcSender, TrySendError as MpmcTrySendError};
+use itertools::Itertools;
 use kaspa_core::{debug, info, trace, warn};
 use kaspa_grpc_core::{
     ops::KaspadPayloadOps,
@@ -12,32 +14,30 @@ use kaspa_grpc_core::{
 use kaspa_notify::{
     connection::Connection as ConnectionT, error::Error as NotificationError, listener::ListenerId, notifier::Notifier,
 };
-use kaspa_rpc_core::Notification;
+use kaspa_rpc_core::{Notification, SubmitBlockRejectReason, SubmitBlockReport, SubmitBlockResponse};
 use parking_lot::Mutex;
 use std::{
     collections::{hash_map::Entry, HashMap},
     fmt::Display,
     net::SocketAddr,
+    ops::Deref,
     sync::{
         atomic::{AtomicBool, Ordering},
         Arc,
     },
 };
-use tokio::sync::mpsc::{channel as mpsc_channel, Receiver as MpscReceiver, Sender as MpscSender};
+use tokio::sync::mpsc::Sender as MpscSender;
 use tokio::sync::oneshot::{channel as oneshot_channel, Sender as OneshotSender};
 use tokio::{select, sync::mpsc::error::TrySendError};
 use tonic::Streaming;
 use uuid::Uuid;
 
-pub type IncomingRoute = MpscReceiver<KaspadRequest>;
+pub type IncomingRoute = MpmcReceiver<KaspadRequest>;
 pub type GrpcNotifier = Notifier<Notification, Connection>;
 pub type GrpcSender = MpscSender<KaspadResponse>;
 pub type StatusResult<T> = Result<T, tonic::Status>;
 pub type ConnectionId = Uuid;
 
-type RequestSender = MpscSender<KaspadRequest>;
-type RoutingMap = HashMap<KaspadPayloadOps, RequestSender>;
-
 #[derive(Debug, Default)]
 struct InnerMutableState {
     /// Used on connection close to signal the connection receive loop to exit
@@ -85,6 +85,30 @@ impl Drop for Inner {
     }
 }
 
+type RequestSender = MpmcSender<KaspadRequest>;
+
+#[derive(Clone)]
+struct Route {
+    sender: RequestSender,
+    policy: RoutingPolicy,
+}
+
+impl Route {
+    fn new(sender: RequestSender, policy: RoutingPolicy) -> Self {
+        Self { sender, policy }
+    }
+}
+
+impl Deref for Route {
+    type Target = RequestSender;
+
+    fn deref(&self) -> &Self::Target {
+        &self.sender
+    }
+}
+
+type RoutingMap = HashMap<KaspadPayloadOps, Route>;
+
 struct Router {
     /// Routing map for mapping messages to RPC op handlers
     routing_map: RoutingMap,
@@ -101,15 +125,39 @@ impl Router {
         Self { routing_map: Default::default(), server_context, interface }
     }
 
-    fn subscribe(&mut self, connection: &Connection, rpc_op: KaspadPayloadOps) -> RequestSender {
+    fn subscribe(&mut self, connection: &Connection, rpc_op: KaspadPayloadOps) -> Route {
         match self.routing_map.entry(rpc_op) {
             Entry::Vacant(entry) => {
-                let (sender, receiver) = mpsc_channel(Connection::request_channel_size());
-                let handler = Factory::new_handler(rpc_op, receiver, self.server_context.clone(), &self.interface, connection.clone());
-                handler.launch();
-                entry.insert(sender.clone());
-                trace!("GRPC, Connection::subscribe - {:?} route is registered, client:{:?}", rpc_op, connection.identity());
-                sender
+                let method = self.interface.get_method(&rpc_op);
+                let (sender, receiver) = bounded(method.queue_size());
+                let handlers = (0..method.tasks())
+                    .map(|_| {
+                        Factory::new_handler(
+                            rpc_op,
+                            receiver.clone(),
+                            self.server_context.clone(),
+                            &self.interface,
+                            connection.clone(),
+                        )
+                    })
+                    .collect_vec();
+                handlers.into_iter().for_each(|x| x.launch());
+                let route = Route::new(sender, method.routing_policy());
+                entry.insert(route.clone());
+                match method.tasks() {
+                    1 => {
+                        trace!("GRPC, Connection::subscribe - {:?} route is registered, client:{:?}", rpc_op, connection.identity());
+                    }
+                    n => {
+                        trace!(
+                            "GRPC, Connection::subscribe - {:?} route is registered with {} workers, client:{:?}",
+                            rpc_op,
+                            n,
+                            connection.identity()
+                        );
+                    }
+                }
+                route
             }
             Entry::Occupied(entry) => entry.get().clone(),
         }
@@ -121,11 +169,27 @@ impl Router {
             return Err(GrpcServerError::InvalidRequestPayload);
         }
         let rpc_op = request.payload.as_ref().unwrap().into();
-        let sender = self.routing_map.get(&rpc_op).cloned();
-        let sender = sender.unwrap_or_else(|| self.subscribe(connection, rpc_op));
-        match sender.send(request).await {
-            Ok(_) => Ok(()),
-            Err(_) => Err(GrpcServerError::ClosedHandler(rpc_op)),
+        let route = self.routing_map.get(&rpc_op).cloned();
+        let route = route.unwrap_or_else(|| self.subscribe(connection, rpc_op));
+        match route.policy {
+            RoutingPolicy::Enqueue => match route.send(request).await {
+                Ok(_) => Ok(()),
+                Err(_) => Err(GrpcServerError::ClosedHandler(rpc_op)),
+            },
+            RoutingPolicy::DropIfFull => match route.try_send(request) {
+                Ok(_) => Ok(()),
+                Err(MpmcTrySendError::Full(_)) => match rpc_op {
+                    KaspadPayloadOps::SubmitBlock => {
+                        let response = SubmitBlockResponse { report: SubmitBlockReport::Reject(SubmitBlockRejectReason::RouteIsFull) };
+                        connection.enqueue(Ok(response).into()).await?;
+                        Ok(())
+                    }
+                    _ => {
+                        panic!("RPC op {:?} is lacking support for the DropIfFull routing policy", rpc_op);
+                    }
+                },
+                Err(MpmcTrySendError::Closed(_)) => Err(GrpcServerError::ClosedHandler(rpc_op)),
+            },
         }
     }
 

rpc/grpc/server/src/connection_handler.rs

@@ -75,6 +75,7 @@ const GRPC_SERVER: &str = "grpc-server";
 
 impl ConnectionHandler {
     pub(crate) fn new(
+        network_bps: u64,
         manager_sender: MpscSender<ManagerEvent>,
         core_service: DynRpcService,
         core_notifier: Arc<Notifier<Notification, ChannelConnection>>,
@@ -93,7 +94,7 @@ impl ConnectionHandler {
         let notifier: Arc<Notifier<Notification, Connection>> =
             Arc::new(Notifier::new(GRPC_SERVER, core_events, vec![collector], vec![subscriber], 10));
         let server_context = ServerContext::new(core_service, notifier);
-        let interface = Arc::new(Factory::new_interface(server_context.clone()));
+        let interface = Arc::new(Factory::new_interface(server_context.clone(), network_bps));
         let running = Default::default();
 
         Self { manager_sender, server_context, interface, running, counters }

rpc/grpc/server/src/request_handler/factory.rs

@@ -1,4 +1,9 @@
-use super::{handler::RequestHandler, handler_trait::Handler, interface::Interface, method::Method};
+use super::{
+    handler::RequestHandler,
+    handler_trait::Handler,
+    interface::Interface,
+    method::{Method, RoutingPolicy},
+};
 use crate::{
     connection::{Connection, IncomingRoute},
     connection_handler::ServerContext,
@@ -22,7 +27,7 @@ impl Factory {
         Box::new(RequestHandler::new(rpc_op, incoming_route, server_context, interface, connection))
     }
 
-    pub fn new_interface(server_ctx: ServerContext) -> Interface {
+    pub fn new_interface(server_ctx: ServerContext, network_bps: u64) -> Interface {
         // The array as last argument in the macro call below must exactly match the full set of
         // KaspadPayloadOps variants.
         let mut interface = build_grpc_server_interface!(
@@ -122,6 +127,10 @@ impl Factory {
             });
         interface.replace_method(KaspadPayloadOps::NotifyFinalityConflict, method);
 
+        // Methods with special properties
+        let bps = network_bps as usize;
+        interface.replace_method_properties(KaspadPayloadOps::SubmitBlock, bps, 10.max(bps * 2), RoutingPolicy::DropIfFull);
+
         interface
     }
 }

rpc/grpc/server/src/request_handler/handler.rs

@@ -45,7 +45,7 @@ impl RequestHandler {
 impl Handler for RequestHandler {
     async fn start(&mut self) {
         debug!("GRPC, Starting request handler {:?} for client {}", self.rpc_op, self.connection);
-        while let Some(request) = self.incoming_route.recv().await {
+        while let Ok(request) = self.incoming_route.recv().await {
             let response = self.handle_request(request).await;
             match response {
                 Ok(response) => {

rpc/grpc/server/src/request_handler/interface.rs

@@ -1,4 +1,4 @@
-use super::method::{Method, MethodTrait};
+use super::method::{Method, MethodTrait, RoutingPolicy};
 use crate::{
     connection::Connection,
     connection_handler::ServerContext,
@@ -55,6 +55,15 @@ impl Interface {
         let _ = self.methods.insert(op, method);
     }
 
+    pub fn replace_method_properties(&mut self, op: KaspadPayloadOps, tasks: usize, queue_size: usize, routing_policy: RoutingPolicy) {
+        self.methods.entry(op).and_modify(|x| {
+            let method: Method<ServerContext, Connection, KaspadRequest, KaspadResponse> =
+                Method::with_properties(x.method_fn(), tasks, queue_size, routing_policy);
+            let method: Arc<dyn MethodTrait<ServerContext, Connection, KaspadRequest, KaspadResponse>> = Arc::new(method);
+            *x = method;
+        });
+    }
+
     pub async fn call(
         &self,
         op: &KaspadPayloadOps,