iceoryx2/node/

mod.rs

// Copyright (c) 2024 Contributors to the Eclipse Foundation
//
// See the NOTICE file(s) distributed with this work for additional
// information regarding copyright ownership.
//
// This program and the accompanying materials are made available under the
// terms of the Apache Software License 2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0, or the MIT license
// which is available at https://opensource.org/licenses/MIT.
//
// SPDX-License-Identifier: Apache-2.0 OR MIT

//! The [`Node`](crate::node::Node) is the central entry point of iceoryx2. It is the owner of all communication
//! entities and provides additional memory to them to perform reference counting amongst other
//! things.
//!
//! It allows also the system to monitor the state of processes and cleanup stale resources of
//! dead processes.
//!
//! # Create a [`Node`](crate::node::Node)
//!
//! ```
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! let node = NodeBuilder::new()
//!                 .name(&"my_little_node".try_into()?)
//!                 .create::<ipc::Service>()?;
//!
//! println!("created node {:?}", node);
//! # Ok(())
//! # }
//! ```
//!
//! # List all existing [`Node`](crate::node::Node)s
//!
//! ```
//! use iceoryx2::prelude::*;
//!
//! Node::<ipc::Service>::list(Config::global_config(), |node_state| {
//!     println!("found node {:?}", node_state);
//!     CallbackProgression::Continue
//! });
//! ```
//!
//! # Cleanup stale resources of all dead [`Node`](crate::node::Node)s
//!
//! ```
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! Node::<ipc::Service>::list(Config::global_config(), |node_state| {
//!     if let NodeState::<ipc::Service>::Dead(view) = node_state {
//!         println!("cleanup resources of dead node {:?}", view);
//!         if let Err(e) = view.try_remove_stale_resources() {
//!             println!("failed to cleanup resources due to {:?}", e);
//!         }
//!     }
//!     CallbackProgression::Continue
//! })?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Simple Event Loop
//!
//! ```no_run
//! use core::time::Duration;
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! const CYCLE_TIME: Duration = Duration::from_secs(1);
//! let node = NodeBuilder::new()
//!                 .name(&"my_little_node".try_into()?)
//!                 .create::<ipc::Service>()?;
//!
//! while node.wait(CYCLE_TIME).is_ok() {
//!     // your algorithm in here
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Simple Event Loop With Disabled [`Signal`](iceoryx2_bb_posix::signal::Signal) Handling
//!
//! This example demonstrates how the [`Node`](crate::node::Node) can be used when system signals
//! are being handled elsewhere. The builder parameter
//! [`NodeBuilder::signal_handling_mode()`](crate::node::NodeBuilder::signal_handling_mode())
//! can be used to disable signal handling in all [`Node`](crate::node::Node) calls like
//! [`Node::wait()`](crate::node::Node::wait()).
//!
//! ```no_run
//! use core::time::Duration;
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! const CYCLE_TIME: Duration = Duration::from_secs(1);
//! let node = NodeBuilder::new()
//!                 .name(&"my_little_node".try_into()?)
//!                 // disable signal handling
//!                 .signal_handling_mode(SignalHandlingMode::Disabled)
//!                 .create::<ipc::Service>()?;
//!
//! while node.wait(CYCLE_TIME).is_ok() {
//!     // your algorithm in here
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Advanced Event Loop
//!
//! ```no_run
//! use core::time::Duration;
//! use iceoryx2::node::NodeWaitFailure;
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! const CYCLE_TIME: Duration = Duration::from_secs(1);
//! let node = NodeBuilder::new()
//!                 .name(&"my_little_node".try_into()?)
//!                 .create::<ipc::Service>()?;
//!
//! loop {
//!     match node.wait(CYCLE_TIME) {
//!         Ok(()) => {
//!             println!("entered next cycle");
//!         }
//!         Err(NodeWaitFailure::TerminationRequest) => {
//!             println!("User pressed CTRL+c, terminating");
//!             break;
//!         }
//!         Err(NodeWaitFailure::Interrupt) => {
//!             println!("Someone send an interrupt signal ...");
//!         }
//!     }
//! }
//! # Ok(())
//! # }
//! ```

pub(crate) mod global_management_segment;
/// The name for a node.
pub mod node_name;

use core::fmt::Debug;
use core::marker::PhantomData;
use core::ptr::NonNull;
use core::time::Duration;
use iceoryx2_bb_concurrency::atomic::Ordering;

use alloc::collections::BTreeMap;
use alloc::format;
use alloc::string::String;
use alloc::string::ToString;
use alloc::sync::Arc;
use alloc::vec;
use alloc::vec::Vec;

use iceoryx2_bb_concurrency::atomic::AtomicBool;
use iceoryx2_bb_concurrency::cell::UnsafeCell;
use iceoryx2_bb_container::semantic_string::SemanticString;
use iceoryx2_bb_derive_macros::ZeroCopySend;
use iceoryx2_bb_elementary::CallbackProgression;
use iceoryx2_bb_elementary::scope_guard::ScopeGuardBuilder;
use iceoryx2_bb_elementary_traits::testing::abandonable::Abandonable;
use iceoryx2_bb_elementary_traits::zero_copy_send::ZeroCopySend;
use iceoryx2_bb_posix::adaptive_wait::{AdaptiveWaitBuilder, AdaptiveWaitStrategy};
use iceoryx2_bb_posix::clock::Time;
use iceoryx2_bb_posix::clock::{NanosleepError, nanosleep};
use iceoryx2_bb_posix::mutex::Handle;
use iceoryx2_bb_posix::mutex::Mutex;
use iceoryx2_bb_posix::mutex::MutexBuilder;
use iceoryx2_bb_posix::mutex::MutexHandle;
use iceoryx2_bb_posix::mutex::MutexType;
use iceoryx2_bb_posix::process::Process;
use iceoryx2_bb_posix::process::ProcessId;
use iceoryx2_bb_posix::signal::SignalHandler;
use iceoryx2_bb_system_types::file_name::FileName;
use iceoryx2_cal::bag::BagFamily;
use iceoryx2_cal::bag::BagHandleFamily;
use iceoryx2_cal::named_concept::{NamedConceptPathHintRemoveError, NamedConceptRemoveError};
use iceoryx2_cal::{
    monitoring::*, named_concept::NamedConceptListError, serialize::*, static_storage::*,
};
use iceoryx2_log::{debug, fail, fatal_panic, trace, warn};

use crate::identifiers::UniqueNodeId;
use crate::node::node_name::NodeName;
use crate::prelude::MessagingPattern;
use crate::service::ServiceRemoveError;
use crate::service::builder::{Builder, OpenDynamicStorageFailure};
use crate::service::config_scheme::port_tag_config;
use crate::service::config_scheme::{
    node_details_path, node_monitoring_config, service_tag_config,
};
use crate::service::service_hash::ServiceHash;
use crate::service::service_name::ServiceName;
use crate::service::stale_resource_cleanup::RemoveStalePortResourcesError;
use crate::service::stale_resource_cleanup::remove_stale_port_resources;
use crate::service::{self, ServiceRemoveNodeError};
use crate::signal_handling_mode::SignalHandlingMode;
use crate::unique_id_generator::*;
use crate::{config::Config, service::config_scheme::node_details_config};

impl UniqueNodeId {
    pub(crate) fn as_file_name(&self) -> FileName {
        fatal_panic!(from self, when FileName::new(self.0.value().to_string().as_bytes()),
                        "This should never happen! The NodeId shall be always a valid FileName.")
    }
}

/// The failures that can occur when a [`Node`] is created with the [`NodeBuilder`].
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum NodeCreationFailure {
    /// The [`Node`] could not be created since the process does not have sufficient permissions.
    InsufficientPermissions,
    /// Errors that indicate either an implementation issue or a wrongly configured system.
    InternalError,
    /// Indicates that another "instance" on the system removed the resource required by the [`Node`].
    SystemCorrupted,
    /// The [`UniqueNodeId`] could not be generated.
    UnableToGenerateUniqueNodeId,
}

impl core::fmt::Display for NodeCreationFailure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(f, "NodeCreationFailure::{self:?}")
    }
}

