Skip to content

chore(fpss): remove dead wait-strategy escape hatch + hand-rolled splitmix #1200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
userFRM merged 1 commit into from
Jul 11, 2026
Merged

chore(fpss): remove dead wait-strategy escape hatch + hand-rolled splitmix #1200

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 0 additions & 37 deletions thetadatadx-rs/src/backoff.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,36 +133,6 @@ pub(crate) fn uniform_duration(lo: Duration, hi: Duration) -> Duration {
Duration::from_nanos(sampled)
}

/// Process-local entropy word for seeding shuffles and cursor offsets
/// where no caller-supplied seed exists. Folds the wall clock with a
/// process-local counter so two clients constructed in the same
/// nanosecond still diverge.
pub(crate) fn entropy_u64() -> u64 {
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let tick = COUNTER.fetch_add(1, Ordering::Relaxed);
let now = u64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_nanos()),
)
.unwrap_or(u64::MAX);
let pid = u64::from(std::process::id());
splitmix64(now ^ pid.rotate_left(32) ^ tick.wrapping_mul(0x9E37_79B9_7F4A_7C15))
}

/// splitmix64 finaliser — documented mixer, used here only to spread
/// seed bits, never for cryptographic purposes.
pub(crate) fn splitmix64(mut seed: u64) -> u64 {
seed = seed.wrapping_add(0x9E37_79B9_7F4A_7C15);
seed ^= seed >> 30;
seed = seed.wrapping_mul(0xBF58_476D_1CE4_E5B9);
seed ^= seed >> 27;
seed = seed.wrapping_mul(0x94D0_49BB_1331_11EB);
seed ^= seed >> 31;
seed
}

#[cfg(test)]
mod tests {
use super::*;
Expand Down Expand Up @@ -224,11 +194,4 @@ mod tests {
assert_eq!(JitterMode::parse("bogus"), None);
assert_eq!(JitterMode::default(), JitterMode::Full);
}

#[test]
fn entropy_words_diverge() {
let a = entropy_u64();
let b = entropy_u64();
assert_ne!(a, b, "consecutive entropy words must differ");
}
}
78 changes: 0 additions & 78 deletions thetadatadx-rs/src/fpss/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2089,55 +2089,6 @@ impl StreamingClient {
}
}

/// Drain events through `on_event`, applying a caller-supplied
/// [`crate::streaming::wait::WaitStrategy`] on each momentarily
/// empty ring instead of the default low-latency wait.
///
/// This is the Rust-native bring-your-own-strategy escape hatch: the
/// default wait ([`crate::streaming::wait::BusySpinWithSpinLoopHint`]-style
/// spin) suits real-time market data, but a Rust caller with an
/// exotic backoff (e.g. an adaptive PID-controlled park, or a
/// strategy that coordinates with another subsystem) can supply any
/// `W: WaitStrategy` here. Use a strategy from
/// [`crate::streaming::wait`] (e.g.
/// [`crate::streaming::wait::BusySpin`]) or implement the trait on
/// your own type.
///
/// `W` is monomorphised into the loop, so the per-poll cost is the
/// caller's `wait_for` body with no indirection. Delivery semantics
/// match [`Self::for_each`]: `on_event` fires exactly once per event
/// and the loop returns on terminal shutdown after the ring drains.
///
/// # Why Rust-only
///
/// `wait_for` fires on every ring-empty poll on the hot path. Routing
/// that per-poll callback across the C ABI, the CPython interpreter
/// lock, or the JavaScript event loop would add call-boundary
/// overhead to the single tightest loop in the consumer — a latency
/// regression, not a tuning knob. The bindings therefore run the
/// fixed low-latency wait with no override.
pub fn for_each_with_wait_strategy<W, F>(&self, mut on_event: F, strategy: W) -> PollOutcome
where
W: crate::streaming::wait::WaitStrategy,
F: FnMut(&StreamEvent),
{
// The drain owner + CPU pin are recorded inside `poll_batch`, after
// the staging lock is acquired, so the identity reflects the proven
// drainer rather than whichever thread merely entered this loop.
loop {
match self.poll_batch(&mut on_event) {
// Return the terminal outcome so the caller can tell a clean
// shutdown from an I/O-thread fault (`Failed`).
terminal @ (PollOutcome::Shutdown | PollOutcome::Failed) => return terminal,
// `Busy` cannot arise from this loop's own poll (it holds no
// lock when it polls); back off and retry defensively in case
// an `on_event` callback re-entered a drain.
PollOutcome::Drained(0) | PollOutcome::Busy => strategy.wait_for(0),
PollOutcome::Drained(_) => {}
}
}
}

