Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion lean_client/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 6 additions & 0 deletions lean_client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,9 @@ name = "lean_client"
version = "0.1.0"
edition = { workspace = true }

[features]
shadow-integration = ["xmss/shadow-integration"]

[dependencies]
anyhow = { workspace = true }
bls = { workspace = true }
Expand Down Expand Up @@ -341,3 +344,6 @@ tikv-jemallocator = { workspace = true }
[profile.release]
lto = true
codegen-units = 1

[profile.shadow]
inherits = "release"
33 changes: 33 additions & 0 deletions lean_client/Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,39 @@ docker-local: ./target/x86_64-unknown-linux-gnu/release/lean_client
$(DOCKER_TAGS) \
.

### Shadow-simulator support
# `rust/patch/quinn-udp` is a vendored fallback-only build of quinn-udp so QUIC
# runs under the Shadow simulator (which does not emulate sendmsg cmsg / GRO /
# GSO / ECN). Applied only for the shadow-* targets via cargo's --config
# override, so prod / devnet builds see the crates.io version.
SHADOW_CONFIG := --config 'patch.crates-io."quinn-udp".path="rust/patch/quinn-udp"'
SHADOW_FEATURES := --features shadow-integration

.PHONY: shadow-build
shadow-build:
cargo build --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)

.PHONY: shadow-x86_64-unknown-linux-gnu
shadow-x86_64-unknown-linux-gnu:
CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS="-C target-cpu=x86-64-v3" \
LEAN_REPO_ROOT=$(LEAN_REPO_ROOT) cross build --bin lean_client --target x86_64-unknown-linux-gnu --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)

.PHONY: shadow-aarch64-unknown-linux-gnu
shadow-aarch64-unknown-linux-gnu:
LEAN_REPO_ROOT=$(LEAN_REPO_ROOT) cross build --bin lean_client --target aarch64-unknown-linux-gnu --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)

.PHONY: shadow-docker-local
shadow-docker-local: shadow-x86_64-unknown-linux-gnu
@mkdir -p ./bin/amd64
@cp ./target/x86_64-unknown-linux-gnu/shadow/lean_client ./bin/amd64/lean_client
docker build \
--file Dockerfile \
--build-arg COMMIT_SHA=$(COMMIT_SHA) \
--build-arg BUILD_DATE=$(BUILD_DATE) \
--build-arg GIT_BRANCH=$(GIT_BRANCH) \
$(DOCKER_TAGS) \
.

.PHONY: help
help:
@awk '/^## / { \
Expand Down
3 changes: 1 addition & 2 deletions lean_client/networking/src/network/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -636,9 +636,8 @@ where
METRICS
.get()
.map(|m| m.lean_gossip_block_size_bytes.observe(data_len as f64));
info!(block_root = %signed_block.block.hash_tree_root(), "received block via gossip");

let slot = signed_block.block.slot.0;
info!(slot, block_root = %signed_block.block.hash_tree_root(), "received block via gossip");

if let Err(err) = self
.chain_message_sink
Expand Down
21 changes: 21 additions & 0 deletions lean_client/rust/patch/quinn-udp/Cargo.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
[package]
name = "quinn-udp"
version = "0.5.14"
edition = "2021"
license = "MIT OR Apache-2.0"
description = "UDP sockets for QUIC — Shadow-compatible fallback-only build"

[features]
default = ["tracing", "log"]
log = ["tracing/log"]
direct-log = ["dep:log"]

[dependencies]
log = { version = "0.4", optional = true }
tracing = { version = "0.1", optional = true }

[target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dependencies]
socket2 = { version = "0.6" }

[lib]
bench = false
126 changes: 126 additions & 0 deletions lean_client/rust/patch/quinn-udp/src/fallback.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
use std::{
io::{self, IoSliceMut},
sync::Mutex,
time::Instant,
};

use super::{IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, log_sendmsg_error};

/// Fallback UDP socket interface that stubs out all special functionality
///
/// Used when a better implementation is not available for a particular target, at the cost of
/// reduced performance compared to that enabled by some target-specific interfaces.
#[derive(Debug)]
pub struct UdpSocketState {
last_send_error: Mutex<Instant>,
}

