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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
//! Ping to the event loop
//!
//! This is an event source that just produces `()` events whevener the associated
//! [`Ping::ping`](Ping#method.ping) method is called. If the event source is pinged multiple times
//! between a single dispatching, it'll only generate one event.
//!
//! This event loop is a simple way of waking up the event loop from an other part of your program
//! (and is what backs the [`LoopSignal`](crate::LoopSignal)). It can also be used as a building
//! block to construct event sources whose source of event is not file descriptor, but rather an
//! userspace source (like an other thread).

use std::{os::unix::io::RawFd, sync::Arc};

use nix::{
    fcntl::OFlag,
    unistd::{close, pipe2, read, write},
};

use super::generic::{Fd, Generic};
use crate::{no_nix_err, EventSource, Interest, Mode, Poll, Readiness, Token};

/// Create a new ping event source
///
/// you are given a `Ping` instance, which can be cloned and used to ping the
/// event loop, and a `PingSource`, which you can insert in your event loop to
/// receive the pings.
pub fn make_ping() -> std::io::Result<(Ping, PingSource)> {
    let (read, write) = pipe2(OFlag::O_CLOEXEC | OFlag::O_NONBLOCK).map_err(no_nix_err)?;
    let source = PingSource {
        pipe: Generic::from_fd(read, Interest::READ, Mode::Level),
    };
    let ping = Ping {
        pipe: Arc::new(CloseOnDrop(write)),
    };
    Ok((ping, source))
}

/// The ping event source
///
/// You can insert it in your event loop to receive pings.
pub struct PingSource {
    pipe: Generic<Fd>,
}

impl EventSource for PingSource {
    type Event = ();
    type Metadata = ();
    type Ret = ();

    fn process_events<C>(
        &mut self,
        readiness: Readiness,
        token: Token,
        mut callback: C,
    ) -> std::io::Result<()>
    where
        C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
    {
        self.pipe.process_events(readiness, token, |_, fd| {
            let mut buf = [0u8; 32];
            let mut read_something = false;
            // If we have already disabled ourselves, do nothing.
            if fd.0 == -1 {
                return Ok(());
            }
            loop {
                match read(fd.0, &mut buf) {
                    Ok(0) => {
                        // The other end of the pipe was closed, close the file descriptor
                        // to disable ourselves
                        if let Err(e) = close(fd.0) {
                            log::warn!("[calloop] Failed to close read ping: {:?}", e);
                        }
                        fd.0 = -1;
                        break;
                    }
                    Ok(_) => read_something = true,
                    Err(e) => {
                        let e = no_nix_err(e);
                        if e.kind() == std::io::ErrorKind::WouldBlock {
                            break;
                        // nothing more to read
                        } else {
                            // propagate error
                            return Err(e);
                        }
                    }
                }
            }
            if read_something {
                callback((), &mut ());
            }
            Ok(())
        })
    }

    fn register(&mut self, poll: &mut Poll, token: Token) -> std::io::Result<()> {
        if self.pipe.file.0 >= 0 {
            self.pipe.register(poll, token)
        } else {
            // do nothing if the other end was already closed
            Ok(())
        }
    }

    fn reregister(&mut self, poll: &mut Poll, token: Token) -> std::io::Result<()> {
        if self.pipe.file.0 >= 0 {
            self.pipe.reregister(poll, token)
        } else {
            // do nothing if the other end was already closed
            Ok(())
        }
    }

    fn unregister(&mut self, poll: &mut Poll) -> std::io::Result<()> {
        if self.pipe.file.0 >= 0 {
            self.pipe.unregister(poll)
        } else {
            // do nothing if the other end was already closed
            Ok(())
        }
    }
}

impl Drop for PingSource {
    fn drop(&mut self) {
        // close the fd if it wasn't already
        if self.pipe.file.0 >= 0 {
            if let Err(e) = close(self.pipe.file.0) {
                log::warn!("[calloop] Failed to close read ping: {:?}", e);
            }
        }
    }
}

/// The Ping handle
///
/// This handle can be cloned and sent accross threads. It can be used to
/// send pings to the `PingSource`.
#[derive(Clone)]
pub struct Ping {
    pipe: Arc<CloseOnDrop>,
}

impl Ping {
    /// Send a ping to the `PingSource`
    pub fn ping(&self) {
        if let Err(e) = write(self.pipe.0, &[0u8]) {
            log::warn!("[calloop] Failed to write a ping: {:?}", e);
        }
    }
}

struct CloseOnDrop(RawFd);

impl Drop for CloseOnDrop {
    fn drop(&mut self) {
        if let Err(e) = close(self.0) {
            log::warn!("[calloop] Failed to close write ping: {:?}", e);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ping() {
        let mut event_loop = crate::EventLoop::<bool>::try_new().unwrap();

        let (ping, source) = make_ping().unwrap();

        event_loop
            .handle()
            .insert_source(source, |(), &mut (), dispatched| *dispatched = true)
            .unwrap();

        ping.ping();

        let mut dispatched = false;
        event_loop
            .dispatch(std::time::Duration::from_millis(0), &mut dispatched)
            .unwrap();
        assert!(dispatched);

        // Ping has been drained an no longer generates events
        let mut dispatched = false;
        event_loop
            .dispatch(std::time::Duration::from_millis(0), &mut dispatched)
            .unwrap();
        assert!(!dispatched);
    }

    #[test]
    fn ping_closed() {
        let mut event_loop = crate::EventLoop::<bool>::try_new().unwrap();

        let (_, source) = make_ping().unwrap();
        event_loop
            .handle()
            .insert_source(source, |(), &mut (), dispatched| *dispatched = true)
            .unwrap();

        let mut dispatched = false;

        // If the sender is closed from the start, the ping should first trigger
        // once, disabling itself but now invoking the callback
        event_loop
            .dispatch(std::time::Duration::from_millis(0), &mut dispatched)
            .unwrap();
        assert!(!dispatched);

        // Then it should not trigger any more, so this dispatch should wait the whole 100ms
        let now = std::time::Instant::now();
        event_loop
            .dispatch(std::time::Duration::from_millis(100), &mut dispatched)
            .unwrap();
        assert!(now.elapsed() >= std::time::Duration::from_millis(100));
    }
}