zenoh_sync/
event.rs

1//
2// Copyright (c) 2024 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14use std::{
15    fmt,
16    sync::{
17        atomic::{AtomicU16, AtomicU8, Ordering},
18        Arc,
19    },
20    time::{Duration, Instant},
21};
22
23use event_listener::{Event as EventLib, Listener};
24
25// Error types
26const WAIT_ERR_STR: &str = "No notifier available";
27pub struct WaitError;
28
29impl fmt::Display for WaitError {
30    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
31        write!(f, "{self:?}")
32    }
33}
34
35impl fmt::Debug for WaitError {
36    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
37        f.write_str(WAIT_ERR_STR)
38    }
39}
40
41impl std::error::Error for WaitError {}
42
43#[repr(u8)]
44pub enum WaitDeadlineError {
45    Deadline,
46    WaitError,
47}
48
49impl fmt::Display for WaitDeadlineError {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        write!(f, "{self:?}")
52    }
53}
54
55impl fmt::Debug for WaitDeadlineError {
56    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
57        match self {
58            Self::Deadline => f.write_str("Deadline reached"),
59            Self::WaitError => f.write_str(WAIT_ERR_STR),
60        }
61    }
62}
63
64impl std::error::Error for WaitDeadlineError {}
65
66#[repr(u8)]
67pub enum WaitTimeoutError {
68    Timeout,
69    WaitError,
70}
71
72impl fmt::Display for WaitTimeoutError {
73    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74        write!(f, "{self:?}")
75    }
76}
77
78impl fmt::Debug for WaitTimeoutError {
79    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
80        match self {
81            Self::Timeout => f.write_str("Timeout expired"),
82            Self::WaitError => f.write_str(WAIT_ERR_STR),
83        }
84    }
85}
86
87impl std::error::Error for WaitTimeoutError {}
88
89const NOTIFY_ERR_STR: &str = "No waiter available";
90pub struct NotifyError;
91
92impl fmt::Display for NotifyError {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        write!(f, "{self:?}")
95    }
96}
97
98impl fmt::Debug for NotifyError {
99    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
100        f.write_str(NOTIFY_ERR_STR)
101    }
102}
103
104impl std::error::Error for NotifyError {}
105
106// Inner
107struct EventInner {
108    event: EventLib,
109    flag: AtomicU8,
110    notifiers: AtomicU16,
111    waiters: AtomicU16,
112}
113
114const UNSET: u8 = 0;
115const OK: u8 = 1;
116const ERR: u8 = 1 << 1;
117
118#[repr(u8)]
119enum EventCheck {
120    Unset = UNSET,
121    Ok = OK,
122    Err = ERR,
123}
124
125#[repr(u8)]
126enum EventSet {
127    Ok = OK,
128    Err = ERR,
129}
130
131impl EventInner {
132    fn check(&self) -> EventCheck {
133        let f = self.flag.fetch_and(!OK, Ordering::SeqCst);
134        if f & ERR != 0 {
135            return EventCheck::Err;
136        }
137        if f == OK {
138            return EventCheck::Ok;
139        }
140        EventCheck::Unset
141    }
142
143    fn set(&self) -> EventSet {
144        let f = self.flag.fetch_or(OK, Ordering::SeqCst);
145        if f & ERR != 0 {
146            return EventSet::Err;
147        }
148        EventSet::Ok
149    }
150
151    fn err(&self) {
152        self.flag.store(ERR, Ordering::SeqCst);
153    }
154}
155
156/// Creates a new lock-free event variable. Every time a [`Notifier`] calls ['Notifier::notify`], one [`Waiter`] will be waken-up.
157/// If no waiter is waiting when the `notify` is called, the notification will not be lost. That means the next waiter will return
158/// immediately when calling `wait`.
159pub fn new() -> (Notifier, Waiter) {
160    let inner = Arc::new(EventInner {
161        event: EventLib::new(),
162        flag: AtomicU8::new(UNSET),
163        notifiers: AtomicU16::new(1),
164        waiters: AtomicU16::new(1),
165    });
166    (Notifier(inner.clone()), Waiter(inner))
167}
168
169/// A [`Notifier`] is used to notify and wake up one and only one [`Waiter`].
170#[repr(transparent)]
171pub struct Notifier(Arc<EventInner>);
172
173impl std::fmt::Debug for Notifier {
174    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175        f.debug_tuple("Notifier").field(&"..").finish()
176    }
177}
178
179impl Notifier {
180    /// Notifies one pending listener
181    #[inline]
182    pub fn notify(&self) -> Result<(), NotifyError> {
183        // Set the flag.
184        match self.0.set() {
185            EventSet::Ok => {
186                self.0.event.notify_additional_relaxed(1);
187                Ok(())
188            }
189            EventSet::Err => Err(NotifyError),
190        }
191    }
192}
193
194impl Clone for Notifier {
195    fn clone(&self) -> Self {
196        let n = self.0.notifiers.fetch_add(1, Ordering::SeqCst);
197        // Panic on overflow
198        assert!(n != 0);
199        Self(self.0.clone())
200    }
201}
202
203impl Drop for Notifier {
204    fn drop(&mut self) {
205        let n = self.0.notifiers.fetch_sub(1, Ordering::SeqCst);
206        if n == 1 {
207            // The last Notifier has been dropped, close the event and notify everyone
208            self.0.err();
209            self.0.event.notify(usize::MAX);
210        }
211    }
212}
213
214#[repr(transparent)]
215pub struct Waiter(Arc<EventInner>);
216
217impl std::fmt::Debug for Waiter {
218    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
219        f.debug_tuple("Waiter").field(&"..").finish()
220    }
221}
222
223impl Waiter {
224    /// Waits for the condition to be notified
225    #[inline]
226    pub async fn wait_async(&self) -> Result<(), WaitError> {
227        // Wait until the flag is set.
228        loop {
229            // Check the flag.
230            match self.0.check() {
231                EventCheck::Ok => break,
232                EventCheck::Unset => {}
233                EventCheck::Err => return Err(WaitError),
234            }
235
236            // Start listening for events.
237            let listener = self.0.event.listen();
238
239            // Check the flag again after creating the listener.
240            match self.0.check() {
241                EventCheck::Ok => break,
242                EventCheck::Unset => {}
243                EventCheck::Err => return Err(WaitError),
244            }
245
246            // Wait for a notification and continue the loop.
247            listener.await;
248        }
249
250        Ok(())
251    }
252
253    /// Waits for the condition to be notified
254    #[inline]
255    pub fn wait(&self) -> Result<(), WaitError> {
256        // Wait until the flag is set.
257        loop {
258            // Check the flag.
259            match self.0.check() {
260                EventCheck::Ok => break,
261                EventCheck::Unset => {}
262                EventCheck::Err => return Err(WaitError),
263            }
264
265            // Start listening for events.
266            let listener = self.0.event.listen();
267
268            // Check the flag again after creating the listener.
269            match self.0.check() {
270                EventCheck::Ok => break,
271                EventCheck::Unset => {}
272                EventCheck::Err => return Err(WaitError),
273            }
274
275            // Wait for a notification and continue the loop.
276            listener.wait();
277        }
278
279        Ok(())
280    }
281
282    /// Waits for the condition to be notified or returns an error when the deadline is reached
283    #[inline]
284    pub fn wait_deadline(&self, deadline: Instant) -> Result<(), WaitDeadlineError> {
285        // Wait until the flag is set.
286        loop {
287            // Check the flag.
288            match self.0.check() {
289                EventCheck::Ok => break,
290                EventCheck::Unset => {}
291                EventCheck::Err => return Err(WaitDeadlineError::WaitError),
292            }
293
294            // Start listening for events.
295            let listener = self.0.event.listen();
296
297            // Check the flag again after creating the listener.
298            match self.0.check() {
299                EventCheck::Ok => break,
300                EventCheck::Unset => {}
301                EventCheck::Err => return Err(WaitDeadlineError::WaitError),
302            }
303
304            // Wait for a notification and continue the loop.
305            if listener.wait_deadline(deadline).is_none() {
306                return Err(WaitDeadlineError::Deadline);
307            }
308        }
309
310        Ok(())
311    }
312
313    /// Waits for the condition to be notified or returns an error when the timeout is expired
314    #[inline]
315    pub fn wait_timeout(&self, timeout: Duration) -> Result<(), WaitTimeoutError> {
316        // Wait until the flag is set.
317        loop {
318            // Check the flag.
319            match self.0.check() {
320                EventCheck::Ok => break,
321                EventCheck::Unset => {}
322                EventCheck::Err => return Err(WaitTimeoutError::WaitError),
323            }
324
325            // Start listening for events.
326            let listener = self.0.event.listen();
327
328            // Check the flag again after creating the listener.
329            match self.0.check() {
330                EventCheck::Ok => break,
331                EventCheck::Unset => {}
332                EventCheck::Err => return Err(WaitTimeoutError::WaitError),
333            }
334
335            // Wait for a notification and continue the loop.
336            if listener.wait_timeout(timeout).is_none() {
337                return Err(WaitTimeoutError::Timeout);
338            }
339        }
340
341        Ok(())
342    }
343}
344
345impl Clone for Waiter {
346    fn clone(&self) -> Self {
347        let n = self.0.waiters.fetch_add(1, Ordering::Relaxed);
348        // Panic on overflow
349        assert!(n != 0);
350        Self(self.0.clone())
351    }
352}
353
354impl Drop for Waiter {
355    fn drop(&mut self) {
356        let n = self.0.waiters.fetch_sub(1, Ordering::SeqCst);
357        if n == 1 {
358            // The last Waiter has been dropped, close the event
359            self.0.err();
360        }
361    }
362}
363
364mod tests {
365    #[test]
366    fn event_timeout() {
367        use std::{
368            sync::{Arc, Barrier},
369            time::Duration,
370        };
371
372        use crate::WaitTimeoutError;
373
374        let barrier = Arc::new(Barrier::new(2));
375        let (notifier, waiter) = super::new();
376        let tslot = Duration::from_secs(1);
377
378        let bs = barrier.clone();
379        let s = std::thread::spawn(move || {
380            // 1 - Wait one notification
381            match waiter.wait_timeout(tslot) {
382                Ok(()) => {}
383                Err(WaitTimeoutError::Timeout) => panic!("Timeout {tslot:#?}"),
384                Err(WaitTimeoutError::WaitError) => panic!("Event closed"),
385            }
386
387            bs.wait();
388
389            // 2 - Being notified twice but waiting only once
390            bs.wait();
391
392            match waiter.wait_timeout(tslot) {
393                Ok(()) => {}
394                Err(WaitTimeoutError::Timeout) => panic!("Timeout {tslot:#?}"),
395                Err(WaitTimeoutError::WaitError) => panic!("Event closed"),
396            }
397
398            match waiter.wait_timeout(tslot) {
399                Ok(()) => panic!("Event Ok but it should be Timeout"),
400                Err(WaitTimeoutError::Timeout) => {}
401                Err(WaitTimeoutError::WaitError) => panic!("Event closed"),
402            }
403
404            bs.wait();
405
406            // 3 - Notifier has been dropped
407            bs.wait();
408
409            waiter.wait().unwrap_err();
410
411            bs.wait();
412        });
413
414        let bp = barrier.clone();
415        let p = std::thread::spawn(move || {
416            // 1 - Notify once
417            notifier.notify().unwrap();
418
419            bp.wait();
420
421            // 2 - Notify twice
422            notifier.notify().unwrap();
423            notifier.notify().unwrap();
424
425            bp.wait();
426            bp.wait();
427
428            // 3 - Drop notifier yielding an error in the waiter
429            drop(notifier);
430
431            bp.wait();
432            bp.wait();
433        });
434
435        s.join().unwrap();
436        p.join().unwrap();
437    }
438
439    #[test]
440    fn event_deadline() {
441        use std::{
442            sync::{Arc, Barrier},
443            time::{Duration, Instant},
444        };
445
446        use crate::WaitDeadlineError;
447
448        let barrier = Arc::new(Barrier::new(2));
449        let (notifier, waiter) = super::new();
450        let tslot = Duration::from_secs(1);
451
452        let bs = barrier.clone();
453        let s = std::thread::spawn(move || {
454            // 1 - Wait one notification
455            match waiter.wait_deadline(Instant::now() + tslot) {
456                Ok(()) => {}
457                Err(WaitDeadlineError::Deadline) => panic!("Timeout {tslot:#?}"),
458                Err(WaitDeadlineError::WaitError) => panic!("Event closed"),
459            }
460
461            bs.wait();
462
463            // 2 - Being notified twice but waiting only once
464            bs.wait();
465
466            match waiter.wait_deadline(Instant::now() + tslot) {
467                Ok(()) => {}
468                Err(WaitDeadlineError::Deadline) => panic!("Timeout {tslot:#?}"),
469                Err(WaitDeadlineError::WaitError) => panic!("Event closed"),
470            }
471
472            match waiter.wait_deadline(Instant::now() + tslot) {
473                Ok(()) => panic!("Event Ok but it should be Timeout"),
474                Err(WaitDeadlineError::Deadline) => {}
475                Err(WaitDeadlineError::WaitError) => panic!("Event closed"),
476            }
477
478            bs.wait();
479
480            // 3 - Notifier has been dropped
481            bs.wait();
482
483            waiter.wait().unwrap_err();
484
485            bs.wait();
486        });
487
488        let bp = barrier.clone();
489        let p = std::thread::spawn(move || {
490            // 1 - Notify once
491            notifier.notify().unwrap();
492
493            bp.wait();
494
495            // 2 - Notify twice
496            notifier.notify().unwrap();
497            notifier.notify().unwrap();
498
499            bp.wait();
500            bp.wait();
501
502            // 3 - Drop notifier yielding an error in the waiter
503            drop(notifier);
504
505            bp.wait();
506            bp.wait();
507        });
508
509        s.join().unwrap();
510        p.join().unwrap();
511    }
512
513    #[test]
514    fn event_loop() {
515        use std::{
516            sync::{
517                atomic::{AtomicUsize, Ordering},
518                Arc, Barrier,
519            },
520            time::{Duration, Instant},
521        };
522
523        const N: usize = 1_000;
524        static COUNTER: AtomicUsize = AtomicUsize::new(0);
525
526        let (notifier, waiter) = super::new();
527        let barrier = Arc::new(Barrier::new(2));
528
529        let bs = barrier.clone();
530        let s = std::thread::spawn(move || {
531            for _ in 0..N {
532                waiter.wait().unwrap();
533                COUNTER.fetch_add(1, Ordering::Relaxed);
534                bs.wait();
535            }
536        });
537        let p = std::thread::spawn(move || {
538            for _ in 0..N {
539                notifier.notify().unwrap();
540                barrier.wait();
541            }
542        });
543
544        let start = Instant::now();
545        let tout = Duration::from_secs(60);
546        loop {
547            let n = COUNTER.load(Ordering::Relaxed);
548            if n == N {
549                break;
550            }
551            if start.elapsed() > tout {
552                panic!("Timeout {tout:#?}. Counter: {n}/{N}");
553            }
554
555            std::thread::sleep(Duration::from_millis(100));
556        }
557
558        s.join().unwrap();
559        p.join().unwrap();
560    }
561
562    #[test]
563    fn event_multiple() {
564        use std::{
565            sync::atomic::{AtomicUsize, Ordering},
566            time::{Duration, Instant},
567        };
568
569        const N: usize = 1_000;
570        static COUNTER: AtomicUsize = AtomicUsize::new(0);
571
572        let (notifier, waiter) = super::new();
573
574        let w1 = waiter.clone();
575        let s1 = std::thread::spawn(move || {
576            let mut n = 0;
577            while COUNTER.fetch_add(1, Ordering::Relaxed) < N - 2 {
578                w1.wait().unwrap();
579                n += 1;
580            }
581            println!("S1: {n}");
582        });
583        let s2 = std::thread::spawn(move || {
584            let mut n = 0;
585            while COUNTER.fetch_add(1, Ordering::Relaxed) < N - 2 {
586                waiter.wait().unwrap();
587                n += 1;
588            }
589            println!("S2: {n}");
590        });
591
592        let n1 = notifier.clone();
593        let p1 = std::thread::spawn(move || {
594            let mut n = 0;
595            while COUNTER.load(Ordering::Relaxed) < N {
596                n1.notify().unwrap();
597                n += 1;
598                std::thread::sleep(Duration::from_millis(1));
599            }
600            println!("P1: {n}");
601        });
602        let p2 = std::thread::spawn(move || {
603            let mut n = 0;
604            while COUNTER.load(Ordering::Relaxed) < N {
605                notifier.notify().unwrap();
606                n += 1;
607                std::thread::sleep(Duration::from_millis(1));
608            }
609            println!("P2: {n}");
610        });
611
612        std::thread::spawn(move || {
613            let start = Instant::now();
614            let tout = Duration::from_secs(60);
615            loop {
616                let n = COUNTER.load(Ordering::Relaxed);
617                if n == N {
618                    break;
619                }
620                if start.elapsed() > tout {
621                    panic!("Timeout {tout:#?}. Counter: {n}/{N}");
622                }
623
624                std::thread::sleep(Duration::from_millis(100));
625            }
626        });
627
628        p1.join().unwrap();
629        p2.join().unwrap();
630
631        s1.join().unwrap();
632        s2.join().unwrap();
633    }
634}
zenoh_sync/event.rs

zenoh_sync/
event.rs