impl core::error::Error for NodeCreationFailure {}

/// The failures that can occur when a list of [`NodeState`]s is created with [`Node::list()`].
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum NodeWaitFailure {
    /// The process received an interrupt signal while acquiring the list of all [`Node`]s.
    Interrupt,
    /// A termination signal `SIGTERM` was received.
    TerminationRequest,
}

impl core::fmt::Display for NodeWaitFailure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(f, "NodeWaitFailure::{self:?}")
    }
}

impl core::error::Error for NodeWaitFailure {}

/// The failures that can occur when a list of [`NodeState`]s is created with [`Node::list()`].
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum NodeListFailure {
    /// A list of all [`Node`]s could not be created since the process does not have sufficient permissions.
    InsufficientPermissions,
    /// The process received an interrupt signal while acquiring the list of all [`Node`]s.
    Interrupt,
    /// Errors that indicate either an implementation issue or a wrongly configured system.
    InternalError,
}

impl core::fmt::Display for NodeListFailure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(f, "NodeListFailure::{self:?}")
    }
}

impl core::error::Error for NodeListFailure {}

/// Failures of [`DeadNodeView::try_remove_stale_resources()`] that occur when the stale resources of
/// a dead [`Node`] are removed.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum NodeCleanupFailure {
    /// The process received an interrupt signal while cleaning up all stale resources of a dead [`Node`].
    Interrupt,
    /// Errors that indicate either an implementation issue or a wrongly configured system.
    InternalError,
    /// The stale resources of a dead [`Node`] could not be removed since the process does not have sufficient permissions.
    InsufficientPermissions,
    /// Trying to cleanup resources from a dead [`Node`] which was using a different iceoryx2 version.
    VersionMismatch,
    /// Another instance has successfully cleaned up all resources.
    ResourcesAlreadyCleanedUp,
    /// Another instance has acquired the ownership of all resources and is currently cleaning up.
    AnotherInstanceIsCleaningUpTheNode,
}

impl core::fmt::Display for NodeCleanupFailure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(f, "NodeCleanupFailure::{self:?}")
    }
}

impl core::error::Error for NodeCleanupFailure {}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum NodeReadStorageFailure {
    ReadError,
    InsufficientPermissions,
    Corrupted,
    Interrupt,
    InternalError,
}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum NodeReadServiceTagsFailure {
    InsufficientPermissions,
    InternalError,
}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum NodeReadPortTagsFailure {
    InsufficientPermissions,
    InternalError,
}

/// Optional detailed information that a [`Node`] can have. They can only be obtained when the
/// process has sufficient access permissions.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct NodeDetails {
    executable: FileName,
    process: ProcessId,
    name: NodeName,
    config: Config,
}

impl NodeDetails {
    #[doc(hidden)]
    pub fn __internal_new(node_name: &Option<NodeName>, config: &Config) -> Self {
        Self::new(node_name, config)
    }

    fn new(node_name: &Option<NodeName>, config: &Config) -> Self {
        let process = Process::from_self();

        let executable = match process.executable() {
            Ok(n) => n.file_name(),
            Err(e) => {
                debug!(from "NodeDetails::new()", "Unable to acquire executable name of the Node's process ({:?}).", e);
                const FALLBACK_EXEC: &[u8] = b"undefined";
                unsafe { FileName::new_unchecked(FALLBACK_EXEC) }
            }
        };

        Self {
            executable,
            process: process.id(),
            name: if let Some(name) = node_name {
                name.clone()
            } else {
                NodeName::new("").expect("An empty NodeName is always valid.")
            },
            config: config.clone(),
        }
    }

    /// Returns the executable [`FileName`] of the [`Node`]s owner process.
    pub fn executable(&self) -> &FileName {
        &self.executable
    }

    /// Returns the [`ProcessId`] of the [`Node`]s owner process.
    pub fn process_id(&self) -> ProcessId {
        self.process
    }

    /// Returns the [`NodeName`]. Multiple [`Node`]s are allowed to have the same [`NodeName`], it
    /// is not unique!
    pub fn name(&self) -> &NodeName {
        &self.name
    }

    /// Returns the [`Config`] the [`Node`] uses to create all entities.
    pub fn config(&self) -> &Config {
        &self.config
    }
}

/// The current state of the [`Node`]. If the [`Node`] is dead all of its resources can be removed
/// with [`DeadNodeView::try_remove_stale_resources()`].
#[derive(Debug)]
pub enum NodeState<Service: service::Service> {
    /// The [`Node`]s process is still alive.
    Alive(AliveNodeView<Service>),
    /// The [`Node`]s process died without cleaning up the [`Node`]s resources. Another process has
    /// now the responsibility to cleanup all the stale resources.
    Dead(DeadNodeView<Service>),
    /// The process does not have sufficient permissions to identify the [`Node`] as dead or alive.
    Inaccessible(UniqueNodeId),
    /// The [`Node`] is in an undefined state, meaning that certain elements are missing,
    /// misconfigured or inconsistent. This can only happen due to an implementation failure or
    /// when the corresponding [`Node`] resources were altered.
    Undefined(UniqueNodeId),
}

impl<Service: service::Service> Clone for NodeState<Service> {
    fn clone(&self) -> Self {
        match self {
            NodeState::Alive(n) => NodeState::Alive(n.clone()),
            NodeState::Dead(n) => NodeState::Dead(n.clone()),
            NodeState::Inaccessible(n) => NodeState::Inaccessible(*n),
            NodeState::Undefined(n) => NodeState::Undefined(*n),
        }
    }
}

impl<Service: service::Service> NodeState<Service> {
    pub(crate) fn new(
        node_id: &UniqueNodeId,
        config: &Config,
    ) -> Result<Option<Self>, NodeListFailure> {
        let details = Node::<Service>::get_node_details(config, node_id).unwrap_or_default();

        let node_view = AliveNodeView::<Service> {
            id: *node_id,
            details,
            _service: PhantomData,
        };

        match Node::<Service>::get_node_state(config, node_id) {
            Ok(State::DoesNotExist) => Ok(None),
            Ok(State::Alive) => Ok(Some(NodeState::Alive(node_view))),
            Ok(State::Dead) => Ok(Some(NodeState::Dead(DeadNodeView(node_view)))),
            Err(NodeListFailure::InsufficientPermissions) => {
                Ok(Some(NodeState::Inaccessible(*node_id)))
            }
            Err(NodeListFailure::InternalError) => Ok(Some(NodeState::Undefined(*node_id))),
            Err(e) => Err(e),
        }
    }

    /// Returns the [`UniqueNodeId`] of the corresponding [`Node`].
    pub fn node_id(&self) -> &UniqueNodeId {
        match self {
            NodeState::Dead(node) => node.id(),
            NodeState::Alive(node) => node.id(),
            NodeState::Inaccessible(node_id) => node_id,
            NodeState::Undefined(node_id) => node_id,
        }
    }
}

/// Returned by [`Node::try_cleanup_dead_nodes()`]. Contains the cleanup report of the call
/// and contains the number of dead nodes that were successfully cleaned up and how many
/// could not be cleaned up.
/// This does not have to be an error, for instance when the current process does not
/// have the permission to access the corresponding resources.
#[derive(Debug, Clone, Copy, PartialEq, Eq, ZeroCopySend)]
#[repr(C)]
pub struct CleanupState {
    /// The number of successful dead node cleanups
    pub cleanups: u64,
    /// The number of failed dead node cleanups
    pub failed_cleanups: u64,
}

