calloop/sources/

generic.rs

//! A generic event source wrapping an IO objects or file descriptor
//!
//! You can use this general purpose adapter around file-descriptor backed objects to
//! insert into an [`EventLoop`](crate::EventLoop).
//!
//! The event generated by this [`Generic`] event source are the [`Readiness`](crate::Readiness)
//! notification itself, and the monitored object is provided to your callback as the second
//! argument.
//!
#![cfg_attr(unix, doc = "```")]
#![cfg_attr(not(unix), doc = "```no_run")]
//! # extern crate calloop;
//! use calloop::{generic::Generic, Interest, Mode, PostAction};
//!
//! # fn main() {
//! # let mut event_loop = calloop::EventLoop::<()>::try_new()
//! #                .expect("Failed to initialize the event loop!");
//! # let handle = event_loop.handle();
//! # #[cfg(unix)]
//! # let io_object = std::io::stdin();
//! # #[cfg(windows)]
//! # let io_object: std::net::TcpStream = panic!();
//! handle.insert_source(
//!     // wrap your IO object in a Generic, here we register for read readiness
//!     // in level-triggering mode
//!     Generic::new(io_object, Interest::READ, Mode::Level),
//!     |readiness, io_object, shared_data| {
//!         // The first argument of the callback is a Readiness
//!         // The second is a &mut reference to your object
//!
//!         // your callback needs to return a Result<PostAction, std::io::Error>
//!         // if it returns an error, the event loop will consider this event
//!         // event source as erroring and report it to the user.
//!         Ok(PostAction::Continue)
//!     }
//! );
//! # }
//! ```
//!
//! It can also help you implementing your own event sources: just have
//! these `Generic<_>` as fields of your event source, and delegate the
//! [`EventSource`](crate::EventSource) implementation to them.

use polling::Poller;
use std::{borrow, marker::PhantomData, ops, sync::Arc};

#[cfg(unix)]
use std::os::unix::io::{AsFd, AsRawFd, BorrowedFd};
#[cfg(windows)]
use std::os::windows::io::{
    AsRawSocket as AsRawFd, AsSocket as AsFd, BorrowedSocket as BorrowedFd,
};

use crate::{EventSource, Interest, Mode, Poll, PostAction, Readiness, Token, TokenFactory};

/// Wrapper to use a type implementing `AsRawFd` but not `AsFd` with `Generic`
#[derive(Debug)]
pub struct FdWrapper<T: AsRawFd>(T);

impl<T: AsRawFd> FdWrapper<T> {
    /// Wrap `inner` with an `AsFd` implementation.
    ///
    /// # Safety
    /// This is safe if the `AsRawFd` implementation of `inner` always returns
    /// a valid fd. This should usually be true for types implementing
    /// `AsRawFd`. But this isn't guaranteed with `FdWrapper<RawFd>`.
    pub unsafe fn new(inner: T) -> Self {
        Self(inner)
    }
}

impl<T: AsRawFd> ops::Deref for FdWrapper<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<T: AsRawFd> ops::DerefMut for FdWrapper<T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

impl<T: AsRawFd> AsFd for FdWrapper<T> {
    #[cfg(unix)]
    fn as_fd(&self) -> BorrowedFd {
        unsafe { BorrowedFd::borrow_raw(self.0.as_raw_fd()) }
    }

    #[cfg(windows)]
    fn as_socket(&self) -> BorrowedFd {
        unsafe { BorrowedFd::borrow_raw(self.0.as_raw_socket()) }
    }
}

/// A wrapper around a type that doesn't expose it mutably safely.
///
/// The [`EventSource`] trait's `Metadata` type demands mutable access to the inner I/O source.
/// However, the inner polling source used by `calloop` keeps the handle-based equivalent of an
/// immutable pointer to the underlying object's I/O handle. Therefore, if the inner source is
/// dropped, this leaves behind a dangling pointer which immediately invokes undefined behavior
/// on the next poll of the event loop.
///
/// In order to prevent this from happening, the [`Generic`] I/O source must not directly expose
/// a mutable reference to the underlying handle. This type wraps around the underlying handle and
/// easily allows users to take immutable (`&`) references to the type, but makes mutable (`&mut`)
/// references unsafe to get. Therefore, it prevents the source from being moved out and dropped
/// while it is still registered in the event loop.
///
/// [`EventSource`]: crate::EventSource
#[derive(Debug)]
pub struct NoIoDrop<T>(T);

impl<T> NoIoDrop<T> {
    /// Get a mutable reference.
    ///
    /// # Safety
    ///
    /// The inner type's I/O source must not be dropped.
    pub unsafe fn get_mut(&mut self) -> &mut T {
        &mut self.0
    }
}

impl<T> AsRef<T> for NoIoDrop<T> {
    fn as_ref(&self) -> &T {
        &self.0
    }
}

impl<T> borrow::Borrow<T> for NoIoDrop<T> {
    fn borrow(&self) -> &T {
        &self.0
    }
}

