use crate::error::{Error, Result};
use crate::sys;
use crate::task::{self, JoinError};
use std::mem;
use std::time::Duration;
#[derive(Clone, Copy, Debug)]
pub enum Profile {
Balanced,
ReleaseFast,
DebugSafe,
IoLatency,
}
impl Profile {
pub(crate) fn raw(self) -> u32 {
match self {
Self::Balanced => sys::LLAM_RUNTIME_PROFILE_BALANCED,
Self::ReleaseFast => sys::LLAM_RUNTIME_PROFILE_RELEASE_FAST,
Self::DebugSafe => sys::LLAM_RUNTIME_PROFILE_DEBUG_SAFE,
Self::IoLatency => sys::LLAM_RUNTIME_PROFILE_IO_LATENCY,
}
}
}
#[derive(Clone, Debug)]
pub struct RuntimeBuilder {
opts: sys::llam_runtime_opts_t,
}
impl RuntimeBuilder {
pub fn new() -> Self {
let mut opts = unsafe { mem::zeroed::<sys::llam_runtime_opts_t>() };
let rc = unsafe { sys::llam_runtime_opts_init(&mut opts, mem::size_of_val(&opts)) };
assert_eq!(rc, 0, "llam_runtime_opts_init failed");
Self { opts }
}
pub fn profile(mut self, profile: Profile) -> Self {
self.opts.profile = profile.raw();
self
}
pub fn deterministic(mut self, enabled: bool) -> Self {
self.opts.deterministic = u32::from(enabled);
self
}
pub fn forced_yield_every(mut self, every: u32) -> Self {
self.opts.forced_yield_every = every;
self
}
pub fn idle_spin(mut self, duration: Duration) -> Self {
self.opts.idle_spin_ns = duration.as_nanos().min(u128::from(u64::MAX)) as u64;
self
}
pub fn idle_spin_max_iters(mut self, iters: u32) -> Self {
self.opts.idle_spin_max_iters = iters;
self
}
pub fn sqpoll_cpu(mut self, cpu: i32) -> Self {
self.opts.sqpoll_cpu = cpu;
self
}
pub fn dynamic_workers(mut self, enabled: bool) -> Self {
self.set_flag(sys::LLAM_RUNTIME_EXPERIMENTAL_F_DYNAMIC_WORKERS, enabled);
self
}
pub fn worker_rings(mut self, enabled: bool) -> Self {
self.set_flag(sys::LLAM_RUNTIME_EXPERIMENTAL_F_WORKER_RINGS, enabled);
self
}
pub fn worker_rings_multishot(mut self, enabled: bool) -> Self {
self.set_flag(
sys::LLAM_RUNTIME_EXPERIMENTAL_F_WORKER_RINGS_MULTISHOT,
enabled,
);
self
}
pub fn lockfree_normq(mut self, enabled: bool) -> Self {
self.set_flag(sys::LLAM_RUNTIME_EXPERIMENTAL_F_LOCKFREE_NORMQ, enabled);
self
}
pub fn huge_alloc(mut self, enabled: bool) -> Self {
self.set_flag(sys::LLAM_RUNTIME_EXPERIMENTAL_F_HUGE_ALLOC, enabled);
self
}
pub fn sqpoll(mut self, enabled: bool) -> Self {
self.set_flag(sys::LLAM_RUNTIME_EXPERIMENTAL_F_SQPOLL, enabled);
self
}
pub fn experimental_flags(mut self, flags: u64) -> Self {
self.opts.experimental_flags = flags;
self
}
pub fn raw_options(&self) -> &sys::llam_runtime_opts_t {
&self.opts
}
pub fn init(self) -> Result<Runtime> {
let rc = unsafe { sys::llam_runtime_init_ex(&self.opts, mem::size_of_val(&self.opts)) };
if rc == 0 {
Ok(Runtime { active: true })
} else {
Err(Error::last())
}
}
pub fn create_handle(self) -> Result<RuntimeHandle> {
let mut raw = std::ptr::null_mut();
let rc =
unsafe { sys::llam_runtime_create(&self.opts, mem::size_of_val(&self.opts), &mut raw) };
if rc == 0 {
Ok(RuntimeHandle { raw, active: true })
} else {
Err(Error::last())
}
}
pub fn run<F>(self, f: F) -> Result<()>
where
F: FnOnce() -> Result<()> + Send + 'static,
{
let runtime = self.init()?;
runtime.run(f)
}
fn set_flag(&mut self, flag: u64, enabled: bool) {
if enabled {
self.opts.experimental_flags |= flag;
} else {
self.opts.experimental_flags &= !flag;
}
}
}
impl Default for RuntimeBuilder {
fn default() -> Self {
Self::new()
}
}
pub struct Runtime {
active: bool,
}
impl Runtime {
pub fn builder() -> RuntimeBuilder {
RuntimeBuilder::new()
}
pub fn init() -> Result<Self> {
RuntimeBuilder::new().init()
}
pub fn run<F>(self, f: F) -> Result<()>
where
F: FnOnce() -> Result<()> + Send + 'static,
{
let handle = task::try_spawn(f)?;
let run_rc = unsafe { sys::llam_run() };
let join_result = handle.join();
self.shutdown();
if run_rc != 0 {
return Err(Error::last());
}
match join_result {
Ok(result) => result,
Err(JoinError::Runtime(error)) => Err(error),
Err(JoinError::Panic(payload)) => std::panic::resume_unwind(payload),
Err(JoinError::MissingResult) => Err(Error::from_errno(libc::EIO)),
}
}
pub fn request_stop(&self) -> Result<()> {
let rc = unsafe { sys::llam_runtime_request_stop() };
if rc == 0 {
Ok(())
} else {
Err(Error::last())
}
}
pub fn stats(&self) -> Result<RuntimeStats> {
let mut raw = unsafe { mem::zeroed::<sys::llam_runtime_stats_t>() };
let rc = unsafe { sys::llam_runtime_collect_stats_ex(&mut raw, mem::size_of_val(&raw)) };
if rc == 0 {
Ok(RuntimeStats(raw))
} else {
Err(Error::last())
}
}
pub fn shutdown(mut self) {
self.shutdown_inner();
}
fn shutdown_inner(&mut self) {
if self.active {
unsafe { sys::llam_runtime_shutdown() };
self.active = false;
}
}
}
impl Drop for Runtime {
fn drop(&mut self) {
self.shutdown_inner();
}
}
#[derive(Clone, Copy, Debug)]
pub struct RuntimeStats(pub(crate) sys::llam_runtime_stats_t);
impl RuntimeStats {
pub fn raw(&self) -> &sys::llam_runtime_stats_t {
&self.0
}
pub fn ctx_switches(&self) -> u64 {
self.0.ctx_switches
}
pub fn yields(&self) -> u64 {
self.0.yields
}
pub fn parks(&self) -> u64 {
self.0.parks
}
pub fn wakes(&self) -> u64 {
self.0.wakes
}
pub fn steals(&self) -> u64 {
self.0.steals
}
pub fn migrations(&self) -> u64 {
self.0.migrations
}
pub fn blocking_calls(&self) -> u64 {
self.0.blocking_calls
}
pub fn blocking_completions(&self) -> u64 {
self.0.blocking_completions
}
pub fn io_submits(&self) -> u64 {
self.0.io_submits
}
pub fn io_submit_calls(&self) -> u64 {
self.0.io_submit_calls
}
pub fn io_submit_syscalls(&self) -> u64 {
self.0.io_submit_syscalls
}
pub fn io_completions(&self) -> u64 {
self.0.io_completions
}
pub fn idle_polls(&self) -> u64 {
self.0.idle_polls
}
pub fn idle_spin_loops(&self) -> u64 {
self.0.idle_spin_loops
}
pub fn idle_spin_hits(&self) -> u64 {
self.0.idle_spin_hits
}
pub fn idle_spin_fallbacks(&self) -> u64 {
self.0.idle_spin_fallbacks
}
pub fn idle_spin_ns(&self) -> u64 {
self.0.idle_spin_ns
}
pub fn active_workers(&self) -> u32 {
self.0.active_workers
}
pub fn online_workers(&self) -> u32 {
self.0.online_workers
}
pub fn online_workers_floor(&self) -> u32 {
self.0.online_workers_floor
}
pub fn online_workers_min(&self) -> u32 {
self.0.online_workers_min
}
pub fn online_workers_max(&self) -> u32 {
self.0.online_workers_max
}
pub fn active_nodes(&self) -> u32 {
self.0.active_nodes
}
pub fn dynamic_workers(&self) -> bool {
self.0.dynamic_workers != 0
}
pub fn worker_rings(&self) -> bool {
self.0.worker_rings != 0
}
pub fn worker_rings_multishot(&self) -> bool {
self.0.worker_rings_multishot != 0
}
pub fn lockfree_normq(&self) -> bool {
self.0.lockfree_normq != 0
}
pub fn huge_alloc(&self) -> bool {
self.0.huge_alloc != 0
}
pub fn sqpoll(&self) -> bool {
self.0.sqpoll != 0
}
pub fn queue_overflows(&self) -> u64 {
self.0.queue_overflows
}
pub fn overflow_depth(&self) -> u64 {
self.0.overflow_depth
}
pub fn opaque_block_ns(&self) -> u64 {
self.0.opaque_block_ns
}
pub fn opaque_block_samples(&self) -> u64 {
self.0.opaque_block_samples
}
pub fn opaque_block_max_ns(&self) -> u64 {
self.0.opaque_block_max_ns
}
pub fn opaque_enter_wait_ns(&self) -> u64 {
self.0.opaque_enter_wait_ns
}
pub fn opaque_enter_wait_samples(&self) -> u64 {
self.0.opaque_enter_wait_samples
}
pub fn opaque_enter_wait_max_ns(&self) -> u64 {
self.0.opaque_enter_wait_max_ns
}
pub fn opaque_leave_wait_ns(&self) -> u64 {
self.0.opaque_leave_wait_ns
}
pub fn opaque_leave_wait_samples(&self) -> u64 {
self.0.opaque_leave_wait_samples
}
pub fn opaque_leave_wait_max_ns(&self) -> u64 {
self.0.opaque_leave_wait_max_ns
}
pub fn yield_direct_attempts(&self) -> u64 {
self.0.yield_direct_attempts
}
pub fn yield_direct_fast_hits(&self) -> u64 {
self.0.yield_direct_fast_hits
}
pub fn yield_direct_locked_hits(&self) -> u64 {
self.0.yield_direct_locked_hits
}
pub fn yield_direct_fail_context(&self) -> u64 {
self.0.yield_direct_fail_context
}
pub fn yield_direct_fail_policy(&self) -> u64 {
self.0.yield_direct_fail_policy
}
pub fn yield_direct_fail_no_work(&self) -> u64 {
self.0.yield_direct_fail_no_work
}
pub fn yield_direct_fail_self(&self) -> u64 {
self.0.yield_direct_fail_self
}
pub fn yield_direct_fail_push(&self) -> u64 {
self.0.yield_direct_fail_push
}
}
pub fn current_runtime_raw() -> *mut sys::llam_runtime_t {
unsafe { sys::llam_runtime_default() }
}
pub struct RuntimeHandle {
raw: *mut sys::llam_runtime_t,
active: bool,
}
unsafe impl Send for RuntimeHandle {}
impl RuntimeHandle {
pub fn run(&self) -> Result<()> {
let rc = unsafe { sys::llam_runtime_run_handle(self.raw) };
if rc == 0 {
Ok(())
} else {
Err(Error::last())
}
}
pub fn raw(&self) -> *mut sys::llam_runtime_t {
self.raw
}
pub fn destroy(mut self) {
self.destroy_inner();
}
fn destroy_inner(&mut self) {
if self.active {
unsafe { sys::llam_runtime_destroy(self.raw) };
self.active = false;
self.raw = std::ptr::null_mut();
}
}
}
impl Drop for RuntimeHandle {
fn drop(&mut self) {
self.destroy_inner();
}
}