/// Contains all available details of a [`Node`].
pub trait NodeView {
    /// Returns the [`UniqueNodeId`] of the [`Node`].
    fn id(&self) -> &UniqueNodeId;
    /// Returns the [`NodeDetails`].
    fn details(&self) -> &Option<NodeDetails>;
}

/// All the information of a [`Node`] that is alive.
#[derive(Debug)]
pub struct AliveNodeView<Service: service::Service> {
    id: UniqueNodeId,
    details: Option<NodeDetails>,
    _service: PhantomData<Service>,
}

impl<Service: service::Service> Clone for AliveNodeView<Service> {
    fn clone(&self) -> Self {
        Self {
            id: self.id,
            details: self.details.clone(),
            _service: PhantomData,
        }
    }
}

impl<Service: service::Service> NodeView for AliveNodeView<Service> {
    fn id(&self) -> &UniqueNodeId {
        &self.id
    }

    fn details(&self) -> &Option<NodeDetails> {
        &self.details
    }
}

/// All the information and management operations belonging to a dead [`Node`].
#[derive(Debug)]
pub struct DeadNodeView<Service: service::Service>(AliveNodeView<Service>);

impl<Service: service::Service> Clone for DeadNodeView<Service> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

impl<Service: service::Service> NodeView for DeadNodeView<Service> {
    fn id(&self) -> &UniqueNodeId {
        self.0.id()
    }

    fn details(&self) -> &Option<NodeDetails> {
        self.0.details()
    }
}

impl<Service: service::Service> DeadNodeView<Service> {
    #[doc(hidden)]
    pub fn __internal_try_remove_stale_resources(
        id: UniqueNodeId,
        details: NodeDetails,
    ) -> Result<(), NodeCleanupFailure> {
        DeadNodeView(AliveNodeView {
            id,
            details: Some(details),
            _service: PhantomData::<Service>,
        })
        .try_remove_stale_resources()
    }

    #[doc(hidden)]
    pub fn __internal_blocking_remove_stale_resources(
        id: UniqueNodeId,
        details: NodeDetails,
        timeout: Duration,
    ) -> Result<(), NodeCleanupFailure> {
        DeadNodeView(AliveNodeView {
            id,
            details: Some(details),
            _service: PhantomData::<Service>,
        })
        .blocking_remove_stale_resources(timeout)
    }

    /// Removes all stale resources of a dead [`Node`]. If another instance
    /// is already removing the dead [`Node`] it waits until the other instance
    /// has cleaned up the dead [`Node`] completely. If the other cleanup instance
    /// crashes, it will take over the ownership and continue the cleanup.
    /// If the process does not have the permission to cleanup all resources it
    /// aborts with an error.
    ///
    /// If the provided timeout is expired it will return with
    /// [`NodeCleanupFailure::AnotherInstanceIsCleaningUpTheNode`].
    pub fn blocking_remove_stale_resources(
        self,
        timeout: Duration,
    ) -> Result<(), NodeCleanupFailure> {
        let msg = "Unable to block until the stale resources of the dead node are removed";
        let mut adaptive_wait = fail!(from self,
                                        when AdaptiveWaitBuilder::new()
                                                .strategy(AdaptiveWaitStrategy::FixedTicks(Duration::from_millis(1)))
                                                .create(),
                                        with NodeCleanupFailure::InternalError,
                                       "{msg} since the adaptive wait builder could not be initiated.");
        let start = fail!(from self,
                          when Time::now(),
                          with NodeCleanupFailure::InternalError,
                          "{msg} since the current system time could not be acquired.");

        loop {
            match self.remove_stale_resources_impl() {
                Ok(()) | Err(NodeCleanupFailure::ResourcesAlreadyCleanedUp) => return Ok(()),
                Err(NodeCleanupFailure::AnotherInstanceIsCleaningUpTheNode) => (),
                Err(e) => return Err(e),
            }

            fail!(from self,
                  when adaptive_wait.wait(),
                  with NodeCleanupFailure::InternalError,
                  "{msg} since the adaptive wait failed.");

            let elapsed = fail!(from self,
                                when start.elapsed(),
                                with NodeCleanupFailure::InternalError,
                                "{msg} due to a failure while acquiring the elapsed time.");

            if elapsed > timeout {
                fail!(from self, with NodeCleanupFailure::AnotherInstanceIsCleaningUpTheNode,
                    "{msg} since another instance requires longer than {timeout:?} to cleanup the resources.");
            }
        }
    }

    /// Tries to remove all stale resources of a dead [`Node`]. If another instance
    /// is already removing the dead [`Node`] or the [`Node`] is already removed it will
    /// return immediately.
    pub fn try_remove_stale_resources(self) -> Result<(), NodeCleanupFailure> {
        self.remove_stale_resources_impl()
    }

