flyby 0.1.1

A high-performance Rust framework for composable data-ingestion pipelines.
Documentation 
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

//! SPDK storage backend.
//!
//! Enabled by the `spdk` Cargo feature.  Currently a design stub — all
//! methods return [`ErrorKind::NotImplemented`].
//!
//! ## SPDK primer
//!
//! SPDK (Storage Performance Development Kit) is an Intel-originated
//! userspace storage stack.  It bypasses the kernel block layer entirely:
//!
//! - **Userspace NVMe driver**: maps the NVMe PCIe BAR directly into the
//!   process address space; I/O is submitted by writing to MMIO registers
//!   without any syscall or kernel involvement.
//! - **DMA buffers**: packet/record buffers must be physically contiguous and
//!   pinned.  SPDK allocates them from hugepages via `spdk_dma_malloc`.
//! - **Queue pairs**: each NVMe namespace exposes one or more I/O queue
//!   pairs, each with a Submission Queue and a Completion Queue.  FlyBy
//!   dedicates one queue pair per polling thread.
//! - **Polling**: unlike io_uring there is no CQ interrupt — the driver must
//!   busy-poll the completion queue.  This is only appropriate when a CPU
//!   core can be dedicated to storage I/O.
//!
//! ## When to use SPDK vs io_uring
//!
//! | Criterion | io_uring | SPDK |
//! |---|---|---|
//! | Kernel required | ≥ 5.1 | Any (MMIO) |
//! | Interrupt-driven | Yes (or kernel poll) | No (busy poll only) |
//! | CPU dedication | Not required | Required per queue |
//! | NVMe latency | ~10–20 µs | ~2–5 µs |
//! | Deployment complexity | Low | High |
//!
//! SPDK is appropriate only after the io_uring backend is validated and a
//! measured latency requirement cannot be met by io_uring (ADR-0006).
//!
//! ## Planned read path
//!
//! ```text
//! poll_batch()
//!   → poll NVMe CQ for completed DMA transfers
//!   → copy completed DMA buffers into RawRecordBatch slots
//!   → submit new NVMe read commands for the next block range
//! ```
//!
//! ## Hugepage requirement
//!
//! SPDK needs hugepages for DMA memory.  Provision before starting:
//!
//! ```sh
//! echo 1024 | sudo tee /sys/kernel/mm/hugepages/hugepages-2048kB/nr_hugepages
//! ```
//!
//! ## References
//!
//! - SPDK documentation: <https://spdk.io/doc/>
//! - NVMe specification: <https://nvmexpress.org/specifications/>

use crate::core::{Error, ErrorKind, Lifecycle, Result};

use crate::storage::batch::RawRecordBatch;
use crate::storage::config::SpdkConfig;
use crate::storage::source::StorageSource;

/// SPDK-backed storage source.
///
/// All methods return [`ErrorKind::NotImplemented`] until the real SPDK
/// binding is implemented.
pub struct SpdkSource {
    #[allow(dead_code)]
    config: SpdkConfig,
}

impl SpdkSource {
    /// Create a new SPDK source with the given config.
    pub fn new(config: SpdkConfig) -> Self {
        Self { config }
    }
}

impl Lifecycle for SpdkSource {
    fn init(&mut self) -> Result<()> {
        Err(Error::new(
            ErrorKind::NotImplemented,
            "SPDK backend: not yet implemented (deferred after io_uring; see ADR-0006)",
        ))
    }

    fn shutdown(&mut self) -> Result<()> {
        Ok(())
    }
}

impl StorageSource for SpdkSource {
    fn poll_batch(&mut self, _batch: &mut RawRecordBatch) -> Result<usize> {
        Err(Error::new(
            ErrorKind::NotImplemented,
            "SPDK backend: not yet implemented",
        ))
    }

    fn backend_name() -> &'static str {
        "spdk"
    }

    fn is_exhausted(&self) -> bool {
        false
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::batch::RawRecordBatch;

    #[test]
    fn init_returns_not_implemented() {
        let mut src = SpdkSource::new(SpdkConfig::default());
        let err = src.init().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::NotImplemented);
    }

    #[test]
    fn poll_returns_not_implemented() {
        let mut src = SpdkSource::new(SpdkConfig::default());
        let mut batch = RawRecordBatch::new(4, 64);
        let err = src.poll_batch(&mut batch).unwrap_err();
        assert_eq!(err.kind(), ErrorKind::NotImplemented);
    }

    #[test]
    fn backend_name_is_spdk() {
        assert_eq!(SpdkSource::backend_name(), "spdk");
    }
}