diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index bc880de38..1de0f7bd1 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -317,7 +317,7 @@ jobs: -i mxl_build_container_with_source \ bash -c " cd /workspace/mxl/rust && \ - cargo build --release --all-targets --locked + cargo build --release --features mxl-fabrics-ofi --all-targets --locked " - name: Test the Rust bindings @@ -336,7 +336,7 @@ jobs: -i mxl_build_container_with_source \ bash -c " cd /workspace/mxl/rust && \ - cargo test --release --all-targets --locked -- --test-threads=1 --show-output + cargo test --release --all-targets --features mxl-fabrics-ofi --locked -- --test-threads=1 --show-output " 2>&1 | tee test.log status=${PIPESTATUS[0]} set -e @@ -354,3 +354,4 @@ jobs: } >> "$GITHUB_STEP_SUMMARY" fi exit "${status}" + diff --git a/rust/.gitignore b/rust/.gitignore index 91712988d..a07dfa212 100644 --- a/rust/.gitignore +++ b/rust/.gitignore @@ -4,4 +4,4 @@ .DS_Store .idea .vscode -target +/target diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 58fd4e0ea..1961ff16c 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -48,6 +48,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bindgen" version = "0.72.1" @@ -84,6 +90,15 @@ dependencies = [ "no_std_io2", ] +[[package]] +name = "block2" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5" +dependencies = [ + "objc2", +] + [[package]] name = "bumpalo" version = "3.19.1" @@ -147,6 +162,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + [[package]] name = "chrono" version = "0.4.42" @@ -222,6 +243,29 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "ctrlc" +version = "3.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0b1fab2ae45819af2d0731d60f2afe17227ebb1a1538a236da84c93e9a60162" +dependencies = [ + "dispatch2", + "nix", + "windows-sys", +] + +[[package]] +name = "dispatch2" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" +dependencies = [ + "bitflags", + "block2", + "libc", + "objc2", +] + [[package]] name = "either" version = "1.15.0" @@ -686,9 +730,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "libc" -version = "0.2.178" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37c93d8daa9d8a012fd8ab92f088405fb202ea0b6ab73ee2482ae66af4f42091" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libloading" @@ -737,7 +781,9 @@ checksum = "956787520e75e9bd233246045d19f42fb73242759cc57fba9611d940ae96d4b0" name = "mxl" version = "0.1.0" dependencies = [ + "base64", "clap", + "ctrlc", "libloading", "mxl-sys", "serde", @@ -757,6 +803,18 @@ dependencies = [ "libloading", ] +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "no_std_io2" version = "0.9.3" @@ -813,6 +871,21 @@ dependencies = [ "autocfg", ] +[[package]] +name = "objc2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a12a8ed07aefc768292f076dc3ac8c48f3781c8f2d5851dd3d98950e8c5a89f" +dependencies = [ + "objc2-encode", +] + +[[package]] +name = "objc2-encode" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" + [[package]] name = "once_cell" version = "1.21.3" diff --git a/rust/mxl-sys/Cargo.toml b/rust/mxl-sys/Cargo.toml index 0fe4555c2..ddd01476d 100644 --- a/rust/mxl-sys/Cargo.toml +++ b/rust/mxl-sys/Cargo.toml @@ -17,3 +17,4 @@ cmake = "0.1.54" [features] mxl-not-built = [] +mxl-fabrics-ofi = [] diff --git a/rust/mxl-sys/build.rs b/rust/mxl-sys/build.rs index b3a2774b0..6581105fa 100644 --- a/rust/mxl-sys/build.rs +++ b/rust/mxl-sys/build.rs @@ -63,6 +63,61 @@ fn get_bindgen_specs() -> BindgenSpecs { } } +fn get_bindgen_specs_fabrics() -> BindgenSpecs { + let header = "wrapper-fabrics.h".to_string(); + + let manifest_dir = + PathBuf::from(env::var("CARGO_MANIFEST_DIR").expect("failed to get current directory")); + let repo_root = manifest_dir.parent().unwrap().parent().unwrap(); + let mut includes_dirs = vec![ + repo_root + .join("lib") + .join("include") + .to_string_lossy() + .to_string(), + repo_root + .join("lib") + .join("fabrics") + .join("include") + .to_string_lossy() + .to_string(), + ]; + if cfg!(not(feature = "mxl-not-built")) { + let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap()); + let build_version_dir = out_dir.join("include").to_string_lossy().to_string(); + + includes_dirs.push(build_version_dir); + + // Rebuild if any file in lib/ changes + let lib_root = repo_root.join("lib"); + println!("cargo:rerun-if-changed={}", lib_root.display()); + + let dst = cmake::Config::new(repo_root) + .generator("Ninja") + .configure_arg("--preset") + .configure_arg(BUILD_VARIANT) + .configure_arg("-B") + .configure_arg(out_dir.join("build")) + .define("MXL_ENABLE_FABRICS_OFI", "ON") + .define("BUILD_DOCS", "OFF") + .define("BUILD_TESTS", "OFF") + .define("BUILD_TOOLS", "OFF") + .build(); + + println!( + "cargo:rustc-link-search={}", + dst.join("lib/fabrics/ofi").display() + ); + println!("cargo:rustc-link-lib=mxl"); + println!("cargo:rustc-link-lib=mxl-fabrics"); + } + + BindgenSpecs { + header, + includes_dirs, + } +} + fn main() { let bindgen_specs = get_bindgen_specs(); for include_dir in &bindgen_specs.includes_dirs { @@ -90,6 +145,28 @@ fn main() { bindings .write_to_file(out_path.join("bindings.rs")) .expect("Could not write bindings"); + + if cfg!(feature = "mxl-fabrics-ofi") { + let fabrics_bindings = bindgen::builder() + .clang_args( + get_bindgen_specs_fabrics() + .includes_dirs + .iter() + .map(|dir| format!("-I{dir}")), + ) + .header("wrapper-fabrics.h") + .derive_default(true) + .derive_debug(true) + .prepend_enum_name(false) + .dynamic_library_name("libmxlfabrics") + .dynamic_link_require_all(false) + .parse_callbacks(Box::new(CB)) + .generate() + .unwrap(); + fabrics_bindings + .write_to_file(out_path.join("fabrics_bindings.rs")) + .expect("Could not write fabrics bindings"); + } } #[derive(Debug)] diff --git a/rust/mxl-sys/src/lib.rs b/rust/mxl-sys/src/lib.rs index 86c46c47d..020045fa7 100644 --- a/rust/mxl-sys/src/lib.rs +++ b/rust/mxl-sys/src/lib.rs @@ -17,3 +17,8 @@ extern crate libloading; include!(concat!(env!("OUT_DIR"), "/bindings.rs")); + +#[cfg(feature = "mxl-fabrics-ofi")] +pub mod fabrics { + include!(concat!(env!("OUT_DIR"), "/fabrics_bindings.rs")); +} diff --git a/rust/mxl-sys/wrapper-fabrics.h b/rust/mxl-sys/wrapper-fabrics.h new file mode 100644 index 000000000..2e74e3cff --- /dev/null +++ b/rust/mxl-sys/wrapper-fabrics.h @@ -0,0 +1,4 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +#include "mxl/fabrics.h" diff --git a/rust/mxl/Cargo.toml b/rust/mxl/Cargo.toml index 951770af0..91a853f55 100644 --- a/rust/mxl/Cargo.toml +++ b/rust/mxl/Cargo.toml @@ -21,6 +21,13 @@ serde.workspace = true clap.workspace = true serde_json.workspace = true tracing-subscriber.workspace = true +ctrlc = "3" +base64 = "0.22" [features] mxl-not-built = ["mxl-sys/mxl-not-built"] +mxl-fabrics-ofi = ["mxl-sys/mxl-fabrics-ofi"] + +[[example]] +name = "fabrics-demo" +required-features = ["mxl-fabrics-ofi"] diff --git a/rust/mxl/examples/fabrics-demo.rs b/rust/mxl/examples/fabrics-demo.rs new file mode 100644 index 000000000..ef443113b --- /dev/null +++ b/rust/mxl/examples/fabrics-demo.rs @@ -0,0 +1,530 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +mod common; + +use std::{ + sync::{ + Arc, + atomic::{self, AtomicBool}, + }, + time::Duration, +}; + +use clap::{Parser, Subcommand}; + +use base64::{Engine as _, prelude::BASE64_STANDARD}; + +use mxl::{ + Error, FlowConfigInfo, FlowInfo, FlowReader, FlowWriter, GrainReader, GrainWriter, MxlInstance, + SamplesReader, SamplesWriter, + config::{get_mxl_fabrics_ofi_so_path, get_mxl_so_path}, + fabrics::{ + Capabilities, EndpointAddress, FabricsInstance, InterfaceConfig, ProviderType, TargetInfo, + initiator::{self, Initiator}, + target::{self, Target}, + }, +}; + +#[derive(Debug, Parser)] +#[command( + version = clap::crate_version!(), + author = clap::crate_authors!(), + subcommand_required = true, + arg_required_else_help = true +)] +pub struct Cli { + #[arg(short, long, help = "The MXL domain directory")] + pub domain: String, + + #[arg( + short, + long, + help = "Force a specific provider (tcp, verbs, efa or shm). Auto-selected if not specified." + )] + pub provider: Option, + + #[arg( + short, + long, + help = "Filter interface selection by node address. If not set, the best available interface is chosen automatically." + )] + pub node: Option, + + #[arg( + short, + long, + help = "Service identifier for the fabrics endpoint (e.g. a port number)." + )] + pub service: Option, + + #[command(subcommand)] + pub command: Command, +} + +#[derive(Debug, Subcommand)] +pub enum Command { + /// Run as a receiver (fabrics target + flow writer). + Target { + #[arg( + long, + help = "The JSON file which contains the NMOS Flow configuration." + )] + flow_file: String, + #[arg( + long, + help = "Output file path for raw target info (optional, always logged as base64)." + )] + target_info_path: Option, + //TODO: flow options? + }, + /// Run as an initiator (flow reader + fabrics initiator). + Initiator { + #[arg(long, help = "The flow ID to read from.")] + flow_id: String, + #[arg( + long, + help = "Base64-encoded target info, or a path prefixed with '@' to read raw target info from a file." + )] + target_info: String, + }, +} + +struct TargetEndpoint<'a> { + _instance: &'a MxlInstance, + flow_config: FlowConfigInfo, + flow_writer: FlowWriter, + target: Target, +} + +impl<'a> TargetEndpoint<'a> { + pub fn new( + instance: &'a MxlInstance, + fabrics_instance: &FabricsInstance, + interface: InterfaceConfig, + flow_file: &str, + ) -> Result<(Self, TargetInfo), mxl::Error> { + let flow_config_str = std::fs::read_to_string(flow_file).expect("Failed to read flow file"); + + let (flow_writer, flow_config, _) = instance.create_flow_writer(&flow_config_str, None)?; + + let target_config = target::Config::new(interface, &flow_writer); + + let target = fabrics_instance.create_target()?; + let (target, target_info) = target.setup(&target_config)?; + + Ok(( + Self { + _instance: instance, + flow_config, + flow_writer, + target, + }, + target_info, + )) + } + + pub fn run(self, running: Arc) -> Result<(), mxl::Error> { + match self.target.specialize(&self.flow_config) { + target::Either::Grain(target) => { + Self::run_discrete(target, self.flow_writer.to_grain_writer()?, running)?; + } + target::Either::Sample(target) => { + Self::run_continuous(target, self.flow_writer.to_samples_writer()?, running)?; + } + } + Ok(()) + } + + fn run_discrete( + target: Target, + writer: GrainWriter, + running: Arc, + ) -> Result<(), mxl::Error> { + while running.load(atomic::Ordering::SeqCst) { + match target.read(Duration::from_millis(200)) { + Ok(read_result) => { + let grain = writer.open_grain(read_result.grain_index)?; + let valid_slices = grain.valid_slices(); + grain.commit(valid_slices)?; + + tracing::debug!( + "Commited grain index {}, slice index {}.", + read_result.grain_index, + valid_slices + ); + } + Err(mxl::Error::NotReady) => { + continue; + } + Err(mxl::Error::Interrupted) => { + tracing::info!("Interrupted, exiting."); + break; + } + Err(e) => { + return Err(e); + } + } + } + Ok(()) + } + + fn run_continuous( + target: Target, + writer: SamplesWriter, + running: Arc, + ) -> Result<(), mxl::Error> { + while running.load(atomic::Ordering::SeqCst) { + match target.read(Duration::from_millis(200)) { + Ok(read_result) => { + let samples = writer.open_samples(read_result.head_index, read_result.count)?; + samples.commit()?; + + tracing::debug!( + "Commited samples, head index {}, count {}.", + read_result.head_index, + read_result.count + ); + } + Err(mxl::Error::NotReady) => { + continue; + } + Err(mxl::Error::Interrupted) => { + tracing::info!("Interrupted, exiting."); + break; + } + Err(e) => { + return Err(e); + } + } + } + Ok(()) + } +} + +struct InitiatorEndpoint<'a> { + instance: &'a MxlInstance, + fabrics_instance: FabricsInstance, + flow_reader: FlowReader, + initiator: Initiator, +} + +impl<'a> InitiatorEndpoint<'a> { + pub fn new( + instance: &'a MxlInstance, + fabrics_instance: FabricsInstance, + interface: InterfaceConfig, + flow_id: &str, + ) -> Result { + let flow_reader = instance.create_flow_reader(flow_id)?; + let initiator = fabrics_instance.create_initiator()?; + let initiator_config = initiator::Config::new(interface, &flow_reader); + let initiator = initiator.setup(&initiator_config)?; + + Ok(Self { + instance, + fabrics_instance, + initiator, + flow_reader, + }) + } + + pub fn run(self, target_info_str: &str, running: Arc) -> Result<(), mxl::Error> { + let flow_info = self.flow_reader.get_info()?; + + let target_info_str = + String::from_utf8(BASE64_STANDARD.decode(target_info_str).map_err(|e| { + Error::Other(format!("Failed to decode target_info from base64: {e}")) + })?) + .map_err(|e| Error::Other(format!("Decoded target_info is not valid UTF-8: {e}")))?; + + let target_info = self + .fabrics_instance + .target_info_from_str(&target_info_str)?; + + match self.initiator.specialize(&flow_info.config) { + initiator::Either::Grain(initiator) => { + initiator.add_target(&target_info)?; + // Wait to be connected + loop { + if !running.load(atomic::Ordering::SeqCst) { + return Ok(()); + } + + if initiator.make_progress(Duration::from_millis(250)).is_ok() { + break; + } + } + Self::run_discrete( + self.instance, + initiator, + self.flow_reader.to_grain_reader()?, + &flow_info, + running, + )?; + } + initiator::Either::Samples(initiator) => { + initiator.add_target(&target_info)?; + // Wait to be connected + loop { + if !running.load(atomic::Ordering::SeqCst) { + return Ok(()); + } + + if initiator.make_progress(Duration::from_millis(250)).is_ok() { + break; + } + } + Self::run_continuous( + self.instance, + initiator, + self.flow_reader.to_samples_reader()?, + &flow_info, + running, + )?; + } + } + + tracing::info!("Stopping as requested."); + + Ok(()) + } + + fn run_discrete( + instance: &MxlInstance, + initiator: Initiator, + reader: GrainReader, + flow_info: &FlowInfo, + running: Arc, + ) -> Result<(), mxl::Error> { + let rate = flow_info.config.common().grain_rate()?; + let mut index = instance.get_current_index(&rate); + while running.load(atomic::Ordering::SeqCst) { + match reader.get_complete_grain(index, Duration::from_millis(200)) { + Ok(grain) => { + match initiator.transfer(index, 0, grain.total_slices) { + Err(Error::NotReady) => { + // Retry the same grain + continue; + } + Err(e) => { + return Err(e); + } + Ok(_) => {} + }; + + // Transfer was posted, now wait for completion + loop { + match initiator.make_progress(Duration::from_millis(10)) { + Ok(_) => { + // we're done exiting the loop + break; + } + Err(Error::Interrupted) => { + return Ok(()); + } + Err(Error::NotReady) => { + // Retry + continue; + } + Err(e) => { + return Err(e); + } + } + } + index += 1; + } + Err(Error::OutOfRangeTooLate) => { + // We are too late, move to the next grain + index = instance.get_current_index(&rate); + } + Err(Error::OutOfRangeTooEarly) => { + // We are too early, retry the same grain + } + Err(e) => { + tracing::error!("Error reading from flow: {}.", e); + } + } + } + Ok(()) + } + + fn run_continuous( + instance: &MxlInstance, + initiator: Initiator, + reader: SamplesReader, + flow_info: &FlowInfo, + running: Arc, + ) -> Result<(), mxl::Error> { + let rate = flow_info.config.common().grain_rate()?; + let count = flow_info.config.common().max_sync_batch_size_hint() as usize; + let mut index = instance.get_current_index(&rate); + while running.load(atomic::Ordering::SeqCst) { + match reader.get_samples_non_blocking(index, count) { + Ok(_sample) => { + match initiator.transfer(index, count) { + Err(Error::NotReady) => { + // Retry the same grain + continue; + } + Err(e) => { + return Err(e); + } + Ok(_) => {} + }; + + // Transfer was posted, now wait for completion + loop { + match initiator.make_progress(Duration::from_millis(10)) { + Ok(_) => { + // we're done exiting the loop + break; + } + Err(Error::Interrupted) => { + return Ok(()); + } + Err(Error::NotReady) => { + // Retry + continue; + } + Err(e) => { + return Err(e); + } + } + } + index += 1; + } + Err(Error::OutOfRangeTooLate) => { + // We are too late, move to the next grain + index = instance.get_current_index(&rate); + } + Err(Error::OutOfRangeTooEarly) => { + // We are too early, retry the same grain + } + Err(e) => { + tracing::error!("Error reading from flow: {}.", e); + } + } + } + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ProviderPrio(ProviderType); +impl ProviderPrio { + fn priority(&self) -> u32 { + match self { + ProviderPrio(ProviderType::Efa) => 4, + ProviderPrio(ProviderType::Verbs) => 3, + ProviderPrio(ProviderType::Tcp) => 2, + ProviderPrio(ProviderType::Shm) => 1, + _ => 0, + } + } +} +impl Ord for ProviderPrio { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.priority().cmp(&other.priority()) + } +} +impl PartialOrd for ProviderPrio { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +fn main() -> Result<(), mxl::Error> { + common::setup_logging(); + + let cli = Cli::parse(); + + tracing::info!(domain = %cli.domain, provider = ?cli.provider, node = ?cli.node, service = ?cli.service, command = ?cli.command, "Starting fabrics demo"); + + let running = Arc::new(AtomicBool::new(true)); + let running2 = running.clone(); + ctrlc::set_handler(move || { + running2.store(false, atomic::Ordering::SeqCst); + }) + .expect("Error setting Ctrl-C handler"); + + let api = mxl::load_api(get_mxl_so_path())?; + + let fabrics_api = mxl::load_fabrics_api(get_mxl_fabrics_ofi_so_path())?; + + let instance = mxl::MxlInstance::new(api, &cli.domain, "")?; + + let fabrics_instance = instance.create_fabrics_instance(&fabrics_api)?; + + let endpoint_address = EndpointAddress { + node: cli.node.as_deref(), + service: cli.service.as_deref(), + }; + + let provider = + fabrics_instance.provider_from_str(&cli.provider.unwrap_or("any".to_string()))?; + + let interface_config = mxl::fabrics::InterfaceConfig::builder(endpoint_address) + .provider(provider.prov_type().clone()) + .caps(Capabilities::default()) + .build(); + + let interfaces = fabrics_instance + .get_interfaces(Some(interface_config)) + .expect("Failed to get interfaces"); + + let mut interface = interfaces + .iter() + .max_by_key(|k| ProviderPrio(k.provider.clone())) + .ok_or(Error::Other("No suitable interface found".to_string()))?; + interface.set_endpoint_address(EndpointAddress { + node: cli.node.as_deref(), + service: cli.service.as_deref(), + }); + + tracing::info!( + provider = ?interface.provider, + node = ?interface.endpoint_address.node, + service = ?interface.endpoint_address.service, + caps = ?interface.caps, + "Selected interface"); + + match &cli.command { + Command::Initiator { + flow_id, + target_info, + } => { + let initiator = + InitiatorEndpoint::new(&instance, fabrics_instance, interface, flow_id)?; + initiator.run(target_info, running)?; + } + Command::Target { + flow_file, + target_info_path, + } => { + let (target, target_info) = + TargetEndpoint::new(&instance, &fabrics_instance, interface, flow_file)?; + + if let Some(target_info_file) = target_info_path + && target_info_file.starts_with('@') + { + let file = if target_info_file.starts_with("@") { + target_info_file + .strip_prefix('@') + .expect("impossible to fail.") //SAFETY: we already checked that the string starts with '@'; + } else { + target_info_file.as_str() + }; + std::fs::write(file, target_info.to_string()?.as_bytes()) + .map_err(|e| mxl::Error::Other(format!("fail to write to file: {}", e)))?; + } + tracing::info!( + "Target Info: {}", + BASE64_STANDARD.encode(target_info.to_string()?) + ); + target.run(running)?; + } + } + + Ok(()) +} diff --git a/rust/mxl/src/api.rs b/rust/mxl/src/api.rs index 57418f8f1..018c82e25 100644 --- a/rust/mxl/src/api.rs +++ b/rust/mxl/src/api.rs @@ -15,3 +15,15 @@ pub fn load_api(path_to_so_file: impl AsRef) -> Result { libmxl::new(path_to_so_file.as_ref().as_os_str())? })) } + +#[cfg(feature = "mxl-fabrics-ofi")] +pub type MxlFabricsApi = mxl_sys::fabrics::libmxlfabrics; +#[cfg(feature = "mxl-fabrics-ofi")] +pub type MxlFabricsAPiHandle = Arc; + +#[cfg(feature = "mxl-fabrics-ofi")] +pub fn load_fabrics_api(path_to_so_file: impl AsRef) -> Result { + Ok(Arc::new(unsafe { + mxl_sys::fabrics::libmxlfabrics::new(path_to_so_file.as_ref().as_os_str())? + })) +} diff --git a/rust/mxl/src/config.rs b/rust/mxl/src/config.rs index b80fd0c2b..e39836903 100644 --- a/rust/mxl/src/config.rs +++ b/rust/mxl/src/config.rs @@ -20,6 +20,23 @@ pub fn get_mxl_so_path() -> std::path::PathBuf { .join("libmxl.so") } +#[cfg(not(feature = "mxl-not-built"))] +pub fn get_mxl_fabrics_ofi_so_path() -> std::path::PathBuf { + // The mxl-sys build script ensures that the build directory is in the library path + // so we can just return the library name here. + "libmxl-fabrics.so".into() +} + +#[cfg(feature = "mxl-not-built")] +pub fn get_mxl_fabrics_ofi_so_path() -> std::path::PathBuf { + std::path::PathBuf::from_str(MXL_BUILD_DIR) + .expect("build error: 'MXL_FABRICS_SO_PATH' is invalid") + .join("lib") + .join("fabrics") + .join("ofi") + .join("libmxl-fabrics.so") +} + pub fn get_mxl_repo_root() -> std::path::PathBuf { std::path::PathBuf::from_str(MXL_REPO_ROOT).expect("build error: 'MXL_REPO_ROOT' is invalid") } diff --git a/rust/mxl/src/error.rs b/rust/mxl/src/error.rs index 4b6aa4ce2..bb1309400 100644 --- a/rust/mxl/src/error.rs +++ b/rust/mxl/src/error.rs @@ -23,6 +23,12 @@ pub enum Error { InvalidArg, #[error("Conflict")] Conflict, + #[error("Not ready")] + NotReady, + #[error("Not found")] + NotFound, + #[error("Interrupted")] + Interrupted, /// The error is not defined in the MXL API, but it is used to wrap other errors. #[error("Other error: {0}")] Other(String), @@ -47,6 +53,9 @@ impl Error { mxl_sys::MXL_ERR_TIMEOUT => Err(Error::Timeout), mxl_sys::MXL_ERR_INVALID_ARG => Err(Error::InvalidArg), mxl_sys::MXL_ERR_CONFLICT => Err(Error::Conflict), + mxl_sys::MXL_ERR_NOT_READY => Err(Error::NotReady), + mxl_sys::MXL_ERR_NOT_FOUND => Err(Error::NotFound), + mxl_sys::MXL_ERR_INTERRUPTED => Err(Error::Interrupted), other => Err(Error::Unknown(other)), } } diff --git a/rust/mxl/src/fabrics/capabilities.rs b/rust/mxl/src/fabrics/capabilities.rs new file mode 100644 index 000000000..784557607 --- /dev/null +++ b/rust/mxl/src/fabrics/capabilities.rs @@ -0,0 +1,138 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +/// Capabilities of the fabric interface +#[derive(Debug)] +pub struct Capabilities { + blocking_operations: bool, + remote_write: bool, + send_recv: bool, + + max_message_size: u64, +} +impl Capabilities { + pub fn builder() -> CapabilitiesBuilder { + CapabilitiesBuilder::default() + } + + /// The interface supports blocking operations. + pub fn supports_blocking_operations(&self) -> bool { + self.blocking_operations + } + + /// The interface supports remote-write (RDMA) operations. + pub fn supports_remote_write(&self) -> bool { + self.remote_write + } + + /// The interface supports send/receive message operations. + pub fn supports_send_recv(&self) -> bool { + self.send_recv + } + + /// Maximum message size supported on this interface. + pub fn max_message_size(&self) -> u64 { + self.max_message_size + } +} +impl Default for Capabilities { + fn default() -> Self { + Self { + blocking_operations: true, + remote_write: true, + send_recv: false, + max_message_size: u64::MAX, + } + } +} +impl From<&Capabilities> for mxl_sys::fabrics::FabricsInterfaceCaps { + fn from(value: &Capabilities) -> Self { + let flags = (if value.blocking_operations { + mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_BLOCKING_OPERATIONS as u64 + } else { + 0 + }) | (if value.remote_write { + mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_REMOTE_WRITE as u64 + } else { + 0 + }) | (if value.send_recv { + mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_SEND_RECEIVE as u64 + } else { + 0 + }); + + Self { + version: mxl_sys::fabrics::MXL_FABRICS_API_VERSION as i32, + flags, + maxMessageSize: value.max_message_size, + } + } +} +impl From for mxl_sys::fabrics::FabricsInterfaceCaps { + fn from(value: Capabilities) -> Self { + (&value).into() + } +} +impl From for Capabilities { + fn from(value: mxl_sys::fabrics::FabricsInterfaceCaps) -> Self { + let flags = value.flags; + Self { + blocking_operations: (flags + & mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_BLOCKING_OPERATIONS as u64) + != 0, + remote_write: (flags & mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_REMOTE_WRITE as u64) + != 0, + send_recv: (flags & mxl_sys::fabrics::MXL_FABRICS_IFACE_CAP_SEND_RECEIVE as u64) != 0, + max_message_size: value.maxMessageSize, + } + } +} + +pub struct CapabilitiesBuilder { + // flags for the capabilities of the fabric + blocking_operations: bool, + remote_write: bool, + send_recv: bool, + + max_message_size: u64, +} +impl CapabilitiesBuilder { + pub fn with_blocking_operations(mut self, value: bool) -> Self { + self.blocking_operations = value; + self + } + + pub fn with_remote_write(mut self, value: bool) -> Self { + self.remote_write = value; + self + } + + pub fn with_send_recv(mut self, value: bool) -> Self { + self.send_recv = value; + self + } + + pub fn max_message_size(mut self, value: u64) -> Self { + self.max_message_size = value; + self + } + + pub fn build(self) -> Capabilities { + Capabilities { + blocking_operations: self.blocking_operations, + remote_write: self.remote_write, + send_recv: self.send_recv, + max_message_size: self.max_message_size, + } + } +} +impl Default for CapabilitiesBuilder { + fn default() -> Self { + Self { + blocking_operations: true, + remote_write: true, + send_recv: false, + max_message_size: u64::MAX, + } + } +} diff --git a/rust/mxl/src/fabrics/endpoint_address.rs b/rust/mxl/src/fabrics/endpoint_address.rs new file mode 100644 index 000000000..1dbd55f49 --- /dev/null +++ b/rust/mxl/src/fabrics/endpoint_address.rs @@ -0,0 +1,46 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::ffi::CString; + +use crate::Error; + +/// Address of a logical network endpoint. This is analogous to a hostname and port number in classic ipv4 networking. +/// The actual values for node and service vary between providers, but often an ip address as the node value and a port number as the service +/// value are sufficient. +#[derive(Debug)] +pub struct EndpointAddress<'a> { + pub node: Option<&'a str>, + pub service: Option<&'a str>, +} + +/// A wrapper around the FFI representation of an EndpointAddress, which owns the underlying CStrings for node and service. +pub(crate) struct OwnedEndpointAddress { + inner: mxl_sys::fabrics::FabricsEndpointAddress, + _node: Option, + _service: Option, +} + +impl OwnedEndpointAddress { + pub(crate) fn new(value: &EndpointAddress<'_>) -> Result { + let node = value.node.map(CString::new).transpose()?; + let service = value.service.map(CString::new).transpose()?; + + Ok(Self { + inner: mxl_sys::fabrics::FabricsEndpointAddress { + node: node + .as_ref() + .map_or(std::ptr::null_mut(), |value| value.as_ptr() as *mut i8), + service: service + .as_ref() + .map_or(std::ptr::null_mut(), |value| value.as_ptr() as *mut i8), + }, + _node: node, + _service: service, + }) + } + + pub(crate) fn as_ffi(&self) -> mxl_sys::fabrics::FabricsEndpointAddress { + self.inner + } +} diff --git a/rust/mxl/src/fabrics/initiator/config.rs b/rust/mxl/src/fabrics/initiator/config.rs new file mode 100644 index 000000000..b77fc6518 --- /dev/null +++ b/rust/mxl/src/fabrics/initiator/config.rs @@ -0,0 +1,62 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ + Error, FlowReader, + fabrics::{InterfaceConfig, interface::config::OwnedInterfaceConfig}, +}; + +/// Configuration object required to set up an initiator. +pub struct Config<'a> { + version: i32, + interface: InterfaceConfig<'a>, + pub(crate) flow_reader: &'a FlowReader, +} + +impl<'a> Config<'a> { + pub fn new(interface: InterfaceConfig<'a>, flow_reader: &'a FlowReader) -> Self { + Self { + version: 0, + interface, + flow_reader, + } + } +} + +impl<'a> TryFrom<&Config<'a>> for mxl_sys::fabrics::FabricsInitiatorConfig { + type Error = Error; + + fn try_from(value: &Config) -> Result { + Ok(Self { + version: value.version, + interface: *OwnedInterfaceConfig::try_from(&value.interface)?.as_ffi(), + // SAFETY: Both types are equivalent opaque reader handles from different bindgen modules. + reader: value.flow_reader.inner().cast(), + }) + } +} + +pub(crate) struct OwnedInitiatorConfig { + inner: mxl_sys::fabrics::FabricsInitiatorConfig, + _interface: OwnedInterfaceConfig, +} + +impl OwnedInitiatorConfig { + pub(crate) fn new(value: &Config<'_>) -> Result { + let interface = OwnedInterfaceConfig::new(&value.interface)?; + + Ok(Self { + inner: mxl_sys::fabrics::FabricsInitiatorConfig { + version: value.version, + interface: *interface.as_ffi(), + // SAFETY: Both types are equivalent opaque reader handles from different bindgen modules. + reader: value.flow_reader.inner().cast(), + }, + _interface: interface, + }) + } + + pub(crate) fn as_ffi(&self) -> &mxl_sys::fabrics::FabricsInitiatorConfig { + &self.inner + } +} diff --git a/rust/mxl/src/fabrics/initiator/grain.rs b/rust/mxl/src/fabrics/initiator/grain.rs new file mode 100644 index 000000000..c5ba7cecf --- /dev/null +++ b/rust/mxl/src/fabrics/initiator/grain.rs @@ -0,0 +1,74 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::time::Duration; + +use crate::{ + Error, Result, + fabrics::{TargetInfo, initiator::Grain, initiator::Initiator}, +}; + +impl Initiator { + /// Add a target to the initiator. This will allow the initiator to send data to the target in subsequent calls to + /// mxlFabricsInitiatorTransferGrain(). This function is always non-blocking. If additional connection setup is required + /// by the underlying implementation, it will only happen during a call to make_progress*(). + pub fn add_target(&self, target: &TargetInfo) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_add_target(self.instance.inner, target.inner) + }) + } + + /// Remove a target from the initiator. This function is always non-blocking. If any additional communication for a graceful shutdown is + /// required it will happend during a call to make_progress*(). It is guaranteed that no new grain transfer operations will + /// be queued for this target during calls to transfer() after the target was removed, but it is only guaranteed that + /// the connection shutdown has completed after make_progress*() no longer returns Error::NotReady. + pub fn remove_target(&self, target: &TargetInfo) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_remove_target(self.instance.inner, target.inner) + }) + } + + /// This function must be called regularly for the initiator to make progress on queued transfer operations, connection establishment + /// operations and connection shutdown operations. + pub fn make_progress_non_blocking(&self) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_make_progress_non_blocking(self.instance.inner) + }) + } + + /// This function must be called regularly for the initiator to make progress on queued transfer operations, connection establishment + /// operations and connection shutdown operations. + pub fn make_progress(&self, timeout: Duration) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_make_progress_blocking( + self.instance.inner, + timeout.as_millis() as u16, + ) + }) + } + + /// Enqueue a transfer operation to all added targets. This function is always non-blocking. The transfer operation might be started right + /// away, but is only guaranteed to have completed after mxlFabricsInitiatorMakeProgress*() no longer returns Error::NotReady. + pub fn transfer(&self, grain_index: u64, start_slice: u16, end_slice: u16) -> Result<()> { + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_initiator_transfer_grain( + self.instance.inner, + grain_index, + start_slice, + end_slice, + ) + }) + } +} diff --git a/rust/mxl/src/fabrics/initiator/mod.rs b/rust/mxl/src/fabrics/initiator/mod.rs new file mode 100644 index 000000000..7586bc320 --- /dev/null +++ b/rust/mxl/src/fabrics/initiator/mod.rs @@ -0,0 +1,148 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +mod config; +mod grain; +mod samples; + +use crate::{ + FlowConfigInfo, + error::{Error, Result}, + fabrics::{initiator::config::OwnedInitiatorConfig, instance::FabricsInstanceContext}, +}; + +pub use config::Config; + +use std::{marker::PhantomData, sync::Arc}; + +use states::*; + +pub mod states { + /// Used to create a new initiator + pub struct New {} + + /// Waiting for the initiator to be initialized with the setup function + pub struct Initializing {} + + /// The setup function has been called, but the initiator has not yet been specialized into a + /// grain or samples initiator + pub struct Specializing {} + + /// The initiator has been specialized into a grain initiator. It can only transfer grains to + /// targets. + pub struct Grain {} + + /// The initiator has been specialized into a samples initiator. It can only transfer samples to + pub struct Samples {} + + impl InitiatorState for New {} + impl InitiatorState for Initializing {} + impl InitiatorState for Specializing {} + impl InitiatorState for Grain {} + impl InitiatorState for Samples {} + + pub trait InitiatorState {} +} + +/// Wrapper class that holds a reference count to the Fabrics Instance and the actual initiator instance. +struct InitiatorInstance { + ctx: Arc, + inner: mxl_sys::fabrics::FabricsInitiator, +} +unsafe impl Send for InitiatorInstance {} + +impl Drop for InitiatorInstance { + fn drop(&mut self) { + if !self.inner.is_null() { + unsafe { + self.ctx + .api() + .fabrics_destroy_initiator(self.ctx.inner, self.inner); + } + } + } +} + +pub struct Initiator { + instance: InitiatorInstance, + _marker: std::marker::PhantomData, +} +//SAFETY: An initiator is safe to be sent across threads, but it's not thread-safe to use its API functions. +unsafe impl Send for Initiator {} + +pub enum Either { + Grain(Initiator), + Samples(Initiator), +} + +impl Initiator { + /// Create a new initiator + pub(crate) fn new( + ctx: Arc, + initiator: mxl_sys::fabrics::FabricsInitiator, + ) -> Initiator { + let instance = InitiatorInstance { + ctx, + inner: initiator, + }; + Initiator { + instance, + _marker: std::marker::PhantomData, + } + } +} + +impl Initiator { + /// Configure the initiator. + pub fn setup(self, config: &Config) -> Result> { + let config = OwnedInitiatorConfig::new(config)?; + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_initiator_setup( + self.instance.inner, + config.as_ffi(), + std::ptr::null(), // Unused for now + ) + })?; + Ok(Initiator { + instance: self.instance, + _marker: PhantomData, + }) + } +} + +impl Initiator { + /// Specialize the initator into a concrete grain or samples initator + pub fn specialize(self, flow_config: &FlowConfigInfo) -> Either { + if flow_config.is_discrete_flow() { + Either::Grain(Initiator { + instance: self.instance, + _marker: PhantomData, + }) + } else { + Either::Samples(Initiator { + instance: self.instance, + _marker: PhantomData, + }) + } + } +} + +/// Create a new initiator +#[doc(hidden)] +pub(crate) fn create_initiator( + ctx: Arc, +) -> Result> { + let mut initiator = mxl_sys::fabrics::FabricsInitiator::default(); + unsafe { + Error::from_status( + ctx.api() + .fabrics_create_initiator(ctx.inner, &mut initiator), + )? + } + if initiator.is_null() { + return Err(Error::Other( + "Failed to create fabrics initiator.".to_string(), + )); + } + Ok(Initiator::new(ctx.clone(), initiator)) +} diff --git a/rust/mxl/src/fabrics/initiator/samples.rs b/rust/mxl/src/fabrics/initiator/samples.rs new file mode 100644 index 000000000..751abb727 --- /dev/null +++ b/rust/mxl/src/fabrics/initiator/samples.rs @@ -0,0 +1,73 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::time::Duration; + +use crate::{ + Error, Result, + fabrics::{TargetInfo, initiator::Initiator, initiator::Samples}, +}; + +impl Initiator { + /// Add a target to the initiator. This will allow the initiator to send data to the target in subsequent calls to + /// mxlFabricsInitiatorTransferGrain(). This function is always non-blocking. If additional connection setup is required + /// by the underlying implementation, it will only happen during a call to make_progress*(). + pub fn add_target(&self, target: &TargetInfo) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_add_target(self.instance.inner, target.inner) + }) + } + + /// Remove a target from the initiator. This function is always non-blocking. If any additional communication for a graceful shutdown is + /// required it will happend during a call to make_progress*(). It is guaranteed that no new grain transfer operations will + /// be queued for this target during calls to transfer() after the target was removed, but it is only guaranteed that + /// the connection shutdown has completed after make_progress*() no longer returns Error::NotReady. + pub fn remove_target(&self, target: &TargetInfo) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_remove_target(self.instance.inner, target.inner) + }) + } + + /// This function must be called regularly for the initiator to make progress on queued transfer operations, connection establishment + /// operations and connection shutdown operations. + pub fn make_progress_non_blocking(&self) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_make_progress_non_blocking(self.instance.inner) + }) + } + + /// This function must be called regularly for the initiator to make progress on queued transfer operations, connection establishment + /// operations and connection shutdown operations. + pub fn make_progress(&self, timeout: Duration) -> Result<()> { + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_initiator_make_progress_blocking( + self.instance.inner, + timeout.as_millis() as u16, + ) + }) + } + + /// Enqueue a transfer operation to all added targets. This function is always non-blocking. The transfer operation might be started right + /// away, but is only guaranteed to have completed after mxlFabricsInitiatorMakeProgress*() no longer returns Error::NotReady. + pub fn transfer(&self, head_index: u64, count: usize) -> Result<()> { + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_initiator_transfer_samples( + self.instance.inner, + head_index, + count, + ) + }) + } +} diff --git a/rust/mxl/src/fabrics/instance.rs b/rust/mxl/src/fabrics/instance.rs new file mode 100644 index 000000000..592c6b75c --- /dev/null +++ b/rust/mxl/src/fabrics/instance.rs @@ -0,0 +1,109 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::sync::Arc; + +use crate::{ + api::MxlFabricsAPiHandle, + error::{Error, Result}, + fabrics::{ + InterfaceConfig, + initiator::{self, Initiator, create_initiator}, + interface::Interfaces, + provider::Provider, + target::{self, Target, create_target}, + target_info::TargetInfo, + }, + instance::InstanceContext, +}; + +pub(crate) fn create_instance( + ctx: Arc, + fabrics_api: &MxlFabricsAPiHandle, +) -> Result { + let mut inst = std::ptr::null_mut(); + Error::from_status(unsafe { + fabrics_api.fabrics_create_instance( + //SAFETY: Both types are equivalent opaque handles from different bindgen modules. + ctx.instance.cast(), + std::ptr::null(), // Unused for now + &mut inst, + ) + })?; + if inst.is_null() { + return Err(Error::Other( + "Failed to create fabrics instance.".to_string(), + )); + } + + #[allow(clippy::arc_with_non_send_sync)] + // This is intentional, this Arc only implement Send, because fabric API as a whole is not thread-safe to use. + let ctx = Arc::new(FabricsInstanceContext { + _parent_ctx: ctx.clone(), + api: fabrics_api.clone(), + inner: inst, + }); + + Ok(FabricsInstance::new(ctx)) +} + +pub(crate) struct FabricsInstanceContext { + _parent_ctx: Arc, + api: MxlFabricsAPiHandle, + pub(crate) inner: mxl_sys::fabrics::FabricsInstance, +} +unsafe impl Send for FabricsInstanceContext {} + +impl FabricsInstanceContext { + pub(crate) fn api(&self) -> &MxlFabricsAPiHandle { + &self.api + } +} + +impl Drop for FabricsInstanceContext { + fn drop(&mut self) { + if !self.inner.is_null() { + unsafe { + let _ = self.api.fabrics_destroy_instance(self.inner); + } + } + } +} + +/// This is just a factory type for creating Fabrics related objects such as Targets, Initiators, etc. +/// The fabrics instance and its pointer are held in the `FabricsInstanceContext`` object. +/// This is created via an [MxlInstance](crate::MxlInstance). +pub struct FabricsInstance { + /// The fabric API is not-thread safe (Sync). + ctx: Arc, +} +/// SAFETY: FabricsInstance is safe to send to another thread, but fabric API as a whole is not thread-safe +unsafe impl Send for FabricsInstance {} + +impl FabricsInstance { + fn new(ctx: Arc) -> Self { + Self { ctx } + } + + /// Create a fabrics target. The target is the receiver of write operations from an initiator. + pub fn create_target(&self) -> Result> { + create_target(self.ctx.clone()) + } + + /// Create a fabrics initiator instance. + pub fn create_initiator(&self) -> Result> { + create_initiator(self.ctx.clone()) + } + + pub fn provider_from_str(&self, provider: &str) -> Result { + Provider::from_str(self.ctx.clone(), provider) + } + + pub fn target_info_from_str(&self, target_info: &str) -> Result { + TargetInfo::from_str(self.ctx.clone(), target_info) + } + + pub fn get_interfaces(&self, query: Option) -> Result { + Interfaces::get(self.ctx.clone(), query) + } +} diff --git a/rust/mxl/src/fabrics/interface/config.rs b/rust/mxl/src/fabrics/interface/config.rs new file mode 100644 index 000000000..af627d89d --- /dev/null +++ b/rust/mxl/src/fabrics/interface/config.rs @@ -0,0 +1,141 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ + Error, + fabrics::{ + EndpointAddress, capabilities::Capabilities, endpoint_address::OwnedEndpointAddress, + provider::ProviderType, + }, +}; + +use std::ffi::CString; + +pub struct InterfaceConfigBuilder<'a> { + provider: Option, + caps: Option, + endpoint_address: EndpointAddress<'a>, + attr: Option<&'a str>, +} + +impl<'a> InterfaceConfigBuilder<'a> { + pub(crate) fn new(endpoint_address: EndpointAddress<'a>) -> Self { + Self { + provider: None, + caps: None, + endpoint_address, + attr: None, + } + } + + /// Sets the provider type for the interface configuration. + pub fn provider(mut self, provider: ProviderType) -> Self { + self.provider = Some(provider); + self + } + + /// Sets the capabilities for the interface configuration. + pub fn caps(mut self, caps: Capabilities) -> Self { + self.caps = Some(caps); + self + } + + pub fn attr(mut self, attr: &'a str) -> Self { + self.attr = Some(attr); + self + } + + /// Builds the `InterfaceConfig` + pub fn build(self) -> InterfaceConfig<'a> { + InterfaceConfig { + provider: self.provider.unwrap_or(ProviderType::Any), + caps: self.caps.unwrap_or_default(), + endpoint_address: self.endpoint_address, + attr: self.attr, + } + } +} + +/// A configuration for a network interface, including the provider type, capabilities, endpoint address, and optional attributes. +#[derive(Debug)] +pub struct InterfaceConfig<'a> { + pub provider: ProviderType, + pub caps: Capabilities, + pub endpoint_address: EndpointAddress<'a>, + pub attr: Option<&'a str>, +} +impl<'a> InterfaceConfig<'a> { + pub fn builder(endpoint_address: EndpointAddress<'a>) -> InterfaceConfigBuilder<'a> { + InterfaceConfigBuilder::new(endpoint_address) + } + pub fn set_endpoint_address(&mut self, endpoint_address: EndpointAddress<'a>) { + self.endpoint_address = endpoint_address; + } +} +impl TryFrom<&InterfaceConfig<'_>> for OwnedInterfaceConfig { + type Error = Error; + fn try_from(value: &InterfaceConfig<'_>) -> Result { + OwnedInterfaceConfig::new(value) + } +} + +/// A wrapper around `mxl_sys::fabrics::FabricsInterfaceConfig` that owns the memory for the endpoint address and attribute strings. +pub(crate) struct OwnedInterfaceConfig { + inner: mxl_sys::fabrics::FabricsInterfaceConfig, + _address: OwnedEndpointAddress, + _attr: Option, +} + +impl OwnedInterfaceConfig { + pub(crate) fn new(value: &InterfaceConfig<'_>) -> Result { + let address = OwnedEndpointAddress::new(&value.endpoint_address)?; + let attr = value.attr.map(CString::new).transpose()?; + + Ok(Self { + inner: mxl_sys::fabrics::FabricsInterfaceConfig { + version: mxl_sys::fabrics::MXL_FABRICS_API_VERSION as i32, + provider: (&value.provider).into(), + caps: (&value.caps).into(), + address: address.as_ffi(), + attr: attr + .as_ref() + .map_or(std::ptr::null_mut(), |value| value.as_ptr() as *mut i8), + }, + _address: address, + _attr: attr, + }) + } + + pub(crate) fn as_ffi(&self) -> &mxl_sys::fabrics::FabricsInterfaceConfig { + &self.inner + } +} +impl<'a> TryFrom for InterfaceConfig<'a> { + type Error = crate::Error; + fn try_from(value: mxl_sys::fabrics::FabricsInterfaceConfig) -> Result { + let provider = (value.provider as mxl_sys::fabrics::FabricsProvider).into(); + let caps = value.caps.into(); + let endpoint_address = EndpointAddress { + node: (!value.address.node.is_null()) + .then(|| unsafe { std::ffi::CStr::from_ptr(value.address.node) }.to_str()) + .transpose() + .map_err(|e| Error::Other(e.to_string()))?, + service: (!value.address.service.is_null()) + .then(|| unsafe { std::ffi::CStr::from_ptr(value.address.service) }.to_str()) + .transpose() + .map_err(|e| Error::Other(e.to_string()))?, + }; + + let attr = (!value.attr.is_null()) + .then(|| unsafe { std::ffi::CStr::from_ptr(value.attr) }.to_str()) + .transpose() + .map_err(|e| Error::Other(e.to_string()))?; + + Ok(Self { + provider, + caps, + endpoint_address, + attr, + }) + } +} diff --git a/rust/mxl/src/fabrics/interface/mod.rs b/rust/mxl/src/fabrics/interface/mod.rs new file mode 100644 index 000000000..77e17e6aa --- /dev/null +++ b/rust/mxl/src/fabrics/interface/mod.rs @@ -0,0 +1,86 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 +// +use std::sync::Arc; + +use crate::{ + Error, + fabrics::{ + instance::FabricsInstanceContext, + interface::config::{InterfaceConfig, OwnedInterfaceConfig}, + }, +}; + +pub mod config; + +pub struct Interfaces { + ctx: Arc, + inner: *mut mxl_sys::fabrics::FabricsInterfaceList, +} + +impl Interfaces { + pub(crate) fn get( + ctx: Arc, + query: Option, + ) -> Result { + let query_storage = match query { + Some(q) => Some(OwnedInterfaceConfig::new(&q)?), + None => None, + }; + let query_ptr = query_storage + .as_ref() + .map_or(std::ptr::null(), |q| q.as_ffi() as *const _); + + let mut out_list = std::ptr::null_mut(); + + Error::from_status(unsafe { + ctx.api() + .fabrics_get_interfaces(ctx.inner, query_ptr, &mut out_list) + })?; + + Ok(Self { + ctx, + inner: out_list, + }) + } + pub fn iter(&self) -> InterfaceIter<'_> { + InterfaceIter { + it: self.inner, + _marker: std::marker::PhantomData, + } + } +} + +impl Drop for Interfaces { + fn drop(&mut self) { + if !self.inner.is_null() { + unsafe { + self.ctx.api().fabrics_free_interface_list( + self.inner as *mut mxl_sys::fabrics::FabricsInterfaceList, + ); + } + } + } +} + +pub struct InterfaceIter<'a> { + it: *mut mxl_sys::fabrics::FabricsInterfaceList, + _marker: std::marker::PhantomData<&'a ()>, +} + +impl<'a> Iterator for InterfaceIter<'a> { + type Item = InterfaceConfig<'a>; + + fn next(&mut self) -> Option { + if self.it.is_null() { + return None; + } + + let iface = unsafe { &*self.it }; + let out = iface.interface.try_into().ok(); + + self.it = iface.next; + + out + } +} diff --git a/rust/mxl/src/fabrics/mod.rs b/rust/mxl/src/fabrics/mod.rs new file mode 100644 index 000000000..be06f265e --- /dev/null +++ b/rust/mxl/src/fabrics/mod.rs @@ -0,0 +1,37 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +//! This module provides the Fabrics API extension for this library. The main type is the +//! [FabricsInstance], which is used to create Targets and Initiators for remote data transfers. +//! This module is gated by the `mxl-fabrics-ofi` feature flag. +//! +//! # Details +//! - To get a FabricsInstance, you must create it from MXL instance and a loaded Fabrics API. +//! ``` +//! let mxl_api = mxl::load_api(mxl::config::get_mxl_so_path()) .unwrap(); +//! let instance = mxl::MxlInstance::new(mxl_api, "/dev/shm","").unwrap(); +//! +//! let mxl_fabrics_api = mxl::load_fabrics_api(mxl::config::get_mxl_fabrics_ofi_so_path()); +//! let fabrics_instance = instance.create_fabrics_instance(&mxl_fabrics_api.unwrap()).unwrap(); +//! +//! // You can now create Targets and Initiators from the fabrics_instance +//! let target = fabrics_instance.create_target().unwrap(); +//! let initiator = fabrics_instance.create_initiator().unwrap(); +//! ```` +mod capabilities; +mod endpoint_address; +pub mod initiator; +mod instance; +mod interface; +mod provider; +pub mod target; +mod target_info; + +pub use capabilities::Capabilities; +pub use endpoint_address::EndpointAddress; +pub use instance::FabricsInstance; +pub use interface::config::InterfaceConfig; +pub use provider::{Provider, ProviderType}; +pub use target_info::TargetInfo; + +pub(crate) use instance::create_instance; diff --git a/rust/mxl/src/fabrics/provider.rs b/rust/mxl/src/fabrics/provider.rs new file mode 100644 index 000000000..40c416fca --- /dev/null +++ b/rust/mxl/src/fabrics/provider.rs @@ -0,0 +1,120 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use mxl_sys::fabrics::FabricsProvider; + +use crate::error::{Error, Result}; +use std::{ffi::CString, sync::Arc}; + +use crate::fabrics::instance::FabricsInstanceContext; + +/// The provider corresponds to the transport used for transfers. This is created from a +/// [FabricsInstance](crate::fabrics::FabricsInstance). +#[derive(Clone)] +pub struct Provider { + inner: ProviderType, + ctx: Arc, +} +unsafe impl Send for Provider {} +/// SAFETY: Although the `FabricsInstanceContext` type as a whole is not thread-safe, the subset of functions that this `Provider type uses is thread-safe: `fabrics_provider_from_string` and `fabrics_provider_to_string` +unsafe impl Sync for Provider {} + +/// The available transports +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ProviderType { + /// Any provider. Currently useful as an input to `FabricsInstance::get_interfaces` + Any, + /// Provider that use linux TCP sockets. + Tcp, + /// Provider for userspace verbs (libibverbs and librdmacm) + Verbs, + /// Provider for AWS Elastic Fabric Adapter + Efa, + /// Provider used for moving data between 2 memory regions inside the same system. + Shm, +} + +impl From for ProviderType { + fn from(value: mxl_sys::fabrics::FabricsProvider) -> Self { + match value { + mxl_sys::fabrics::MXL_FABRICS_PROVIDER_ANY => ProviderType::Any, + mxl_sys::fabrics::MXL_FABRICS_PROVIDER_TCP => ProviderType::Tcp, + mxl_sys::fabrics::MXL_FABRICS_PROVIDER_VERBS => ProviderType::Verbs, + mxl_sys::fabrics::MXL_FABRICS_PROVIDER_EFA => ProviderType::Efa, + mxl_sys::fabrics::MXL_FABRICS_PROVIDER_SHM => ProviderType::Shm, + _ => panic!("Unknown FabricsProvider value"), + } + } +} + +impl From<&ProviderType> for mxl_sys::fabrics::FabricsProvider { + fn from(value: &ProviderType) -> Self { + match value { + ProviderType::Any => mxl_sys::fabrics::MXL_FABRICS_PROVIDER_ANY, + ProviderType::Tcp => mxl_sys::fabrics::MXL_FABRICS_PROVIDER_TCP, + ProviderType::Verbs => mxl_sys::fabrics::MXL_FABRICS_PROVIDER_VERBS, + ProviderType::Efa => mxl_sys::fabrics::MXL_FABRICS_PROVIDER_EFA, + ProviderType::Shm => mxl_sys::fabrics::MXL_FABRICS_PROVIDER_SHM, + } + } +} + +impl From<&Provider> for mxl_sys::fabrics::FabricsProvider { + fn from(value: &Provider) -> Self { + (&value.inner).into() + } +} + +impl Provider { + fn new(ctx: Arc, inner: FabricsProvider) -> Self { + Self { + inner: inner.into(), + ctx, + } + } + + pub fn prov_type(&self) -> &ProviderType { + &self.inner + } + + /// Convert a string to a fabrics provider enum value. + /// Public visibility is set to crate only, because a `FabricsInstanceContext` is required. + /// See [FabricsInstance](crate::FabricsInstance). + pub(crate) fn from_str(ctx: Arc, s: &str) -> Result { + let mut inner = FabricsProvider::default(); + + Error::from_status(unsafe { + ctx.api() + .fabrics_provider_from_string(CString::new(s)?.as_ptr(), &mut inner) + })?; + + Ok(Self::new(ctx, inner)) + } + + /// Convert a fabrics provider enum value to a string. + pub fn to_string(&self) -> Result { + let mut size = 0; + + Error::from_status(unsafe { + self.ctx.api().fabrics_provider_to_string( + (&self.inner).into(), + std::ptr::null_mut(), + &mut size, + ) + })?; + + let mut out_string = vec![0u8; size]; + Error::from_status(unsafe { + self.ctx.api().fabrics_provider_to_string( + (&self.inner).into(), + out_string.as_mut_ptr() as *mut i8, + &mut size, + ) + })?; + + CString::from_vec_with_nul(out_string) + .map_err(|e| Error::Other(e.to_string()))? + .into_string() + .map_err(|e| Error::Other(e.to_string())) + } +} diff --git a/rust/mxl/src/fabrics/target/config.rs b/rust/mxl/src/fabrics/target/config.rs new file mode 100644 index 000000000..1f4209c03 --- /dev/null +++ b/rust/mxl/src/fabrics/target/config.rs @@ -0,0 +1,49 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use crate::FlowWriter; + +use crate::Error; +use crate::fabrics::{InterfaceConfig, interface::config::OwnedInterfaceConfig}; + +/// Configuration object required to set up a target. +pub struct Config<'a> { + version: i32, + interface: InterfaceConfig<'a>, + pub(crate) flow_writer: &'a FlowWriter, +} + +impl<'a> Config<'a> { + pub fn new(interface: InterfaceConfig<'a>, flow_writer: &'a FlowWriter) -> Self { + Self { + version: 0, + interface, + flow_writer, + } + } +} + +pub(crate) struct OwnedTargetConfig { + inner: mxl_sys::fabrics::FabricsTargetConfig, + _interface: OwnedInterfaceConfig, +} + +impl OwnedTargetConfig { + pub(crate) fn new(value: &Config<'_>) -> Result { + let interface = OwnedInterfaceConfig::new(&value.interface)?; + + Ok(Self { + inner: mxl_sys::fabrics::FabricsTargetConfig { + version: value.version, + interface: *interface.as_ffi(), + // SAFETY: Both types are equivalent opaque writer handles from different bindgen modules. + writer: value.flow_writer.inner().cast(), + }, + _interface: interface, + }) + } + + pub(crate) fn as_ffi(&self) -> &mxl_sys::fabrics::FabricsTargetConfig { + &self.inner + } +} diff --git a/rust/mxl/src/fabrics/target/grain.rs b/rust/mxl/src/fabrics/target/grain.rs new file mode 100644 index 000000000..e8bed0cfa --- /dev/null +++ b/rust/mxl/src/fabrics/target/grain.rs @@ -0,0 +1,42 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +// use std::time::Duration; +// +use crate::{ + Error, Result, + fabrics::target::{Target, states::Grain}, +}; +use std::time::Duration; + +/// Returned value from calling read* methods. +pub struct GrainReadResult { + pub grain_index: u64, +} + +impl Target { + /// Blocking accessor for a new grain. + pub fn read(&self, timeout: Duration) -> Result { + let mut grain_index = 0; + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_target_read_grain( + self.instance.inner, + u16::try_from(timeout.as_millis()).map_err(|_| Error::InvalidArg)?, + &mut grain_index, + ) + })?; + Ok(GrainReadResult { grain_index }) + } + + /// Non-blocking accessor for a new grain. + pub fn read_non_blocking(&self) -> Result { + let mut grain_index = 0; + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_target_read_grain_non_blocking(self.instance.inner, &mut grain_index) + })?; + Ok(GrainReadResult { grain_index }) + } +} diff --git a/rust/mxl/src/fabrics/target/mod.rs b/rust/mxl/src/fabrics/target/mod.rs new file mode 100644 index 000000000..acb2ca823 --- /dev/null +++ b/rust/mxl/src/fabrics/target/mod.rs @@ -0,0 +1,150 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +mod config; +mod grain; +mod samples; + +use std::{marker::PhantomData, sync::Arc}; + +use crate::{ + FlowConfigInfo, + error::{Error, Result}, + fabrics::{ + instance::FabricsInstanceContext, target::config::OwnedTargetConfig, + target_info::TargetInfo, + }, +}; +pub use config::Config; + +use states::*; + +pub mod states { + /// Used to create a new target + pub struct New {} + + /// Waiting for the target to be initialized with the setup function + pub struct Initializing {} + + /// The setup function has been called, but the target has not yet been specialized into a + /// grain or samples target + pub struct Specializing {} + + /// The target has been specialized into a grain target. It can only receive grains + pub struct Grain {} + + /// The target has been specialized into a samples target. It can only receive samples + pub struct Sample {} + + impl TargetState for New {} + impl TargetState for Initializing {} + impl TargetState for Specializing {} + impl TargetState for Grain {} + impl TargetState for Sample {} + + pub trait TargetState {} +} + +/// Wrapper class that holds a reference count to the Fabrics Instance and the actual target +/// instance. +pub struct TargetInstance { + ctx: Arc, + inner: mxl_sys::fabrics::FabricsTarget, +} +unsafe impl Send for TargetInstance {} + +impl Drop for TargetInstance { + fn drop(&mut self) { + if !self.inner.is_null() { + unsafe { + self.ctx + .api() + .fabrics_destroy_target(self.ctx.inner, self.inner); + } + } + } +} + +pub struct Target { + instance: TargetInstance, + _marker: PhantomData, +} +//SAFETY: A target is safe to be sent across threads, but it's not thread-safe to use its API functions. +unsafe impl Send for Target {} + +pub enum Either { + Grain(Target), + Sample(Target), +} + +impl Target { + pub(crate) fn new( + ctx: Arc, + target: mxl_sys::fabrics::FabricsTarget, + ) -> Target { + let instance = TargetInstance { ctx, inner: target }; + Target { + instance, + _marker: PhantomData, + } + } +} + +impl Target { + /// Configure the target. After the target has been configured, it is ready to receive transfers from an initiator. + /// If additional connection setup is required by the underlying implementation it might not happen during the call to + /// setup(), but be deferred until the first call to mxlFabricsTargetTryNewGrain(). + pub fn setup(self, config: &Config) -> Result<(Target, TargetInfo)> { + let mut info = mxl_sys::fabrics::FabricsTargetInfo::default(); + let config = OwnedTargetConfig::new(config)?; + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_target_setup( + self.instance.inner, + config.as_ffi(), + std::ptr::null(), + &mut info, + ) + })?; + + let ctx = self.instance.ctx.clone(); + + Ok(( + Target { + instance: self.instance, + _marker: PhantomData, + }, + TargetInfo::new(ctx, info), + )) + } +} + +impl Target { + /// Specialize the target into a concrete grain or samples target + pub fn specialize(self, flow_config: &FlowConfigInfo) -> Either { + if flow_config.is_discrete_flow() { + Either::Grain(Target { + instance: self.instance, + _marker: PhantomData, + }) + } else { + Either::Sample(Target { + instance: self.instance, + _marker: PhantomData, + }) + } + } +} + +/// Create a new target. +#[doc(hidden)] +pub(crate) fn create_target(ctx: Arc) -> Result> { + let mut target = mxl_sys::fabrics::FabricsTarget::default(); + unsafe { + Error::from_status(ctx.api().fabrics_create_target(ctx.inner, &mut target))?; + } + if target.is_null() { + return Err(Error::Other("Failed to create fabrics target.".to_string())); + } + + Ok(Target::new(ctx.clone(), target)) +} diff --git a/rust/mxl/src/fabrics/target/samples.rs b/rust/mxl/src/fabrics/target/samples.rs new file mode 100644 index 000000000..8ea491da3 --- /dev/null +++ b/rust/mxl/src/fabrics/target/samples.rs @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::time::Duration; + +use crate::{ + Error, + error::Result, + fabrics::target::{Target, states::Sample}, +}; + +pub struct SampleReadResult { + pub head_index: u64, + pub count: usize, +} + +impl Target { + ///Blocking accessor for a new grain. + pub fn read(&self, timeout: Duration) -> Result { + let mut head_index = 0; + let mut count = 0; + Error::from_status(unsafe { + self.instance.ctx.api().fabrics_target_read_samples( + self.instance.inner, + u16::try_from(timeout.as_millis()).map_err(|_| Error::InvalidArg)?, + &mut head_index, + &mut count, + ) + })?; + Ok(SampleReadResult { head_index, count }) + } + /// Non-blocking accessor for a new grain. + pub fn read_non_blocking(&self) -> Result { + let mut head_index = 0; + let mut count = 0; + Error::from_status(unsafe { + self.instance + .ctx + .api() + .fabrics_target_read_samples_non_blocking( + self.instance.inner, + &mut head_index, + &mut count, + ) + })?; + Ok(SampleReadResult { head_index, count }) + } +} diff --git a/rust/mxl/src/fabrics/target_info.rs b/rust/mxl/src/fabrics/target_info.rs new file mode 100644 index 000000000..243fd57da --- /dev/null +++ b/rust/mxl/src/fabrics/target_info.rs @@ -0,0 +1,73 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use std::{ffi::CString, sync::Arc}; + +use crate::error::{Error, Result}; +use mxl_sys::fabrics::FabricsTargetInfo; + +use crate::fabrics::instance::FabricsInstanceContext; + +/// The TargetInfo object holds the local fabric address, keys and memory region addresses for a target. +/// It is returned after setting up a new target and must be passed to the initiator to connect it. +pub struct TargetInfo { + ctx: Arc, + pub(crate) inner: FabricsTargetInfo, +} +unsafe impl Send for TargetInfo {} +/// SAFETY: Although the `FabricsInstanceContext` type as a whole is not thread-safe, the subset of functions that this `TargetInfo` type uses is thread-safe: `fabrics_target_info_from_string` and `fabrics_target_info_to_string` +unsafe impl Sync for TargetInfo {} + +impl TargetInfo { + pub(crate) fn new(ctx: Arc, inner: FabricsTargetInfo) -> Self { + Self { ctx, inner } + } + + /// Parse a targetInfo object from its string representation. + /// Public visibility is set to crate only, because a `FabricsInstanceContext` is required. + /// See [FabricsInstance](crate::FabricsInstance). + pub(crate) fn from_str(ctx: Arc, s: &str) -> Result { + let mut inner = FabricsTargetInfo::default(); + + Error::from_status(unsafe { + ctx.api() + .fabrics_target_info_from_string(CString::new(s)?.as_ptr(), &mut inner) + })?; + + Ok(Self::new(ctx, inner)) + } + + /// Serialize a target info object obtained from mxlFabricsTargetSetup() into a string representation. + pub fn to_string(&self) -> Result { + let mut size = 0; + Error::from_status(unsafe { + self.ctx.api().fabrics_target_info_to_string( + self.inner, + std::ptr::null_mut(), + &mut size, + ) + })?; + + let mut out_string = vec![0u8; size]; + Error::from_status(unsafe { + self.ctx.api().fabrics_target_info_to_string( + self.inner, + out_string.as_mut_ptr() as *mut i8, + &mut size, + ) + })?; + + CString::from_vec_with_nul(out_string) + .map_err(|e| Error::Other(e.to_string()))? + .into_string() + .map_err(|e| Error::Other(e.to_string())) + } +} + +impl Drop for TargetInfo { + fn drop(&mut self) { + if !self.inner.is_null() { + unsafe { self.ctx.api().fabrics_free_target_info(self.inner) }; + } + } +} diff --git a/rust/mxl/src/flow/reader.rs b/rust/mxl/src/flow/reader.rs index ebc5ab638..902401bc3 100644 --- a/rust/mxl/src/flow/reader.rs +++ b/rust/mxl/src/flow/reader.rs @@ -71,6 +71,11 @@ impl FlowReader { Self { context, reader } } + #[allow(dead_code)] + pub(crate) fn inner(&self) -> mxl_sys::FlowReader { + self.reader + } + pub fn get_info(&self) -> Result { get_flow_info(&self.context, self.reader) } diff --git a/rust/mxl/src/flow/writer.rs b/rust/mxl/src/flow/writer.rs index 5c8cfd271..231c74175 100644 --- a/rust/mxl/src/flow/writer.rs +++ b/rust/mxl/src/flow/writer.rs @@ -34,6 +34,11 @@ impl FlowWriter { } } + #[allow(dead_code)] + pub(crate) fn inner(&self) -> mxl_sys::FlowWriter { + self.writer + } + pub fn to_grain_writer(mut self) -> Result { let flow_type = self.get_flow_type()?; if !is_discrete_data_format(flow_type) { diff --git a/rust/mxl/src/grain/data.rs b/rust/mxl/src/grain/data.rs index d1d2e65c6..c66ba59c3 100644 --- a/rust/mxl/src/grain/data.rs +++ b/rust/mxl/src/grain/data.rs @@ -16,6 +16,10 @@ pub struct GrainData<'a> { /// index means `R` has not been produced yet, a newer one means the writer /// lapped the reader and `R` was evicted. pub index: u64, + + /// The number of slices in the full grain. This is does not change depending on whether the + /// grain is partial or complete. + pub total_slices: u16, } impl<'a> GrainData<'a> { diff --git a/rust/mxl/src/grain/reader.rs b/rust/mxl/src/grain/reader.rs index 97906c7da..e0c14a3cc 100644 --- a/rust/mxl/src/grain/reader.rs +++ b/rust/mxl/src/grain/reader.rs @@ -85,6 +85,7 @@ impl GrainReader { total_size: grain_info.grainSize as usize, flags: grain_info.flags, index: grain_info.index, + total_slices: grain_info.totalSlices, }) } @@ -119,6 +120,7 @@ impl GrainReader { total_size: grain_info.grainSize as usize, flags: grain_info.flags, index: grain_info.index, + total_slices: grain_info.totalSlices, }) } diff --git a/rust/mxl/src/grain/write_access.rs b/rust/mxl/src/grain/write_access.rs index f93043cb3..17efee8e5 100644 --- a/rust/mxl/src/grain/write_access.rs +++ b/rust/mxl/src/grain/write_access.rs @@ -50,6 +50,9 @@ impl<'a> GrainWriteAccess<'a> { pub fn total_slices(&self) -> u16 { self.grain_info.totalSlices } + pub fn valid_slices(&self) -> u16 { + self.grain_info.validSlices + } pub fn commit(mut self, valid_slices: u16) -> Result<()> { self.committed_or_canceled = true; diff --git a/rust/mxl/src/instance.rs b/rust/mxl/src/instance.rs index bc6253d88..d7cc51e58 100644 --- a/rust/mxl/src/instance.rs +++ b/rust/mxl/src/instance.rs @@ -3,6 +3,8 @@ use std::{ffi::CString, sync::Arc}; +#[cfg(feature = "mxl-fabrics-ofi")] +use crate::api::MxlFabricsAPiHandle; use crate::{Error, FlowConfigInfo, FlowReader, FlowWriter, Result, api::MxlApiHandle}; /// This struct stores the context that is shared by all objects. @@ -240,4 +242,14 @@ impl MxlInstance { .ok_or_else(|| Error::Other("Instance is still in use.".to_string()))?; context.destroy() } + + #[cfg(feature = "mxl-fabrics-ofi")] + pub fn create_fabrics_instance( + &self, + fabrics_api: &MxlFabricsAPiHandle, + ) -> Result { + use crate::fabrics; + + fabrics::create_instance(self.context.clone(), fabrics_api) + } } diff --git a/rust/mxl/src/lib.rs b/rust/mxl/src/lib.rs index 8c327cedf..10c1683e4 100644 --- a/rust/mxl/src/lib.rs +++ b/rust/mxl/src/lib.rs @@ -22,3 +22,8 @@ pub use mxl_sys::Rational; pub use samples::{ data::*, reader::SamplesReader, write_access::SamplesWriteAccess, writer::SamplesWriter, }; + +#[cfg(feature = "mxl-fabrics-ofi")] +pub mod fabrics; +#[cfg(feature = "mxl-fabrics-ofi")] +pub use api::{MxlFabricsApi, load_fabrics_api}; diff --git a/rust/mxl/tests/basic_tests.rs b/rust/mxl/tests/basic_tests.rs index d08b81c80..72cf1efa8 100644 --- a/rust/mxl/tests/basic_tests.rs +++ b/rust/mxl/tests/basic_tests.rs @@ -5,84 +5,14 @@ /// /// The tests now require an MXL library of a specific name to be present in the system. This should /// change in the future. For now, feel free to just edit the path to your library. +mod common; + use std::time::Duration; -use mxl::{MxlInstance, OwnedGrainData, OwnedSamplesData, config::get_mxl_so_path}; +use common::{read_flow_def, setup_test}; +use mxl::{OwnedGrainData, OwnedSamplesData}; use tracing::info; -static LOG_ONCE: std::sync::Once = std::sync::Once::new(); - -struct TestDomainGuard { - dir: std::path::PathBuf, -} - -impl TestDomainGuard { - fn new(test: &str) -> Self { - let dir = std::path::PathBuf::from(format!( - "/dev/shm/mxl_rust_unit_tests_domain_{}_{}", - test, - uuid::Uuid::new_v4() - )); - std::fs::create_dir_all(dir.as_path()).unwrap_or_else(|_| { - panic!( - "Failed to create test domain directory \"{}\".", - dir.display() - ) - }); - Self { dir } - } - - fn domain(&self) -> String { - self.dir.to_string_lossy().to_string() - } -} - -impl Drop for TestDomainGuard { - fn drop(&mut self) { - std::fs::remove_dir_all(self.dir.as_path()).unwrap_or_else(|_| { - panic!( - "Failed to remove test domain directory \"{}\".", - self.dir.display() - ) - }); - } -} - -fn setup_test(test: &str) -> (MxlInstance, TestDomainGuard) { - // Set up the logging to use the RUST_LOG environment variable and if not present, print INFO - // and higher. - LOG_ONCE.call_once(|| { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::builder() - .with_default_directive(tracing::level_filters::LevelFilter::INFO.into()) - .from_env_lossy(), - ) - .init(); - }); - - let mxl_api = mxl::load_api(get_mxl_so_path()).unwrap(); - let domain_guard = TestDomainGuard::new(test); - ( - MxlInstance::new(mxl_api, domain_guard.domain().as_str(), "").unwrap(), - domain_guard, - ) -} - -fn read_flow_def>(path: P) -> String { - let flow_config_file = mxl::config::get_mxl_repo_root().join(path); - - std::fs::read_to_string(flow_config_file.as_path()) - .map_err(|error| { - mxl::Error::Other(format!( - "Error while reading flow definition from \"{}\": {}", - flow_config_file.display(), - error - )) - }) - .unwrap() -} - #[test] fn basic_mxl_grain_writing_reading() { let (mxl_instance, _domain_guard) = setup_test("grains"); diff --git a/rust/mxl/tests/common/mod.rs b/rust/mxl/tests/common/mod.rs new file mode 100644 index 000000000..8e76f2876 --- /dev/null +++ b/rust/mxl/tests/common/mod.rs @@ -0,0 +1,84 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +use mxl::{MxlInstance, config::get_mxl_so_path}; + +#[cfg(feature = "mxl-fabrics-ofi")] +use mxl::{MxlFabricsApi, config::get_mxl_fabrics_ofi_so_path}; + +static LOG_ONCE: std::sync::Once = std::sync::Once::new(); + +pub struct TestDomainGuard { + dir: std::path::PathBuf, +} + +impl TestDomainGuard { + fn new(test: &str) -> Self { + let dir = std::path::PathBuf::from(format!( + "/dev/shm/mxl_rust_tests_domain_{}_{}", + test, + uuid::Uuid::new_v4() + )); + std::fs::create_dir_all(dir.as_path()).unwrap_or_else(|_| { + panic!( + "Failed to create test domain directory \"{}\".", + dir.display() + ) + }); + Self { dir } + } + + fn domain(&self) -> String { + self.dir.to_string_lossy().to_string() + } +} + +impl Drop for TestDomainGuard { + fn drop(&mut self) { + std::fs::remove_dir_all(self.dir.as_path()).unwrap_or_else(|_| { + panic!( + "Failed to remove test domain directory \"{}\".", + self.dir.display() + ) + }); + } +} + +pub fn setup_test(test: &str) -> (MxlInstance, TestDomainGuard) { + LOG_ONCE.call_once(|| { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::builder() + .with_default_directive(tracing::level_filters::LevelFilter::INFO.into()) + .from_env_lossy(), + ) + .init(); + }); + + let mxl_api = mxl::load_api(get_mxl_so_path()).unwrap(); + let domain_guard = TestDomainGuard::new(test); + ( + MxlInstance::new(mxl_api, domain_guard.domain().as_str(), "").unwrap(), + domain_guard, + ) +} + +#[cfg(feature = "mxl-fabrics-ofi")] +#[allow(dead_code)] +pub fn load_fabrics_test_api() -> std::sync::Arc { + mxl::load_fabrics_api(get_mxl_fabrics_ofi_so_path()).unwrap() +} + +pub fn read_flow_def>(path: P) -> String { + let flow_config_file = mxl::config::get_mxl_repo_root().join(path); + + std::fs::read_to_string(flow_config_file.as_path()) + .map_err(|error| { + mxl::Error::Other(format!( + "Error while reading flow definition from \"{}\": {}", + flow_config_file.display(), + error + )) + }) + .unwrap() +} diff --git a/rust/mxl/tests/fabrics_ofi_tests.rs b/rust/mxl/tests/fabrics_ofi_tests.rs new file mode 100644 index 000000000..cc304c394 --- /dev/null +++ b/rust/mxl/tests/fabrics_ofi_tests.rs @@ -0,0 +1,533 @@ +// SPDX-FileCopyrightText: 2026 Contributors to the Media eXchange Layer project. +// SPDX-License-Identifier: Apache-2.0 + +#![cfg(feature = "mxl-fabrics-ofi")] + +mod common; + +use std::time::{Duration, Instant}; + +use common::{load_fabrics_test_api, read_flow_def, setup_test}; +use mxl::{ + Error, FlowReader, FlowWriter, GrainReader, GrainWriter, OwnedGrainData, OwnedSamplesData, + SamplesReader, SamplesWriter, + fabrics::{ + EndpointAddress, InterfaceConfig, Provider, + initiator::{self, Initiator}, + target::{self, Target}, + }, +}; + +const POLL_TIMEOUT: Duration = Duration::from_secs(5); +const BLOCKING_WAIT: Duration = Duration::from_millis(20); +const AUDIO_SAMPLE_COUNT: usize = 42; + +fn tcp_endpoint() -> EndpointAddress<'static> { + EndpointAddress { + node: Some("127.0.0.1"), + service: Some("0"), + } +} + +fn tcp_interface(provider: &Provider) -> InterfaceConfig<'static> { + InterfaceConfig::builder(tcp_endpoint()) + .provider(provider.prov_type().clone()) + .build() +} + +fn poll_until_success(mut step: F, timeout_message: &str) +where + F: FnMut() -> bool, +{ + let deadline = Instant::now() + POLL_TIMEOUT; + while Instant::now() < deadline { + if step() { + return; + } + } + + panic!("{timeout_message}"); +} + +fn wait_for_grain_connection( + target: &Target, + initiator: &Initiator, +) { + poll_until_success( + || { + match target.read_non_blocking() { + Ok(_) | Err(Error::NotReady) => {} + Err(error) => { + panic!("unexpected target status while waiting for connection: {error}") + } + } + + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) => true, + Err(Error::NotReady) => false, + Err(error) => panic!( + "unexpected initiator status while waiting for grain connection: {error}" + ), + } + }, + "failed to connect grain initiator and target in 5 seconds", + ); +} + +fn wait_for_samples_connection( + target: &Target, + initiator: &Initiator, +) { + poll_until_success( + || { + match target.read_non_blocking() { + Ok(_) | Err(Error::NotReady) => {} + Err(error) => { + panic!("unexpected target status while waiting for sample connection: {error}") + } + } + + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) => true, + Err(Error::NotReady) => false, + Err(error) => panic!( + "unexpected initiator status while waiting for sample connection: {error}" + ), + } + }, + "failed to connect sample initiator and target in 5 seconds", + ); +} + +fn wait_for_grain_transfer_start( + target: &Target, + initiator: &Initiator, + grain_index: u64, + end_slice: u16, +) { + poll_until_success( + || { + match target.read_non_blocking() { + Ok(_) | Err(Error::NotReady) => {} + Err(error) => panic!("unexpected target status before grain transfer: {error}"), + } + + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) | Err(Error::NotReady) => {} + Err(error) => panic!("unexpected initiator status before grain transfer: {error}"), + } + + match initiator.transfer(grain_index, 0, end_slice) { + Ok(()) => true, + Err(Error::NotReady) => false, + Err(error) => panic!("failed to start grain transfer: {error}"), + } + }, + "failed to start grain transfer in 5 seconds", + ); +} + +fn wait_for_samples_transfer_start( + target: &Target, + initiator: &Initiator, + head_index: u64, + count: usize, +) { + poll_until_success( + || { + match target.read_non_blocking() { + Ok(_) | Err(Error::NotReady) => {} + Err(error) => panic!("unexpected target status before sample transfer: {error}"), + } + + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) | Err(Error::NotReady) => {} + Err(error) => panic!("unexpected initiator status before sample transfer: {error}"), + } + + match initiator.transfer(head_index, count) { + Ok(()) => true, + Err(Error::NotReady) => false, + Err(error) => panic!("failed to start samples transfer: {error}"), + } + }, + "failed to start samples transfer in 5 seconds", + ); +} + +fn wait_for_grain_transfer_completion( + target: &Target, + initiator: &Initiator, + expected_grain_index: u64, +) -> u64 { + let mut completed_index = None; + poll_until_success( + || { + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) | Err(Error::NotReady) => {} + Err(error) => { + panic!("unexpected initiator status while completing grain transfer: {error}") + } + } + + match target.read(BLOCKING_WAIT) { + Ok(result) => { + assert_eq!(result.grain_index, expected_grain_index); + completed_index = Some(result.grain_index); + true + } + Err(Error::NotReady) => false, + Err(Error::Interrupted) => { + panic!("grain target disconnected before transfer completed") + } + Err(error) => panic!("unexpected grain completion status: {error}"), + } + }, + "grain transfer did not complete in 5 seconds", + ); + completed_index.unwrap() +} + +fn wait_for_samples_transfer_completion( + target: &Target, + initiator: &Initiator, + expected_head_index: u64, + expected_count: usize, +) -> (u64, usize) { + let mut completed = None; + poll_until_success( + || { + match initiator.make_progress(BLOCKING_WAIT) { + Ok(()) | Err(Error::NotReady) => {} + Err(error) => { + panic!("unexpected initiator status while completing samples transfer: {error}") + } + } + + match target.read(BLOCKING_WAIT) { + Ok(result) => { + assert_eq!(result.head_index, expected_head_index); + assert_eq!(result.count, expected_count); + completed = Some((result.head_index, result.count)); + true + } + Err(Error::NotReady) => false, + Err(Error::Interrupted) => { + panic!("samples target disconnected before transfer completed") + } + Err(error) => panic!("unexpected samples completion status: {error}"), + } + }, + "samples transfer did not complete in 5 seconds", + ); + completed.unwrap() +} + +fn wait_for_target_grain(reader: &GrainReader, grain_index: u64) -> OwnedGrainData { + let mut result = None; + poll_until_success( + || match reader.get_grain_non_blocking(grain_index) { + Ok(grain) => { + result = Some(grain.into()); + true + } + Err(Error::OutOfRangeTooEarly) | Err(Error::NotReady) => false, + Err(error) => panic!("unexpected target grain read status: {error}"), + }, + "target grain did not become visible in 5 seconds", + ); + result.unwrap() +} + +fn wait_for_target_samples( + reader: &SamplesReader, + head_index: u64, + count: usize, +) -> OwnedSamplesData { + let mut result = None; + poll_until_success( + || match reader.get_samples_non_blocking(head_index, count) { + Ok(samples) => { + result = Some(samples.into()); + true + } + Err(Error::OutOfRangeTooEarly) | Err(Error::NotReady) => false, + Err(error) => panic!("unexpected target samples read status: {error}"), + }, + "target samples did not become visible in 5 seconds", + ); + result.unwrap() +} + +fn create_video_flow( + mxl_instance: &mxl::MxlInstance, +) -> (FlowWriter, FlowReader, mxl::FlowConfigInfo) { + create_flow_with_def(mxl_instance, read_flow_def("lib/tests/data/v210_flow.json")) +} + +fn create_video_flow_with_unique_id( + mxl_instance: &mxl::MxlInstance, +) -> (FlowWriter, FlowReader, mxl::FlowConfigInfo) { + create_flow_with_def( + mxl_instance, + flow_def_with_fresh_id(&read_flow_def("lib/tests/data/v210_flow.json")), + ) +} + +fn create_audio_flow( + mxl_instance: &mxl::MxlInstance, +) -> (FlowWriter, FlowReader, mxl::FlowConfigInfo) { + create_flow_with_def( + mxl_instance, + read_flow_def("lib/tests/data/audio_flow.json"), + ) +} + +fn create_audio_flow_with_unique_id( + mxl_instance: &mxl::MxlInstance, +) -> (FlowWriter, FlowReader, mxl::FlowConfigInfo) { + create_flow_with_def( + mxl_instance, + flow_def_with_fresh_id(&read_flow_def("lib/tests/data/audio_flow.json")), + ) +} + +fn flow_def_with_fresh_id(flow_def: &str) -> String { + let mut value: serde_json::Value = serde_json::from_str(flow_def).unwrap(); + value["id"] = serde_json::Value::String(uuid::Uuid::new_v4().to_string()); + serde_json::to_string(&value).unwrap() +} + +fn create_flow_with_def( + mxl_instance: &mxl::MxlInstance, + flow_def: String, +) -> (FlowWriter, FlowReader, mxl::FlowConfigInfo) { + let (flow_writer, flow_config, was_created) = mxl_instance + .create_flow_writer(flow_def.as_str(), None) + .unwrap(); + assert!(was_created); + + let flow_id = flow_config.common().id().to_string(); + let flow_reader = mxl_instance.create_flow_reader(flow_id.as_str()).unwrap(); + + (flow_writer, flow_reader, flow_config) +} + +fn fill_grain_payload(payload: &mut [u8]) { + for (index, byte) in payload.iter_mut().enumerate() { + *byte = (index % 251) as u8; + } +} + +fn fill_samples_payload(writer: &mut mxl::SamplesWriteAccess<'_>) { + for channel in 0..writer.channels() { + let (first, second) = writer.channel_data_mut(channel).unwrap(); + for (index, byte) in first.iter_mut().enumerate() { + *byte = (channel as u8).wrapping_mul(17).wrapping_add(index as u8); + } + for (index, byte) in second.iter_mut().enumerate() { + *byte = (channel as u8) + .wrapping_mul(29) + .wrapping_add(index as u8) + .wrapping_add(3); + } + } +} + +#[test] +fn provider_tcp_roundtrip() { + let (mxl_instance, _domain_guard) = setup_test("provider_tcp_roundtrip"); + + { + let fabrics_api = load_fabrics_test_api(); + let fabrics_instance = mxl_instance.create_fabrics_instance(&fabrics_api).unwrap(); + let provider = fabrics_instance.provider_from_str("tcp").unwrap(); + + assert_eq!(provider.to_string().unwrap(), "tcp"); + } + + mxl_instance.destroy().unwrap(); +} + +#[test] +fn target_info_roundtrip() { + let (mxl_instance, _domain_guard) = setup_test("target_info_roundtrip"); + + { + let fabrics_api = load_fabrics_test_api(); + let fabrics_instance = mxl_instance.create_fabrics_instance(&fabrics_api).unwrap(); + let (flow_writer, _flow_reader, _flow_config) = create_video_flow(&mxl_instance); + + let provider = fabrics_instance.provider_from_str("tcp").unwrap(); + let target = fabrics_instance.create_target().unwrap(); + let config = target::Config::new(tcp_interface(&provider), &flow_writer); + let (_target, target_info) = target.setup(&config).unwrap(); + + let serialized = target_info.to_string().unwrap(); + let deserialized = fabrics_instance.target_info_from_str(&serialized).unwrap(); + + assert_eq!(serialized, deserialized.to_string().unwrap()); + } + + mxl_instance.destroy().unwrap(); +} + +#[test] +fn tcp_grain_transfer_delivers_payload_to_target_flow() { + let (mxl_instance, _domain_guard) = setup_test("tcp_grain_transfer"); + + { + let fabrics_api = load_fabrics_test_api(); + let fabrics_instance = mxl_instance.create_fabrics_instance(&fabrics_api).unwrap(); + + let (source_flow_writer, source_flow_reader, source_flow_config) = + create_video_flow(&mxl_instance); + let (target_flow_writer, target_flow_reader, _target_flow_config) = + create_video_flow_with_unique_id(&mxl_instance); + + let source_grain_writer: GrainWriter = source_flow_writer.to_grain_writer().unwrap(); + let source_grain_reader: GrainReader = source_flow_reader.to_grain_reader().unwrap(); + let target_grain_reader: GrainReader = target_flow_reader.to_grain_reader().unwrap(); + + let (target, target_info) = { + let target_provider = fabrics_instance.provider_from_str("tcp").unwrap(); + let target = fabrics_instance.create_target().unwrap(); + let target_config = + target::Config::new(tcp_interface(&target_provider), &target_flow_writer); + target.setup(&target_config).unwrap() + }; + let target_grain_writer: GrainWriter = target_flow_writer.to_grain_writer().unwrap(); + let target = match target.specialize(&source_flow_config) { + target::Either::Grain(target) => target, + target::Either::Sample(_) => panic!("expected grain target for video flow"), + }; + + let initiator_flow_reader = mxl_instance + .create_flow_reader(source_flow_config.common().id().to_string().as_str()) + .unwrap(); + let initiator = { + let initiator_provider = fabrics_instance.provider_from_str("tcp").unwrap(); + let initiator = fabrics_instance.create_initiator().unwrap(); + let initiator_config = + initiator::Config::new(tcp_interface(&initiator_provider), &initiator_flow_reader); + initiator.setup(&initiator_config).unwrap() + }; + let initiator = match initiator.specialize(&source_flow_config) { + initiator::Either::Grain(initiator) => initiator, + initiator::Either::Samples(_) => panic!("expected grain initiator for video flow"), + }; + + initiator.add_target(&target_info).unwrap(); + wait_for_grain_connection(&target, &initiator); + + let grain_index = + mxl_instance.get_current_index(&source_flow_config.common().grain_rate().unwrap()); + let mut grain = source_grain_writer.open_grain(grain_index).unwrap(); + fill_grain_payload(grain.payload_mut()); + let total_slices = grain.total_slices(); + grain.commit(total_slices).unwrap(); + + let expected: OwnedGrainData = source_grain_reader + .get_complete_grain(grain_index, POLL_TIMEOUT) + .unwrap() + .into(); + + wait_for_grain_transfer_start(&target, &initiator, grain_index, total_slices); + let completed_index = wait_for_grain_transfer_completion(&target, &initiator, grain_index); + + let committed_grain = target_grain_writer.open_grain(completed_index).unwrap(); + let committed_slices = committed_grain.valid_slices(); + committed_grain.commit(committed_slices).unwrap(); + + let actual = wait_for_target_grain(&target_grain_reader, grain_index); + assert_eq!(actual.payload, expected.payload); + + initiator.remove_target(&target_info).unwrap(); + } + + mxl_instance.destroy().unwrap(); +} + +#[test] +fn tcp_samples_transfer_delivers_payload_to_target_flow() { + let (mxl_instance, _domain_guard) = setup_test("tcp_samples_transfer"); + + { + let fabrics_api = load_fabrics_test_api(); + let fabrics_instance = mxl_instance.create_fabrics_instance(&fabrics_api).unwrap(); + + let (source_flow_writer, source_flow_reader, source_flow_config) = + create_audio_flow(&mxl_instance); + let (target_flow_writer, target_flow_reader, _target_flow_config) = + create_audio_flow_with_unique_id(&mxl_instance); + + let source_samples_writer: SamplesWriter = source_flow_writer.to_samples_writer().unwrap(); + let source_samples_reader: SamplesReader = source_flow_reader.to_samples_reader().unwrap(); + let target_samples_reader: SamplesReader = target_flow_reader.to_samples_reader().unwrap(); + + let (target, target_info) = { + let target_provider = fabrics_instance.provider_from_str("tcp").unwrap(); + let target = fabrics_instance.create_target().unwrap(); + let target_config = + target::Config::new(tcp_interface(&target_provider), &target_flow_writer); + target.setup(&target_config).unwrap() + }; + let target_samples_writer: SamplesWriter = target_flow_writer.to_samples_writer().unwrap(); + let target = match target.specialize(&source_flow_config) { + target::Either::Sample(target) => target, + target::Either::Grain(_) => panic!("expected samples target for audio flow"), + }; + + let initiator_flow_reader = mxl_instance + .create_flow_reader(source_flow_config.common().id().to_string().as_str()) + .unwrap(); + let initiator = { + let initiator_provider = fabrics_instance.provider_from_str("tcp").unwrap(); + let initiator = fabrics_instance.create_initiator().unwrap(); + let initiator_config = + initiator::Config::new(tcp_interface(&initiator_provider), &initiator_flow_reader); + initiator.setup(&initiator_config).unwrap() + }; + let initiator = match initiator.specialize(&source_flow_config) { + initiator::Either::Samples(initiator) => initiator, + initiator::Either::Grain(_) => panic!("expected samples initiator for audio flow"), + }; + + initiator.add_target(&target_info).unwrap(); + wait_for_samples_connection(&target, &initiator); + + let head_index = + mxl_instance.get_current_index(&source_flow_config.common().sample_rate().unwrap()); + let mut samples = source_samples_writer + .open_samples(head_index, AUDIO_SAMPLE_COUNT) + .unwrap(); + fill_samples_payload(&mut samples); + samples.commit().unwrap(); + + let expected: OwnedSamplesData = source_samples_reader + .get_samples(head_index, AUDIO_SAMPLE_COUNT, POLL_TIMEOUT) + .unwrap() + .into(); + + wait_for_samples_transfer_start(&target, &initiator, head_index, AUDIO_SAMPLE_COUNT); + let (completed_head_index, completed_count) = wait_for_samples_transfer_completion( + &target, + &initiator, + head_index, + AUDIO_SAMPLE_COUNT, + ); + + let committed_samples = target_samples_writer + .open_samples(completed_head_index, completed_count) + .unwrap(); + committed_samples.commit().unwrap(); + + let actual = + wait_for_target_samples(&target_samples_reader, head_index, AUDIO_SAMPLE_COUNT); + assert_eq!(actual.payload, expected.payload); + + initiator.remove_target(&target_info).unwrap(); + } + + mxl_instance.destroy().unwrap(); +}