    fn remove_stale_resources_impl(&self) -> Result<(), NodeCleanupFailure> {
        let msg = "Unable to remove stale resources";
        let monitor_name = fatal_panic!(from self,
                            when FileName::new(self.id().0.value().to_string().as_bytes()),
                            "This should never happen! {msg} since the NodeId is not a valid file name.");

        // The cleaner guarantees that the lock can be acquired only once in the inter-process context.
        // But the same process could acquire the same cleaner multiple times. To avoid intra-process
        // races an additional lock is introduced so that only one thread can call
        // remove_stale_resources.
        static IN_CLEANUP_SECTION: AtomicBool = AtomicBool::new(false);

        let _cleanup_section_guard = ScopeGuardBuilder::new(&IN_CLEANUP_SECTION)
            .on_init(|v| {
                // if swap returns true, someone else is holding the lock
                if v.swap(true, Ordering::Relaxed) {
                    fail!(from self, with NodeCleanupFailure::AnotherInstanceIsCleaningUpTheNode,
                        "{msg} since another instance is already cleaning up the dead nodes resources.");
                }

                Ok(())
            })
            .on_drop(|v| v.store(false, Ordering::Relaxed))
            .create()?;

        let config = if let Some(d) = self.details() {
            d.config()
        } else {
            Config::global_config()
        };

        let cleaner = fail!(from self, when self.acquire_cleaner_lock(&monitor_name, config),
                        "{} since the monitor cleaner lock could not be acquired.", msg);

        let mut cleanup_failure = Ok(());
        let remove_node_from_service = |service_hash: &ServiceHash| {
            match Service::__internal_remove_node_from_service(self.id(), service_hash, config) {
                Ok(()) => (),
                Err(ServiceRemoveNodeError::VersionMismatch) => {
                    cleanup_failure = Err(NodeCleanupFailure::VersionMismatch);
                    debug!(from self,
                        "{msg} since the dead node was using a different iceoryx2 version.");
                }
                Err(ServiceRemoveNodeError::Interrupt) => {
                    cleanup_failure = Err(NodeCleanupFailure::Interrupt);
                    debug!(from self,
                        "{msg} since an interrupt signal was raised while removing the node from the service.");
                }
                Err(ServiceRemoveNodeError::InsufficientPermissions) => {
                    cleanup_failure = Err(NodeCleanupFailure::InsufficientPermissions);
                    debug!(from self,
                        "{msg} since an interrupt signal was raised while removing the node from the service.");
                }
                Err(ServiceRemoveNodeError::InternalError) => {
                    cleanup_failure = Err(NodeCleanupFailure::InternalError);
                    debug!(from self,
                        "{msg} since an internal failure occurred while removing the node from the service.");
                }
            }
            CallbackProgression::Continue
        };

        // the tags need to be removed at the end of the cleanup process; at first the service tags
        match Node::<Service>::service_tags(config, self.id(), remove_node_from_service) {
            Ok(()) => (),
            Err(NodeReadServiceTagsFailure::InsufficientPermissions) => {
                cleaner.abandon();
                fail!(from self, with NodeCleanupFailure::InsufficientPermissions,
                    "{} since the service tags could not be read due to insufficient permissions.", msg);
            }
            Err(NodeReadServiceTagsFailure::InternalError) => {
                cleaner.abandon();
                fail!(from self, with NodeCleanupFailure::InternalError,
                    "{} since the service tags could not be read due to an internal error.", msg);
            }
        };

        cleanup_failure?;

        // remove the port tags last, after the ports have been removed from the service;
        // now, everything not belonging to a service can be removed
        match Node::<Service>::port_tags(config, self.id(), |port_id| {
            match unsafe { remove_stale_port_resources::<Service>(self.id(), port_id, config) } {
                Ok(()) => CallbackProgression::Continue,
                Err(RemoveStalePortResourcesError::InsufficientPermissions) => {
                    cleanup_failure = Err(NodeCleanupFailure::InsufficientPermissions);
                    debug!(from self,
                        "{} since the stale resources of the port {port_id} could not be removed due to insufficient permissions.", msg);
                    CallbackProgression::Stop
                }
                Err(RemoveStalePortResourcesError::VersionMismatch) => {
                    cleanup_failure = Err(NodeCleanupFailure::VersionMismatch);
                    debug!(from self,
                        "{} since the stale resources of the port {port_id} could not be removed since the iceoryx2 version does not match.", msg);
                    CallbackProgression::Stop
                }
                Err(RemoveStalePortResourcesError::InternalError) => {
                    cleanup_failure = Err(NodeCleanupFailure::InternalError);
                    debug!(from self,
                        "{} since the stale resources of the port {port_id} could not be removed due to an internal failure.", msg);
                    CallbackProgression::Stop
                }
                Err(RemoveStalePortResourcesError::Interrupt) => {
                    cleanup_failure = Err(NodeCleanupFailure::Interrupt);
                    debug!(from self,
                        "{} since the stale resources of the port {port_id} could not be removed due to an interrupt signal.", msg);
                    CallbackProgression::Stop
                }
            }
        }) {
            Ok(()) => (),
            Err(NodeReadPortTagsFailure::InsufficientPermissions) => {
                cleaner.abandon();
                fail!(from self, with NodeCleanupFailure::InsufficientPermissions,
                    "{} since the port tags could not be read due to insufficient permissions.", msg);
            }
            Err(NodeReadPortTagsFailure::InternalError) => {
                cleaner.abandon();
                fail!(from self, with NodeCleanupFailure::InternalError,
                    "{} since the port tags could not be read due to an internal error.", msg);
            }
        }

        cleanup_failure?;

        match remove_node::<Service>(*self.id(), config) {
            Ok(_) => {
                drop(cleaner);
                Ok(())
            }
            Err(e) => {
                cleaner.abandon();
                fail!(from self, with e, "{} since the node itself could not be removed.", msg);
            }
        }
    }

    fn acquire_cleaner_lock(
        &self,
        monitor_name: &FileName,
        config: &Config,
    ) -> Result<<Service::Monitoring as Monitoring>::Cleaner, NodeCleanupFailure> {
        let msg = "Unable to acquire monitor cleaner";

        match <Service::Monitoring as Monitoring>::Builder::new(monitor_name)
            .config(&node_monitoring_config::<Service>(config))
            .cleaner()
        {
            Ok(cleaner) => Ok(cleaner),
            Err(MonitoringCreateCleanerError::AlreadyOwnedByAnotherInstance)
            | Err(
                MonitoringCreateCleanerError::IsBeingCleanedUpOrAnotherCleanerCrashedDuringCleanup,
            ) => {
                fail!(from self, with NodeCleanupFailure::AnotherInstanceIsCleaningUpTheNode,
                    "{} since another instance is already cleaning up all resources.", msg);
            }
            Err(MonitoringCreateCleanerError::DoesNotExist) => {
                fail!(from self, with NodeCleanupFailure::ResourcesAlreadyCleanedUp,
                    "{} since another instance has already cleaned up all resources.", msg);
            }
            Err(MonitoringCreateCleanerError::Interrupt) => {
                fail!(from self, with NodeCleanupFailure::Interrupt,
                    "{} since an interrupt signal was received.", msg);
            }
            Err(MonitoringCreateCleanerError::InternalError) => {
                fail!(from self, with NodeCleanupFailure::InternalError,
                    "{} due to an internal error while acquiring monitoring cleaner.", msg);
            }
            Err(MonitoringCreateCleanerError::InstanceStillAlive) => {
                fatal_panic!(from self,
                        "This should never happen! {} since the Node is still alive.", msg);
            }
        }
    }
}

fn acquire_all_node_detail_storages<Service: service::Service>(
    origin: &str,
    config: &<Service::StaticStorage as NamedConceptMgmt>::Configuration,
) -> Result<Vec<FileName>, NodeCleanupFailure> {
    let msg = "Unable to list all node detail storages";
    match <Service::StaticStorage as NamedConceptMgmt>::list_cfg(config) {
        Ok(v) => Ok(v),
        Err(NamedConceptListError::InsufficientPermissions) => {
            fail!(from origin, with NodeCleanupFailure::InsufficientPermissions,
                "{} due to insufficient permissions.", msg);
        }
        Err(NamedConceptListError::InternalError) => {
            fail!(from origin, with NodeCleanupFailure::InternalError,
                "{} due to an internal error.", msg);
        }
    }
}

fn remove_detail_storages<Service: service::Service>(
    origin: &str,
    storages: Vec<FileName>,
    config: &<Service::StaticStorage as NamedConceptMgmt>::Configuration,
) -> Result<(), NodeCleanupFailure> {
    let msg = "Unable to remove node detail storage";
    for entry in storages {
        match unsafe { <Service::StaticStorage as NamedConceptMgmt>::remove_cfg(&entry, config) } {
            Ok(_) => (),
            Err(NamedConceptRemoveError::InsufficientPermissions) => {
                fail!(from origin, with NodeCleanupFailure::InsufficientPermissions,
                    "{} {} due to insufficient permissions.", msg, entry);
            }
            Err(NamedConceptRemoveError::InternalError) => {
                fail!(from origin, with NodeCleanupFailure::InternalError,
                    "{} {} due to an internal failure.", msg, entry);
            }
            Err(NamedConceptRemoveError::Interrupt) => {
                fail!(from origin, with NodeCleanupFailure::Interrupt,
                    "{} {} since an interrupt signal was raised.", msg, entry);
            }
        }
    }

    Ok(())
}

fn remove_node_details_directory<Service: service::Service>(
    config: &Config,
    node_id: &UniqueNodeId,
) -> Result<(), NodeCleanupFailure> {
    let origin = format!("remove_node_details_directory({config:?}, {node_id:?})");
    let msg = "Unable to remove node details directory";
    let path = node_details_path(config, node_id);
    match <Service::StaticStorage as NamedConceptMgmt>::remove_path_hint(&path) {
        Ok(()) => Ok(()),
        Err(NamedConceptPathHintRemoveError::InsufficientPermissions) => {
            fail!(from origin, with NodeCleanupFailure::InsufficientPermissions,
                "{} due to insufficient permissions.", msg);
        }
        Err(NamedConceptPathHintRemoveError::InternalError) => {
            fail!(from origin, with NodeCleanupFailure::InternalError,
                "{} due to an internal error.", msg);
        }
    }
}

