Initial attempt at implementing enif_select

rustler/src/lib.rs

@@ -47,7 +47,10 @@ pub use crate::types::{
 pub use crate::types::BigInt;
 
 mod resource;
-pub use crate::resource::{Monitor, Resource, ResourceArc, ResourceInitError};
+pub use crate::resource::{
+    Event, Monitor, Resource, ResourceArc, ResourceInitError, SelectError, SelectMode,
+    SelectResult, SelectReturn,
+};
 
 #[doc(hidden)]
 pub mod dynamic;

rustler/src/resource/mod.rs

@@ -8,6 +8,7 @@ mod arc;
 mod error;
 mod monitor;
 mod registration;
+mod select;
 mod term;
 mod traits;
 mod util;
@@ -16,5 +17,6 @@ pub use arc::ResourceArc;
 pub use error::*;
 pub use monitor::Monitor;
 pub use registration::Registration;
+pub use select::{Event, SelectError, SelectMode, SelectResult, SelectReturn};
 pub use traits::Resource;
 use traits::ResourceExt;

rustler/src/resource/registration.rs

@@ -3,8 +3,9 @@ use super::util::align_alloced_mem_for_struct;
 use super::ResourceInitError;
 use crate::env::EnvKind;
 use crate::sys::{
-    c_char, c_void, ErlNifEnv, ErlNifMonitor, ErlNifPid, ErlNifResourceDown, ErlNifResourceDtor,
-    ErlNifResourceFlags, ErlNifResourceType, ErlNifResourceTypeInit,
+    c_char, c_int, c_void, ErlNifEnv, ErlNifEvent, ErlNifMonitor, ErlNifPid, ErlNifResourceDown,
+    ErlNifResourceDtor, ErlNifResourceFlags, ErlNifResourceStop, ErlNifResourceType,
+    ErlNifResourceTypeInit,
 };
 use crate::{Env, LocalPid, Monitor, Resource};
 use std::any::TypeId;
@@ -65,6 +66,7 @@ impl Registration {
             type_name: None,
         }
         .maybe_add_destructor_callback::<T>()
+        .maybe_add_stop_callback::<T>()
         .maybe_add_down_callback::<T>()
         .maybe_add_dyncall_callback::<T>()
     }
@@ -92,6 +94,22 @@ impl Registration {
         }
     }
 
+    const fn maybe_add_stop_callback<T: Resource>(self) -> Self {
+        if T::IMPLEMENTS_STOP {
+            Self {
+                init: ErlNifResourceTypeInit {
+                    stop: resource_stop::<T> as *const ErlNifResourceStop,
+                    #[cfg(feature = "nif_version_2_16")]
+                    members: max(self.init.members, 2),
+                    ..self.init
+                },
+                ..self
+            }
+        } else {
+            self
+        }
+    }
+
     const fn maybe_add_down_callback<T: Resource>(self) -> Self {
         if T::IMPLEMENTS_DOWN {
             Self {
@@ -175,6 +193,21 @@ where
     });
 }
 
