feat(multiplexer): Allow fine-grained control of concurrency strategy

examples/block-download/src/main.rs

@@ -5,7 +5,7 @@ use pallas::network::{
         handshake::{n2n::VersionTable, Initiator},
         run_agent, Point, MAINNET_MAGIC,
     },
-    multiplexer::Multiplexer,
+    multiplexer::{spawn_demuxer, spawn_muxer, Multiplexer},
 };
 
 use pallas::network::miniprotocols::blockfetch::{BatchClient, Observer};
@@ -30,11 +30,15 @@ fn main() {
     bearer.set_nodelay(true).unwrap();
     bearer.set_keepalive_ms(Some(30_000u32)).unwrap();
 
-    let mut muxer = Multiplexer::setup(bearer, &[0, 3]).unwrap();
+    let mut plexer = Multiplexer::new(bearer);
+    let mut channel0 = plexer.use_channel(0);
+    let mut channel3 = plexer.use_channel(3);
+
+    spawn_muxer(plexer.muxer);
+    spawn_demuxer(plexer.demuxer);
 
-    let mut hs_channel = muxer.use_channel(0);
     let versions = VersionTable::v4_and_above(MAINNET_MAGIC);
-    let _last = run_agent(Initiator::initial(versions), &mut hs_channel).unwrap();
+    let _last = run_agent(Initiator::initial(versions), &mut channel0).unwrap();
 
     let range = (
         Point::Specific(
@@ -49,8 +53,7 @@ fn main() {
         ),
     );
 
-    let mut bf_channel = muxer.use_channel(3);
     let bf = BatchClient::initial(range, BlockPrinter {});
-    let bf_last = run_agent(bf, &mut bf_channel);
+    let bf_last = run_agent(bf, &mut channel3);
     println!("{:?}", bf_last);
 }

examples/n2c-miniprotocols/src/main.rs

@@ -1,6 +1,6 @@
 use pallas::network::{
     miniprotocols::{chainsync, handshake, localstate, run_agent, Point, MAINNET_MAGIC},
-    multiplexer::Multiplexer,
+    multiplexer::{spawn_demuxer, spawn_muxer, Channel, Multiplexer},
 };
 
 use std::os::unix::net::UnixStream;
@@ -45,15 +45,12 @@ impl chainsync::Observer<chainsync::HeaderContent> for LoggingObserver {
     }
 }
 
-fn do_handshake(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(0);
+fn do_handshake(mut channel: Channel) {
     let versions = handshake::n2c::VersionTable::v1_and_above(MAINNET_MAGIC);
     let _last = run_agent(handshake::Initiator::initial(versions), &mut channel).unwrap();
 }
 
-fn do_localstate_query(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(7);
-
+fn do_localstate_query(mut channel: Channel) {
     let agent = run_agent(
         localstate::OneShotClient::<localstate::queries::QueryV10>::initial(
             None,
@@ -65,9 +62,7 @@ fn do_localstate_query(muxer: &mut Multiplexer) {
     log::info!("state query result: {:?}", agent);
 }
 
-fn do_chainsync(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(5);
-
+fn do_chainsync(mut channel: Channel) {
     let known_points = vec![Point::Specific(
         43847831u64,
         hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45").unwrap(),
@@ -95,14 +90,20 @@ fn main() {
 
     // setup the multiplexer by specifying the bearer and the IDs of the
     // miniprotocols to use
-    let mut muxer = Multiplexer::setup(bearer, &[0, 4, 5]).unwrap();
+    let mut plexer = Multiplexer::new(bearer);
+    let channel0 = plexer.use_channel(0);
+    let channel7 = plexer.use_channel(7);
+    let channel5 = plexer.use_channel(5);
+
+    spawn_muxer(plexer.muxer);
+    spawn_demuxer(plexer.demuxer);
 
     // execute the required handshake against the relay
-    do_handshake(&mut muxer);
+    do_handshake(channel0);
 
     // execute an arbitrary "Local State" query against the node
-    do_localstate_query(&mut muxer);
+    do_localstate_query(channel7);
 
     // execute the chainsync flow from an arbitrary point in the chain
-    do_chainsync(&mut muxer);
+    do_chainsync(channel5);
 }

examples/n2n-miniprotocols/src/main.rs

@@ -2,7 +2,7 @@ use net2::TcpStreamExt;
 
 use pallas::network::{
     miniprotocols::{blockfetch, chainsync, handshake, run_agent, Point, MAINNET_MAGIC},
-    multiplexer::Multiplexer,
+    multiplexer::{spawn_demuxer, spawn_muxer, Channel, Multiplexer},
 };
 
 use std::net::TcpStream;
@@ -54,15 +54,12 @@ impl chainsync::Observer<chainsync::HeaderContent> for LoggingObserver {
     }
 }
 
-fn do_handshake(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(0);
+fn do_handshake(mut channel: Channel) {
     let versions = handshake::n2n::VersionTable::v4_and_above(MAINNET_MAGIC);
     let _last = run_agent(handshake::Initiator::initial(versions), &mut channel).unwrap();
 }
 
-fn do_blockfetch(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(3);
-
+fn do_blockfetch(mut channel: Channel) {
     let range = (
         Point::Specific(
             43847831,
@@ -84,9 +81,7 @@ fn do_blockfetch(muxer: &mut Multiplexer) {
     println!("{:?}", agent);
 }
 
-fn do_chainsync(muxer: &mut Multiplexer) {
-    let mut channel = muxer.use_channel(2);
-
+fn do_chainsync(mut channel: Channel) {
     let known_points = vec![Point::Specific(
         43847831u64,
         hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45").unwrap(),
@@ -116,14 +111,20 @@ fn main() {
 
     // setup the multiplexer by specifying the bearer and the IDs of the
     // miniprotocols to use
-    let mut muxer = Multiplexer::setup(bearer, &[0, 2, 3, 4]).unwrap();
+    let mut plexer = Multiplexer::new(bearer);
+    let channel0 = plexer.use_channel(0);
+    let channel3 = plexer.use_channel(3);
+    let channel2 = plexer.use_channel(2);
+
+    spawn_muxer(plexer.muxer);
+    spawn_demuxer(plexer.demuxer);
 
     // execute the required handshake against the relay
-    do_handshake(&mut muxer);
+    do_handshake(channel0);
 
     // fetch an arbitrary batch of block
-    do_blockfetch(&mut muxer);
+    do_blockfetch(channel3);
 
     // execute the chainsync flow from an arbitrary point in the chain
-    do_chainsync(&mut muxer);
+    do_chainsync(channel2);
 }

pallas-miniprotocols/src/machines.rs

@@ -147,7 +147,7 @@ where
 
             let mut payload = Vec::new();
             minicbor::encode(&msg, &mut payload)?;
-            tx.send(payload)?;
+            tx.send_payload(payload)?;
 
             agent.apply_outbound(msg)
         }

pallas-multiplexer/Cargo.toml

@@ -16,7 +16,7 @@ authors = [
 log = "0.4.14"
 byteorder = "1.4.3"
 hex = "0.4.3"
+rand = "0.8.4"
 
 [dev-dependencies]
-rand = "0.8.4"
 env_logger = "0.9.0"