fn remove_node<Service: service::Service>(
    id: UniqueNodeId,
    config: &Config,
) -> Result<bool, NodeCleanupFailure> {
    let origin = format!(
        "remove_node<{}>({:?})",
        core::any::type_name::<Service>(),
        id
    );

    let details_config = node_details_config::<Service>(config, &id);
    let detail_storages = acquire_all_node_detail_storages::<Service>(&origin, &details_config)?;
    remove_detail_storages::<Service>(&origin, detail_storages, &details_config)?;
    remove_node_details_directory::<Service>(config, &id)?;

    Ok(true)
}

#[derive(Debug)]
pub(crate) struct RegisteredServices<BagHandle: BagHandleFamily> {
    handle: MutexHandle<BTreeMap<ServiceHash, (BagHandle, u64)>>,
}

impl<BagHandle: BagHandleFamily> RegisteredServices<BagHandle> {
    pub(crate) fn new() -> Self {
        let origin = "RegisteredServices::new()";
        let handle = MutexHandle::new();

        fatal_panic!(
            from origin,
            when MutexBuilder::new()
                .is_interprocess_capable(false)
                .mutex_type(MutexType::Normal)
                .create(BTreeMap::new(), &handle),
            "Failed to create mutex"
        );

        Self { handle }
    }

    fn insert(
        services: &mut BTreeMap<ServiceHash, (BagHandle, u64)>,
        service_hash: ServiceHash,
        handle: BagHandle,
    ) {
        if services.insert(service_hash, (handle, 1)).is_some() {
            fatal_panic!(from "RegisteredServices::insert()",
                "This should never happen! The service with the {:?} was already registered.",
                service_hash);
        }
    }

    pub(crate) fn add(&self, service_hash: &ServiceHash, handle: BagHandle) {
        let mut guard = fatal_panic!(
            from self,
            when self.mutex().lock(),
            "Failed to lock mutex"
        );

        Self::insert(&mut guard, *service_hash, handle);
    }

    pub(crate) fn add_or<F: FnMut() -> Result<BagHandle, OpenDynamicStorageFailure>>(
        &self,
        service_hash: &ServiceHash,
        mut or_callback: F,
    ) -> Result<(), OpenDynamicStorageFailure> {
        let mut guard = fatal_panic!(
            from self,
            when self.mutex().lock(),
            "Failed to lock mutex"
        );

        match guard.get_mut(service_hash) {
            Some(entry) => {
                entry.1 += 1;
            }
            None => {
                let new_handle = or_callback()?;
                Self::insert(&mut guard, *service_hash, new_handle);
            }
        };
        Ok(())
    }

    pub(crate) fn remove<F: FnMut(BagHandle)>(
        &self,
        service_hash: &ServiceHash,
        mut cleanup_call: F,
    ) {
        let mut guard = self.mutex().lock().expect("Failed to lock mutex");

        if let Some(entry) = guard.get_mut(service_hash) {
            entry.1 -= 1;
            if entry.1 == 0 {
                let handle = entry.0;
                guard.remove(service_hash);
                cleanup_call(handle);
            }
        } else {
            fatal_panic!(from "RegisteredServices::remove()",
                "This should never happen! The service with the {:?} was not registered.", service_hash);
        }

        drop(guard);
    }

    fn mutex(&self) -> Mutex<'_, '_, BTreeMap<ServiceHash, (BagHandle, u64)>> {
        // Safe - the mutex is initialized when constructing the struct and
        // not interacted with by anything else.
        unsafe { Mutex::from_handle(&self.handle) }
    }
}

#[derive(Debug)]
struct SharedNodeState<Service: service::Service> {
    id: UniqueNodeId,
    details: NodeDetails,
    monitoring_token: UnsafeCell<Option<<Service::Monitoring as Monitoring>::Token>>,
    registered_services: RegisteredServices<<Service::Bag as BagFamily>::BagHandle>,
    signal_handling_mode: SignalHandlingMode,
    details_storage: Service::StaticStorage,
}

unsafe impl<Service: service::Service> Send for SharedNodeState<Service> {}
unsafe impl<Service: service::Service> Sync for SharedNodeState<Service> {}

impl<Service: service::Service> Abandonable for SharedNodeState<Service> {
    unsafe fn abandon_in_place(mut this: NonNull<Self>) {
        let this = unsafe { this.as_mut() };
        unsafe {
            <Service::StaticStorage as Abandonable>::abandon_in_place(NonNull::from_mut(
                &mut this.details_storage,
            ))
        };
        if let Some(token) = this.monitoring_token.get_mut() {
            unsafe {
                <<Service::Monitoring as Monitoring>::Token as Abandonable>::abandon_in_place(
                    NonNull::from_mut(token),
                )
            };
        }
    }
}

impl<Service: service::Service> SharedNodeState<Service> {
    pub(crate) fn blocking_cleanup_dead_nodes(&self, timeout: Duration) -> CleanupState {
        let mut cleanup_state = CleanupState {
            cleanups: 0,
            failed_cleanups: 0,
        };
        let origin = format!(
            "Node::<{}>::cleanup_dead_nodes()",
            core::any::type_name::<Service>()
        );

        let cleanup_call = |node_state| {
            if let NodeState::Dead(dead_node) = node_state {
                let node_id = *dead_node.id();
                debug!(from origin, "Dead node ({:?}) detected", node_id);
                match dead_node.blocking_remove_stale_resources(timeout) {
                    Ok(_) => {
                        cleanup_state.cleanups += 1;
                        trace!(from origin, "The dead node ({:?}) was successfully removed.", node_id)
                    }
                    Err(e) => {
                        cleanup_state.failed_cleanups += 1;
                        trace!(from origin, "Unable to remove dead node {:?} ({:?}).", node_id, e)
                    }
                }
            }

            CallbackProgression::Continue
        };

        match Node::<Service>::list(&self.details.config, cleanup_call) {
            Ok(()) => cleanup_state,
            Err(e) => {
                debug!(from origin, "Unable to perform a full scan for dead nodes since the all existing nodes could not be listed ({:?}).", e);
                cleanup_state
            }
        }
    }
}

impl<Service: service::Service> Drop for SharedNodeState<Service> {
    fn drop(&mut self) {
        let config = self.details.config();
        if self.monitoring_token.get_mut().is_some() {
            if config.global.node.cleanup_dead_nodes_on_destruction {
                self.blocking_cleanup_dead_nodes(Duration::ZERO);
            }

            warn!(from self, when remove_node::<Service>(self.id, config),
                "Unable to remove node resources.");
        }

        trace!(from self, "removed");
    }
}

#[derive(Debug, Clone)]
pub(crate) struct SharedNode<Service: service::Service> {
    state: Arc<SharedNodeState<Service>>,
}

impl<Service: service::Service> Abandonable for SharedNode<Service> {
    unsafe fn abandon_in_place(mut this: NonNull<Self>) {
        let this = unsafe { this.as_mut() };
        if let Some(state) = Arc::get_mut(&mut this.state) {
            unsafe { SharedNodeState::abandon_in_place(NonNull::from_mut(state)) };
        } else {
            unsafe { core::ptr::drop_in_place(&mut this.state) };
        }
    }
}

impl<Service: service::Service> SharedNode<Service> {
    pub(crate) fn config(&self) -> &Config {
        &self.state.details.config
    }

    pub(crate) fn id(&self) -> &UniqueNodeId {
        &self.state.id
    }

    pub(crate) fn registered_services(
        &self,
    ) -> &RegisteredServices<<Service::Bag as BagFamily>::BagHandle> {
        &self.state.registered_services
    }

    pub(crate) fn name(&self) -> &NodeName {
        &self.state.details.name
    }