impl<T> ops::Deref for NoIoDrop<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<T: AsFd> AsFd for NoIoDrop<T> {
    #[cfg(unix)]
    fn as_fd(&self) -> BorrowedFd<'_> {
        // SAFETY: The innter type is not mutated.
        self.0.as_fd()
    }

    #[cfg(windows)]
    fn as_socket(&self) -> BorrowedFd<'_> {
        // SAFETY: The innter type is not mutated.
        self.0.as_socket()
    }
}

/// A generic event source wrapping a FD-backed type
#[derive(Debug)]
pub struct Generic<F: AsFd, E = std::io::Error> {
    /// The wrapped FD-backed type.
    ///
    /// This must be deregistered before it is dropped.
    file: Option<NoIoDrop<F>>,
    /// The programmed interest
    pub interest: Interest,
    /// The programmed mode
    pub mode: Mode,

    /// Back-reference to the poller.
    ///
    /// This is needed to drop the original file.
    poller: Option<Arc<Poller>>,

    // This token is used by the event loop logic to look up this source when an
    // event occurs.
    token: Option<Token>,

    // This allows us to make the associated error and return types generic.
    _error_type: PhantomData<E>,
}

impl<F: AsFd> Generic<F, std::io::Error> {
    /// Wrap a FD-backed type into a `Generic` event source that uses
    /// [`std::io::Error`] as its error type.
    pub fn new(file: F, interest: Interest, mode: Mode) -> Generic<F, std::io::Error> {
        Generic {
            file: Some(NoIoDrop(file)),
            interest,
            mode,
            token: None,
            poller: None,
            _error_type: PhantomData,
        }
    }

    /// Wrap a FD-backed type into a `Generic` event source using an arbitrary error type.
    pub fn new_with_error<E>(file: F, interest: Interest, mode: Mode) -> Generic<F, E> {
        Generic {
            file: Some(NoIoDrop(file)),
            interest,
            mode,
            token: None,
            poller: None,
            _error_type: PhantomData,
        }
    }
}

impl<F: AsFd, E> Generic<F, E> {
    /// Unwrap the `Generic` source to retrieve the underlying type
    pub fn unwrap(mut self) -> F {
        let NoIoDrop(file) = self.file.take().unwrap();

        // Remove it from the poller.
        if let Some(poller) = self.poller.take() {
            poller
                .delete(
                    #[cfg(unix)]
                    file.as_fd(),
                    #[cfg(windows)]
                    file.as_socket(),
                )
                .ok();
        }

        file
    }

    /// Get a reference to the underlying type.
    pub fn get_ref(&self) -> &F {
        &self.file.as_ref().unwrap().0
    }

    /// Get a mutable reference to the underlying type.
    ///
    /// # Safety
    ///
    /// This is unsafe because it allows you to modify the underlying type, which
    /// allows you to drop the underlying event source. Dropping the underlying source
    /// leads to a dangling reference.
    pub unsafe fn get_mut(&mut self) -> &mut F {
        self.file.as_mut().unwrap().get_mut()
    }
}

impl<F: AsFd, E> Drop for Generic<F, E> {
    fn drop(&mut self) {
        // Remove it from the poller.
        if let (Some(file), Some(poller)) = (self.file.take(), self.poller.take()) {
            poller
                .delete(
                    #[cfg(unix)]
                    file.as_fd(),
                    #[cfg(windows)]
                    file.as_socket(),
                )
                .ok();
        }
    }
}

impl<F, E> EventSource for Generic<F, E>
where
    F: AsFd,
    E: Into<Box<dyn std::error::Error + Send + Sync>>,
{
    type Event = Readiness;
    type Metadata = NoIoDrop<F>;
    type Ret = Result<PostAction, E>;
    type Error = E;

    fn process_events<C>(
        &mut self,
        readiness: Readiness,
        token: Token,
        mut callback: C,
    ) -> Result<PostAction, Self::Error>
    where
        C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
    {
        // If the token is invalid or not ours, skip processing.
        if self.token != Some(token) {
            return Ok(PostAction::Continue);
        }

        callback(readiness, self.file.as_mut().unwrap())
    }

    fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
        let token = token_factory.token();

        // SAFETY: We ensure that we have a poller to deregister with (see below).
        unsafe {
            poll.register(
                &self.file.as_ref().unwrap().0,
                self.interest,
                self.mode,
                token,
            )?;
        }

        // Make sure we can use the poller to deregister if need be.
        // But only if registration actually succeeded
        // So that we don't try to unregister the FD on drop if it wasn't registered
        // in the first place (for example if registration failed because of a duplicate insertion)
        self.poller = Some(poll.poller().clone());
        self.token = Some(token);

        Ok(())
    }

    fn reregister(
        &mut self,
        poll: &mut Poll,
        token_factory: &mut TokenFactory,
    ) -> crate::Result<()> {
        let token = token_factory.token();

        poll.reregister(
            &self.file.as_ref().unwrap().0,
            self.interest,
            self.mode,
            token,
        )?;

        self.token = Some(token);
        Ok(())
    }

    fn unregister(&mut self, poll: &mut Poll) -> crate::Result<()> {
        poll.unregister(&self.file.as_ref().unwrap().0)?;
        self.poller = None;
        self.token = None;
        Ok(())
    }
}

