raptorcast: expose priority api for publishing to validators

this change doesn't use priority api yet, also we never publish
to secondary nodes with priority, therefore no similar api is
added for them

Author: dshulyak

Cargo.lock

@@ -9,6 +9,8 @@ edition = "2021"
 bench = false
 
 [dependencies]
+monad-types = { workspace = true }
+
 bytes = { workspace = true }
 env_logger = { workspace = true }
 futures = { workspace = true }

monad-dataplane/Cargo.toml

@@ -27,6 +27,7 @@ use std::{
 use addrlist::Addrlist;
 use bytes::Bytes;
 use futures::channel::oneshot;
+use monad_types::UdpPriority;
 use monoio::{spawn, time::Instant, IoUringDriver, RuntimeBuilder};
 use tcp::{TcpConfig, TcpControl, TcpRateLimit};
 use tokio::sync::mpsc::{self, error::TrySendError};
@@ -40,13 +41,6 @@ pub mod udp;
 
 pub(crate) use udp::UdpMessageType;
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-#[repr(usize)]
-pub enum UdpPriority {
-    High = 0,
-    Regular = 1,
-}
-
 pub struct DataplaneBuilder {
     local_addr: SocketAddr,
     trusted_addresses: Vec<IpAddr>,

monad-dataplane/src/lib.rs

@@ -27,7 +27,7 @@ use monoio::{net::udp::UdpSocket, spawn, time};
 use tokio::sync::mpsc;
 use tracing::{debug, error, trace, warn};
 
-use super::{RecvUdpMsg, UdpMsg, UdpPriority};
+use super::{RecvUdpMsg, UdpMsg};
 use crate::buffer_ext::SocketBufferExt;
 
 #[derive(Debug, Clone, Copy)]
@@ -36,18 +36,6 @@ pub(crate) enum UdpMessageType {
     Direct,
 }
 
-impl TryFrom<usize> for UdpPriority {
-    type Error = &'static str;
-
-    fn try_from(value: usize) -> Result<Self, Self::Error> {
-        match value {
-            0 => Ok(UdpPriority::High),
-            1 => Ok(UdpPriority::Regular),
-            _ => Err("invalid priority index"),
-        }
-    }
-}
-
 struct PriorityQueues {
     queues: [VecDeque<UdpMsg>; 2],
 }

monad-dataplane/src/udp.rs

@@ -26,8 +26,9 @@ use futures::{channel::oneshot, executor, FutureExt};
 use monad_dataplane::{
     tcp::tx::{MSG_WAIT_TIMEOUT, QUEUED_MESSAGE_LIMIT},
     udp::DEFAULT_SEGMENT_SIZE,
-    BroadcastMsg, DataplaneBuilder, RecvUdpMsg, TcpMsg, UdpPriority, UnicastMsg,
+    BroadcastMsg, DataplaneBuilder, RecvUdpMsg, TcpMsg, UnicastMsg,
 };
+use monad_types::UdpPriority;
 use ntest::timeout;
 use rand::Rng;
 use rstest::*;

monad-dataplane/tests/tests.rs

@@ -60,6 +60,12 @@ pub enum RouterCommand<ST: CertificateSignatureRecoverable, OM> {
         target: RouterTarget<CertificateSignaturePubKey<ST>>,
         message: OM,
     },
+    PublishWithPriority {
+        // NOTE(dshulyak) priority for tcp messages is ignored
+        target: RouterTarget<CertificateSignaturePubKey<ST>>,
+        message: OM,
+        priority: monad_types::UdpPriority,
+    },
     PublishToFullNodes {
         epoch: Epoch, // Epoch gets embedded into the raptorcast message
         message: OM,
@@ -87,6 +93,15 @@ impl<ST: CertificateSignatureRecoverable, OM> Debug for RouterCommand<ST, OM> {
             Self::Publish { target, message: _ } => {
                 f.debug_struct("Publish").field("target", target).finish()
             }
+            Self::PublishWithPriority {
+                target,
+                message: _,
+                priority,
+            } => f
+                .debug_struct("PublishWithPriority")
+                .field("target", target)
+                .field("priority", priority)
+                .finish(),
             Self::PublishToFullNodes { epoch, message: _ } => f
                 .debug_struct("PublishToFullNodes")
                 .field("epoch", epoch)

monad-executor-glue/src/lib.rs

@@ -400,6 +400,13 @@ impl<S: SwarmRelation> Executor for MockExecutor<S> {
                 RouterCommand::PublishToFullNodes { .. } => {
                     // TODO
                 }
+                RouterCommand::PublishWithPriority {
+                    target,
+                    message,
+                    priority: _,
+                } => {
+                    self.router.send_outbound(self.tick, target, message);
+                }
             }
         }
     }

monad-mock-swarm/src/mock.rs

@@ -148,6 +148,11 @@ impl<S: PeerDiscSwarmRelation> Executor for MockPeerDiscExecutor<S> {
                 RouterCommand::Publish { target, message } => {
                     self.router.send_outbound(self.tick, target, message)
                 }
+                RouterCommand::PublishWithPriority {
+                    target,
+                    message,
+                    priority: _,
+                } => self.router.send_outbound(self.tick, target, message),
                 RouterCommand::AddEpochValidatorSet { .. } => {}
                 RouterCommand::UpdateCurrentRound(..) => {}
                 RouterCommand::GetPeers => {}