    pub(crate) fn create_port_tag(
        &self,
        origin: &str,
        msg: &str,
        port_id: u128,
    ) -> Result<Service::StaticStorage, StaticStorageCreateError> {
        let name = FileName::new(port_id.to_string().as_bytes())
            .expect("A number is always a valid file name.");

        match <<Service::StaticStorage as StaticStorage>::Builder as NamedConceptBuilder<
            Service::StaticStorage,
        >>::new(&name)
        .config(&port_tag_config::<Service>(self.config(), self.id()))
        .has_ownership(true)
        .create(&[])
        {
            Ok(static_storage) => Ok(static_storage),
            Err(e) => {
                fail!(from origin, with e,
                    "{msg} since the port tag could not be created. [{e:?}]");
            }
        }
    }

    pub(crate) fn create_service_tag<T: Debug + ?Sized>(
        &self,
        origin: &T,
        msg: &str,
        service_hash: &ServiceHash,
    ) -> Result<Option<Service::StaticStorage>, StaticStorageCreateError> {
        match <<Service::StaticStorage as StaticStorage>::Builder as NamedConceptBuilder<
            Service::StaticStorage,
        >>::new(&service_hash.0.into())
        .config(&service_tag_config::<Service>(self.config(), self.id()))
        .has_ownership(true)
        .create(&[])
        {
            Ok(static_storage) => Ok(Some(static_storage)),
            Err(StaticStorageCreateError::AlreadyExists) => Ok(None),
            Err(e) => {
                fail!(from origin, with e,
                    "{msg} since the service tag could not be created. [{e:?}]");
            }
        }
    }
}

/// The [`Node`] is the entry point to the whole iceoryx2 infrastructure and owns all entities.
///
/// As soon as a process crashes other processes can detect dead [`Node`]s via [`Node::list()`]
/// and clean up the stale resources - the entities that
/// were created via the [`Node`].
///
/// Can be created via the [`NodeBuilder`].
#[derive(Debug)]
pub struct Node<Service: service::Service> {
    pub(crate) shared: SharedNode<Service>,
}

unsafe impl<Service: service::Service> Send for Node<Service> {}

impl<Service: service::Service> Abandonable for Node<Service> {
    unsafe fn abandon_in_place(mut this: NonNull<Self>) {
        let this = unsafe { this.as_mut() };
        unsafe { SharedNode::abandon_in_place(NonNull::from_mut(&mut this.shared)) };
    }
}

impl<Service: service::Service> Node<Service> {
    /// Returns the [`NodeName`].
    pub fn name(&self) -> &NodeName {
        self.shared.name()
    }

    /// Returns the [`Config`] that the [`Node`] will use to create any iceoryx2 entity.
    pub fn config(&self) -> &Config {
        self.shared.config()
    }

    /// Returns the [`UniqueNodeId`] of the [`Node`].
    pub fn id(&self) -> &UniqueNodeId {
        self.shared.id()
    }

    /// Instantiates a [`ServiceBuilder`](Builder) for a service with the provided name.
    pub fn service_builder(&self, name: &ServiceName) -> Builder<Service> {
        Builder::new(name, self.shared.clone())
    }

    /// Returns the [`NodeState`] of a specific [`Node`] that has the provided [`UniqueNodeId`].
    /// If this [`Node`] does not exist, the function returns [`None`].
    pub fn state_of(
        config: &Config,
        node_id: UniqueNodeId,
    ) -> Result<Option<NodeState<Service>>, NodeListFailure> {
        let mut node_state = None;
        match Node::list(config, |v| {
            if *v.node_id() == node_id {
                node_state = Some(v);
                CallbackProgression::Stop
            } else {
                CallbackProgression::Continue
            }
        }) {
            Ok(()) => Ok(node_state),
            Err(e) => {
                fail!(from "Node::state_of()", with e,
                    "Unable to acquire the node state of \"{node_id}\" due to a failure while listing all nodes. [{e:?}]");
            }
        }
    }

    /// Calls the provided callback for all [`Node`]s in the system under a given [`Config`] and
    /// provides [`NodeState<Service>`] as input argument. With every iteration the callback has to
    /// return [`CallbackProgression::Continue`] to perform the next iteration or
    /// [`CallbackProgression::Stop`] to stop the iteration immediately.
    /// ```
    /// # use iceoryx2::prelude::*;
    /// Node::<ipc::Service>::list(Config::global_config(), |node_state| {
    ///     println!("found node {:?}", node_state);
    ///     CallbackProgression::Continue
    /// });
    /// ```
    pub fn list<F: FnMut(NodeState<Service>) -> CallbackProgression>(
        config: &Config,
        mut callback: F,
    ) -> Result<(), NodeListFailure> {
        let msg = "Unable to iterate over Node list";
        let origin = "Node::list()";
        let monitoring_config = node_monitoring_config::<Service>(config);

        match Self::list_all_nodes(&monitoring_config) {
            Ok(node_list) => {
                for node_name in node_list {
                    // not bad, just found a file that is not a node
                    let Ok(node_id) = core::str::from_utf8(node_name.as_bytes()) else {
                        continue;
                    };
                    let node_id = match node_id.parse::<u128>() {
                        Ok(v) => UniqueNodeId(unsafe { UniqueId::from_raw_id(v) }),
                        Err(_) => continue,
                    };

                    match NodeState::new(&node_id, config) {
                        Ok(Some(node_state)) => {
                            if callback(node_state) == CallbackProgression::Stop {
                                break;
                            }
                        }
                        Ok(None) => (),
                        Err(e) => {
                            fail!(from origin, with e,
                                "{msg} since the following error occurred ({:?}).", e);
                        }
                    }
                }
            }
            Err(e) => {
                fail!(from origin, with e,
                    "{msg} since the node list could not be acquired ({:?}).", e);
            }
        }

        Ok(())
    }

    fn handle_termination_request(&self, error_msg: &str) -> Result<(), NodeWaitFailure> {
        if self.signal_handling_mode() == SignalHandlingMode::HandleTerminationRequests
            && SignalHandler::termination_requested()
        {
            fail!(from self, with NodeWaitFailure::TerminationRequest,
                "{error_msg} since a termination request was received.");
        }

        Ok(())
    }

    /// Waits until the cycle time has passed. It returns [`NodeWaitFailure::TerminationRequest`]
    /// when a `SIGTERM` signal was received or [`NodeWaitFailure::Interrupt`] when a `SIGINT`
    /// signal was received.
    pub fn wait(&self, cycle_time: Duration) -> Result<(), NodeWaitFailure> {
        let msg = "Unable to wait on node";
        self.handle_termination_request(msg)?;

        match nanosleep(cycle_time) {
            Ok(()) => {
                self.handle_termination_request(msg)?;
                Ok(())
            }
            Err(NanosleepError::InterruptedBySignal(_)) => {
                fail!(from self, with NodeWaitFailure::Interrupt,
                        "{msg} since a interrupt signal was received.");
            }
            Err(v) => {
                fatal_panic!(from self,
                    "Failed to wait with cycle time {:?} in main event look, caused by ({:?}).",
                    cycle_time, v);
            }
        }
    }

    /// Returns the [`SignalHandlingMode`] with which the [`Node`] was created.
    pub fn signal_handling_mode(&self) -> SignalHandlingMode {
        self.shared.state.signal_handling_mode
    }

    /// Removes the stale system resources of all dead [`Node`]s. The dead [`Node`]s are also
    /// removed from all registered [`Service`](crate::service::Service)s.
    ///
    /// If a [`Node`] cannot be cleaned up since the process has insufficient permissions or it
    /// is currently being cleaned up by another process then the [`Node`] is skipped.
    pub fn try_cleanup_dead_nodes(&self) -> CleanupState {
        self.shared
            .state
            .blocking_cleanup_dead_nodes(Duration::ZERO)
    }