/// Polymorphic subscribe — wire-level entry point.
///
/// Accepts a typed [`protocol::Subscription`] value built via
Expand Down Expand Up @@ -4144,35 +4095,6 @@ mod ring_occupancy_tests {
);
}

/// The bring-your-own-strategy drain is reachable through the
/// crate-owned wait-strategy re-export, so a Rust caller never names
/// the underlying ring crate. Dropping the producer terminates the
/// drain; `BusySpin`'s `wait_for` is a no-op, so the loop spins only
/// until the ring reports shutdown.
#[test]
fn for_each_with_wait_strategy_accepts_crate_owned_preset() {
use crate::streaming::wait::BusySpin;

let (client, mut producer) = StreamingClient::for_ring_occupancy_test(64);
for _ in 0..3 {
assert!(
publish_one(&mut producer),
"fresh ring must accept a publish"
);
}
// Drop the producer so the poller observes terminal shutdown once
// the three published events drain, and the loop returns.
drop(producer);

let mut delivered = 0usize;
client.for_each_with_wait_strategy(|_event| delivered += 1, BusySpin);

assert_eq!(
delivered, 3,
"every published event must be delivered before shutdown returns"
);
}

/// A faulted I/O thread — the `io_loop` unwound and its fault guard set
/// `io_faulted` before the ring published its shutdown sequence — must
/// surface through the CALLBACK drain path, not read as a clean shutdown.
Expand Down
5 changes: 2 additions & 3 deletions thetadatadx-rs/src/grpc/pool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,9 +209,8 @@ impl ChannelPool {
/// together after an outage lands its entire first burst on the same
/// upstream connection.
fn seeded_cursor(len: usize) -> AtomicUsize {
let seed = crate::backoff::entropy_u64();
let offset = (seed as usize) % len.max(1);
AtomicUsize::new(offset)
use rand::RngExt;
AtomicUsize::new(rand::rng().random_range(0..len.max(1)))
}

/// Pre-dispatch reservation on a pooled [`Channel`].
Expand Down
32 changes: 0 additions & 32 deletions thetadatadx-rs/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -344,38 +344,6 @@ pub mod streaming {
#[cfg(feature = "arrow")]
#[doc(hidden)]
pub use crate::fpss::batch_schema::stream_batch_schema;

/// Consumer wait strategies for the streaming ring.
///
/// The streaming client drains the ring with a fixed low-latency
/// wait, which every language binding uses. A Rust caller that needs
/// an exotic backoff can instead supply any type implementing
/// `WaitStrategy` to
/// [`crate::streaming::StreamingClient::for_each_with_wait_strategy`]. The
/// strategy is monomorphised into the drain loop, so the per-poll
/// cost is the caller's `wait_for` body with no indirection.
///
/// `BusySpin` is the lowest-latency strategy (a true busy spin);
/// `BusySpinWithSpinLoopHint` adds a `spin_loop` hint so the core
/// can save power or switch hyper-threads; `Sleep` parks the thread
/// for a fixed duration between polls.
///
/// ```rust,ignore
/// use thetadatadx::streaming::wait::BusySpin;
///
/// client.for_each_with_wait_strategy(
/// |event| { /* handle event */ },
/// BusySpin,
/// );
/// ```
pub mod wait {
// VOCAB-OK: re-exporting the ring's wait-strategy surface under a
// crate-owned path so callers never name the underlying ring
// crate in their own `use` statements or trait bounds.
pub use disruptor::wait_strategies::{
BusySpin, BusySpinWithSpinLoopHint, Sleep, WaitStrategy,
};
}
}

// ─── Market-data queries ──────────────────────────────────────────────────────
Expand Down
Loading