photon-ring 3.0.0

Ultra-low-latency SPMC/MPMC pub/sub using stamped ring buffers. Formally sound with atomic-slots feature. no_std compatible.
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

// Copyright 2026 Photon Ring Contributors
// SPDX-License-Identifier: MIT OR Apache-2.0

use super::mp_publisher::MpPublisher;
use super::publisher::Publisher;
use super::subscribable::Subscribable;
use crate::pod::Pod;
use crate::ring::SharedRing;
use alloc::sync::Arc;

/// Build the single-producer `Publisher` view over a freshly created ring.
/// Shared by [`channel`] (lossy) and [`channel_bounded`] — they differ only
/// in how the ring is allocated, not in the publisher wiring.
fn make_publisher<T: Pod>(ring: &Arc<SharedRing<T>>) -> Publisher<T> {
    Publisher {
        has_backpressure: ring.backpressure.is_some(),
        ring: ring.clone(),
        slots_ptr: ring.slots_ptr(),
        index: ring.index,
        cursor_ptr: ring.cursor_ptr(),
        seq: 0,
        cached_slowest: 0,
    }
}

/// Create a Photon SPMC channel.
///
/// `capacity` must be >= 2. Any positive integer is accepted; power-of-two
/// capacities use a single-cycle AND for slot indexing, while arbitrary
/// capacities use Lemire fastmod (~1.5 ns overhead per indexing operation).
///
/// # Example
/// ```
/// let (mut pub_, subs) = photon_ring::channel::<u64>(64);
/// let mut sub = subs.subscribe();
/// pub_.publish(42);
/// assert_eq!(sub.try_recv(), Ok(42));
/// ```
pub fn channel<T: Pod>(capacity: usize) -> (Publisher<T>, Subscribable<T>) {
    let ring = Arc::new(SharedRing::new(capacity));
    let publisher = make_publisher(&ring);
    (publisher, Subscribable { ring })
}

/// Create a backpressure-capable SPMC channel.
///
/// The publisher will refuse to publish (returning
/// [`PublishError::Full`](super::errors::PublishError::Full))
/// when it would overwrite a slot that the slowest subscriber hasn't
/// read yet, minus `watermark` slots of headroom.
///
/// Unlike the default lossy [`channel()`], no messages are ever dropped.
///
/// # Arguments
/// - `capacity` -- ring size, must be >= 2.
/// - `watermark` -- headroom slots; must be less than `capacity`.
///   A watermark of 0 means the publisher blocks as soon as all slots are
///   occupied. A watermark of `capacity - 1` means it blocks when only one
///   slot is free.
///
/// # Example
/// ```
/// use photon_ring::channel_bounded;
/// use photon_ring::PublishError;
///
/// let (mut p, s) = channel_bounded::<u64>(4, 0);
/// let mut sub = s.subscribe();
///
/// // Fill the ring (4 slots).
/// for i in 0u64..4 {
///     p.try_publish(i).unwrap();
/// }
///
/// // Ring is full — backpressure kicks in.
/// assert_eq!(p.try_publish(99u64), Err(PublishError::Full(99)));
///
/// // Drain one slot — publisher can continue.
/// assert_eq!(sub.try_recv(), Ok(0));
/// p.try_publish(99).unwrap();
/// ```
pub fn channel_bounded<T: Pod>(
    capacity: usize,
    watermark: usize,
) -> (Publisher<T>, Subscribable<T>) {
    let ring = Arc::new(SharedRing::new_bounded(capacity, watermark));
    let publisher = make_publisher(&ring);
    (publisher, Subscribable { ring })
}

/// Create a Photon MPMC (multi-producer, multi-consumer) channel.
///
/// `capacity` must be >= 2. Returns a clone-able [`MpPublisher`] and the
/// same [`Subscribable`] factory used by SPMC channels.
///
/// Multiple threads can clone the publisher and publish concurrently.
/// Subscribers work identically to the SPMC case.
///
/// # Example
/// ```
/// let (pub_, subs) = photon_ring::channel_mpmc::<u64>(64);
/// let mut sub = subs.subscribe();
///
/// let pub2 = pub_.clone();
/// pub_.publish(1);
/// pub2.publish(2);
///
/// assert_eq!(sub.try_recv(), Ok(1));
/// assert_eq!(sub.try_recv(), Ok(2));
/// ```
pub fn channel_mpmc<T: Pod>(capacity: usize) -> (MpPublisher<T>, Subscribable<T>) {
    use core::sync::atomic::AtomicU64;
    let ring = Arc::new(SharedRing::new_mpmc(capacity));
    let slots_ptr = ring.slots_ptr();
    let idx = ring.index;
    let cursor_ptr = ring.cursor_ptr();
    let next_seq_ptr = &ring
        .next_seq
        .as_ref()
        .expect("MPMC ring must have next_seq")
        .0 as *const AtomicU64;
    (
        MpPublisher {
            ring: ring.clone(),
            slots_ptr,
            index: idx,
            cursor_ptr,
            next_seq_ptr,
        },
        Subscribable { ring },
    )
}