dma-libfabric 0.1.0

Tools for building server-initiated RDMA side-channels
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

//! like `one_transfer`, but async.
//!
//! Run `one_target` first to print the three advertisement fields
//!
//! ```text
//! cargo run -p dma-libfabric --example one_target     -- 127.0.0.1
//! cargo run -p dma-libfabric --example async_transfer -- 127.0.0.1 <address-hex> <rkey> <remote-addr>
//! ```

use std::sync::Arc;

use dma_libfabric::asynchronous::FabricService;
use dma_libfabric::{Direction, Operands, Pool, TransferRequest};
use dma_libfabric_protocol::{DmaError, encode_hex};

mod common;

/// The per-op context. It owns the payload, which keeps the buffer alive for the whole
/// transfer. Since the vec is inline it also trivially handles cancellation-safety.
struct Operation {
    payload: Vec<u8>,
}

impl Operands for Operation {
    /// The bytes this transfer writes into the peer. `FromPeer` would implement `allocate` instead.
    fn source(&self) -> Option<&[u8]> {
        Some(&self.payload)
    }
}

/// tokio here for brevity; a [`Transfer`](dma_libfabric::asynchronous::Transfer) is a plain future
/// and this crate depends on no particular runtime.
#[tokio::main]
async fn main() -> Result<(), DmaError> {
    let common::Arguments {
        configuration,
        peer,
    } = common::parse()?;

    let pool = Arc::new(Pool::new(2).map_err(|error| DmaError::Fabric(format!("pool: {error}")))?);
    let server: FabricService<Operation> = FabricService::start(&configuration, pool)?;

    // What a peer must hold to be RMA'd against. Carry it on your own control channel.
    println!("local address: {}", encode_hex(server.local_address()));

    let peer = match peer {
        Some(peer) => peer,
        None => {
            println!("paste the target's <address-hex> <rkey> <remote-addr>:");
            common::peer_from_stdin()?
        }
    };

    // `one_target` waits for a full buffer of this byte.
    let payload = vec![0xab_u8; 4096];
    let request = TransferRequest {
        client_id: 1,
        peer_address: peer.address,
        remote_key: peer.remote_key,
        remote_address: peer.remote_address,
        direction: Direction::ToPeer,
        want_checksum: true,
        caller_context: Operation { payload },
        parent_id: None,
    };
    let Ok(transfer) = server.transfer(request) else {
        return Err(DmaError::Fabric("fabric worker is gone".into()));
    };

    // The context comes back with the outcome, so the payload is yours again here.
    let (outcome, operation) = transfer.await;
    match outcome {
        Ok(transferred) => println!(
            "wrote {} of {} bytes, crc {:?}",
            transferred.bytes,
            operation.payload.len(),
            transferred.checksum
        ),
        Err(error) => println!("transfer failed: {error}"),
    }
    Ok(())
}