disrupt_rs/
builder.rs

1//! Module for building the Disruptor and adding event handlers.
2//!
3//! Use [build_single_producer] or [build_multi_producer] to get started.
4
5use crate::{
6    affinity::cpu_has_core_else_panic,
7    barrier::{Barrier, NONE},
8    consumer::{event_poller::EventPoller, start_processor, start_processor_with_state, Consumer},
9    cursor::Cursor,
10    event_handler::{EventHandler, EventHandlerWithState},
11    producer::{multi::MultiProducerBarrier, single::SingleProducerBarrier},
12    ringbuffer::RingBuffer,
13    wait_strategies::WaitStrategy,
14};
15use core_affinity::CoreId;
16use crossbeam_utils::CachePadded;
17use std::sync::{atomic::AtomicI64, Arc};
18
19use self::{multi::MPBuilder, single::SPBuilder};
20
21pub mod multi;
22pub mod single;
23
24/// State: No consumers (yet).
25pub struct NC;
26/// State: Single consumer.
27pub struct SC;
28/// State: Multiple consumers.
29pub struct MC;
30
31/// Build a single producer Disruptor. Use this if you only need to publish events from one thread.
32///
33/// For using a producer see [Producer](crate::producer::Producer).
34///
35/// # Examples
36///
37/// ```
38///# use disrupt_rs::*;
39///#
40/// // The example data entity on the ring buffer.
41/// struct Event {
42///     price: f64
43/// }
44/// let factory = || { Event { price: 0.0 }};
45///# let processor1 = |e: &Event, _, _| {};
46///# let processor2 = |e: &Event, _, _| {};
47///# let processor3 = |e: &Event, _, _| {};
48/// let mut producer = build_single_producer(8, factory, BusySpin)
49///    .handle_events_with(processor1)
50///    .handle_events_with(processor2)
51///    .and_then()
52///        // `processor3` only reads events after the other two processors are done reading.
53///        .handle_events_with(processor3)
54///    .build();
55///
56/// // Now use the `producer` to publish events.
57/// ```
58pub fn build_single_producer<E, W, F>(
59    size: usize,
60    event_factory: F,
61    wait_strategy: W,
62) -> SPBuilder<NC, E, W, SingleProducerBarrier>
63where
64    F: FnMut() -> E,
65    E: 'static + Send + Sync,
66    W: 'static + WaitStrategy,
67{
68    let producer_barrier = Arc::new(SingleProducerBarrier::new());
69    let dependent_barrier = Arc::clone(&producer_barrier);
70    SPBuilder::new(
71        size,
72        event_factory,
73        wait_strategy,
74        producer_barrier,
75        dependent_barrier,
76    )
77}
78
79/// Build a multi producer Disruptor. Use this if you need to publish events from many threads.
80///
81/// For using a producer see [Producer](crate::producer::Producer).
82///
83/// # Examples
84///
85/// ```
86///# use disrupt_rs::*;
87///#
88/// // The example data entity on the ring buffer.
89/// struct Event {
90///     price: f64
91/// }
92/// let factory = || { Event { price: 0.0 }};
93///# let processor1 = |e: &Event, _, _| {};
94///# let processor2 = |e: &Event, _, _| {};
95///# let processor3 = |e: &Event, _, _| {};
96/// let mut producer1 = build_multi_producer(64, factory, BusySpin)
97///    .handle_events_with(processor1)
98///    .handle_events_with(processor2)
99///    .and_then()
100///        // `processor3` only reads events after the other two processors are done reading.
101///        .handle_events_with(processor3)
102///    .build();
103///
104/// let mut producer2 = producer1.clone();
105///
106/// // Now two threads can get a Producer each.
107/// ```
108pub fn build_multi_producer<E, W, F>(
109    size: usize,
110    event_factory: F,
111    wait_strategy: W,
112) -> MPBuilder<NC, E, W, MultiProducerBarrier>
113where
114    F: FnMut() -> E,
115    E: 'static + Send + Sync,
116    W: 'static + WaitStrategy,
117{
118    let producer_barrier = Arc::new(MultiProducerBarrier::new(size));
119    let dependent_barrier = Arc::clone(&producer_barrier);
120    MPBuilder::new(
121        size,
122        event_factory,
123        wait_strategy,
124        producer_barrier,
125        dependent_barrier,
126    )
127}
128
129/// The processor's thread name and CPU affinity can be set via the builders that implement this trait.
130///
131/// # Examples
132///
133/// ```
134///# #[cfg(miri)] fn main() {}
135///# #[cfg(not(miri))]
136///# fn main() {
137///# use disrupt_rs::*;
138///#
139/// // The example data entity on the ring buffer.
140/// struct Event {
141///     price: f64
142/// }
143/// let factory = || { Event { price: 0.0 }};
144///# let processor1 = |e: &Event, _, _| {};
145///# let processor2 = |e: &Event, _, _| {};
146///# let processor3 = |e: &Event, _, _| {};
147/// let mut producer = build_single_producer(8, factory, BusySpin)
148///    // Processor 1 is pinned and has a custom name.
149///    .pin_at_core(1).thread_name("my_processor").handle_events_with(processor1)
150///    // Processor 2 is not pinned and gets a generic name.
151///    .handle_events_with(processor2)
152///    // Processor 3 is pinned and gets a generic name.
153///    .pin_at_core(2).handle_events_with(processor3)
154///    .build();
155///# }
156/// ```
157pub trait ProcessorSettings<E, W>: Sized
158where
159    W: WaitStrategy,
160{
161    #[doc(hidden)]
162    fn shared(&mut self) -> &mut Shared<E, W>;
163
164    /// Pin processor thread on the core with `id` for the next added event handler.
165    /// Outputs an error on stderr if the thread could not be pinned.
166    fn pin_at_core(mut self, id: usize) -> Self {
167        self.shared().pin_at_core(id);
168        self
169    }
170
171    /// Set a name for the processor thread for the next added event handler.
172    fn thread_name(mut self, name: &'static str) -> Self {
173        self.shared().thread_named(name);
174        self
175    }
176}
177
178trait Builder<E, W, B>: ProcessorSettings<E, W>
179where
180    E: 'static + Send + Sync,
181    B: 'static + Barrier,
182    W: 'static + WaitStrategy,
183{
184    fn add_event_handler<EH>(&mut self, event_handler: EH)
185    where
186        EH: 'static + EventHandler<E>,
187    {
188        let barrier = self.dependent_barrier();
189        let notifier = self.shared().notifier.clone();
190        let (cursor, consumer) = start_processor(event_handler, self.shared(), barrier, notifier);
191        self.shared().add_consumer_and_cursor(consumer, cursor);
192    }
193
194    fn get_event_poller(&mut self) -> EventPoller<E, B, W::Notifier> {
195        let cursor = Arc::new(Cursor::new(-1)); // Initially, the consumer has not read slot 0 yet.
196        self.shared().add_cursor(Arc::clone(&cursor));
197        let notifier = self.shared().notifier.clone();
198
199        EventPoller::new(
200            Arc::clone(&self.shared().ring_buffer),
201            Arc::clone(&self.dependent_barrier()),
202            Arc::clone(&self.shared().shutdown_at_sequence),
203            cursor,
204            notifier,
205        )
206    }
207
208    fn add_event_handler_with_state<EH, S, IS>(&mut self, event_handler: EH, initialize_state: IS)
209    where
210        EH: 'static + EventHandlerWithState<E, S>,
211        IS: 'static + Send + FnOnce() -> S,
212    {
213        let barrier = self.dependent_barrier();
214        let notifier = self.shared().notifier.clone();
215        let (cursor, consumer) = start_processor_with_state(
216            event_handler,
217            self.shared(),
218            barrier,
219            initialize_state,
220            notifier,
221        );
222        self.shared().add_consumer_and_cursor(consumer, cursor);
223    }
224
225    fn dependent_barrier(&self) -> Arc<B>;
226}
227
228#[doc(hidden)]
229pub struct Shared<E, W>
230where
231    W: crate::wait_strategies::WaitStrategy,
232{
233    pub(crate) shutdown_at_sequence: Arc<CachePadded<AtomicI64>>,
234    pub(crate) ring_buffer: Arc<RingBuffer<E>>,
235    pub(crate) consumers: Vec<Consumer>,
236    current_consumer_cursors: Option<Vec<Arc<Cursor>>>,
237    pub(crate) notifier: W::Notifier,
238    pub(crate) wait_strategy: W,
239    pub(crate) thread_context: ThreadContext,
240}
241
242impl<E, W> Shared<E, W>
243where
244    W: crate::wait_strategies::WaitStrategy,
245{
246    fn new<F>(size: usize, event_factory: F, wait_strategy: W) -> Self
247    where
248        F: FnMut() -> E,
249    {
250        let notifier = wait_strategy.notifier();
251        let ring_buffer = Arc::new(RingBuffer::new(size, event_factory));
252        let shutdown_at_sequence = Arc::new(CachePadded::new(AtomicI64::new(NONE)));
253        let current_consumer_cursors = Some(vec![]);
254
255        Self {
256            ring_buffer,
257            notifier,
258            wait_strategy,
259            shutdown_at_sequence,
260            current_consumer_cursors,
261            consumers: vec![],
262            thread_context: ThreadContext::default(),
263        }
264    }
265
266    fn add_consumer_and_cursor(&mut self, consumer: Consumer, cursor: Arc<Cursor>) {
267        self.consumers.push(consumer);
268        self.add_cursor(cursor);
269    }
270
271    fn add_cursor(&mut self, cursor: Arc<Cursor>) {
272        self.current_consumer_cursors.as_mut().unwrap().push(cursor);
273    }
274
275    fn pin_at_core(&mut self, id: usize) {
276        cpu_has_core_else_panic(id);
277        self.thread_context.affinity = Some(CoreId { id });
278    }
279
280    fn thread_named(&mut self, name: &'static str) {
281        self.thread_context.name = Some(name.to_owned());
282    }
283}
284
285#[derive(Default)]
286pub(crate) struct ThreadContext {
287    affinity: Option<CoreId>,
288    name: Option<String>,
289    id: usize,
290}
291
292impl ThreadContext {
293    pub(crate) fn name(&mut self) -> String {
294        self.name
295            .take()
296            .or_else(|| {
297                self.id += 1;
298                Some(format!("processor-{}", self.id))
299            })
300            .unwrap()
301    }
302
303    pub(crate) fn affinity(&mut self) -> Option<CoreId> {
304        self.affinity.take()
305    }
306}
disrupt_rs/builder.rs

disrupt_rs/
builder.rs
