Sulkta-OSS/pallas
3
0
Fork 0
Code Issues 1 Pull requests 4 Projects Releases Packages Wiki Activity Actions

feat: Migrate to asynchronous I/O (#241)

Browse source 
This commit updates the networking stack to use asynchronous I/O for improved performance and concurrency. We have replaced synchronous I/O calls with their asynchronous counterparts and refactored the code to use async/await and Tokio runtime.
This commit is contained in:
Santiago Carmuega 2023-04-09 13:50:56 +02:00 • committed by GitHub
parent ae7e1f5c82
commit b63d2052cb
29 changed files with 924 additions and 523 deletions
Download patch file Download diff file Expand all files Collapse all files

8
pallas-upstream/Cargo.toml
View file

@ -11,7 +11,11 @@ readme = "README.md"
authors = ["Santiago Carmuega <santiago@carmuega.me>"]
[dependencies]
byteorder = "1.4.3"
gasket = { git = "https://github.com/construkts/gasket-rs" }
# gasket = { path = "../../../construkts/gasket-rs" }
hex = "0.4.3"
mio = { version = "0.8.6", features = ["net", "os-poll"] }
# gasket = { version = "0.1.0", path = "../../../construkts/gasket-rs" }
pallas-codec = { version = "0.18.0", path = "../pallas-codec" }
pallas-crypto = { version = "0.18.0", path = "../pallas-crypto" }
@ -21,4 +25,8 @@ pallas-traverse = { version = "0.18.0", path = "../pallas-traverse" }
rayon = "1.7.0"
serde = { version = "1.0.154", features = ["derive"] }
thiserror = "1.0.31"
tokio = { version = "1", features = ["net", "macros", "io-util"] }
tracing = "0.1.37"
[dev-dependencies]
tracing-subscriber = "0.3.16"

29
pallas-upstream/src/api.rs
View file

@ -1,11 +1,8 @@
pub use crate::cursor;
pub use crate::framework::BlockFetchEvent;
pub use crate::framework::DownstreamPort;
pub use crate::framework::{BlockFetchEvent, Cursor, DownstreamPort, Intersection};
pub mod n2n {
use crate::{blockfetch, chainsync, cursor::Cursor, framework::*, plexer};
use crate::{blockfetch, chainsync, framework::*, plexer};
use gasket::{
messaging::{SendAdapter, SendPort},
runtime::Tether,
@ -17,21 +14,23 @@ pub mod n2n {
pub blockfetch_tether: Tether,
}
pub struct Bootstrapper<A>
pub struct Bootstrapper<A, C>
where
A: SendAdapter<BlockFetchEvent>,
C: Cursor,
{
cursor: Cursor,
cursor: C,
peer_address: String,
network_magic: u64,
output: super::DownstreamPort<A>,
}
impl<A> Bootstrapper<A>
impl<A, C> Bootstrapper<A, C>
where
A: SendAdapter<BlockFetchEvent> + 'static,
C: Cursor + 'static,
{
pub fn new(cursor: Cursor, peer_address: String, network_magic: u64) -> Self {
pub fn new(cursor: C, peer_address: String, network_magic: u64) -> Self {
Bootstrapper {
cursor,
peer_address,
@ -67,15 +66,15 @@ pub mod n2n {
let mut demux2_out = DemuxOutputPort::default();
let mut demux2_in = DemuxInputPort::default();
gasket::messaging::crossbeam::connect_ports(&mut demux2_out, &mut demux2_in, 1000);
gasket::messaging::tokio::connect_ports(&mut demux2_out, &mut demux2_in, 1000);
let mut demux3_out = DemuxOutputPort::default();
let mut demux3_in = DemuxInputPort::default();
gasket::messaging::crossbeam::connect_ports(&mut demux3_out, &mut demux3_in, 1000);
gasket::messaging::tokio::connect_ports(&mut demux3_out, &mut demux3_in, 1000);
let mut mux2_out = MuxOutputPort::default();
let mut mux3_out = MuxOutputPort::default();
gasket::messaging::crossbeam::funnel_ports(
gasket::messaging::tokio::funnel_ports(
vec![&mut mux2_out, &mut mux3_out],
&mut mux_input,
1000,
@ -83,10 +82,10 @@ pub mod n2n {
let mut chainsync_downstream = chainsync::DownstreamPort::default();
let mut blockfetch_upstream = blockfetch::UpstreamPort::default();
gasket::messaging::crossbeam::connect_ports(
gasket::messaging::tokio::connect_ports(
&mut chainsync_downstream,
&mut blockfetch_upstream,
20,
100,
);
let plexer_tether = gasket::runtime::spawn_stage(

37
pallas-upstream/src/blockfetch.rs
View file

@ -1,4 +1,5 @@
use gasket::messaging::SendAdapter;
use gasket::runtime::WorkSchedule;
use tracing::{error, info, instrument};
use pallas_crypto::hash::Hash;
@ -7,7 +8,7 @@ use pallas_miniprotocols::Point;
use crate::framework::*;
pub type UpstreamPort = gasket::messaging::crossbeam::TwoPhaseInputPort<ChainSyncEvent>;
pub type UpstreamPort = gasket::messaging::tokio::InputPort<ChainSyncEvent>;
pub type OuroborosClient = blockfetch::Client<ProtocolChannel>;
pub struct Worker<T>
@ -40,12 +41,17 @@ where
}
#[instrument(skip(self), fields(slot, %hash))]
fn fetch_block(&mut self, slot: u64, hash: Hash<32>) -> Result<Vec<u8>, gasket::error::Error> {
async fn fetch_block(
&mut self,
slot: u64,
hash: &Hash<32>,
) -> Result<Vec<u8>, gasket::error::Error> {
info!("fetching block");
match self
.client
.fetch_single(Point::Specific(slot, hash.to_vec()))
.await
{
Ok(x) => {
info!("block fetch succeeded");
@ -73,23 +79,28 @@ where
.build()
}
fn work(&mut self) -> gasket::runtime::WorkResult {
let msg = self.upstream.recv_or_idle()?;
type WorkUnit = ChainSyncEvent;
let msg = match msg.payload {
async fn schedule(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
let msg = self.upstream.recv().await?;
info!("scheduling block betch");
Ok(WorkSchedule::Unit(msg.payload))
}
async fn execute(&mut self, unit: &Self::WorkUnit) -> Result<(), gasket::error::Error> {
let output = match unit {
ChainSyncEvent::RollForward(s, h) => {
let body = self.fetch_block(s, h)?;
let body = self.fetch_block(*s, h).await?;
self.block_count.inc(1);
BlockFetchEvent::RollForward(s, h, body)
BlockFetchEvent::RollForward(*s, h.clone(), body)
}
ChainSyncEvent::Rollback(x) => BlockFetchEvent::Rollback(x),
ChainSyncEvent::Rollback(x) => BlockFetchEvent::Rollback(x.clone()),
};
self.downstream.send(msg.into())?;
self.downstream.send(output.into()).await?;
// remove the processed event from the queue
self.upstream.commit();
Ok(gasket::runtime::WorkOutcome::Partial)
Ok(())
}
}

159
pallas-upstream/src/chainsync.rs
View file

@ -1,11 +1,10 @@
use gasket::error::AsWorkError;
use tracing::{debug, info};
use pallas_miniprotocols::chainsync::{HeaderContent, NextResponse};
use pallas_miniprotocols::chainsync::{HeaderContent, NextResponse, Tip};
use pallas_miniprotocols::{chainsync, Point};
use pallas_traverse::MultiEraHeader;
use crate::cursor::{Cursor, Intersection};
use crate::framework::*;
fn to_traverse(header: &chainsync::HeaderContent) -> Result<MultiEraHeader<'_>, Error> {
@ -17,20 +16,26 @@ fn to_traverse(header: &chainsync::HeaderContent) -> Result<MultiEraHeader<'_>,
out.map_err(Error::parse)
}
pub type DownstreamPort = gasket::messaging::crossbeam::OutputPort<ChainSyncEvent>;
pub type DownstreamPort = gasket::messaging::tokio::OutputPort<ChainSyncEvent>;
pub type OuroborosClient = chainsync::N2NClient<ProtocolChannel>;
pub struct Worker {
chain_cursor: Cursor,
pub struct Worker<C>
where
C: Cursor,
{
chain_cursor: C,
client: OuroborosClient,
downstream: DownstreamPort,
block_count: gasket::metrics::Counter,
chain_tip: gasket::metrics::Gauge,
}
impl Worker {
pub fn new(chain_cursor: Cursor, plexer: ProtocolChannel, downstream: DownstreamPort) -> Self {
impl<C> Worker<C>
where
C: Cursor,
{
pub fn new(chain_cursor: C, plexer: ProtocolChannel, downstream: DownstreamPort) -> Self {
let client = OuroborosClient::new(plexer);
Self {
@ -42,45 +47,71 @@ impl Worker {
}
}
fn intersect(&mut self) -> Result<Option<Point>, gasket::error::Error> {
let value = self.chain_cursor.read();
match value {
Intersection::Origin => {
let point = self.client.intersect_origin().or_restart()?;
Ok(Some(point))
}
Intersection::Tip => {
let point = self.client.intersect_tip().or_restart()?;
Ok(Some(point))
}
Intersection::Breadcrumbs(points) => {
let (point, _) = self.client.find_intersect(Vec::from(points)).or_restart()?;
Ok(point)
}
}
fn notify_tip(&self, tip: Tip) {
self.chain_tip.set(tip.0.slot_or_default() as i64);
}
fn process_next(
async fn intersect(&mut self) -> Result<(), gasket::error::Error> {
let value = self.chain_cursor.intersection();
let intersect = match value {
Intersection::Origin => {
info!("intersecting origin");
self.client.intersect_origin().await.or_restart()?.into()
}
Intersection::Tip => {
info!("intersecting tip");
self.client.intersect_tip().await.or_restart()?.into()
}
Intersection::Breadcrumbs(points) => {
info!("intersecting breadcrumbs");
let (point, tip) = self
.client
.find_intersect(Vec::from(points))
.await
.or_restart()?;
self.notify_tip(tip);
point
}
};
info!(?intersect, "intersected");
Ok(())
}
async fn process_next(
&mut self,
next: NextResponse<HeaderContent>,
) -> Result<(), gasket::error::Error> {
match next {
chainsync::NextResponse::RollForward(h, t) => {
let h = to_traverse(&h).or_panic()?;
self.downstream
.send(ChainSyncEvent::RollForward(h.slot(), h.hash()).into())?;
chainsync::NextResponse::RollForward(header, tip) => {
let header = to_traverse(&header).or_panic()?;
debug!(slot = header.slot(), hash = %header.hash(), "chain sync roll forward");
self.downstream
.send(ChainSyncEvent::RollForward(header.slot(), header.hash()).into())
.await?;
self.notify_tip(tip);
debug!(slot = h.slot(), hash = %h.hash(), "chain sync roll forward");
self.chain_tip.set(t.1 as i64);
Ok(())
}
chainsync::NextResponse::RollBackward(p, t) => {
self.downstream.send(ChainSyncEvent::Rollback(p).into())?;
self.chain_tip.set(t.1 as i64);
chainsync::NextResponse::RollBackward(point, tip) => {
match &point {
Point::Origin => debug!("rollback to origin"),
Point::Specific(slot, _) => debug!(slot, "rollback"),
};
self.downstream
.send(ChainSyncEvent::Rollback(point).into())
.await?;
self.notify_tip(tip);
Ok(())
}
chainsync::NextResponse::Await => {
@ -90,20 +121,31 @@ impl Worker {
}
}
fn request_next(&mut self) -> Result<(), gasket::error::Error> {
async fn request_next(&mut self) -> Result<(), gasket::error::Error> {
info!("requesting next block");
let next = self.client.request_next().or_restart()?;
self.process_next(next)
let next = self.client.request_next().await.or_restart()?;
self.process_next(next).await
}
fn await_next(&mut self) -> Result<(), gasket::error::Error> {
async fn await_next(&mut self) -> Result<(), gasket::error::Error> {
info!("awaiting next block (blocking)");
let next = self.client.recv_while_must_reply().or_restart()?;
self.process_next(next)
let next = self.client.recv_while_must_reply().await.or_restart()?;
self.process_next(next).await
}
}
impl gasket::runtime::Worker for Worker {
pub enum WorkUnit {
Intersect,
RequestNext,
AwaitNext,
}
impl<C> gasket::runtime::Worker for Worker<C>
where
C: Cursor + Sync + Send,
{
type WorkUnit = WorkUnit;
fn metrics(&self) -> gasket::metrics::Registry {
gasket::metrics::Builder::new()
.with_counter("received_blocks", &self.block_count)
@ -111,19 +153,24 @@ impl gasket::runtime::Worker for Worker {
.build()
}
fn bootstrap(&mut self) -> Result<(), gasket::error::Error> {
let intersect = self.intersect()?;
info!(?intersect, "chain-sync intersected");
async fn bootstrap(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::Intersect))
}
async fn schedule(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
match self.client.has_agency() {
true => Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::RequestNext)),
false => Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::AwaitNext)),
}
}
async fn execute(&mut self, unit: &Self::WorkUnit) -> Result<(), gasket::error::Error> {
match unit {
WorkUnit::Intersect => self.intersect().await?,
WorkUnit::RequestNext => self.request_next().await?,
WorkUnit::AwaitNext => self.await_next().await?,
};
Ok(())
}
fn work(&mut self) -> gasket::runtime::WorkResult {
match self.client.has_agency() {
true => self.request_next()?,
false => self.await_next()?,
};
Ok(gasket::runtime::WorkOutcome::Partial)
}
}

56
pallas-upstream/src/cursor.rs
View file

@ -1,56 +0,0 @@
use std::{
collections::VecDeque,
sync::{Arc, RwLock},
};
use pallas_miniprotocols::Point;
#[derive(Clone)]
pub enum Intersection {
Tip,
Origin,
Breadcrumbs(VecDeque<Point>),
}
const HARDCODED_BREADCRUMBS: usize = 20;
// TODO: include exponential breadcrumbs logic here
#[derive(Clone)]
pub struct Cursor(Arc<RwLock<Intersection>>);
impl Cursor {
pub fn new(value: Intersection) -> Self {
Self(Arc::new(RwLock::new(value)))
}
pub fn read(&self) -> Intersection {
let v = self.0.read().unwrap();
v.clone()
}
pub fn latest_known_point(&self) -> Option<Point> {
let guard = self.0.read().unwrap();
match &*guard {
Intersection::Breadcrumbs(v) => v.front().cloned(),
_ => None,
}
}
pub fn add_breadcrumb(&self, value: Point) {
let mut guard = self.0.write().unwrap();
match &mut *guard {
Intersection::Tip | Intersection::Origin => {
*guard = Intersection::Breadcrumbs(VecDeque::from(vec![value]));
}
Intersection::Breadcrumbs(crumbs) => {
crumbs.push_front(value);
if crumbs.len() > HARDCODED_BREADCRUMBS {
crumbs.pop_back();
}
}
}
}
}

47
pallas-upstream/src/framework.rs
View file

@ -2,12 +2,23 @@ use pallas_crypto::hash::Hash;
use pallas_miniprotocols::Point;
use pallas_multiplexer as multiplexer;
use thiserror::Error;
use tracing::error;
use tracing::{error, trace};
pub type BlockSlot = u64;
pub type BlockHash = Hash<32>;
pub type RawBlock = Vec<u8>;
#[derive(Clone)]
pub enum Intersection {
Tip,
Origin,
Breadcrumbs(Vec<Point>),
}
pub trait Cursor: Send + Sync {
fn intersection(&self) -> Intersection;
}
#[derive(Debug, Clone)]
pub enum ChainSyncEvent {
RollForward(BlockSlot, BlockHash),
@ -21,12 +32,12 @@ pub enum BlockFetchEvent {
}
// ports used by plexer
pub type MuxOutputPort = gasket::messaging::crossbeam::OutputPort<(u16, multiplexer::Payload)>;
pub type DemuxInputPort = gasket::messaging::crossbeam::InputPort<multiplexer::Payload>;
pub type MuxOutputPort = gasket::messaging::tokio::OutputPort<(u16, multiplexer::Payload)>;
pub type DemuxInputPort = gasket::messaging::tokio::InputPort<multiplexer::Payload>;
// ports used by mini-protocols
pub type MuxInputPort = gasket::messaging::crossbeam::InputPort<(u16, multiplexer::Payload)>;
pub type DemuxOutputPort = gasket::messaging::crossbeam::OutputPort<multiplexer::Payload>;
pub type MuxInputPort = gasket::messaging::tokio::InputPort<(u16, multiplexer::Payload)>;
pub type DemuxOutputPort = gasket::messaging::tokio::OutputPort<multiplexer::Payload>;
// final output port
pub type DownstreamPort<A> = gasket::messaging::OutputPort<A, BlockFetchEvent>;
@ -34,14 +45,22 @@ pub type DownstreamPort<A> = gasket::messaging::OutputPort<A, BlockFetchEvent>;
pub struct ProtocolChannel(pub u16, pub MuxOutputPort, pub DemuxInputPort);
impl multiplexer::agents::Channel for ProtocolChannel {
fn enqueue_chunk(
async fn enqueue_chunk(
&mut self,
payload: multiplexer::Payload,
) -> Result<(), multiplexer::agents::ChannelError> {
match self
trace!(
protocol = self.0,
payload = hex::encode(&payload),
"enqueing"
);
let res = self
.1
.send(gasket::messaging::Message::from((self.0, payload)))
{
.await;
match res {
Ok(_) => Ok(()),
Err(error) => {
error!(?error, "enqueue chunk failed");
@ -50,8 +69,12 @@ impl multiplexer::agents::Channel for ProtocolChannel {
}
}
fn dequeue_chunk(&mut self) -> Result<multiplexer::Payload, multiplexer::agents::ChannelError> {
match self.2.recv() {
async fn dequeue_chunk(
&mut self,
) -> Result<multiplexer::Payload, multiplexer::agents::ChannelError> {
let res = self.2.recv().await;
match res {
Ok(msg) => Ok(msg.payload),
Err(error) => {
error!(?error, "dequeue chunk failed");
@ -96,8 +119,8 @@ impl Error {
Error::Message(error.to_string())
}
pub fn custom(error: Box<dyn std::error::Error>) -> Error {
Error::Custom(format!("{error}"))
pub fn custom(error: impl Into<Box<dyn std::error::Error>>) -> Error {
Error::Custom(format!("{}", error.into()))
}
}

4
pallas-upstream/src/lib.rs
View file

@ -1,10 +1,10 @@
#![feature(async_fn_in_trait)]
pub(crate) mod blockfetch;
pub(crate) mod chainsync;
pub(crate) mod framework;
pub(crate) mod plexer;
pub mod cursor;
mod api;
pub use api::*;

527
pallas-upstream/src/plexer.rs
View file

@ -1,84 +1,319 @@
use std::future::ready;
use byteorder::{ByteOrder, NetworkEndian};
use gasket::error::AsWorkError;
use tracing::{debug, error, info, warn};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf, ReadHalf, WriteHalf};
use tokio::net::{TcpStream, ToSocketAddrs};
use tokio::select;
use tokio::time::Instant;
use tracing::{debug, error, info, trace, warn};
use pallas_miniprotocols::handshake;
use pallas_multiplexer as multiplexer;
use pallas_multiplexer::bearers::Bearer;
use pallas_multiplexer::demux::{Demuxer, Egress};
use pallas_multiplexer::mux::{Ingress, Muxer};
use pallas_multiplexer::sync::SyncPlexer;
use crate::framework::*;
struct GasketEgress(DemuxOutputPort);
const HEADER_LEN: usize = 8;
impl Egress for GasketEgress {
fn send(
&mut self,
payload: multiplexer::Payload,
) -> Result<(), multiplexer::demux::EgressError> {
self.0
.send(gasket::messaging::Message::from(payload))
.map_err(|_| multiplexer::demux::EgressError(vec![]))
}
pub type Timestamp = u32;
pub type Payload = Vec<u8>;
pub type Protocol = u16;
/// A `Header` struct represents an Ouroboros segment header.
///
/// # Examples
///
/// Converting a `Header` to bytes:
///
/// ```
/// use byteorder::{BigEndian, ByteOrder};
/// use pallas_upstream::plexer::Header;
///
/// let header = Header {
/// protocol: 0x01,
/// timestamp: 1619804871,
/// payload_len: 42,
/// };
///
/// let header_bytes: [u8; 8] = header.into();
/// assert_eq!(header_bytes, [97, 75, 168, 15, 128, 1, 0, 42]);
/// ```
///
/// Converting bytes to a `Header`:
///
/// ```
/// use byteorder::{BigEndian, ByteOrder};
/// use pallas_upstream::plexer::Header;
///
/// let bytes = [97, 75, 168, 15, 128, 1, 0, 42];
/// let header: Header = (&bytes[..]).into();
///
/// assert_eq!(header.protocol, 0x01);
/// assert_eq!(header.timestamp, 1619804871);
/// assert_eq!(header.payload_len, 42);
/// ```
#[derive(Debug)]
pub struct Header {
pub protocol: Protocol,
pub timestamp: Timestamp,
pub payload_len: u16,
}
struct GasketIngress(MuxInputPort);
impl From<&[u8]> for Header {
fn from(value: &[u8]) -> Self {
let timestamp = NetworkEndian::read_u32(&value[0..4]);
let protocol = NetworkEndian::read_u16(&value[4..6]) ^ 0x8000;
let payload_len = NetworkEndian::read_u16(&value[6..8]);
impl Ingress for GasketIngress {
fn recv_timeout(
&mut self,
duration: std::time::Duration,
) -> Result<multiplexer::Message, multiplexer::mux::IngressError> {
self.0
.recv_timeout(duration)
.map(|msg| msg.payload)
.map_err(|err| match err {
gasket::error::Error::RecvIdle => multiplexer::mux::IngressError::Empty,
_ => multiplexer::mux::IngressError::Disconnected,
})
}
}
type IsBusy = bool;
fn handle_demux_outcome(
outcome: Result<multiplexer::demux::TickOutcome, multiplexer::demux::DemuxError>,
) -> Result<IsBusy, gasket::error::Error> {
match outcome {
Ok(x) => match x {
multiplexer::demux::TickOutcome::Busy => Ok(true),
multiplexer::demux::TickOutcome::Idle => Ok(false),
},
Err(err) => match err {
multiplexer::demux::DemuxError::BearerError(err) => {
error!("{}", err.kind());
Err(gasket::error::Error::ShouldRestart)
}
multiplexer::demux::DemuxError::EgressDisconnected(x, _) => {
error!(protocol = x, "egress disconnected");
Err(gasket::error::Error::WorkPanic)
}
multiplexer::demux::DemuxError::EgressUnknown(x, _) => {
error!(protocol = x, "unknown egress");
Err(gasket::error::Error::WorkPanic)
}
},
}
}
fn handle_mux_outcome(
outcome: multiplexer::mux::TickOutcome,
) -> Result<IsBusy, gasket::error::Error> {
match outcome {
multiplexer::mux::TickOutcome::Busy => Ok(true),
multiplexer::mux::TickOutcome::Idle => Ok(false),
multiplexer::mux::TickOutcome::BearerError(err) => {
warn!(%err);
Err(gasket::error::Error::ShouldRestart)
Self {
timestamp,
protocol,
payload_len,
}
multiplexer::mux::TickOutcome::IngressDisconnected => {
error!("ingress disconnected");
}
}
impl From<Header> for [u8; 8] {
fn from(value: Header) -> Self {
let mut out = [0u8; 8];
NetworkEndian::write_u32(&mut out[0..4], value.timestamp);
NetworkEndian::write_u16(&mut out[4..6], value.protocol);
NetworkEndian::write_u16(&mut out[6..8], value.payload_len);
out
}
}
pub struct Segment {
pub header: Header,
pub payload: Payload,
}
use tokio::io::{AsyncReadExt, AsyncWriteExt};
struct AsyncBearer(OwnedReadHalf, OwnedWriteHalf, Instant);
impl AsyncBearer {
async fn connect_tcp(addr: impl ToSocketAddrs) -> Result<Self, std::io::Error> {
let stream = TcpStream::connect(addr).await?;
let (read, write) = stream.into_split();
Ok(Self(read, write, Instant::now()))
}
}
impl AsyncBearer {
async fn readable(&self) -> tokio::io::Result<()> {
self.0.readable().await
}
/// Peek the available data in search for a frame header
async fn peek_header(&mut self) -> tokio::io::Result<Option<Header>> {
let mut buf = [0u8; HEADER_LEN];
let len = self.0.peek(&mut buf).await?;
if len < HEADER_LEN {
return Ok(None);
}
Ok(Some(Header::from(buf.as_slice())))
}
async fn has_payload(&mut self, payload_len: usize) -> tokio::io::Result<bool> {
let segment_size = HEADER_LEN + payload_len;
let mut buf = vec![0u8; segment_size];
let available = self.0.peek(&mut buf).await?;
return Ok(available >= segment_size);
}
/// Peeks the bearer to see if a full segment is available to be read
async fn has_segment(&mut self) -> std::io::Result<bool> {
let header = match self.peek_header().await? {
Some(x) => x,
None => return Ok(false),
};
self.has_payload(header.payload_len as usize).await
}
/// Reads a full segment from the bearer while consuming the bytes
///
/// This function is NOT "cancel safe", meaning that it shouldn't be used
/// inside the context of a select!. Only call this function once you're
/// sure that you can await until all the required bytes are available.
async fn read_segment(&mut self) -> tokio::io::Result<(Protocol, Payload)> {
let mut buf = [0u8; HEADER_LEN];
self.0.read_exact(&mut buf).await?;
let header = Header::from(buf.as_slice());
// TODO: assert any business invariants regarding timestamp from the other party
let mut payload = vec![0u8; header.payload_len as usize];
self.0.read_exact(&mut payload).await?;
Ok((header.protocol, payload))
}
async fn write_segment(&mut self, protocol: u16, payload: &[u8]) -> Result<(), std::io::Error> {
let header = Header {
protocol,
timestamp: self.2.elapsed().as_micros() as u32,
payload_len: payload.len() as u16,
};
let buf: [u8; 8] = header.into();
self.1.write_all(&buf).await?;
self.1.write_all(&payload).await?;
Ok(())
}
}
pub struct AsyncAgentChannel(
Protocol,
tokio::sync::mpsc::Sender<(Protocol, Payload)>,
tokio::sync::broadcast::Receiver<(Protocol, Payload)>,
);
impl pallas_multiplexer::agents::Channel for AsyncAgentChannel {
async fn enqueue_chunk(
&mut self,
chunk: pallas_multiplexer::Payload,
) -> Result<(), pallas_multiplexer::agents::ChannelError> {
let res = self.1.send((self.0, chunk)).await;
res.map_err(|err| pallas_multiplexer::agents::ChannelError::NotConnected(Some(err.0 .1)))
}
async fn dequeue_chunk(
&mut self,
) -> Result<pallas_multiplexer::Payload, pallas_multiplexer::agents::ChannelError> {
loop {
let (protocol, payload) = self
.2
.recv()
.await
.map_err(|err| pallas_multiplexer::agents::ChannelError::NotConnected(None))?;
if protocol == self.0 {
break Ok(payload);
}
}
}
}
pub type AsyncIngress = (
tokio::sync::mpsc::Sender<(Protocol, Payload)>,
tokio::sync::mpsc::Receiver<(Protocol, Payload)>,
);
pub type AsyncEgress = (
tokio::sync::broadcast::Sender<(Protocol, Payload)>,
tokio::sync::broadcast::Receiver<(Protocol, Payload)>,
);
struct AsyncPlexer {
bearer: AsyncBearer,
ingress: AsyncIngress,
egress: AsyncEgress,
}
impl AsyncPlexer {
pub fn new(bearer: AsyncBearer) -> Self {
Self {
bearer,
ingress: tokio::sync::mpsc::channel(100), // TODO: define buffer
egress: tokio::sync::broadcast::channel(100),
}
}
async fn mux(&mut self, msg: (Protocol, Payload)) -> tokio::io::Result<()> {
self.bearer.write_segment(msg.0, &msg.1).await?;
Ok(())
}
async fn demux(&mut self) -> tokio::io::Result<()> {
let (protocol, payload) = self.bearer.read_segment().await?;
self.egress.0.send((protocol, payload)).unwrap();
Ok(())
}
pub fn subscribe(&mut self, protocol: Protocol) -> AsyncAgentChannel {
let agent_tx = self.ingress.0.clone();
let agent_rx = self.egress.0.subscribe();
AsyncAgentChannel(protocol, agent_tx, agent_rx)
}
pub async fn run(&mut self) -> tokio::io::Result<()> {
loop {
select! {
Ok(_) = self.bearer.readable() => {
if let Ok(true) = self.bearer.has_segment().await {
trace!("demux selected");
self.demux().await?
}
},
Some(x) = self.ingress.1.recv() => {
trace!("mux selected");
self.mux(x).await?
},
}
}
}
}
impl From<AsyncBearer> for AsyncPlexer {
fn from(value: AsyncBearer) -> Self {
Self::new(value)
}
}
impl From<AsyncPlexer> for AsyncBearer {
fn from(value: AsyncPlexer) -> Self {
value.bearer
}
}
async fn handshake(
plexer: &mut AsyncPlexer,
network_magic: u64,
) -> Result<(), gasket::error::Error> {
info!("executing handshake");
let channel0 = plexer.subscribe(0);
let versions = handshake::n2n::VersionTable::v7_and_above(network_magic);
let mut client = handshake::Client::new(channel0);
//let p = tokio::spawn(plexer.run());
//let output = client.handshake(versions).or_restart()?;
let output = select! {
x = client.handshake(versions) => x.or_restart()?,
x = plexer.run() => {
match x.or_restart() {
Err(x) => return Err(x),
_ => unreachable!(),
};
},
};
debug!("handshake output: {:?}", output);
//p.abort();
match output {
handshake::Confirmation::Accepted(version, _) => {
info!(version, "connected to upstream peer");
Ok(())
}
_ => {
error!("couldn't agree on handshake version");
Err(gasket::error::Error::WorkPanic)
}
}
@ -87,11 +322,10 @@ fn handle_mux_outcome(
pub struct Worker {
peer_address: String,
network_magic: u64,
input: MuxInputPort,
bearer: Option<AsyncBearer>,
mux_input: MuxInputPort,
channel2_out: Option<DemuxOutputPort>,
channel3_out: Option<DemuxOutputPort>,
demuxer: Option<Demuxer<GasketEgress>>,
muxer: Option<Muxer<GasketIngress>>,
ops_count: gasket::metrics::Counter,
}
@ -99,48 +333,31 @@ impl Worker {
pub fn new(
peer_address: String,
network_magic: u64,
input: MuxInputPort,
mux_input: MuxInputPort,
channel2_out: Option<DemuxOutputPort>,
channel3_out: Option<DemuxOutputPort>,
) -> Self {
Self {
peer_address,
network_magic,
input,
channel2_out,
channel3_out,
demuxer: None,
muxer: None,
mux_input,
bearer: None,
ops_count: Default::default(),
}
}
fn handshake(&self, bearer: Bearer) -> Result<Bearer, gasket::error::Error> {
info!("excuting handshake");
let plexer = SyncPlexer::new(bearer, 0);
let versions = handshake::n2n::VersionTable::v7_and_above(self.network_magic);
let mut client = handshake::Client::new(plexer);
let output = client.handshake(versions).or_panic()?;
debug!("handshake output: {:?}", output);
let bearer = client.unwrap().unwrap();
match output {
handshake::Confirmation::Accepted(version, _) => {
info!(version, "connected to upstream peer");
Ok(bearer)
}
_ => {
error!("couldn't agree on handshake version");
Err(gasket::error::Error::WorkPanic)
}
}
}
}
pub enum WorkUnit {
Connect,
Mux((u16, Vec<u8>)),
Demux,
}
impl gasket::runtime::Worker for Worker {
type WorkUnit = WorkUnit;
fn metrics(&self) -> gasket::metrics::Registry {
// TODO: define networking metrics (bytes in / out, etc)
gasket::metrics::Builder::new()
@ -148,60 +365,72 @@ impl gasket::runtime::Worker for Worker {
.build()
}
fn bootstrap(&mut self) -> Result<(), gasket::error::Error> {
debug!("connecting muxer");
async fn bootstrap(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::Connect))
}
let bearer = multiplexer::bearers::Bearer::connect_tcp(&self.peer_address).or_restart()?;
let bearer = self.handshake(bearer)?;
let mut demuxer = Demuxer::new(bearer.clone());
if let Some(c2) = &self.channel2_out {
demuxer.register(2, GasketEgress(c2.clone()));
async fn schedule(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
let bearer = self.bearer.as_mut().unwrap();
trace!("selecting");
select! {
Ok(msg) = self.mux_input.recv() => { Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::Mux(msg.payload))) }
Ok(true) = bearer.has_segment() => Ok(gasket::runtime::WorkSchedule::Unit(WorkUnit::Demux)),
_ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => Ok(gasket::runtime::WorkSchedule::Idle),
}
}
if let Some(c3) = &self.channel3_out {
demuxer.register(3, GasketEgress(c3.clone()));
}
async fn execute(&mut self, unit: &Self::WorkUnit) -> Result<(), gasket::error::Error> {
match unit {
WorkUnit::Connect => {
debug!("connecting");
let bearer = AsyncBearer::connect_tcp(&self.peer_address)
.await
.or_retry()?;
self.demuxer = Some(demuxer);
let mut plexer = bearer.into();
let muxer = Muxer::new(bearer, GasketIngress(self.input.clone()));
self.muxer = Some(muxer);
handshake(&mut plexer, self.network_magic).await?;
self.bearer = Some(plexer.into());
}
WorkUnit::Mux(x) => {
trace!("muxing");
self.bearer
.as_mut()
.unwrap()
.write_segment(x.0, &x.1)
.await
.or_restart()?;
}
WorkUnit::Demux => {
trace!("demuxing");
let (protocol, payload) = self
.bearer
.as_mut()
.unwrap()
.read_segment()
.await
.or_restart()?;
match protocol {
2 => {
if let Some(channel) = &mut self.channel2_out {
channel.send(payload.into()).await?;
trace!("sent protocol 2 msg");
}
}
3 => {
if let Some(channel) = &mut self.channel3_out {
channel.send(payload.into()).await?;
trace!("sent protocol 3 msg");
}
}
x => warn!("trying to demux unexpected protocol {x}"),
}
}
};
Ok(())
}
fn work(&mut self) -> gasket::runtime::WorkResult {
let muxer = self.muxer.as_mut().unwrap();
let demuxer = self.demuxer.as_mut().unwrap();
let span = tracing::span::Span::current();
let mut mux_res = None;
let mut demux_res = None;
rayon::scope(|s| {
s.spawn(|_| {
let _guard = span.enter();
info!("mux ticking");
let outcome = muxer.tick();
mux_res = Some(handle_mux_outcome(outcome));
});
s.spawn(|_| {
let _guard = span.enter();
info!("demux ticking");
let outcome = demuxer.tick();
demux_res = Some(handle_demux_outcome(outcome));
});
});
mux_res.unwrap()?;
demux_res.unwrap()?;
self.ops_count.inc(1);
Ok(gasket::runtime::WorkOutcome::Partial)
}
}

87
pallas-upstream/tests/integration.rs Normal file
View file

@ -0,0 +1,87 @@
#![feature(async_fn_in_trait)]
use std::time::Duration;
use gasket::{
messaging::{
tokio::{InputPort, OutputPort},
RecvPort, SendPort,
},
runtime::{ScheduleResult, WorkSchedule, Worker},
};
use pallas_miniprotocols::Point;
use pallas_upstream::{BlockFetchEvent, Cursor};
use tracing::{error, info};
struct Witness {
input: InputPort<pallas_upstream::BlockFetchEvent>,
}
impl Worker for Witness {
type WorkUnit = BlockFetchEvent;
fn metrics(&self) -> gasket::metrics::Registry {
gasket::metrics::Registry::new()
}
async fn schedule(&mut self) -> gasket::runtime::ScheduleResult<Self::WorkUnit> {
error!("dequeing form witness");
let msg = self.input.recv().await?;
Ok(WorkSchedule::Unit(msg.payload))
}
async fn execute(&mut self, unit: &Self::WorkUnit) -> Result<(), gasket::error::Error> {
error!("witnessing block event");
Ok(())
}
}
struct StaticCursor;
impl Cursor for StaticCursor {
fn intersection(&self) -> pallas_upstream::Intersection {
pallas_upstream::Intersection::Origin
}
}
#[test]
fn test_mainnet_upstream() {
tracing::subscriber::set_global_default(
tracing_subscriber::FmtSubscriber::builder()
.with_max_level(tracing::Level::TRACE)
.finish(),
)
.unwrap();
let mut b = pallas_upstream::n2n::Bootstrapper::new(
StaticCursor,
"relays-new.cardano-mainnet.iohk.io:3001".into(),
764824073,
);
let (send, receive) = gasket::messaging::tokio::channel(200);
// let mut f = Faker {
// output: Default::default(),
// };
//f.output.connect(send);
b.connect_output(send);
let b = b.spawn().unwrap();
let mut w = Witness {
input: Default::default(),
};
w.input.connect(receive);
//let f = gasket::runtime::spawn_stage(f, Default::default(), Some("faker"));
let w = gasket::runtime::spawn_stage(w, Default::default(), Some("witness"));
let d = gasket::daemon::Daemon(vec![w]);
d.block();
}