    /// Removes the stale system resources of all dead [`Node`]s. The dead [`Node`]s are also
    /// removed from all registered [`Service`](crate::service::Service)s.
    ///
    /// If a [`Node`] cannot be cleaned up since the process has insufficient permissions then the
    /// [`Node`] is skipped. If it is currently being cleaned up by another process then the
    /// cleaner will wait until the timeout as either passed or the cleaned was finished.
    ///
    /// The timeout is applied to every individual dead [`Node`] the function needs to wait on.
    pub fn blocking_cleanup_dead_nodes(&self, timeout: Duration) -> CleanupState {
        self.shared.state.blocking_cleanup_dead_nodes(timeout)
    }

    /// Removes a [`Service`](crate::service::Service) by force. This shall be used if the
    /// resources could not be removed in a previous run and now it is no longer possible to
    /// open the service.
    ///
    /// # Safety
    ///
    ///  * No other process shall use the service.
    ///
    pub unsafe fn force_remove_service(
        &self,
        name: &ServiceName,
        messaging_pattern: MessagingPattern,
    ) -> Result<bool, ServiceRemoveError> {
        unsafe { Service::__internal_force_remove_service(name, self.config(), messaging_pattern) }
    }

    fn list_all_nodes(
        config: &<Service::Monitoring as NamedConceptMgmt>::Configuration,
    ) -> Result<Vec<FileName>, NodeListFailure> {
        let result = <Service::Monitoring as NamedConceptMgmt>::list_cfg(config);

        if let Ok(result) = result {
            return Ok(result);
        }

        let msg = "Unable to list all nodes";
        let origin = format!("Node::list_all_nodes({config:?})");
        match result.err().unwrap() {
            NamedConceptListError::InsufficientPermissions => {
                fail!(from origin, with NodeListFailure::InsufficientPermissions,
                        "{} due to insufficient permissions while listing all nodes.", msg);
            }
            NamedConceptListError::InternalError => {
                fail!(from origin, with NodeListFailure::InternalError,
                        "{} due to an internal failure while listing all nodes.", msg);
            }
        }
    }

    fn state_from_monitor(
        monitor: &<Service::Monitoring as Monitoring>::Monitor,
    ) -> Result<State, NodeListFailure> {
        let result = monitor.state();

        if let Ok(result) = result {
            return Ok(result);
        }

        let msg = "Unable to acquire node state from monitor";
        let origin = format!("Node::state_from_monitor({monitor:?})");

        match result.err().unwrap() {
            MonitoringStateError::InsufficientPermissions => {
                fail!(from origin, with NodeListFailure::InsufficientPermissions,
                    "{} due to insufficient permissions to acquire the nodes state.", msg);
            }
            MonitoringStateError::Interrupt => {
                fail!(from origin, with NodeListFailure::Interrupt,
                    "{} due to an interrupt signal while acquiring the nodes state.", msg);
            }
            MonitoringStateError::InternalError => {
                fail!(from origin, with NodeListFailure::InternalError,
                    "{} due to an internal error while acquiring the nodes state.", msg);
            }
        }
    }

    fn get_node_state(config: &Config, node_id: &UniqueNodeId) -> Result<State, NodeListFailure> {
        let config = node_monitoring_config::<Service>(config);
        let result = <Service::Monitoring as Monitoring>::Builder::new(&node_id.as_file_name())
            .config(&config)
            .monitor();

        if let Ok(result) = result {
            return Self::state_from_monitor(&result);
        }

        let msg = "Unable to acquire node monitor";
        let origin = format!("Node::get_node_state({config:?}, {node_id:?})");
        match result.err().unwrap() {
            MonitoringCreateMonitorError::InsufficientPermissions => {
                fail!(from origin, with NodeListFailure::InsufficientPermissions,
                        "{} due to insufficient permissions while acquiring the node state.", msg);
            }
            MonitoringCreateMonitorError::Interrupt => {
                fail!(from origin, with NodeListFailure::Interrupt,
                        "{} since an interrupt was received while acquiring the node state.", msg);
            }
            MonitoringCreateMonitorError::InternalError
            | MonitoringCreateMonitorError::ConceptNameNotSupportedOnPlatform => {
                fail!(from origin, with NodeListFailure::InternalError,
                        "{} since an internal failure occurred while acquiring the node state.", msg);
            }
        }
    }

    fn open_node_storage(
        config: &Config,
        node_id: &UniqueNodeId,
    ) -> Result<Option<Service::StaticStorage>, NodeReadStorageFailure> {
        let details_config = node_details_config::<Service>(config, node_id);
        let msg = "Unable to open node config storage";
        let origin = format!("open_node_storage({config:?}, {node_id:?})");

        match <Service::StaticStorage as StaticStorage>::Builder::new(
            &FileName::new(b"node").unwrap(),
        )
        .config(&details_config)
        .has_ownership(false)
        .open(Duration::ZERO)
        {
            Ok(result) => Ok(Some(result)),
            Err(StaticStorageOpenError::DoesNotExist) => Ok(None),
            Err(StaticStorageOpenError::Read) => {
                fail!(from origin, with NodeReadStorageFailure::ReadError,
                        "{} since the node config storage could not be read.", msg);
            }
            Err(StaticStorageOpenError::InitializationNotYetFinalized) => {
                fail!(from origin, with NodeReadStorageFailure::Corrupted,
                        "{} since the node config storage seems to be uninitialized but the state should always be present.", msg);
            }
            Err(StaticStorageOpenError::InternalError) => {
                fail!(from origin, with NodeReadStorageFailure::InternalError,
                        "{} due to an internal failure while opening the node config storage.", msg);
            }
            Err(StaticStorageOpenError::Interrupt) => {
                fail!(from origin, with NodeReadStorageFailure::Interrupt,
                    "{} since an interrupt signal was raised.", msg);
            }
            Err(StaticStorageOpenError::InsufficientPermissions) => {
                fail!(from origin, with NodeReadStorageFailure::InsufficientPermissions,
                    "{} due to insufficient permissions.", msg);
            }
        }
    }

    fn get_node_details(
        config: &Config,
        node_id: &UniqueNodeId,
    ) -> Result<Option<NodeDetails>, NodeReadStorageFailure> {
        let node_storage = if let Some(n) = Self::open_node_storage(config, node_id)? {
            n
        } else {
            return Ok(None);
        };

        let mut read_content =
            String::from_utf8(vec![b' '; node_storage.len() as usize]).expect("");

        let origin = format!("get_node_details({config:?}, {node_id:?})");
        let msg = "Unable to read node details";

        if node_storage
            .read(unsafe { read_content.as_mut_vec() }.as_mut_slice())
            .is_err()
        {
            fail!(from origin, with NodeReadStorageFailure::ReadError,
                "{} since the content of the node config storage could not be read.", msg);
        }

        let node_details = fail!(from origin,
                    when Service::ConfigSerializer::deserialize::<NodeDetails>(unsafe { read_content.as_mut_vec()}),
                    with NodeReadStorageFailure::Corrupted,
                "{} since the contents of the node config storage is corrupted.", msg);

        Ok(Some(node_details))
    }