impl UdpSocketState {
pub fn new(socket: UdpSockRef<'_>) -> io::Result<Self> {
socket.0.set_nonblocking(true)?;
let now = Instant::now();
Ok(Self {
last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
})
}

/// Sends a [`Transmit`] on the given socket.
///
/// This function will only ever return errors of kind [`io::ErrorKind::WouldBlock`].
/// All other errors will be logged and converted to `Ok`.
///
/// UDP transmission errors are considered non-fatal because higher-level protocols must
/// employ retransmits and timeouts anyway in order to deal with UDP's unreliable nature.
/// Thus, logging is most likely the only thing you can do with these errors.
///
/// If you would like to handle these errors yourself, use [`UdpSocketState::try_send`]
/// instead.
pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
match send(socket, transmit) {
Ok(()) => Ok(()),
Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
Err(e) => {
log_sendmsg_error(&self.last_send_error, e, transmit);
Ok(())
}
}
}

/// Sends a [`Transmit`] on the given socket without any additional error handling.
pub fn try_send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
send(socket, transmit)
}

pub fn recv(
&self,
socket: UdpSockRef<'_>,
bufs: &mut [IoSliceMut<'_>],
meta: &mut [RecvMeta],
) -> io::Result<usize> {
// Safety: both `IoSliceMut` and `MaybeUninitSlice` promise to have the
// same layout, that of `iovec`/`WSABUF`. Furthermore `recv_vectored`
// promises to not write unitialised bytes to the `bufs` and pass it
// directly to the `recvmsg` system call, so this is safe.
let bufs = unsafe {
&mut *(bufs as *mut [IoSliceMut<'_>] as *mut [socket2::MaybeUninitSlice<'_>])
};
let (len, _flags, addr) = socket.0.recv_from_vectored(bufs)?;
meta[0] = RecvMeta {
len,
stride: len,
addr: addr.as_socket().unwrap(),
ecn: None,
dst_ip: None,
};
Ok(1)
}

#[inline]
pub fn max_gso_segments(&self) -> usize {
1
}

#[inline]
pub fn gro_segments(&self) -> usize {
1
}

/// Resize the send buffer of `socket` to `bytes`
#[inline]
pub fn set_send_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
socket.0.set_send_buffer_size(bytes)
}

/// Resize the receive buffer of `socket` to `bytes`
#[inline]
pub fn set_recv_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
socket.0.set_recv_buffer_size(bytes)
}

/// Get the size of the `socket` send buffer
#[inline]
pub fn send_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
socket.0.send_buffer_size()
}

/// Get the size of the `socket` receive buffer
#[inline]
pub fn recv_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
socket.0.recv_buffer_size()
}

#[inline]
pub fn may_fragment(&self) -> bool {
true
}
}

fn send(socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
socket.0.send_to(
transmit.contents,
&socket2::SockAddr::from(transmit.destination),
)?;
Ok(())
}

pub(crate) const BATCH_SIZE: usize = 1;
159 changes: 159 additions & 0 deletions lean_client/rust/patch/quinn-udp/src/lib.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
//! quinn-udp — Shadow-compatible fallback-only build
//!
//! This is a vendored replacement for quinn-udp 0.5.14 that always uses the
//! simple fallback UDP path (send_to/recv_from_vectored) instead of the
//! Linux-optimized unix.rs path. This is required for Shadow network simulator
//! compatibility, since Shadow does not fully emulate sendmsg/recvmsg cmsg,
//! GRO/GSO, or ECN socket options.
//!
//! See: https://shadow.github.io/
#![warn(unreachable_pub)]
#![warn(clippy::use_self)]

use std::net::{IpAddr, Ipv6Addr, SocketAddr};
#[cfg(unix)]
use std::os::unix::io::AsFd;
#[cfg(windows)]
use std::os::windows::io::AsSocket;
use std::{
sync::Mutex,
time::{Duration, Instant},
};

// Always use fallback — no unix.rs, no windows.rs, no cmsg
#[path = "fallback.rs"]
mod imp;