+unsafe extern "C" fn resource_stop<T>(
+    caller_env: *mut ErlNifEnv,
+    handle: *mut c_void,
+    event: ErlNifEvent,
+    is_direct_call: c_int,
+) where
+    T: Resource,
+{
+    let env = Env::new_internal(&caller_env, caller_env, EnvKind::Callback);
+    let aligned = align_alloced_mem_for_struct::<T>(handle);
+    let obj = ptr::read::<T>(aligned as *mut T);
+
+    obj.stop(env, event.into(), is_direct_call != 0);
+}
+
 unsafe extern "C" fn resource_down<T: Resource>(
     env: *mut ErlNifEnv,
     obj: *mut c_void,

rustler/src/resource/select.rs

@@ -0,0 +1,185 @@
+use std::ffi::c_int;
+use std::os::fd::AsRawFd;
+
+use crate::sys::{
+    enif_select, ErlNifEvent, ErlNifSelectFlags, ERL_NIF_SELECT_ERROR_CANCELLED,
+    ERL_NIF_SELECT_FAILED, ERL_NIF_SELECT_INVALID_EVENT, ERL_NIF_SELECT_NOTSUP,
+    ERL_NIF_SELECT_READ_CANCELLED, ERL_NIF_SELECT_STOP_CALLED, ERL_NIF_SELECT_STOP_SCHEDULED,
+    ERL_NIF_SELECT_WRITE_CANCELLED,
+};
+use crate::types::atom::undefined;
+use crate::{Encoder, Env, LocalPid, Reference, Resource, ResourceArc};
+
+macro_rules! as_raw {
+    () => {
+        AsRawFd
+    };
+}
+
+#[cfg(windows)]
+macro_rules! as_raw {
+    () => {
+        AsRawHandle
+    };
+}
+
+macro_rules!  {
+    () => {
+
+    };
+}
+
+macro_rules! getter {
+    (pub $name:ident, $flag:ident) => {
+        #[inline]
+        pub fn $name(self) -> bool {
+            self.0 & $flag != 0
+        }
+    };
+    ($name:ident, $flag:ident) => {
+        #[inline]
+        fn $name(self) -> bool {
+            self.0 & $flag != 0
+        }
+    };
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct SelectReturn(c_int);
+
+pub type SelectResult = Result<SelectReturn, SelectError>;
+
+impl From<SelectReturn> for SelectResult {
+    fn from(val: SelectReturn) -> Self {
+        use SelectError::*;
+
+        if val.0 < 0 {
+            if val.invalid_event() {
+                Err(InvalidEvent)
+            } else if val.failed() {
+                Err(Failed)
+            } else if val.not_supported() {
+                Err(NotSupported)
+            } else {
+                Err(Unknown)
+            }
+        } else {
+            Ok(val)
+        }
+    }
+}
+
+impl SelectReturn {
+    pub fn cancelled(self) -> bool {
+        self.read_cancelled() || self.write_cancelled() || self.error_cancelled()
+    }
+
+    getter! {pub stop_called, ERL_NIF_SELECT_STOP_CALLED}
+    getter! {pub stop_scheduled, ERL_NIF_SELECT_STOP_SCHEDULED}
+    getter! {pub read_cancelled, ERL_NIF_SELECT_READ_CANCELLED}
+    getter! {pub write_cancelled, ERL_NIF_SELECT_WRITE_CANCELLED}
+    getter! {pub error_cancelled, ERL_NIF_SELECT_ERROR_CANCELLED}
+    getter! {invalid_event, ERL_NIF_SELECT_INVALID_EVENT}
+    getter! {failed, ERL_NIF_SELECT_FAILED}
+    getter! {not_supported, ERL_NIF_SELECT_NOTSUP}
+}
+
+#[derive(Clone, Copy, Debug)]
+pub enum SelectError {
+    InvalidEvent,
+    Failed,
+    NotSupported,
+    Unknown,
+}
+
+impl From<Event> for ErlNifEvent {
+    fn from(val: Event) -> Self {
+        val.0
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub enum SelectMode {
+    Read,
+    Write,
+    ReadWrite,
+}
+
+impl SelectMode {
+    fn to_flags(self) -> c_int {
+        use ErlNifSelectFlags::*;
+
+        match self {
+            SelectMode::Read => ERL_NIF_SELECT_READ as c_int,
+            SelectMode::Write => ERL_NIF_SELECT_WRITE as c_int,
+            SelectMode::ReadWrite => ERL_NIF_SELECT_READ as c_int | ERL_NIF_SELECT_WRITE as c_int,
+        }
+    }
+}
+
+impl<T> ResourceArc<T>
+where
+    T: Resource,
+{
+    fn select_internal<'a, E: AsRawFd>(
+        &self,
+        env: Env<'a>,
+        event: E,
+        mode: SelectMode,
+        pid: Option<LocalPid>,
+        reference: Option<Reference>,
+    ) -> SelectResult {
+        let reference = match reference {
+            Some(reference) => reference.encode(env),
+            None => undefined().encode(env),
+        }
+        .as_c_arg();
+
+        let pid = pid.map_or(std::ptr::null(), |p| p.as_c_arg());
+
+        let res = unsafe {
+            enif_select(
+                env.as_c_arg(),
+                event.0,
+                mode.to_flags(),
+                self.as_c_arg(),
+                pid,
+                reference,
+            )
+        };
+
+        SelectReturn(res).into()
+    }
+
+    // TODO: select_read/select_write with an optional custom message
+
+    pub fn cancel<'a>(&self, env: Env<'a>, event: &Event, mode: SelectMode) -> SelectResult {
+        let res = unsafe {
+            enif_select(
+                env.as_c_arg(),
+                event.0,
+                mode.to_flags() | ErlNifSelectFlags::ERL_NIF_SELECT_CANCEL as c_int,
+                self.as_c_arg(),
+                std::ptr::null(),
+                0usize,
+            )
+        };
+
+        SelectReturn(res).into()
+    }
+
+    pub fn stop<'a>(&self, env: Env<'a>, event: &Event) -> SelectResult {
+        let res = unsafe {
+            enif_select(
+                env.as_c_arg(),
+                event.0,
+                ErlNifSelectFlags::ERL_NIF_SELECT_STOP as c_int,
+                self.as_c_arg(),
+                std::ptr::null(),
+                0usize,
+            )
+        };
+
+        SelectReturn(res).into()
+    }
+}

rustler/src/resource/traits.rs

@@ -2,6 +2,7 @@ use std::any::TypeId;
 use std::collections::HashMap;
 use std::sync::OnceLock;
 
+use crate::resource::Event;
 use crate::sys::ErlNifResourceType;
 use crate::{Env, LocalPid, Monitor};
 
@@ -30,11 +31,12 @@ pub(crate) unsafe fn register_resource_type(type_id: TypeId, resource_type: NifR
 /// In particular, the type needs to handle all synchronization itself (thus we require it to
 /// implement `Sync`) and callbacks or NIFs can run on arbitrary threads (thus we require `Send`).
 ///
-/// Currently only `destructor` and `down` callbacks are possible. If a callback is implemented,
-/// the respective associated constant `IMPLEMENTS_...` must be set to `true` for the registration
-/// to take it into account. All callbacks provide (empty) default implementations.
+/// If a callback is implemented, the respective associated constant `IMPLEMENTS_...`
+/// must be set to `true` for the registration to take it into account.
+/// All callbacks provide (empty) default implementations.
 pub trait Resource: Sized + Send + Sync + 'static {
     const IMPLEMENTS_DESTRUCTOR: bool = false;
+    const IMPLEMENTS_STOP: bool = false;
     const IMPLEMENTS_DOWN: bool = false;
 
     #[cfg(feature = "nif_version_2_16")]
@@ -49,6 +51,14 @@ pub trait Resource: Sized + Send + Sync + 'static {
     #[allow(unused_mut, unused)]
     fn destructor(mut self, env: Env<'_>) {}
 
+    /// The stop callback of a resource.
+    /// It is called on the behalf of [`enif_select()`](crate::sys::enif_select).
+    ///
+    /// - event is the OS event
+    /// - is_direct_call is true if the call is made directly from enif_select or false if it is a scheduled call (potentially from another thread).
+    #[allow(unused_mut, unused)]
+    fn stop<'a>(&'a self, env: Env<'a>, event: Event, direct_call: bool) {}
+
     /// Callback function to handle process monitoring.
     ///
     /// This callback is called when a process monitored using `Env::monitor` terminates

rustler/src/sys/types.rs

@@ -163,16 +163,29 @@ pub struct ErlNifResourceTypeInit {
 }
 
 /// See [ErlNifSelectFlags](http://erlang.org/doc/man/erl_nif.html#ErlNifSelectFlags) in the Erlang docs.
-pub type ErlNifSelectFlags = c_int;
+#[derive(Debug, Copy, Clone)]
+#[repr(C)]
+pub enum ErlNifSelectFlags {
+    ERL_NIF_SELECT_READ = 1 << 0,
+    ERL_NIF_SELECT_WRITE = 1 << 1,
+    ERL_NIF_SELECT_STOP = 1 << 2,
+    ERL_NIF_SELECT_CANCEL = 1 << 3,
+    ERL_NIF_SELECT_CUSTOM_MSG = 1 << 4,
+    ERL_NIF_SELECT_ERROR = 1 << 5,
+}
+
+/// See [enif_select](https://www.erlang.org/doc/apps/erts/erl_nif.html#enif_select) in the Erlang docs.
 #[allow(clippy::identity_op)]
-pub const ERL_NIF_SELECT_READ: ErlNifSelectFlags = 1 << 0;
-pub const ERL_NIF_SELECT_WRITE: ErlNifSelectFlags = 1 << 1;
-pub const ERL_NIF_SELECT_STOP: ErlNifSelectFlags = 1 << 2;
-pub const ERL_NIF_SELECT_FAILED: ErlNifSelectFlags = 1 << 3;
-pub const ERL_NIF_SELECT_READ_CANCELLED: ErlNifSelectFlags = 1 << 4;
-pub const ERL_NIF_SELECT_WRITE_CANCELLED: ErlNifSelectFlags = 1 << 5;
-pub const ERL_NIF_SELECT_ERROR_CANCELLED: ErlNifSelectFlags = 1 << 6;
-pub const ERL_NIF_SELECT_NOTSUP: ErlNifSelectFlags = 1 << 7;
+pub type ErlNifSelectReturnType = c_int;
+
+pub const ERL_NIF_SELECT_STOP_CALLED: ErlNifSelectReturnType = 1 << 0;
+pub const ERL_NIF_SELECT_STOP_SCHEDULED: ErlNifSelectReturnType = 1 << 1;
+pub const ERL_NIF_SELECT_INVALID_EVENT: ErlNifSelectReturnType = 1 << 2;
+pub const ERL_NIF_SELECT_FAILED: ErlNifSelectReturnType = 1 << 3;
+pub const ERL_NIF_SELECT_READ_CANCELLED: ErlNifSelectReturnType = 1 << 4;
+pub const ERL_NIF_SELECT_WRITE_CANCELLED: ErlNifSelectReturnType = 1 << 5;
+pub const ERL_NIF_SELECT_ERROR_CANCELLED: ErlNifSelectReturnType = 1 << 6;
+pub const ERL_NIF_SELECT_NOTSUP: ErlNifSelectReturnType = 1 << 7;
 
 /// See [ErlNifMonitor](http://www.erlang.org/doc/man/erl_nif.html#ErlNifMonitor) in the Erlang docs.
 #[derive(Debug, Copy, Clone)]