dvcompute_network 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::ops::Deref;
use std::slice;
use std::mem::ManuallyDrop;
use std::ptr;

use libc::*;

/// The network buffer representation.
#[repr(C)]
pub struct NetworkBuffer {

    /// Delete the object.
    delete: unsafe extern "C" fn(data: *mut u8, len: c_ulong, capacity: c_ulong),

    /// Data.
    data: *mut u8,

    /// The data size.
    len: c_ulong,

    /// The data capacity.
    capacity: c_ulong
}

impl Drop for NetworkBuffer {

    fn drop(&mut self) {
        unsafe {
            (self.delete)(self.data, self.len, self.capacity);
        }
    }
}

impl Deref for NetworkBuffer {

    type Target = [u8];

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

impl NetworkBuffer {

    /// Create an empty buffer.
    pub fn empty() -> Self {
        Self {
            delete: delete_empty_network_buffer,
            data: ptr::null_mut(),
            len: 0,
            capacity: 0
        }
    }

    /// Create a new buffer.
    pub fn new(vec: Vec<u8>) -> Self {
        let mut vec = ManuallyDrop::new(vec);
        Self {
            delete: delete_vector_network_buffer,
            data: vec.as_mut_ptr(),
            len: vec.len() as c_ulong,
            capacity: vec.capacity() as c_ulong
        }
    }

    /// Treat the buffer as a slice.
    pub fn as_slice(&self) -> &[u8] {
        unsafe {
            if self.len > 0 {
                slice::from_raw_parts(self.data, self.len as usize)
            } else {
                &[]
            }
        }
    }
}

/// Free the memory allocated for the network buffer.
unsafe extern "C" fn delete_empty_network_buffer(_data: *mut u8, _len: c_ulong, _capacity: c_ulong) {}

/// Free the memory allocated for the network buffer.
unsafe extern "C" fn delete_vector_network_buffer(data: *mut u8, len: c_ulong, capacity: c_ulong) {
    let _: Vec<u8> = Vec::from_raw_parts(data, len as usize, capacity as usize);
}

/// The trait object of the network support.
pub type NetworkTraitObject = *mut c_void;

/// The network support.
#[repr(C)]
pub struct NetworkSupport {

    /// Delete the object.
    pub delete: unsafe extern "C" fn(obj: *mut NetworkTraitObject),

    /// Get the rank.
    pub rank: unsafe extern "C" fn(obj: *const NetworkTraitObject) -> c_int,

    /// Get the network size.
    pub size: unsafe extern "C" fn(obj: *const NetworkTraitObject) -> c_int,

    /// Send the message synchronously.
    pub send: unsafe extern "C" fn(obj: *mut NetworkTraitObject, pid: c_int, msg: *const u8, count: c_ulong),

    /// Send the message asynchronously.
    pub send_async: unsafe extern "C" fn(obj: *mut NetworkTraitObject, pid: c_int, msg: NetworkBuffer),

    /// Try to receive a message and return the result equaled to 1 if the message was received, 0 if there are no messages and -1 if there was an error.
    pub try_recv: unsafe extern "C" fn(obj: *mut NetworkTraitObject, pid: *mut c_int, result: *mut c_int) -> NetworkBuffer,

    /// Receive the message and return the result equaled to 1 if the message was received and -1 if there was an error.
    pub recv: unsafe extern "C" fn(obj: *mut NetworkTraitObject, pid: *mut c_int, result: *mut c_int) -> NetworkBuffer,

    /// Test the requests.
    pub test: unsafe extern "C" fn(obj: *mut NetworkTraitObject),

    /// Wait for the requests.
    pub wait: unsafe extern "C" fn(obj: *mut NetworkTraitObject),

    /// Get the request count.
    pub request_count: unsafe extern "C" fn(obj: *const NetworkTraitObject) -> c_ulong,

    /// Abort the simulation.
    pub abort: unsafe extern "C" fn(obj: *mut NetworkTraitObject, code: i32),

    /// Block until the barrier has passed by all logical processes.
    pub barrier: unsafe extern "C" fn(obj: *mut NetworkTraitObject),

    /// The trait object.
    pub trait_object: NetworkTraitObject
}

impl Drop for NetworkSupport {

    fn drop(&mut self) {
        unsafe {
            (self.delete)(&mut self.trait_object);
        }
    }
}

impl NetworkSupport {

    /// Get the rank.
    pub fn rank(&self) -> c_int {
        unsafe {
            (self.rank)(&self.trait_object)
        }
    }

    /// Get the network size.
    pub fn size(&self) -> c_int {
        unsafe {
            (self.size)(&self.trait_object)
        }
    }

    /// Send the message synchronously.
    pub fn send(&mut self, pid: c_int, msg: &[u8]) {
        unsafe {
            (self.send)(&mut self.trait_object, pid, msg.as_ptr(), msg.len() as c_ulong)
        }
    }

    /// Send the message asynchronously.
    pub fn send_async(&mut self, pid: c_int, msg: NetworkBuffer) {
        unsafe {
            (self.send_async)(&mut self.trait_object, pid, msg)
        }
    }

    /// Try to receive a message.
    pub fn try_recv(&mut self) -> Result<Option<(NetworkBuffer, c_int)>, c_int> {
        unsafe {
            let mut pid = 0 as c_int;
            let mut result = 0 as c_int;
            let buf = (self.try_recv)(&mut self.trait_object, &mut pid, &mut result);
            if result == 0 {
                Result::Ok(None)
            } else if result > 0 {
                Result::Ok(Some((buf, pid)))
            } else {
                Result::Err(pid)
            }
        }
    }

    /// Receive the message.
    pub fn recv(&mut self) -> Result<(NetworkBuffer, c_int), c_int> {
        unsafe {
            let mut pid = 0 as c_int;
            let mut result = 0 as c_int;
            let buf = (self.recv)(&mut self.trait_object, &mut pid, &mut result);
            if result > 0 {
                Result::Ok((buf, pid))
            } else {
                Result::Err(pid)
            }
        }
    }

    /// Test the requests.
    pub fn test(&mut self) {
        unsafe {
            (self.test)(&mut self.trait_object)
        }
    }

    /// Wait for the requests.
    pub fn wait(&mut self) {
        unsafe {
            (self.wait)(&mut self.trait_object)
        }
    }

    /// Get the request count.
    pub fn request_count(&self) -> usize {
        unsafe {
            (self.request_count)(&self.trait_object) as usize
        }
    }

    /// Abort the simulation.
    pub fn abort(&mut self, code: i32) {
        unsafe {
            (self.abort)(&mut self.trait_object, code)
        }
    }

    /// Block until the barrier has passed by all logical processes.
    pub fn barrier(&mut self) {
        unsafe {
            (self.barrier)(&mut self.trait_object)
        }
    }
}