    fn port_tags<F: FnMut(u128) -> CallbackProgression>(
        config: &Config,
        node_id: &UniqueNodeId,
        mut callback: F,
    ) -> Result<(), NodeReadPortTagsFailure> {
        let origin = "Node::service_tags()";
        let msg = format!("Unable to acquire all port tags of the node {node_id:?}");
        match <Service::StaticStorage as NamedConceptMgmt>::list_cfg(&port_tag_config::<Service>(
            config, node_id,
        )) {
            Ok(tags) => {
                for tag in &tags {
                    if let Ok(v) = tag.to_string().parse::<u128>() {
                        if callback(v) == CallbackProgression::Stop {
                            break;
                        }
                    } else {
                        continue;
                    }
                }
                Ok(())
            }
            Err(NamedConceptListError::InsufficientPermissions) => {
                fail!(from origin, with NodeReadPortTagsFailure::InsufficientPermissions,
                    "{} due to insufficient permissions.", msg);
            }
            Err(NamedConceptListError::InternalError) => {
                fail!(from origin, with NodeReadPortTagsFailure::InternalError,
                    "{} due to an internal error.", msg);
            }
        }
    }

    fn service_tags<F: FnMut(&ServiceHash) -> CallbackProgression>(
        config: &Config,
        node_id: &UniqueNodeId,
        mut callback: F,
    ) -> Result<(), NodeReadServiceTagsFailure> {
        let origin = "Node::service_tags()";
        let msg = format!("Unable to acquire all service tags of the node {node_id:?}");
        match <Service::StaticStorage as NamedConceptMgmt>::list_cfg(
            &service_tag_config::<Service>(config, node_id),
        ) {
            Ok(tags) => {
                for tag in &tags {
                    if let Ok(v) = tag.try_into() {
                        if callback(&ServiceHash(v)) == CallbackProgression::Stop {
                            break;
                        }
                    } else {
                        continue;
                    }
                }
                Ok(())
            }
            Err(NamedConceptListError::InsufficientPermissions) => {
                fail!(from origin, with NodeReadServiceTagsFailure::InsufficientPermissions,
                    "{} due to insufficient permissions.", msg);
            }
            Err(NamedConceptListError::InternalError) => {
                fail!(from origin, with NodeReadServiceTagsFailure::InternalError,
                    "{} due to an internal error.", msg);
            }
        }
    }
}

/// Creates a [`Node`].
///
/// ```
/// use iceoryx2::prelude::*;
///
/// # fn main() -> Result<(), Box<dyn core::error::Error>> {
/// let node = NodeBuilder::new()
///                 .name(&"my_little_node".try_into()?)
///                 .create::<ipc::Service>()?;
///
/// // do things with your cool new node
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Default, Clone)]
pub struct NodeBuilder {
    name: Option<NodeName>,
    signal_handling_mode: SignalHandlingMode,
    config: Option<Config>,
}

impl NodeBuilder {
    /// Creates a new [`NodeBuilder`]
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the [`NodeName`] of the to be created [`Node`].
    pub fn name(mut self, value: &NodeName) -> Self {
        self.name = Some(value.clone());
        self
    }

    /// Defines the [`SignalHandlingMode`] for the [`Node`]. It affects the [`Node::wait()`] call
    /// that returns any received [`Signal`](iceoryx2_bb_posix::signal::Signal) via its
    /// [`NodeWaitFailure`]
    pub fn signal_handling_mode(mut self, value: SignalHandlingMode) -> Self {
        self.signal_handling_mode = value;
        self
    }

    /// Sets the config of the [`Node`] that will be used to create all entities owned by the
    /// [`Node`].
    pub fn config(mut self, value: &Config) -> Self {
        self.config = Some(value.clone());
        self
    }

    /// Creates a new [`Node`] for a specific [`service::Service`]. All entities owned by the
    /// [`Node`] will have the same [`service::Service`].
    pub fn create<Service: service::Service>(self) -> Result<Node<Service>, NodeCreationFailure> {
        let msg = "Unable to create node";

        let config = self
            .config
            .as_ref()
            .unwrap_or_else(|| Config::global_config());

        let name = match &self.name {
            Some(n) => n.clone(),
            None => NodeName::default(),
        };
        let node_id = fail!(from self, when UniqueNodeId::new::<Service>(name, config),
            with NodeCreationFailure::UnableToGenerateUniqueNodeId,
            "{msg} since the UniqueNodeId could not be generated.");

        let monitor_name = fatal_panic!(from self, when FileName::new(node_id.value().to_string().as_bytes()),
                                "This should never happen! {msg} since the UniqueNodeId is not a valid file name.");
        let (details_storage, details) =
            self.create_node_details_storage::<Service>(config, &node_id)?;
        let monitoring_token = self.create_token::<Service>(config, &monitor_name)?;

        let state = Arc::new(SharedNodeState {
            id: node_id,
            monitoring_token: UnsafeCell::new(Some(monitoring_token)),
            registered_services: RegisteredServices::new(),
            details_storage,
            signal_handling_mode: self.signal_handling_mode,
            details,
        });

        if config.global.node.cleanup_dead_nodes_on_creation {
            state.blocking_cleanup_dead_nodes(Duration::ZERO);
        }

        let new_node = Node {
            shared: SharedNode { state },
        };

        trace!(from new_node, "created");
        Ok(new_node)
    }

    fn create_token<Service: service::Service>(
        &self,
        config: &Config,
        monitor_name: &FileName,
    ) -> Result<<Service::Monitoring as Monitoring>::Token, NodeCreationFailure> {
        let msg = "Unable to create token for new node";
        let token_result = <Service::Monitoring as Monitoring>::Builder::new(monitor_name)
            .config(&node_monitoring_config::<Service>(config))
            .token();

        match token_result {
            Ok(token) => Ok(token),
            Err(MonitoringCreateTokenError::InsufficientPermissions) => {
                fail!(from self, with NodeCreationFailure::InsufficientPermissions,
                    "{msg} due to insufficient permissions to create a monitor token.");
            }
            Err(MonitoringCreateTokenError::AlreadyExists) => {
                fatal_panic!(from self,
                    "This should never happen! {msg} since a node with the same UniqueNodeId already exists.");
            }
            Err(MonitoringCreateTokenError::InternalError) => {
                fail!(from self, with NodeCreationFailure::InternalError,
                    "{msg} since the monitor token could not be created.");
            }
            Err(MonitoringCreateTokenError::SystemCorrupted) => {
                fail!(from self, with NodeCreationFailure::SystemCorrupted,
                    "{msg} since some external instance removed the underlying resources of the monitoring token.");
            }
        }
    }

    fn create_node_details_storage<Service: service::Service>(
        &self,
        config: &Config,
        node_id: &UniqueNodeId,
    ) -> Result<(Service::StaticStorage, NodeDetails), NodeCreationFailure> {
        let msg = "Unable to create node details storage";
        let details = NodeDetails::new(&self.name, config);

        let details_config = node_details_config::<Service>(&details.config, node_id);
        let serialized_details = match <Service::ConfigSerializer>::serialize(&details) {
            Ok(serialized_details) => serialized_details,
            Err(SerializeError::InternalError) => {
                fail!(from self, with NodeCreationFailure::InternalError,
                    "{msg} since the node details could not be serialized.");
            }
        };

        match <Service::StaticStorage as StaticStorage>::Builder::new(
            &FileName::new(b"node").unwrap(),
        )
        .config(&details_config)
        .has_ownership(false)
        .create(&serialized_details)
        {
            Ok(node_details) => Ok((node_details, details)),
            Err(StaticStorageCreateError::InsufficientPermissions) => {
                fail!(from self, with NodeCreationFailure::InsufficientPermissions,
                    "{msg} due to insufficient permissions to create the node details file.");
            }
            Err(StaticStorageCreateError::AlreadyExists) => {
                fatal_panic!(from self,
                    "This should never happen! {msg} since the node details file already exists.");
            }
            Err(e) => {
                fail!(from self, with NodeCreationFailure::InternalError,
                    "{msg} due to an unknown failure while creating the node details file {:?}.", e);
            }
        }
    }
}