dvcompute 2.0.0

// Copyright (c) 2020-2022  David Sorokin <davsor@mail.ru>, based in Yoshkar-Ola, Russia
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

use std::time::Duration;
use std::ptr;
use std::ffi::CString;
use std::ffi::CStr;

use libc::*;

/// A context of the logical process.
pub type LogicalProcessContext = c_void;

/// The threading mode used for simulation.
#[derive(Clone)]
pub enum LogicalProcessThread {

  /// This is the default single threaded mode that corresponds to `MPI_THREAD_SINGLE`.
  SingleThread,

  /// This is the two threaded mode that corresponds to `MPI_THREAD_FUNNELED`.
  FunneledThread
}

/// The logical process parameters.
#[derive(Clone)]
pub struct LogicalProcessParameters {

    /// The logical process name.
    pub name: String,

    /// The lookahead parameter.
    pub lookahead: f64,

    /// The threshold to test the lookahead time (single-threaded mode only)
    pub lookahead_time_test_threshold: usize,

    /// The timeout used for synchronising the operations.
    pub sync_timeout: Duration,

    /// The timeout used when intiailizing the logical process.
    pub initializing_timeout: Duration,

    /// The idle loop interval when processing the messages.
    pub idle_loop_interval: Option<Duration>,

    /// The test interval for network messages.
    pub network_test_interval: Duration,

    /// The threshold of network requests to test.
    pub network_test_threshold: usize,

    /// The threshold of network requests to wait.
    pub network_wait_threshold: usize,

    /// The thread mode used for running the logical process.
    pub thread_mode: LogicalProcessThread
}

impl LogicalProcessParameters {

    /// Create new logical process parameters.
    pub fn new(lookahead: f64) -> LogicalProcessParameters {
        LogicalProcessParameters {
            name: String::from("LP"),
            lookahead: lookahead,
            lookahead_time_test_threshold: 1000,
            sync_timeout: Duration::from_secs(60),
            initializing_timeout: Duration::from_secs(10),
            idle_loop_interval: None,
            network_test_interval: Duration::from_millis(10),
            network_test_threshold: 256,
            network_wait_threshold: 8192,
            thread_mode: LogicalProcessThread::SingleThread
        }
    }
}

/// A C-friendly representation of the logical process parameters.
#[repr(C)]
pub struct LogicalProcessParametersRepr {

    /// Delete the representation.
    delete_repr: unsafe extern "C" fn(*mut LogicalProcessParametersRepr),

    /// The logical process name.
    name: *mut c_char,

    /// Delete the logical process name.
    delete_name: unsafe extern "C" fn(*mut c_char),

    /// The lookahead parameter.
    lookahead: f64,

    /// The threshold to test the lookahead time (single-threaded mode only)
    lookahead_time_test_threshold: usize,

    /// The timeout used for synchronising the operations.
    sync_timeout: [u8; 16],

    /// The timeout used when intiailizing the logical process.
    initializing_timeout: [u8; 16],

    /// The idle loop interval when processing the messages.
    idle_loop_interval: [u8; 16],

    /// A flag indicating whether the idle loop interval is defined.
    idle_loop_interval_defined: c_int,

    /// The test interval for network messages.
    network_test_interval: [u8; 16],

    /// The threshold of network requests to test.
    network_test_threshold: usize,

    /// The threshold of network requests to wait.
    network_wait_threshold: usize,

    /// The thread mode used for running the logical process.
    thread_mode: c_int
}

impl LogicalProcessParametersRepr {

    /// Convert to a C-friendly representation.
    pub fn new(ps: LogicalProcessParameters) -> Self {
        let name = CString::new(ps.name.clone()).expect("NulError");
        let name = CString::into_raw(name);
        LogicalProcessParametersRepr {
            delete_repr: delete_lp_repr,
            name: name,
            delete_name: delete_lp_name,
            lookahead: ps.lookahead,
            lookahead_time_test_threshold: ps.lookahead_time_test_threshold,
            sync_timeout: u128::to_ne_bytes(ps.sync_timeout.as_nanos()),
            initializing_timeout: u128::to_ne_bytes(ps.initializing_timeout.as_nanos()),
            idle_loop_interval: u128::to_ne_bytes(ps.idle_loop_interval.unwrap_or(Duration::from_micros(3)).as_nanos()),
            idle_loop_interval_defined: if ps.idle_loop_interval.is_some() { 1 } else { 0 },
            network_test_interval: u128::to_ne_bytes(ps.network_test_interval.as_nanos()),
            network_test_threshold: ps.network_test_threshold,
            network_wait_threshold: ps.network_wait_threshold,
            thread_mode: {
                match ps.thread_mode {
                    LogicalProcessThread::SingleThread => 0,
                    LogicalProcessThread::FunneledThread => 1
                }
            }
        }
    }
}

impl Drop for LogicalProcessParametersRepr {

    fn drop(&mut self) {
        unsafe {
            (self.delete_name)(self.name);
        }
    }
}

unsafe extern "C" fn delete_lp_name(name: *mut c_char) {
    if name != ptr::null_mut() {
        let _ = CString::from_raw(name);
    }
}

unsafe extern "C" fn delete_lp_repr(e: *mut LogicalProcessParametersRepr) {
    if e != ptr::null_mut() {
        let _ = Box::from_raw(e);
    }
}

/// Convert the `LogicalProcessParametersRepr` referenced to by the FFI pointer to an `LogicalProcessParameters`.
pub unsafe fn ffi_logical_process_parameters_repr_into(ps: *mut LogicalProcessParametersRepr) -> LogicalProcessParameters {
    let name = CStr::from_ptr((*ps).name);
    let name = name.to_bytes().to_vec();
    let name = String::from_utf8(name).expect("FromUtf8Error");
    let xs   = LogicalProcessParameters {
        name: name,
        lookahead: (*ps).lookahead,
        lookahead_time_test_threshold: (*ps).lookahead_time_test_threshold,
        sync_timeout: Duration::from_nanos(u128::from_ne_bytes((*ps).sync_timeout) as u64),
        initializing_timeout: Duration::from_nanos(u128::from_ne_bytes((*ps).initializing_timeout) as u64),
        idle_loop_interval: if (*ps).idle_loop_interval_defined != 0 { Some(Duration::from_nanos(u128::from_ne_bytes((*ps).idle_loop_interval) as u64)) } else { None },
        network_test_interval: Duration::from_nanos(u128::from_ne_bytes((*ps).network_test_interval) as u64),
        network_test_threshold: (*ps).network_test_threshold,
        network_wait_threshold: (*ps).network_wait_threshold,
        thread_mode: {
            if (*ps).thread_mode != 0 { LogicalProcessThread::FunneledThread } else { LogicalProcessThread::SingleThread }
        }
    };
    ((*ps).delete_repr)(ps);
    xs
}