#[allow(unused_imports, unused_macros)]
mod log {
#[cfg(all(feature = "direct-log", not(feature = "tracing")))]
pub(crate) use log::{debug, error, info, trace, warn};

#[cfg(feature = "tracing")]
pub(crate) use tracing::{debug, error, info, trace, warn};

#[cfg(not(any(feature = "direct-log", feature = "tracing")))]
mod no_op {
macro_rules! trace ( ($($tt:tt)*) => {{}} );
macro_rules! debug ( ($($tt:tt)*) => {{}} );
macro_rules! info ( ($($tt:tt)*) => {{}} );
macro_rules! log_warn ( ($($tt:tt)*) => {{}} );
macro_rules! error ( ($($tt:tt)*) => {{}} );

pub(crate) use {debug, error, info, log_warn as warn, trace};
}

#[cfg(not(any(feature = "direct-log", feature = "tracing")))]
pub(crate) use no_op::*;
}

pub use imp::UdpSocketState;

/// Number of UDP packets to send/receive at a time
pub const BATCH_SIZE: usize = imp::BATCH_SIZE;

/// Metadata for a single buffer filled with bytes received from the network
#[derive(Debug, Copy, Clone)]
pub struct RecvMeta {
pub addr: SocketAddr,
pub len: usize,
pub stride: usize,
pub ecn: Option<EcnCodepoint>,
pub dst_ip: Option<IpAddr>,
}

impl Default for RecvMeta {
fn default() -> Self {
Self {
addr: SocketAddr::new(Ipv6Addr::UNSPECIFIED.into(), 0),
len: 0,
stride: 0,
ecn: None,
dst_ip: None,
}
}
}

/// An outgoing packet
#[derive(Debug, Clone)]
pub struct Transmit<'a> {
pub destination: SocketAddr,
pub ecn: Option<EcnCodepoint>,
pub contents: &'a [u8],
pub segment_size: Option<usize>,
pub src_ip: Option<IpAddr>,
}

/// Log at most 1 IO error per minute
const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60);

#[cfg(any(feature = "tracing", feature = "direct-log"))]
fn log_sendmsg_error(
last_send_error: &Mutex<Instant>,
err: impl core::fmt::Debug,
transmit: &Transmit,
) {
let now = Instant::now();
let last_send_error = &mut *last_send_error.lock().expect("poisoned lock");
if now.saturating_duration_since(*last_send_error) > IO_ERROR_LOG_INTERVAL {
*last_send_error = now;
log::warn!(
"sendmsg error: {:?}, Transmit: {{ destination: {:?}, src_ip: {:?}, ecn: {:?}, len: {:?}, segment_size: {:?} }}",
err,
transmit.destination,
transmit.src_ip,
transmit.ecn,
transmit.contents.len(),
transmit.segment_size
);
}
}

#[cfg(not(any(feature = "tracing", feature = "direct-log")))]
fn log_sendmsg_error(_: &Mutex<Instant>, _: impl core::fmt::Debug, _: &Transmit) {}

/// A borrowed UDP socket
pub struct UdpSockRef<'a>(socket2::SockRef<'a>);

#[cfg(unix)]
impl<'s, S> From<&'s S> for UdpSockRef<'s>
where
S: AsFd,
{
fn from(socket: &'s S) -> Self {
Self(socket.into())
}
}

#[cfg(windows)]
impl<'s, S> From<&'s S> for UdpSockRef<'s>
where
S: AsSocket,
{
fn from(socket: &'s S) -> Self {
Self(socket.into())
}
}

/// Explicit congestion notification codepoint
#[repr(u8)]
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub enum EcnCodepoint {
Ect0 = 0b10,
Ect1 = 0b01,
Ce = 0b11,
}

impl EcnCodepoint {
pub fn from_bits(x: u8) -> Option<Self> {
use EcnCodepoint::*;
Some(match x & 0b11 {
0b10 => Ect0,
0b01 => Ect1,
0b11 => Ce,
_ => {
return None;
}
})
}
}
Loading
Loading