use std::collections::BTreeMap;
use std::fmt;
use std::time::Duration;
use oxdock_func_macro::oxdock_type;
use oxdock_pipe::{PipeHandle, new_handle_in_task};
pub fn type_anchor(name: &str) -> String {
format!("value-type-{}", name.to_lowercase())
}
pub fn startup_descriptors() -> [(&'static str, &'static TypeDescriptor); 12] {
[
("INT", IntValue::descriptor()),
("FLOAT", FloatValue::descriptor()),
("STRING", StringValue::descriptor()),
("BOOL", BoolValue::descriptor()),
("LIST", ListValue::descriptor()),
("MAP", MapValue::descriptor()),
("PATH", PathValue::descriptor()),
("DURATION", DurationValue::descriptor()),
("PIPE", PipeValue::descriptor()),
("HANDLE", HandleValue::descriptor()),
("SEMAPHORE", SemaphoreValue::descriptor()),
("PERMIT", SemaphorePermit::descriptor()),
]
}
#[oxdock_type(crate_path = "::oxdock_parser", name = "INT", inline)]
#[derive(Debug, Clone, Copy, PartialEq)]
struct IntValue(pub i64);
#[oxdock_type(crate_path = "::oxdock_parser", name = "FLOAT", inline)]
#[derive(Debug, Clone, Copy, PartialEq)]
struct FloatValue(pub f64);
#[oxdock_type(
crate_path = "::oxdock_parser",
name = "STRING",
summary = "Arbitrary text."
)]
#[derive(Debug, Clone, PartialEq)]
struct StringValue(pub String);
#[oxdock_type(crate_path = "::oxdock_parser", name = "BOOL", inline)]
#[derive(Debug, Clone, Copy, PartialEq)]
struct BoolValue(pub bool);
#[oxdock_type(crate_path = "::oxdock_parser", name = "LIST", shared)]
#[derive(Debug, Clone, PartialEq)]
struct ListValue(pub Vec<Value>);
#[oxdock_type(crate_path = "::oxdock_parser", name = "MAP", shared)]
#[derive(Debug, Clone, PartialEq)]
struct MapValue(pub BTreeMap<String, Value>);
#[oxdock_type(crate_path = "::oxdock_parser", name = "PATH")]
#[derive(Debug, Clone, PartialEq)]
#[allow(clippy::disallowed_types)]
struct PathValue(#[allow(clippy::disallowed_types)] pub std::path::PathBuf);
#[oxdock_type(
crate_path = "::oxdock_parser",
name = "DURATION",
summary = "Positive time span."
)]
#[derive(Debug, Clone, PartialEq)]
struct DurationValue(pub Duration);
#[oxdock_type(
crate_path = "::oxdock_parser",
name = "PIPE",
summary = "Anonymous pipe handle.",
shared
)]
#[derive(Debug, Clone)]
struct PipeValue(pub PipeHandle);
impl PartialEq for PipeValue {
fn eq(&self, other: &Self) -> bool {
self.0.ptr_eq(&other.0)
}
}
#[oxdock_type(crate_path = "::oxdock_parser", name = "HANDLE", inline)]
#[derive(Debug, Clone, Copy, PartialEq)]
struct HandleValue(pub u64);
#[derive(Debug)]
pub struct SemaphoreState {
max: usize,
held: std::sync::atomic::AtomicUsize,
}
impl SemaphoreState {
fn new(max: usize) -> Self {
Self {
max,
held: std::sync::atomic::AtomicUsize::new(0),
}
}
pub fn try_acquire(&self) -> bool {
use std::sync::atomic::Ordering;
let mut current = self.held.load(Ordering::Acquire);
loop {
if current >= self.max {
return false;
}
match self.held.compare_exchange_weak(
current,
current + 1,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => return true,
Err(actual) => current = actual,
}
}
}
pub fn release(&self) {
use std::sync::atomic::Ordering;
let _ = self
.held
.try_update(Ordering::AcqRel, Ordering::Acquire, |held| {
held.checked_sub(1)
});
}
pub fn available(&self) -> usize {
self.max
.saturating_sub(self.held.load(std::sync::atomic::Ordering::Acquire))
}
}
#[oxdock_type(
crate_path = "::oxdock_parser",
name = "SEMAPHORE",
summary = "Counting semaphore for admission control.",
shared
)]
#[derive(Debug, Clone)]
struct SemaphoreValue(pub std::sync::Arc<SemaphoreState>);
impl PartialEq for SemaphoreValue {
fn eq(&self, other: &Self) -> bool {
std::sync::Arc::ptr_eq(&self.0, &other.0)
}
}
#[derive(Debug)]
struct PermitInner {
sem: std::sync::Arc<SemaphoreState>,
}
impl Drop for PermitInner {
fn drop(&mut self) {
self.sem.release();
}
}
#[oxdock_type(
crate_path = "::oxdock_parser",
name = "PERMIT",
summary = "Opaque admission permit; last drop releases it.",
shared
)]
#[derive(Debug, Clone)]
struct SemaphorePermit(pub std::sync::Arc<PermitInner>);
impl PartialEq for SemaphorePermit {
fn eq(&self, other: &Self) -> bool {
std::sync::Arc::ptr_eq(&self.0, &other.0)
}
}
impl fmt::Display for IntValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.0)
}
}
impl fmt::Display for FloatValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.0)
}
}
impl fmt::Display for BoolValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.0)
}
}
impl fmt::Display for HandleValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "task#{}", self.0)
}
}
impl fmt::Display for StringValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "\"{}\"", self.0)
}
}
impl fmt::Display for ListValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "[")?;
for (i, item) in self.0.iter().enumerate() {
if i > 0 {
write!(f, ", ")?;
}
write!(f, "{}", item)?;
}
write!(f, "]")
}
}
impl fmt::Display for MapValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{{")?;
for (i, (k, v)) in self.0.iter().enumerate() {
if i > 0 {
write!(f, ", ")?;
}
write!(f, "{}: {}", k, v)?;
}
write!(f, "}}")
}
}
impl fmt::Display for DurationValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", crate::command::format_duration(&self.0))
}
}
impl fmt::Display for PathValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.0.display())
}
}
impl fmt::Display for PipeValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "<pipe>")
}
}
impl fmt::Display for SemaphoreValue {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "<semaphore>")
}
}
impl fmt::Display for SemaphorePermit {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "<permit>")
}
}
#[repr(C)]
#[derive(Clone, Copy)]
pub union ValuePayload {
as_u64: u64,
as_ptr: *mut (),
}
unsafe impl Send for ValuePayload {}
unsafe impl Sync for ValuePayload {}
unsafe impl Send for Value {}
unsafe impl Sync for Value {}
#[repr(C)]
pub struct Value {
vtable: &'static TypeDescriptor,
payload: ValuePayload,
}
impl Value {
pub fn descriptor(&self) -> &'static TypeDescriptor {
self.vtable
}
pub fn type_name(&self) -> &'static str {
self.vtable.name
}
pub fn inline_bits(&self) -> u64 {
unsafe { self.payload.as_u64 }
}
pub fn heap_ptr(&self) -> *mut () {
unsafe { self.payload.as_ptr }
}
pub fn mint_inline<T>(descriptor: &'static TypeDescriptor, value: T) -> Self
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
Self {
vtable: descriptor,
payload: store_inline(value),
}
}
pub fn mint_heap<T>(descriptor: &'static TypeDescriptor, value: T) -> Self
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
Self {
vtable: descriptor,
payload: ValuePayload {
as_ptr: Box::into_raw(Box::new(value)) as *mut (),
},
}
}
pub fn mint_heap_shared<T>(descriptor: &'static TypeDescriptor, value: T) -> Self
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
Self {
vtable: descriptor,
payload: ValuePayload {
as_ptr: std::sync::Arc::into_raw(std::sync::Arc::new(value)) as *mut (),
},
}
}
pub fn read_inline<T>(&self, expected: &'static TypeDescriptor) -> Option<T>
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
if !std::ptr::eq(self.vtable, expected) {
return None;
}
Some(unsafe { load_inline::<T>(self.payload) })
}
pub fn read_heap<T>(&self, expected: &'static TypeDescriptor) -> Option<&T>
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
if !std::ptr::eq(self.vtable, expected) {
return None;
}
Some(unsafe { &*(self.payload.as_ptr as *const T) })
}
pub fn read_heap_mut<T>(&mut self, expected: &'static TypeDescriptor) -> Option<&mut T>
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
if !std::ptr::eq(self.vtable, expected) {
return None;
}
Some(unsafe { &mut *((self.vtable.unshare)(&mut self.payload) as *mut T) })
}
pub fn int(n: i64) -> Self {
Self::mint_inline(IntValue::descriptor(), IntValue(n))
}
pub fn float(f: f64) -> Self {
Self::mint_inline(FloatValue::descriptor(), FloatValue(f))
}
pub fn bool(b: bool) -> Self {
Self::mint_inline(BoolValue::descriptor(), BoolValue(b))
}
pub fn handle(id: u64) -> Self {
Self::mint_inline(HandleValue::descriptor(), HandleValue(id))
}
pub fn string(s: String) -> Self {
Self::mint_heap(StringValue::descriptor(), StringValue(s))
}
pub fn list(items: Vec<Value>) -> Self {
Self::mint_heap_shared(ListValue::descriptor(), ListValue(items))
}
pub fn map(entries: BTreeMap<String, Value>) -> Self {
Self::mint_heap_shared(MapValue::descriptor(), MapValue(entries))
}
#[allow(clippy::disallowed_types)]
pub fn path(p: std::path::PathBuf) -> Self {
Self::mint_heap(PathValue::descriptor(), PathValue(p))
}
pub fn duration(d: Duration) -> Self {
Self::mint_heap(DurationValue::descriptor(), DurationValue(d))
}
pub fn pipe_fresh() -> Self {
Self::pipe_fresh_in_task(0)
}
pub fn pipe_fresh_in_task(task_id: u64) -> Self {
Self::mint_heap_shared(
PipeValue::descriptor(),
PipeValue(new_handle_in_task(task_id)),
)
}
pub fn pipe_handle(handle: PipeHandle) -> Self {
Self::mint_heap_shared(PipeValue::descriptor(), PipeValue(handle))
}
pub fn semaphore(max: usize) -> Self {
Self::mint_heap_shared(
SemaphoreValue::descriptor(),
SemaphoreValue(std::sync::Arc::new(SemaphoreState::new(max))),
)
}
pub fn permit(sem: &std::sync::Arc<SemaphoreState>) -> Self {
Self::mint_heap_shared(
SemaphorePermit::descriptor(),
SemaphorePermit(std::sync::Arc::new(PermitInner {
sem: std::sync::Arc::clone(sem),
})),
)
}
pub fn as_semaphore(&self) -> Option<std::sync::Arc<SemaphoreState>> {
self.read_heap::<SemaphoreValue>(SemaphoreValue::descriptor())
.map(|v| std::sync::Arc::clone(&v.0))
}
pub fn as_i64(&self) -> Option<i64> {
self.read_inline::<IntValue>(IntValue::descriptor())
.map(|v| v.0)
}
pub fn as_f64(&self) -> Option<f64> {
self.read_inline::<FloatValue>(FloatValue::descriptor())
.map(|v| v.0)
}
pub fn as_bool(&self) -> Option<bool> {
self.read_inline::<BoolValue>(BoolValue::descriptor())
.map(|v| v.0)
}
pub fn as_handle(&self) -> Option<u64> {
self.read_inline::<HandleValue>(HandleValue::descriptor())
.map(|v| v.0)
}
pub fn as_str(&self) -> Option<&str> {
self.read_heap::<StringValue>(StringValue::descriptor())
.map(|v| v.0.as_str())
}
pub fn as_list(&self) -> Option<&Vec<Value>> {
self.read_heap::<ListValue>(ListValue::descriptor())
.map(|v| &v.0)
}
pub fn as_list_mut(&mut self) -> Option<&mut Vec<Value>> {
self.read_heap_mut::<ListValue>(ListValue::descriptor())
.map(|v| &mut v.0)
}
pub fn as_map(&self) -> Option<&BTreeMap<String, Value>> {
self.read_heap::<MapValue>(MapValue::descriptor())
.map(|v| &v.0)
}
pub fn as_map_mut(&mut self) -> Option<&mut BTreeMap<String, Value>> {
self.read_heap_mut::<MapValue>(MapValue::descriptor())
.map(|v| &mut v.0)
}
pub fn as_pipe_handle(&self) -> Option<PipeHandle> {
self.read_heap::<PipeValue>(PipeValue::descriptor())
.map(|v| v.0.clone())
}
pub fn as_duration(&self) -> Option<Duration> {
self.read_heap::<DurationValue>(DurationValue::descriptor())
.map(|v| v.0)
}
#[allow(clippy::disallowed_types)]
pub fn as_path(&self) -> Option<&std::path::Path> {
self.read_heap::<PathValue>(PathValue::descriptor())
.map(|v| v.0.as_path())
}
}
impl Clone for Value {
fn clone(&self) -> Self {
let payload = unsafe { (self.vtable.clone)(self.payload) };
Self {
vtable: self.vtable,
payload,
}
}
}
impl Drop for Value {
fn drop(&mut self) {
unsafe { (self.vtable.drop)(self.payload) };
}
}
impl PartialEq for Value {
fn eq(&self, other: &Self) -> bool {
if !std::ptr::eq(self.vtable, other.vtable) {
return false;
}
unsafe { (self.vtable.eq)(self.payload, other.payload) }
}
}
impl fmt::Debug for Value {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}(", self.vtable.name)?;
unsafe { (self.vtable.fmt)(self.payload, f) }?;
write!(f, ")")
}
}
impl fmt::Display for Value {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
unsafe { (self.vtable.fmt)(self.payload, f) }
}
}
pub trait OxDockType {
fn descriptor() -> &'static TypeDescriptor;
}
#[derive(Clone, Copy)]
pub struct TypeDescriptor {
pub name: &'static str,
pub summary: &'static str,
pub docs: &'static str,
pub clone: unsafe fn(ValuePayload) -> ValuePayload,
pub drop: unsafe fn(ValuePayload),
pub eq: unsafe fn(ValuePayload, ValuePayload) -> bool,
pub fmt: unsafe fn(ValuePayload, &mut fmt::Formatter<'_>) -> fmt::Result,
pub unshare: unsafe fn(&mut ValuePayload) -> *mut (),
}
pub fn store_inline<T>(value: T) -> ValuePayload
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
assert!(
std::mem::size_of::<T>() <= 8,
"inline payloads hold at most 64 bits"
);
let mut bits: u64 = 0;
unsafe {
std::ptr::copy_nonoverlapping(
&value as *const T as *const u8,
&mut bits as *mut u64 as *mut u8,
std::mem::size_of::<T>(),
);
}
ValuePayload { as_u64: bits }
}
pub unsafe fn load_inline<T>(payload: ValuePayload) -> T
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
#[cfg(debug_assertions)]
if std::mem::size_of::<T>() > 8 {
panic!("inline payloads hold at most 64 bits");
}
let mut value = std::mem::MaybeUninit::<T>::uninit();
unsafe {
std::ptr::copy_nonoverlapping(
&payload.as_u64 as *const u64 as *const u8,
value.as_mut_ptr() as *mut u8,
std::mem::size_of::<T>(),
);
value.assume_init()
}
}
pub unsafe fn clone_copy(payload: ValuePayload) -> ValuePayload {
payload
}
pub unsafe fn drop_noop(_payload: ValuePayload) {}
pub unsafe fn eq_inline<T>(a: ValuePayload, b: ValuePayload) -> bool
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
unsafe { load_inline::<T>(a) == load_inline::<T>(b) }
}
pub unsafe fn fmt_inline<T>(payload: ValuePayload, f: &mut fmt::Formatter) -> fmt::Result
where
T: Copy + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
write!(f, "{}", unsafe { load_inline::<T>(payload) })
}
pub unsafe fn clone_boxed<T>(payload: ValuePayload) -> ValuePayload
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let source = unsafe { &*(payload.as_ptr as *const T) };
ValuePayload {
as_ptr: Box::into_raw(Box::new(source.clone())) as *mut (),
}
}
pub unsafe fn drop_boxed<T>(payload: ValuePayload)
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
drop(unsafe { Box::from_raw(payload.as_ptr as *mut T) });
}
pub unsafe fn eq_boxed<T>(a: ValuePayload, b: ValuePayload) -> bool
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let left = unsafe { &*(a.as_ptr as *const T) };
let right = unsafe { &*(b.as_ptr as *const T) };
left == right
}
pub unsafe fn fmt_boxed<T>(payload: ValuePayload, f: &mut fmt::Formatter) -> fmt::Result
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let value = unsafe { &*(payload.as_ptr as *const T) };
write!(f, "{value}")
}
pub unsafe fn clone_shared<T>(payload: ValuePayload) -> ValuePayload
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
unsafe { std::sync::Arc::increment_strong_count(payload.as_ptr as *const T) };
payload
}
pub unsafe fn drop_shared<T>(payload: ValuePayload)
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
drop(unsafe { std::sync::Arc::from_raw(payload.as_ptr as *const T) });
}
pub unsafe fn eq_shared<T>(a: ValuePayload, b: ValuePayload) -> bool
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let left = unsafe { &*(a.as_ptr as *const T) };
let right = unsafe { &*(b.as_ptr as *const T) };
left == right
}
pub unsafe fn fmt_shared<T>(payload: ValuePayload, f: &mut fmt::Formatter) -> fmt::Result
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let value = unsafe { &*(payload.as_ptr as *const T) };
write!(f, "{value}")
}
pub unsafe fn unshare_inline(payload: &mut ValuePayload) -> *mut () {
let _ = payload;
panic!("inline words have no heap buffer to unshare");
}
pub unsafe fn unshare_boxed<T>(payload: &mut ValuePayload) -> *mut ()
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
unsafe { payload.as_ptr }
}
pub unsafe fn unshare_shared<T>(payload: &mut ValuePayload) -> *mut ()
where
T: Clone + PartialEq + fmt::Display + fmt::Debug + Send + Sync + 'static,
{
let raw = unsafe { payload.as_ptr } as *const T;
let mut shared = unsafe { std::sync::Arc::from_raw(raw) };
let unique = std::sync::Arc::make_mut(&mut shared);
let out = unique as *mut T;
payload.as_ptr = std::sync::Arc::into_raw(shared) as *mut ();
out as *mut ()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn semaphore_counts_exactly_to_cap() {
let sem = Value::semaphore(2);
let inner = sem.as_semaphore().expect("SEMAPHORE word");
assert_eq!(inner.available(), 2);
assert!(inner.try_acquire());
assert_eq!(inner.available(), 1);
assert!(inner.try_acquire());
assert_eq!(inner.available(), 0);
assert!(!inner.try_acquire());
inner.release();
assert_eq!(inner.available(), 1);
}
#[test]
fn semaphore_words_share_one_backend_by_identity() {
let first = Value::semaphore(1);
let alias = first.clone();
assert_eq!(&alias, &first);
assert_eq!(format!("{first}"), "<semaphore>");
assert!(first.as_semaphore().expect("backend").try_acquire());
assert!(!alias.as_semaphore().expect("backend").try_acquire());
assert_ne!(Value::semaphore(1), first);
assert!(Value::int(1).as_semaphore().is_none());
}
#[test]
fn permit_last_drop_releases_exactly_once() {
let sem = Value::semaphore(1);
let inner = sem.as_semaphore().expect("backend");
assert!(inner.try_acquire());
let first = Value::permit(&inner);
assert_eq!(format!("{first}"), "<permit>");
let second = first.clone();
assert_eq!(&first, &second);
drop(first);
assert_eq!(inner.available(), 0);
drop(second);
assert_eq!(inner.available(), 1);
}
#[test]
fn permit_releases_when_holder_panics() {
let sem = Value::semaphore(1);
let inner = sem.as_semaphore().expect("backend");
let worker = {
let inner = std::sync::Arc::clone(&inner);
std::thread::spawn(move || {
assert!(inner.try_acquire());
let _permit = Value::permit(&inner);
panic!("worker fails holding the permit");
})
};
assert!(worker.join().is_err());
assert_eq!(inner.available(), 1);
}
#[test]
fn semaphore_holds_cap_under_contention() {
use std::sync::atomic::Ordering;
let sem = Value::semaphore(4);
let inner = sem.as_semaphore().expect("backend");
let holders = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let peak = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let mut threads = Vec::new();
for _ in 0..8 {
let inner = std::sync::Arc::clone(&inner);
let holders = std::sync::Arc::clone(&holders);
let peak = std::sync::Arc::clone(&peak);
threads.push(std::thread::spawn(move || {
let mut acquired = 0;
while acquired < 25 {
if inner.try_acquire() {
{
let _permit = Value::permit(&inner);
let current = holders.fetch_add(1, Ordering::SeqCst) + 1;
peak.fetch_max(current, Ordering::SeqCst);
std::thread::yield_now();
holders.fetch_sub(1, Ordering::SeqCst);
}
acquired += 1;
} else {
std::thread::yield_now();
}
}
}));
}
for thread in threads {
thread.join().expect("worker joins");
}
assert!(peak.load(Ordering::SeqCst) <= 4);
assert_eq!(holders.load(Ordering::SeqCst), 0);
assert_eq!(inner.available(), 4);
}
}