#![allow(unused)]
use crate::parking::Event;
use std::collections::HashMap;
use std::mem;
use std::rc::Rc;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use cljrs_gc::GcPtr;
use cljrs_gc::mutator::TrackedMutex;
use cljrs_reader::Form;
use crate::TypeHint;
use crate::Value;
#[cfg(all(feature = "no-gc", debug_assertions))]
pub(crate) fn value_gcptr_is_static(value: &Value) -> bool {
use crate::value::MapValue;
use crate::value::SetValue;
match value {
Value::Nil
| Value::Bool(_)
| Value::Long(_)
| Value::Double(_)
| Value::Char(_)
| Value::Uuid(_) => true,
Value::Resource(_) | Value::SharedAtom(_) | Value::ByteBlob(_) => true,
Value::BigInt(p) => p.is_static_alloc(),
Value::BigDecimal(p) => p.is_static_alloc(),
Value::Ratio(p) => p.is_static_alloc(),
Value::Str(p) => p.is_static_alloc(),
Value::Pattern(p) => p.is_static_alloc(),
Value::Matcher(p) => p.is_static_alloc(),
Value::Symbol(p) => p.is_static_alloc(),
Value::Keyword(p) => p.is_static_alloc(),
Value::List(p) => p.is_static_alloc(),
Value::Vector(p) => p.is_static_alloc(),
Value::Queue(p) => p.is_static_alloc(),
Value::Map(m) => match m {
MapValue::Array(p) => p.is_static_alloc(),
MapValue::Hash(p) => p.is_static_alloc(),
MapValue::Sorted(p) => p.is_static_alloc(),
},
Value::Set(s) => match s {
SetValue::Hash(p) => p.is_static_alloc(),
SetValue::Sorted(p) => p.is_static_alloc(),
},
Value::NativeFunction(p) => p.is_static_alloc(),
Value::Fn(p) | Value::Macro(p) => p.is_static_alloc(),
Value::BoundFn(p) => p.is_static_alloc(),
Value::Var(p) => p.is_static_alloc(),
Value::Atom(p) => p.is_static_alloc(),
Value::Namespace(p) => p.is_static_alloc(),
Value::LazySeq(p) => p.is_static_alloc(),
Value::Cons(p) => p.is_static_alloc(),
Value::Protocol(p) => p.is_static_alloc(),
Value::ProtocolFn(p) => p.is_static_alloc(),
Value::MultiFn(p) => p.is_static_alloc(),
Value::Volatile(p) => p.is_static_alloc(),
Value::Delay(p) => p.is_static_alloc(),
Value::Promise(p) => p.is_static_alloc(),
Value::Future(p) => p.is_static_alloc(),
Value::Agent(p) => p.is_static_alloc(),
Value::TypeInstance(p) => p.is_static_alloc(),
Value::ObjectArray(p) => p.is_static_alloc(),
Value::NativeObject(p) => p.is_static_alloc(),
Value::Error(p) => p.is_static_alloc(),
Value::TransientMap(p) => p.is_static_alloc(),
Value::TransientVector(p) => p.is_static_alloc(),
Value::TransientSet(p) => p.is_static_alloc(),
Value::BooleanArray(_)
| Value::ByteArray(_)
| Value::ShortArray(_)
| Value::IntArray(_)
| Value::LongArray(_)
| Value::FloatArray(_)
| Value::DoubleArray(_)
| Value::CharArray(_) => true,
Value::Reduced(inner) | Value::WithMeta(inner, _) => value_gcptr_is_static(inner),
}
}
pub type MethodMap = HashMap<Arc<str>, Value>;
#[derive(Debug)]
pub struct Protocol {
pub name: Arc<str>,
pub ns: Arc<str>,
pub methods: Vec<ProtocolMethod>,
pub impls: TrackedMutex<HashMap<Arc<str>, MethodMap>>,
pub extend_via_metadata: bool,
}
impl Protocol {
pub fn new(
name: Arc<str>,
ns: Arc<str>,
methods: Vec<ProtocolMethod>,
extend_via_metadata: bool,
) -> Self {
Self {
name,
ns,
methods,
impls: TrackedMutex::new(HashMap::new()),
extend_via_metadata,
}
}
}
static PROTOCOL_GENERATION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
pub fn protocol_generation() -> u64 {
PROTOCOL_GENERATION.load(std::sync::atomic::Ordering::Acquire)
}
pub fn bump_protocol_generation() {
PROTOCOL_GENERATION.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
}
impl cljrs_gc::Trace for Protocol {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let impls = self.impls.lock().unwrap();
for method_map in impls.values() {
for v in method_map.values() {
v.trace(visitor);
}
}
}
}
}
#[derive(Debug, Clone)]
pub struct ProtocolMethod {
pub name: Arc<str>,
pub min_arity: usize,
pub variadic: bool,
}
impl cljrs_gc::Trace for ProtocolMethod {
fn trace(&self, _: &mut cljrs_gc::MarkVisitor) {}
}
#[derive(Debug)]
pub struct ProtocolFn {
pub protocol: GcPtr<Protocol>,
pub method_name: Arc<str>,
pub min_arity: usize,
pub variadic: bool,
}
impl cljrs_gc::Trace for ProtocolFn {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
use cljrs_gc::GcVisitor as _;
visitor.visit(&self.protocol);
}
}
#[derive(Debug)]
pub struct MultiFn {
pub name: Arc<str>,
pub dispatch_fn: Value,
pub methods: TrackedMutex<HashMap<String, Value>>,
pub dispatch_vals: TrackedMutex<HashMap<String, Value>>,
pub prefers: TrackedMutex<HashMap<String, Vec<String>>>,
pub preference_vals: TrackedMutex<HashMap<String, Value>>,
pub default_dispatch: String,
method_generation: std::sync::atomic::AtomicU64,
method_cache: TrackedMutex<MultiFnMethodCache>,
}
#[derive(Debug, Default)]
struct MultiFnMethodCache {
hierarchy_generation: u64,
method_generation: u64,
methods: HashMap<String, String>,
}
impl MultiFn {
pub fn new(name: Arc<str>, dispatch_fn: Value, default_dispatch: String) -> Self {
Self {
name,
dispatch_fn,
methods: TrackedMutex::new(HashMap::new()),
dispatch_vals: TrackedMutex::new(HashMap::new()),
prefers: TrackedMutex::new(HashMap::new()),
preference_vals: TrackedMutex::new(HashMap::new()),
default_dispatch,
method_generation: std::sync::atomic::AtomicU64::new(0),
method_cache: TrackedMutex::new(MultiFnMethodCache::default()),
}
}
pub fn method_generation(&self) -> u64 {
self.method_generation
.load(std::sync::atomic::Ordering::Acquire)
}
pub fn bump_method_generation(&self) {
self.method_generation
.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
}
pub fn cached_method(
&self,
dispatch_key: &str,
hierarchy_generation: u64,
method_generation: u64,
) -> Option<String> {
let mut cache = self.method_cache.lock().unwrap();
if cache.hierarchy_generation != hierarchy_generation
|| cache.method_generation != method_generation
{
cache.hierarchy_generation = hierarchy_generation;
cache.method_generation = method_generation;
cache.methods.clear();
}
cache.methods.get(dispatch_key).cloned()
}
pub fn cache_method(
&self,
dispatch_key: String,
method_key: String,
hierarchy_generation: u64,
method_generation: u64,
) {
let mut cache = self.method_cache.lock().unwrap();
if cache.hierarchy_generation != hierarchy_generation
|| cache.method_generation != method_generation
{
cache.hierarchy_generation = hierarchy_generation;
cache.method_generation = method_generation;
cache.methods.clear();
}
cache.methods.insert(dispatch_key, method_key);
}
pub fn cached_method_count(&self) -> usize {
self.method_cache.lock().unwrap().methods.len()
}
}
impl cljrs_gc::Trace for MultiFn {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
self.dispatch_fn.trace(visitor);
{
let methods = self.methods.lock().unwrap();
for v in methods.values() {
v.trace(visitor);
}
}
{
let dispatch_vals = self.dispatch_vals.lock().unwrap();
for v in dispatch_vals.values() {
v.trace(visitor);
}
}
{
let preference_vals = self.preference_vals.lock().unwrap();
for v in preference_vals.values() {
v.trace(visitor);
}
}
}
}
#[derive(Debug)]
pub struct Var {
pub namespace: Arc<str>,
pub name: Arc<str>,
pub value: TrackedMutex<Option<Value>>,
pub shared_root: Arc<arc_swap::ArcSwap<Option<crate::shared::SharedValue>>>,
pub is_macro: bool,
pub meta: TrackedMutex<Option<Value>>,
pub watches: TrackedMutex<Vec<(Value, Value)>>,
binding_generation: std::sync::atomic::AtomicU64,
}
impl Var {
pub fn new(namespace: impl Into<Arc<str>>, name: impl Into<Arc<str>>) -> Self {
Self {
namespace: namespace.into(),
name: name.into(),
value: TrackedMutex::new(None),
shared_root: Arc::new(arc_swap::ArcSwap::new(Arc::new(None))),
is_macro: false,
meta: TrackedMutex::new(None),
watches: TrackedMutex::new(Vec::new()),
binding_generation: std::sync::atomic::AtomicU64::new(0),
}
}
pub fn from_shared_root(
namespace: impl Into<Arc<str>>,
name: impl Into<Arc<str>>,
is_macro: bool,
shared_root: Arc<arc_swap::ArcSwap<Option<crate::shared::SharedValue>>>,
) -> Self {
let local = shared_root
.load()
.as_ref()
.as_ref()
.map(crate::shared::demote);
Self {
namespace: namespace.into(),
name: name.into(),
value: TrackedMutex::new(local),
shared_root,
is_macro,
meta: TrackedMutex::new(None),
watches: TrackedMutex::new(Vec::new()),
binding_generation: std::sync::atomic::AtomicU64::new(0),
}
}
pub fn is_bound(&self) -> bool {
self.value.lock().unwrap().is_some()
}
pub fn deref(&self) -> Option<Value> {
self.value.lock().unwrap().clone()
}
pub fn binding_generation(&self) -> u64 {
self.binding_generation
.load(std::sync::atomic::Ordering::Acquire)
}
pub fn deref_shared(&self) -> Option<Value> {
self.shared_root
.load()
.as_ref()
.as_ref()
.map(crate::shared::demote)
}
pub fn bind(&self, v: Value) {
#[cfg(all(feature = "no-gc", debug_assertions))]
debug_assert!(
value_gcptr_is_static(&v),
"no-gc: Var::bind({}/{}) received a region-local value — store violations \
indicate a missing StaticCtxGuard around the value expression",
self.namespace,
self.name
);
let v = crate::publish::publish_value(v);
let prev = {
let mut slot = self.value.lock().unwrap();
slot.replace(v.clone())
};
let shared = crate::shared::promote(&v).ok();
self.shared_root.store(Arc::new(shared));
self.binding_generation
.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
if let Some(prev) = prev {
crate::jit_hooks::notify_var_rebind(&prev, &v);
}
}
pub fn get_meta(&self) -> Option<Value> {
self.meta.lock().unwrap().clone()
}
pub fn set_meta(&self, m: Value) {
*self.meta.lock().unwrap() = Some(m);
}
pub fn full_name(&self) -> String {
format!("{}/{}", self.namespace, self.name)
}
}
impl cljrs_gc::Trace for Var {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let value = self.value.lock().unwrap();
if let Some(v) = value.as_ref() {
v.trace(visitor);
}
}
{
let meta = self.meta.lock().unwrap();
if let Some(m) = meta.as_ref() {
m.trace(visitor);
}
}
{
let watches = self.watches.lock().unwrap();
for (key, f) in watches.iter() {
key.trace(visitor);
f.trace(visitor);
}
}
}
}
#[derive(Debug)]
pub struct Atom {
pub value: TrackedMutex<Value>,
pub meta: TrackedMutex<Option<Value>>,
pub validator: TrackedMutex<Option<Value>>,
pub watches: TrackedMutex<Vec<(Value, Value)>>,
}
impl Atom {
pub fn new(v: Value) -> Self {
let v = crate::publish::publish_value(v);
Self {
value: TrackedMutex::new(v),
meta: TrackedMutex::new(None),
validator: TrackedMutex::new(None),
watches: TrackedMutex::new(Vec::new()),
}
}
pub fn deref(&self) -> Value {
self.value.lock().unwrap().clone()
}
pub fn reset(&self, v: Value) -> Value {
#[cfg(all(feature = "no-gc", debug_assertions))]
debug_assert!(
value_gcptr_is_static(&v),
"no-gc: Atom::reset() received a region-local value — the new-value \
expression must be computed inside a StaticCtxGuard (i.e. inside \
the swap! / reset! call) so it is allocated in the static arena"
);
let v = crate::publish::publish_value(v);
let mut guard = self.value.lock().unwrap();
*guard = v.clone();
v
}
pub fn get_meta(&self) -> Option<Value> {
self.meta.lock().unwrap().clone()
}
pub fn set_meta(&self, m: Option<Value>) {
*self.meta.lock().unwrap() = m;
}
pub fn get_validator(&self) -> Option<Value> {
self.validator.lock().unwrap().clone()
}
pub fn set_validator(&self, vf: Option<Value>) {
*self.validator.lock().unwrap() = vf;
}
}
impl cljrs_gc::Trace for Atom {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let value = self.value.lock().unwrap();
value.trace(visitor);
}
{
let meta = self.meta.lock().unwrap();
if let Some(m) = meta.as_ref() {
m.trace(visitor);
}
}
{
let validator = self.validator.lock().unwrap();
if let Some(vf) = validator.as_ref() {
vf.trace(visitor);
}
}
{
let watches = self.watches.lock().unwrap();
for (key, f) in watches.iter() {
key.trace(visitor);
f.trace(visitor);
}
}
}
}
#[derive(Debug, Clone, Default)]
pub struct ReferClojureFilter {
pub only: Option<std::collections::HashSet<Arc<str>>>,
pub exclude: std::collections::HashSet<Arc<str>>,
pub rename: HashMap<Arc<str>, Arc<str>>,
}
impl ReferClojureFilter {
pub fn local_name(&self, name: &Arc<str>) -> Option<Arc<str>> {
if self.exclude.contains(name) {
return None;
}
if let Some(only) = &self.only
&& !only.contains(name)
{
return None;
}
Some(
self.rename
.get(name)
.cloned()
.unwrap_or_else(|| name.clone()),
)
}
}
#[derive(Debug)]
pub struct Namespace {
pub name: Arc<str>,
pub interns: TrackedMutex<HashMap<Arc<str>, GcPtr<Var>>>,
pub refers: TrackedMutex<HashMap<Arc<str>, GcPtr<Var>>>,
pub aliases: TrackedMutex<HashMap<Arc<str>, Arc<str>>>,
pub source_file: TrackedMutex<Option<Arc<str>>>,
pub git_repo_root: TrackedMutex<Option<Arc<str>>>,
pub is_versioned: bool,
pub meta: TrackedMutex<Option<Value>>,
pub refer_clojure_filter: TrackedMutex<Option<ReferClojureFilter>>,
}
impl Namespace {
pub fn new(name: impl Into<Arc<str>>) -> Self {
Self {
name: name.into(),
interns: TrackedMutex::new(HashMap::new()),
refers: TrackedMutex::new(HashMap::new()),
aliases: TrackedMutex::new(HashMap::new()),
source_file: TrackedMutex::new(None),
git_repo_root: TrackedMutex::new(None),
is_versioned: false,
meta: TrackedMutex::new(None),
refer_clojure_filter: TrackedMutex::new(None),
}
}
pub fn new_versioned(name: impl Into<Arc<str>>) -> Self {
Self {
is_versioned: true,
..Self::new(name)
}
}
pub fn set_source_location(&self, file: &str, repo_root: Option<&str>) {
*self.source_file.lock().unwrap() = Some(Arc::from(file));
*self.git_repo_root.lock().unwrap() = repo_root.map(Arc::from);
}
pub fn get_meta(&self) -> Option<Value> {
self.meta.lock().unwrap().clone()
}
pub fn set_meta(&self, m: Value) {
*self.meta.lock().unwrap() = Some(m);
}
}
impl cljrs_gc::Trace for Namespace {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
use cljrs_gc::GcVisitor as _;
{
let interns = self.interns.lock().unwrap();
for var in interns.values() {
visitor.visit(var);
}
}
{
let refers = self.refers.lock().unwrap();
for var in refers.values() {
visitor.visit(var);
}
}
{
let meta = self.meta.lock().unwrap();
if let Some(m) = meta.as_ref() {
m.trace(visitor);
}
}
}
}
pub type NativeFnPtr = fn(&[Value]) -> crate::error::ValueResult<Value>;
pub type NativeFnFunc = Arc<dyn Fn(&[Value]) -> crate::error::ValueResult<Value>>;
#[derive(Clone, Debug)]
pub enum Arity {
Fixed(usize),
Variadic { min: usize },
}
pub struct NativeFn {
pub name: Arc<str>,
pub arity: Arity,
pub func: NativeFnFunc,
pub blocking_func: Option<NativeFnFunc>,
}
impl NativeFn {
pub fn new(name: impl Into<Arc<str>>, arity: Arity, func: NativeFnPtr) -> Self {
Self {
name: name.into(),
arity,
blocking_func: None,
func: Arc::new(move |args| {
crate::realization::catch(|| func(args))
.unwrap_or_else(|message| Err(crate::ValueError::Other(message)))
}),
}
}
pub fn with_closure(
name: impl Into<Arc<str>>,
arity: Arity,
func: impl Fn(&[Value]) -> crate::error::ValueResult<Value> + 'static,
) -> Self {
Self {
name: name.into(),
arity,
blocking_func: None,
func: Arc::new(move |args| {
crate::realization::catch(|| func(args))
.unwrap_or_else(|message| Err(crate::ValueError::Other(message)))
}),
}
}
pub fn with_blocking(
name: impl Into<Arc<str>>,
arity: Arity,
func: impl Fn(&[Value]) -> crate::ValueResult<Value> + 'static,
blocking_func: impl Fn(&[Value]) -> crate::ValueResult<Value> + 'static,
) -> Self {
let mut native = Self::with_closure(name, arity, func);
native.blocking_func = Some(Arc::new(move |args| {
crate::realization::catch(|| blocking_func(args))
.unwrap_or_else(|message| Err(crate::ValueError::Other(message)))
}));
native
}
pub fn call_blocking(&self, args: &[Value]) -> crate::ValueResult<Value> {
let expected = match self.arity {
Arity::Fixed(n) if args.len() != n => Some(n.to_string()),
Arity::Variadic { min } if args.len() < min => Some(format!("{min}+")),
_ => None,
};
if let Some(expected) = expected {
return Err(crate::ValueError::ArityError {
name: self.name.to_string(),
expected,
got: args.len(),
});
}
let adapter = self.blocking_func.as_ref().ok_or_else(|| {
crate::ValueError::Other(format!(
"blocking cannot offload {}: this function has no owned-work adapter",
self.name
))
})?;
adapter(args)
}
}
impl std::fmt::Debug for NativeFn {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NativeFn")
.field("name", &self.name)
.field("arity", &self.arity)
.field("func", &"<fn>")
.finish()
}
}
impl cljrs_gc::Trace for NativeFn {
fn trace(&self, _: &mut cljrs_gc::MarkVisitor) {}
}
#[derive(Debug, Clone)]
pub struct CljxFnArity {
pub params: Vec<Arc<str>>,
pub rest_param: Option<Arc<str>>,
pub body: Vec<Form>,
pub destructure_params: Vec<(usize, Form)>,
pub destructure_rest: Option<Form>,
pub ir_arity_id: u64,
pub param_hints: Vec<Option<TypeHint>>,
pub rest_hint: Option<TypeHint>,
}
impl CljxFnArity {
pub fn heap_size(&self) -> usize {
self.params.capacity() * mem::size_of::<Arc<str>>()
+ self.body.capacity() * mem::size_of::<Form>()
+ self.body.iter().map(|f| f.heap_size()).sum::<usize>()
+ self.destructure_params.capacity() * mem::size_of::<(usize, Form)>()
+ self.destructure_params.iter().map(|(_, f)| f.heap_size()).sum::<usize>()
+ self.destructure_rest.as_ref()
.map_or(0, |f| mem::size_of::<Form>() + f.heap_size())
+ self.param_hints.capacity() * mem::size_of::<Option<TypeHint>>()
}
}
#[derive(Debug, Clone)]
pub struct CljxFn {
pub name: Option<Arc<str>>,
pub arities: Vec<CljxFnArity>,
pub closed_over_names: Vec<Arc<str>>,
pub closed_over_vals: Vec<Value>,
pub is_macro: bool,
pub is_async: bool,
pub defining_ns: Arc<str>,
pub self_ptr: Option<GcPtr<CljxFn>>,
}
impl CljxFn {
pub fn new(
name: Option<Arc<str>>,
arities: Vec<CljxFnArity>,
closed_over_names: Vec<Arc<str>>,
closed_over_vals: Vec<Value>,
is_macro: bool,
defining_ns: Arc<str>,
) -> Self {
Self {
name,
arities,
closed_over_names,
closed_over_vals,
is_macro,
is_async: false,
defining_ns,
self_ptr: None,
}
}
}
impl cljrs_gc::Trace for CljxFn {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
use cljrs_gc::GcVisitor as _;
for v in &self.closed_over_vals {
v.trace(visitor);
}
if let Some(ref p) = self.self_ptr {
visitor.visit(p);
}
}
fn gc_size_extra(&self) -> usize {
self.arities.capacity() * mem::size_of::<CljxFnArity>()
+ self
.arities
.iter()
.map(CljxFnArity::heap_size)
.sum::<usize>()
}
}
#[derive(Debug)]
pub struct BoundFn {
pub wrapped: Value,
pub captured_bindings: HashMap<usize, Value>,
}
impl cljrs_gc::Trace for BoundFn {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
self.wrapped.trace(visitor);
for val in self.captured_bindings.values() {
val.trace(visitor);
}
}
fn gc_size_extra(&self) -> usize {
self.captured_bindings.capacity() * (1 + mem::size_of::<usize>() + mem::size_of::<Value>())
}
}
pub trait Thunk: std::fmt::Debug + cljrs_gc::Trace {
fn force(&self) -> Result<Value, String>;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Realizer {
thread: std::thread::ThreadId,
fiber: Option<u64>,
}
impl Realizer {
pub fn current() -> Self {
Self {
thread: std::thread::current().id(),
fiber: crate::parking::current(),
}
}
pub fn ensure_can_wait(self) -> Result<(), String> {
if self.thread == std::thread::current().id()
&& self.fiber.is_some()
&& crate::parking::current().is_none()
{
Err("cannot block the isolate thread waiting for a parked fiber. Run this code inside a fiber.".into())
} else {
Ok(())
}
}
}
struct RealizationGuard<'a> {
lazy: Option<&'a LazySeq>,
delay: Option<&'a Delay>,
}
impl Drop for RealizationGuard<'_> {
fn drop(&mut self) {
if let Some(cell) = self.lazy {
let mut state = cell.state.lock().unwrap();
if let LazySeqState::Realizing { thunk, .. } = &*state {
*state = LazySeqState::Pending(thunk.clone());
}
drop(state);
cell.ready.notify_all();
}
if let Some(cell) = self.delay {
let mut state = cell.state.lock().unwrap();
if let DelayState::Realizing { thunk, .. } = &*state {
*state = DelayState::Pending(thunk.clone());
}
drop(state);
cell.ready.notify_all();
}
}
}
pub enum LazySeqState {
Pending(Rc<dyn Thunk>),
Realizing {
owner: Realizer,
thunk: Rc<dyn Thunk>,
},
Forced(Value),
Error(String),
}
pub struct LazySeq {
ready: Event,
pub state: TrackedMutex<LazySeqState>,
}
impl LazySeq {
pub fn new(thunk: Box<dyn Thunk>) -> Self {
Self {
state: TrackedMutex::new(LazySeqState::Pending(Rc::from(thunk))),
ready: Event::default(),
}
}
pub fn realize(&self) -> Value {
match self.try_realize() {
Ok(value) => value,
Err(error) => {
assert!(self.error().is_some(), "{error}");
Value::Nil
}
}
}
pub fn try_realize(&self) -> Result<Value, String> {
let owner = Realizer::current();
let thunk = loop {
let listener = self.ready.listen();
let mut state = self.state.lock().unwrap();
match &*state {
LazySeqState::Forced(v) => return Ok(v.clone()),
LazySeqState::Error(error) => return Err(error.clone()),
LazySeqState::Realizing { owner: active, .. } => {
if *active == owner {
return Ok(Value::Nil);
}
active.ensure_can_wait()?;
drop(state);
listener.wait(None);
continue;
}
LazySeqState::Pending(_) => {}
}
let LazySeqState::Pending(thunk) =
mem::replace(&mut *state, LazySeqState::Forced(Value::Nil))
else {
unreachable!()
};
*state = LazySeqState::Realizing {
owner,
thunk: thunk.clone(),
};
break thunk;
};
let mut guard = RealizationGuard {
lazy: Some(self),
delay: None,
};
let result = match thunk.force() {
Ok(result) => {
let result = crate::publish::publish_value(result);
*self.state.lock().unwrap() = LazySeqState::Forced(result.clone());
Ok(result)
}
Err(msg) => {
*self.state.lock().unwrap() = LazySeqState::Error(msg.clone());
Err(msg)
}
};
guard.lazy = None;
self.ready.notify_all();
result
}
pub fn error(&self) -> Option<String> {
let guard = self.state.lock().unwrap();
if let LazySeqState::Error(e) = &*guard {
Some(e.clone())
} else {
None
}
}
}
impl std::fmt::Debug for LazySeq {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "LazySeq(...)")
}
}
impl cljrs_gc::Trace for LazySeq {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let state = self.state.lock().unwrap();
match &*state {
LazySeqState::Pending(thunk) => thunk.trace(visitor),
LazySeqState::Realizing { thunk, .. } => thunk.trace(visitor),
LazySeqState::Forced(v) => v.trace(visitor),
LazySeqState::Error(_) => {}
}
}
}
}
#[derive(Debug, Clone)]
pub struct CljxCons {
pub head: Value,
pub tail: Value,
}
impl cljrs_gc::Trace for CljxCons {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
self.head.trace(visitor);
self.tail.trace(visitor);
}
}
pub struct Volatile {
pub value: TrackedMutex<Value>,
}
impl Volatile {
pub fn new(v: Value) -> Self {
let v = crate::publish::publish_value(v);
Self {
value: TrackedMutex::new(v),
}
}
pub fn deref(&self) -> Value {
self.value.lock().unwrap().clone()
}
pub fn reset(&self, v: Value) -> Value {
#[cfg(all(feature = "no-gc", debug_assertions))]
debug_assert!(
value_gcptr_is_static(&v),
"no-gc: Volatile::reset() received a region-local value — ensure the \
new-value expression is inside a StaticCtxGuard (vreset! handles this)"
);
let v = crate::publish::publish_value(v);
*self.value.lock().unwrap() = v.clone();
v
}
}
impl std::fmt::Debug for Volatile {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Volatile")
}
}
impl cljrs_gc::Trace for Volatile {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let value = self.value.lock().unwrap();
value.trace(visitor);
}
}
}
pub enum DelayState {
Error(String),
Pending(Rc<dyn Thunk>),
Realizing {
owner: Realizer,
thunk: Rc<dyn Thunk>,
},
Forced(Value),
}
pub struct Delay {
ready: Event,
pub state: TrackedMutex<DelayState>,
}
impl Delay {
pub fn new(thunk: Box<dyn Thunk>) -> Self {
Self {
state: TrackedMutex::new(DelayState::Pending(Rc::from(thunk))),
ready: Event::default(),
}
}
pub fn force(&self) -> Result<Value, String> {
let owner = Realizer::current();
let thunk = loop {
let listener = self.ready.listen();
let mut state = self.state.lock().unwrap();
match &*state {
DelayState::Forced(v) => return Ok(v.clone()),
DelayState::Error(e) => return Err(e.clone()),
DelayState::Realizing { owner: active, .. } => {
if *active == owner {
return Err("recursive delay realization".into());
}
active.ensure_can_wait()?;
drop(state);
listener.wait(None);
continue;
}
DelayState::Pending(_) => {}
}
let DelayState::Pending(thunk) =
mem::replace(&mut *state, DelayState::Forced(Value::Nil))
else {
unreachable!()
};
*state = DelayState::Realizing {
owner,
thunk: thunk.clone(),
};
break thunk;
};
let mut guard = RealizationGuard {
lazy: None,
delay: Some(self),
};
let result = thunk.force().map(crate::publish::publish_value);
*self.state.lock().unwrap() = match &result {
Ok(value) => DelayState::Forced(value.clone()),
Err(error) => DelayState::Error(error.clone()),
};
guard.delay = None;
self.ready.notify_all();
result
}
pub fn is_realized(&self) -> bool {
matches!(
&*self.state.lock().unwrap(),
DelayState::Forced(_) | DelayState::Error(_)
)
}
}
impl std::fmt::Debug for Delay {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Delay")
}
}
impl cljrs_gc::Trace for Delay {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let state = self.state.lock().unwrap();
match &*state {
DelayState::Pending(thunk) => thunk.trace(visitor),
DelayState::Realizing { thunk, .. } => thunk.trace(visitor),
DelayState::Error(_) => {}
DelayState::Forced(v) => v.trace(visitor),
}
}
}
}
pub struct CljxPromise {
pub ready: Event,
pub value: Mutex<Option<Value>>,
pub cond: Condvar,
}
impl CljxPromise {
pub fn new() -> Self {
Self {
value: Mutex::new(None),
cond: Condvar::new(),
ready: Event::default(),
}
}
pub fn deliver(&self, v: Value) {
let v = crate::publish::publish_value(v);
let mut guard = self.value.lock().unwrap();
if guard.is_none() {
*guard = Some(v);
self.cond.notify_all();
drop(guard);
self.ready.notify_all();
}
}
pub fn deref_blocking(&self) -> Value {
self.wait_value(None).unwrap()
}
pub fn wait_value(&self, timeout: Option<Duration>) -> Option<Value> {
let deadline = timeout.map(|d| Instant::now() + d);
loop {
let listener = self.ready.listen();
if let Some(value) = self.value.lock().unwrap().clone() {
return Some(value);
}
if !listener.wait(deadline.map(|d| d.saturating_duration_since(Instant::now()))) {
return self.value.lock().unwrap().clone();
}
}
}
pub fn is_realized(&self) -> bool {
self.value.lock().unwrap().is_some()
}
}
impl Default for CljxPromise {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for CljxPromise {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Promise")
}
}
impl cljrs_gc::Trace for CljxPromise {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let value = self.value.lock().unwrap();
if let Some(v) = value.as_ref() {
v.trace(visitor);
}
}
}
}
pub enum FutureState {
Running,
Done(Value),
Failed(Value),
GasExhausted,
MemoryLimitExceeded(cljrs_gc::MemoryLimitExceeded),
Cancelled,
}
pub const FUTURE_CANCELLED_MSG: &str = "future was cancelled";
pub struct CljxFuture {
pub ready: Event,
pub state: Mutex<FutureState>,
pub cond: Condvar,
observed: std::sync::atomic::AtomicBool,
}
impl CljxFuture {
pub fn new() -> Self {
Self {
state: Mutex::new(FutureState::Running),
cond: Condvar::new(),
ready: Event::default(),
observed: std::sync::atomic::AtomicBool::new(false),
}
}
pub fn wait_value(&self, timeout: Option<Duration>) -> crate::ValueResult<Option<Value>> {
let deadline = timeout.map(|d| Instant::now() + d);
loop {
let listener = self.ready.listen();
if let Some(result) = self.settled_value() {
return result.map(Some);
}
if !listener.wait(deadline.map(|d| d.saturating_duration_since(Instant::now()))) {
return self.settled_value().transpose();
}
}
}
pub fn settled_value(&self) -> Option<crate::ValueResult<Value>> {
use crate::ValueError;
let state = self.state.lock().unwrap();
let value = match &*state {
FutureState::Running => return None,
FutureState::Done(v) => Ok(v.clone()),
FutureState::Failed(v) => Err(ValueError::Thrown(v.clone())),
FutureState::GasExhausted => Err(ValueError::GasExhausted),
FutureState::MemoryLimitExceeded(f) => Err(ValueError::MemoryLimitExceeded(*f)),
FutureState::Cancelled => {
drop(state);
return Some(Err(ValueError::Thrown(Self::cancelled_error())));
}
};
self.mark_observed();
Some(value)
}
pub fn is_done(&self) -> bool {
!matches!(&*self.state.lock().unwrap(), FutureState::Running)
}
pub fn is_cancelled(&self) -> bool {
matches!(&*self.state.lock().unwrap(), FutureState::Cancelled)
}
pub fn cancel(&self) -> bool {
let mut state = self.state.lock().unwrap();
if matches!(&*state, FutureState::Running) {
*state = FutureState::Cancelled;
self.cond.notify_all();
drop(state);
self.ready.notify_all();
true
} else {
false
}
}
pub fn cancelled_error() -> Value {
Value::Error(GcPtr::new(crate::ExceptionInfo::new(
crate::ValueError::Other(FUTURE_CANCELLED_MSG.to_string()),
FUTURE_CANCELLED_MSG.to_string(),
None,
None,
)))
}
pub fn mark_observed(&self) {
self.observed
.store(true, std::sync::atomic::Ordering::Relaxed);
}
}
impl Drop for CljxFuture {
fn drop(&mut self) {
if !self.observed.load(std::sync::atomic::Ordering::Relaxed)
&& let Ok(state) = self.state.lock()
&& matches!(&*state, FutureState::Failed(_))
{
eprintln!(
"[clojurust warning] a failed future was discarded without its error \
being observed (no await/deref); the thrown exception was lost"
);
}
}
}
impl Default for CljxFuture {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for CljxFuture {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Future")
}
}
impl cljrs_gc::Trace for CljxFuture {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let state = self.state.lock().unwrap();
if let FutureState::Done(v) | FutureState::Failed(v) = &*state {
v.trace(visitor);
}
}
}
}
pub struct Agent {
pub state: Arc<Mutex<Value>>,
pub error: Arc<Mutex<Option<Value>>>,
pub watches: TrackedMutex<Vec<(Value, Value)>>,
}
impl Agent {
pub fn get_state(&self) -> Value {
self.state.lock().unwrap().clone()
}
pub fn get_error(&self) -> Option<Value> {
self.error.lock().unwrap().clone()
}
pub fn clear_error(&self) {
*self.error.lock().unwrap() = None;
}
}
impl std::fmt::Debug for Agent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Agent")
}
}
impl cljrs_gc::Trace for Agent {
fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
{
let state = self.state.lock().unwrap();
state.trace(visitor);
}
{
let error = self.error.lock().unwrap();
if let Some(e) = error.as_ref() {
e.trace(visitor);
}
}
{
let watches = self.watches.lock().unwrap();
for (key, f) in watches.iter() {
key.trace(visitor);
f.trace(visitor);
}
}
}
}
#[cfg(test)]
mod var_tests {
use super::*;
use crate::shared::SharedValue;
#[test]
fn bind_promotable_mirrors_shared_root() {
let var = Var::new("user", "x");
assert!(var.shared_root.load().is_none());
var.bind(Value::Long(7));
assert!(matches!(
var.shared_root.load().as_ref().as_ref(),
Some(SharedValue::Long(7))
));
assert_eq!(var.deref(), Some(Value::Long(7)));
assert_eq!(var.deref_shared(), Some(Value::Long(7)));
}
#[test]
fn bind_nonpromotable_clears_shared_root() {
let var = Var::new("user", "f");
var.bind(Value::Long(1));
assert!(var.shared_root.load().is_some());
let f = Value::NativeFunction(GcPtr::new(NativeFn::new("f", Arity::Fixed(0), |_| {
Ok(Value::Nil)
})));
var.bind(f);
assert!(var.shared_root.load().is_none());
assert!(var.is_bound());
assert_eq!(var.deref_shared(), None);
}
#[test]
fn from_shared_root_seeds_local_slot() {
let src = Var::new("user", "y");
src.bind(Value::Long(99));
let recv = Var::from_shared_root("user", "y", false, src.shared_root.clone());
assert_eq!(recv.deref(), Some(Value::Long(99)));
src.bind(Value::Long(100));
assert_eq!(recv.deref_shared(), Some(Value::Long(100)));
}
}