#[cfg(all(unix, test))]
mod tests {
    use std::io::{Read, Write};

    use super::Generic;
    use crate::{Dispatcher, Interest, Mode, PostAction};
    #[cfg(unix)]
    #[test]
    fn dispatch_unix() {
        use std::os::unix::net::UnixStream;

        let mut event_loop = crate::EventLoop::try_new().unwrap();

        let handle = event_loop.handle();

        let (mut tx, rx) = UnixStream::pair().unwrap();

        let generic = Generic::new(rx, Interest::READ, Mode::Level);

        let mut dispached = false;

        let _generic_token = handle
            .insert_source(generic, move |readiness, file, d| {
                assert!(readiness.readable);
                // we have not registered for writability
                assert!(!readiness.writable);
                let mut buffer = vec![0; 10];
                let ret = (&**file).read(&mut buffer).unwrap();
                assert_eq!(ret, 6);
                assert_eq!(&buffer[..6], &[1, 2, 3, 4, 5, 6]);

                *d = true;
                Ok(PostAction::Continue)
            })
            .unwrap();

        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut dispached)
            .unwrap();

        assert!(!dispached);

        let ret = tx.write(&[1, 2, 3, 4, 5, 6]).unwrap();
        assert_eq!(ret, 6);
        tx.flush().unwrap();

        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut dispached)
            .unwrap();

        assert!(dispached);
    }

    #[test]
    fn register_deregister_unix() {
        use std::os::unix::net::UnixStream;

        let mut event_loop = crate::EventLoop::try_new().unwrap();

        let handle = event_loop.handle();

        let (mut tx, rx) = UnixStream::pair().unwrap();

        let generic = Generic::new(rx, Interest::READ, Mode::Level);
        let dispatcher = Dispatcher::new(generic, move |_, _, d| {
            *d = true;
            Ok(PostAction::Continue)
        });

        let mut dispached = false;

        let generic_token = handle.register_dispatcher(dispatcher.clone()).unwrap();

        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut dispached)
            .unwrap();

        assert!(!dispached);

        // remove the source, and then write something

        event_loop.handle().remove(generic_token);

        let ret = tx.write(&[1, 2, 3, 4, 5, 6]).unwrap();
        assert_eq!(ret, 6);
        tx.flush().unwrap();

        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut dispached)
            .unwrap();

        // the source has not been dispatched, as the source is no longer here
        assert!(!dispached);

        // insert it again
        let generic = dispatcher.into_source_inner();
        let _generic_token = handle
            .insert_source(generic, move |readiness, file, d| {
                assert!(readiness.readable);
                // we have not registered for writability
                assert!(!readiness.writable);
                let mut buffer = vec![0; 10];
                let ret = (&**file).read(&mut buffer).unwrap();
                assert_eq!(ret, 6);
                assert_eq!(&buffer[..6], &[1, 2, 3, 4, 5, 6]);

                *d = true;
                Ok(PostAction::Continue)
            })
            .unwrap();

        event_loop
            .dispatch(Some(::std::time::Duration::ZERO), &mut dispached)
            .unwrap();

        // the has now been properly dispatched
        assert!(dispached);
    }

    // Duplicate insertion does not fail on all platforms, but does on Linux
    #[cfg(target_os = "linux")]
    #[test]
    fn duplicate_insert() {
        use std::os::unix::{
            io::{AsFd, BorrowedFd},
            net::UnixStream,
        };
        let event_loop = crate::EventLoop::<()>::try_new().unwrap();

        let handle = event_loop.handle();

        let (_, rx) = UnixStream::pair().unwrap();

        // Rc only implements AsFd since 1.69...
        struct RcFd<T> {
            rc: std::rc::Rc<T>,
        }

        impl<T: AsFd> AsFd for RcFd<T> {
            fn as_fd(&self) -> BorrowedFd<'_> {
                self.rc.as_fd()
            }
        }

        let rx = std::rc::Rc::new(rx);

        let token = handle
            .insert_source(
                Generic::new(RcFd { rc: rx.clone() }, Interest::READ, Mode::Level),
                |_, _, _| Ok(PostAction::Continue),
            )
            .unwrap();

        // inserting the same FD a second time should fail
        let ret = handle.insert_source(
            Generic::new(RcFd { rc: rx.clone() }, Interest::READ, Mode::Level),
            |_, _, _| Ok(PostAction::Continue),
        );
        assert!(ret.is_err());
        std::mem::drop(ret);

        // but the original token is still registered
        handle.update(&token).unwrap();
    }
}