use crate::context::{Context, RealmKey, Root};
use crate::fiber::Fiber;
use parking_lot::Mutex;
use std::any::Any;
use std::collections::HashMap;
use std::hash::{Hash, Hasher};
use std::sync::{Arc, Weak};
#[cfg(test)]
struct VisiblePublishProbe {
service: &'static str,
reached: Arc<std::sync::Barrier>,
resume: Arc<std::sync::Barrier>,
}
#[cfg(test)]
static VISIBLE_PUBLISH_PROBE: parking_lot::Mutex<Option<Arc<VisiblePublishProbe>>> =
parking_lot::Mutex::new(None);
#[cfg(test)]
fn probe_visible_publish(service: &'static str) {
let probe = VISIBLE_PUBLISH_PROBE.lock().clone();
if let Some(probe) = probe.filter(|probe| probe.service == service) {
probe.reached.wait();
probe.resume.wait();
}
}
pub(crate) struct RealmMembership {
_identity: u8,
}
impl RealmMembership {
pub(crate) fn new() -> Self {
Self { _identity: 0 }
}
}
#[derive(Clone)]
pub struct ServiceRealm {
membership: Arc<RealmMembership>,
key: RealmKey,
}
impl ServiceRealm {
pub(crate) fn new(membership: Arc<RealmMembership>, key: RealmKey) -> Self {
Self { membership, key }
}
pub(crate) fn belongs_to(&self, membership: &Arc<RealmMembership>) -> bool {
Arc::ptr_eq(&self.membership, membership)
}
pub(crate) fn key(&self) -> RealmKey {
self.key
}
}
impl std::fmt::Debug for ServiceRealm {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("ServiceRealm(..)")
}
}
impl PartialEq for ServiceRealm {
fn eq(&self, other: &Self) -> bool {
self.key == other.key && Arc::ptr_eq(&self.membership, &other.membership)
}
}
impl Eq for ServiceRealm {}
impl Hash for ServiceRealm {
fn hash<H: Hasher>(&self, state: &mut H) {
std::ptr::hash(Arc::as_ptr(&self.membership), state);
self.key.hash(state);
}
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum RealmMappingError {
#[error("service `{service}` is mapped more than once")]
DuplicateService {
service: String,
},
#[error("service `{service}` uses a realm from another Runtime")]
ForeignRealm {
service: String,
},
}
pub trait Service: Send + Sync + 'static {
const NAME: &'static str;
}
pub trait ConfigurableService: Service {
type Config;
type Layer: Send + Sync + 'static;
type Resolved: Send + 'static;
type PrepareError: std::error::Error;
type ComposeError: std::error::Error;
fn prepare_config(config: Self::Config)
-> std::result::Result<Self::Layer, Self::PrepareError>;
fn compose_config<'a>(
base: Option<&'a Self::Layer>,
layers: impl IntoIterator<Item = &'a Self::Layer>,
head: Option<&'a Self::Layer>,
) -> std::result::Result<Self::Resolved, Self::ComposeError>;
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ConfigResolutionError<E: std::error::Error> {
#[error("configuration for service `{service}` belongs to a different contract")]
ContractMismatch {
service: &'static str,
},
#[error("service configuration composition failed: {0}")]
Compose(E),
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum ServiceLookupError {
#[error("service `{service}` is unavailable")]
Unavailable {
service: &'static str,
},
#[error("service `{service}` belongs to a different contract")]
ContractMismatch {
service: &'static str,
},
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum ServicePublishError {
#[error("the current Context generation is closed")]
InactiveContext,
#[error("service `{service}` already has a publication in this realm")]
DuplicatePublication {
service: &'static str,
},
#[error("service `{service}` belongs to a different contract")]
ContractMismatch {
service: &'static str,
},
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum ServiceControlError {
#[error("service `{service}` publication is stale")]
StalePublication {
service: &'static str,
},
#[error("service `{service}` publication mutation is closed")]
MutationClosed {
service: &'static str,
},
}
#[derive(Clone)]
pub(crate) struct ServiceOccurrenceId(Arc<u8>);
impl ServiceOccurrenceId {
pub(crate) fn fresh() -> Self {
Self(Arc::new(0))
}
}
impl PartialEq for ServiceOccurrenceId {
fn eq(&self, other: &Self) -> bool {
Arc::ptr_eq(&self.0, &other.0)
}
}
impl Eq for ServiceOccurrenceId {}
impl std::hash::Hash for ServiceOccurrenceId {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
std::ptr::hash(Arc::as_ptr(&self.0), state);
}
}
impl std::fmt::Debug for ServiceOccurrenceId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("ServiceOccurrenceId(..)")
}
}
struct ServiceSlot {
occurrence: ServiceOccurrenceId,
value: Arc<dyn Any + Send + Sync>,
contract: std::any::TypeId,
fiber: Weak<Fiber>,
}
#[derive(Default)]
struct ServiceState {
contracts: HashMap<&'static str, std::any::TypeId>,
slots: HashMap<(RealmKey, &'static str), ServiceSlot>,
}
enum ServiceInstallRefusal {
Duplicate,
ContractMismatch,
}
pub(crate) struct CommittedServiceDrift {
affected: Vec<Arc<Fiber>>,
retained_snapshot: Vec<Arc<Fiber>>,
}
impl CommittedServiceDrift {
fn commit(root: &Root, edges: &[(String, RealmKey)]) -> Self {
let (affected, retained_snapshot) = root.dependents_for_edges_with_retention(edges);
for fiber in &affected {
fiber.slot.commit_recheck();
}
Self {
affected,
retained_snapshot,
}
}
pub(crate) fn kick(self, root: &Arc<Root>) {
let Self {
affected,
retained_snapshot,
} = self;
for fiber in affected {
fiber.kick_committed_recheck(root);
}
drop(retained_snapshot);
}
}
struct ServiceSlotPublish<'a> {
root: &'a Arc<Root>,
owner: &'a Arc<Fiber>,
store: &'a ServiceStore,
key: RealmKey,
name: &'static str,
slot: Option<ServiceSlot>,
evicted: Option<ServiceSlot>,
drift: Option<CommittedServiceDrift>,
refusal: Option<ServicePublishError>,
}
impl crate::gated::PublishStep for ServiceSlotPublish<'_> {
fn publish(&mut self) -> std::result::Result<(), crate::gated::PublishRefused> {
let slot = self.slot.take().expect("publish runs once");
match self
.store
.install(self.root, self.owner, slot, self.key, self.name)
{
Ok((evicted, drift)) => {
self.evicted = evicted;
self.drift = drift;
Ok(())
}
Err((refusal, slot)) => {
self.slot = Some(slot);
self.refusal = Some(match refusal {
ServiceInstallRefusal::Duplicate => {
ServicePublishError::DuplicatePublication { service: self.name }
}
ServiceInstallRefusal::ContractMismatch => {
ServicePublishError::ContractMismatch { service: self.name }
}
});
Err(crate::gated::PublishRefused)
}
}
}
}
#[derive(Default)]
pub(crate) struct ServiceStore {
state: Mutex<ServiceState>,
}
impl ServiceStore {
pub(crate) fn new() -> Self {
Self::default()
}
fn install(
&self,
root: &Root,
owner: &Arc<Fiber>,
slot: ServiceSlot,
key: RealmKey,
name: &'static str,
) -> std::result::Result<
(Option<ServiceSlot>, Option<CommittedServiceDrift>),
(ServiceInstallRefusal, ServiceSlot),
> {
let mut state = self.state.lock();
if let Some(contract) = state.contracts.get(name) {
if *contract != slot.contract {
return Err((ServiceInstallRefusal::ContractMismatch, slot));
}
} else {
state.contracts.insert(name, slot.contract);
}
if let Some(existing) = state.slots.get(&(key, name))
&& existing
.fiber
.upgrade()
.is_some_and(|fiber| fiber.is_alive())
{
return Err((ServiceInstallRefusal::Duplicate, slot));
}
let drift = (owner.state() == crate::fiber::FiberState::Active)
.then(|| CommittedServiceDrift::commit(root, &[(name.to_owned(), key)]));
let evicted = state.slots.remove(&(key, name));
state.slots.insert((key, name), slot);
Ok((evicted, drift))
}
fn visible(slot: &ServiceSlot) -> bool {
slot.fiber.upgrade().is_some_and(|fiber| {
fiber.is_alive() && fiber.state() == crate::fiber::FiberState::Active
})
}
pub(crate) fn occurrence_id(&self, key: &RealmKey, name: &str) -> Option<ServiceOccurrenceId> {
let state = self.state.lock();
let slot = state.slots.get(&(*key, name))?;
Self::visible(slot).then(|| slot.occurrence.clone())
}
fn visible_value<S: Service>(
&self,
key: RealmKey,
) -> std::result::Result<Arc<S>, ServiceLookupError> {
let value = {
let state = self.state.lock();
match state.contracts.get(S::NAME) {
Some(contract) if *contract != std::any::TypeId::of::<S>() => {
return Err(ServiceLookupError::ContractMismatch { service: S::NAME });
}
_ => {}
}
let Some(slot) = state
.slots
.get(&(key, S::NAME))
.filter(|slot| Self::visible(slot))
else {
return Err(ServiceLookupError::Unavailable { service: S::NAME });
};
slot.value.clone()
};
value
.downcast::<S>()
.map_err(|_| ServiceLookupError::ContractMismatch { service: S::NAME })
}
fn mutation_state(owner: &Weak<Fiber>) -> Option<crate::fiber::FiberState> {
let fiber = owner.upgrade()?;
if !fiber.is_alive() {
return None;
}
let state = fiber.state();
matches!(
state,
crate::fiber::FiberState::Loading | crate::fiber::FiberState::Active
)
.then_some(state)
}
fn set_exact(
&self,
key: RealmKey,
name: &'static str,
occurrence: &ServiceOccurrenceId,
owner: &Weak<Fiber>,
value: Arc<dyn Any + Send + Sync>,
) -> std::result::Result<
Arc<dyn Any + Send + Sync>,
(ServiceControlError, Arc<dyn Any + Send + Sync>),
> {
let mut state = self.state.lock();
let Some(slot) = state.slots.get_mut(&(key, name)) else {
return Err((
ServiceControlError::StalePublication { service: name },
value,
));
};
if slot.occurrence != *occurrence {
return Err((
ServiceControlError::StalePublication { service: name },
value,
));
}
if Self::mutation_state(owner).is_none() {
return Err((ServiceControlError::MutationClosed { service: name }, value));
}
Ok(std::mem::replace(&mut slot.value, value))
}
fn remove_exact(
&self,
root: &Root,
key: RealmKey,
name: &'static str,
occurrence: &ServiceOccurrenceId,
owner: &Weak<Fiber>,
) -> std::result::Result<(ServiceSlot, Option<CommittedServiceDrift>), ServiceControlError>
{
let mut state = self.state.lock();
let entry = match state.slots.entry((key, name)) {
std::collections::hash_map::Entry::Occupied(entry)
if entry.get().occurrence == *occurrence =>
{
entry
}
_ => return Err(ServiceControlError::StalePublication { service: name }),
};
let Some(owner_state) = Self::mutation_state(owner) else {
return Err(ServiceControlError::MutationClosed { service: name });
};
let drift = (owner_state == crate::fiber::FiberState::Active).then(|| {
CommittedServiceDrift::commit(root, &[(name.to_owned(), key)])
});
Ok((entry.remove(), drift))
}
fn withdraw_exact(&self, key: RealmKey, name: &'static str, occurrence: &ServiceOccurrenceId) {
let removed = {
let mut state = self.state.lock();
match state.slots.entry((key, name)) {
std::collections::hash_map::Entry::Occupied(entry)
if entry.get().occurrence == *occurrence =>
{
Some(entry.remove())
}
_ => None,
}
};
drop(removed);
}
pub(crate) fn slots_owned_by(&self, fiber: &Arc<Fiber>) -> Vec<(String, RealmKey)> {
let owner = Arc::downgrade(fiber);
self.state
.lock()
.slots
.iter()
.filter(|(_, slot)| Weak::ptr_eq(&slot.fiber, &owner))
.map(|((key, name), _)| ((*name).to_owned(), *key))
.collect()
}
pub(crate) fn commit_fiber_transition(
&self,
root: &Root,
fiber: &Arc<Fiber>,
old: crate::fiber::FiberState,
next: crate::fiber::FiberState,
) -> (
Vec<ServiceVisibilityOccurrence>,
Option<CommittedServiceDrift>,
) {
debug_assert_ne!(old, next);
debug_assert_ne!(
old == crate::fiber::FiberState::Active,
next == crate::fiber::FiberState::Active,
"only Active visibility boundaries use the Service semantic commit"
);
let state = self.state.lock();
let owner = Arc::downgrade(fiber);
let visibility = state
.slots
.iter()
.filter(|(_, slot)| Weak::ptr_eq(&slot.fiber, &owner))
.map(|((key, name), slot)| ServiceVisibilityOccurrence {
service: (*name).to_owned(),
realm: *key,
occurrence: slot.occurrence.clone(),
})
.collect::<Vec<_>>();
let drift = (!visibility.is_empty()).then(|| {
let edges = visibility
.iter()
.map(|slot| (slot.service.clone(), slot.realm))
.collect::<Vec<_>>();
CommittedServiceDrift::commit(root, &edges)
});
fiber.set_state(next);
drop(state);
(visibility, drift)
}
pub(crate) fn snapshot_occurrences(&self) -> Vec<ServiceOccurrenceSnapshot> {
let state = self.state.lock();
state
.slots
.iter()
.filter_map(|((realm, service), slot)| {
let fiber = slot.fiber.upgrade()?;
let fiber_state = fiber.state();
let current = fiber.assert_can_register().is_ok();
current.then(|| ServiceOccurrenceSnapshot {
id: slot.occurrence.clone(),
service: (*service).to_owned(),
realm: *realm,
provider: fiber.id().clone(),
visible: fiber_state == crate::fiber::FiberState::Active,
})
})
.collect()
}
}
pub(crate) struct ServiceVisibilityOccurrence {
pub(crate) service: String,
pub(crate) realm: RealmKey,
pub(crate) occurrence: ServiceOccurrenceId,
}
pub(crate) struct ServiceOccurrenceSnapshot {
pub(crate) id: ServiceOccurrenceId,
pub(crate) service: String,
pub(crate) realm: RealmKey,
pub(crate) provider: crate::fiber::FiberId,
pub(crate) visible: bool,
}
#[must_use = "dropping a ServicePublication leaves generation-owned publication cleanup armed"]
pub struct ServicePublication<S: Service> {
root: Arc<Root>,
key: RealmKey,
occurrence: ServiceOccurrenceId,
owner: Weak<Fiber>,
cleanup: crate::effect::DisposableToken,
_service: std::marker::PhantomData<fn() -> S>,
}
impl<S: Service> ServicePublication<S> {
pub fn set(&self, value: Arc<S>) -> std::result::Result<(), ServiceControlError> {
let erased: Arc<dyn Any + Send + Sync> = value;
match self
.root
.services
.set_exact(self.key, S::NAME, &self.occurrence, &self.owner, erased)
{
Ok(outgoing) => {
drop(outgoing);
Ok(())
}
Err((error, rejected)) => {
drop(rejected);
Err(error)
}
}
}
pub fn remove(self) -> std::result::Result<(), ServiceControlError> {
let (removed, drift) = self.root.services.remove_exact(
&self.root,
self.key,
S::NAME,
&self.occurrence,
&self.owner,
)?;
let was_visible = drift.is_some();
let claimed_cleanup = self
.owner
.upgrade()
.and_then(|fiber| fiber.remove_disposable(self.cleanup));
if let Some(drift) = drift {
drift.kick(&self.root);
}
if was_visible {
self.root.observations.publish(
crate::observation::RuntimeObservation::ServiceVisibility {
service: S::NAME.to_owned(),
realm: ServiceRealm::new(self.root.realm_membership.clone(), self.key),
previous: Some(crate::observation::ServicePublicationId(
self.occurrence.clone(),
)),
current: None,
},
);
}
drop(claimed_cleanup);
drop(removed);
Ok(())
}
}
impl Context {
pub fn provide<S: Service>(
&self,
value: Arc<S>,
) -> std::result::Result<ServicePublication<S>, ServicePublishError> {
let fiber = self.fiber().clone();
fiber
.assert_can_register()
.map_err(|_| ServicePublishError::InactiveContext)?;
let key = self.isolate_key(S::NAME);
let occurrence = ServiceOccurrenceId::fresh();
let cleanup_root = self.root.clone();
let cleanup_occurrence = occurrence.clone();
let mut step = ServiceSlotPublish {
root: &self.root,
owner: &fiber,
store: &self.root.services,
key,
name: S::NAME,
slot: Some(ServiceSlot {
occurrence: occurrence.clone(),
value,
contract: std::any::TypeId::of::<S>(),
fiber: Arc::downgrade(&fiber),
}),
evicted: None,
drift: None,
refusal: None,
};
let committed = crate::gated::push_gated(
&fiber,
crate::effect::sync_cleanup(move || {
cleanup_root
.services
.withdraw_exact(key, S::NAME, &cleanup_occurrence);
}),
&mut step,
);
let cleanup = match committed {
Ok(cleanup) => cleanup,
Err(_) => {
return Err(step
.refusal
.take()
.unwrap_or(ServicePublishError::InactiveContext));
}
};
#[cfg(test)]
if step.drift.is_some() {
probe_visible_publish(S::NAME);
}
let drift = step.drift.take();
let became_visible = drift.is_some();
drop(step);
if let Some(drift) = drift {
drift.kick(&self.root);
}
if became_visible {
self.root.observations.publish(
crate::observation::RuntimeObservation::ServiceVisibility {
service: S::NAME.to_owned(),
realm: ServiceRealm::new(self.root.realm_membership.clone(), key),
previous: None,
current: Some(crate::observation::ServicePublicationId(occurrence.clone())),
},
);
}
Ok(ServicePublication {
root: self.root.clone(),
key,
occurrence,
owner: Arc::downgrade(&fiber),
cleanup,
_service: std::marker::PhantomData,
})
}
pub fn try_service<S: Service>(&self) -> std::result::Result<Arc<S>, ServiceLookupError> {
let key = self.isolate_key(S::NAME);
self.root.services.visible_value::<S>(key)
}
}
#[cfg(test)]
mod semantic_commit_tests {
use super::{VISIBLE_PUBLISH_PROBE, VisiblePublishProbe};
use crate::{Context, FiberState, InjectSpec, Plugin, PreparedPlugin, Service};
use std::convert::Infallible;
use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
struct AtomicVisibility;
impl Service for AtomicVisibility {
const NAME: &'static str = "issue101/atomic-visibility";
}
struct WantsAtomicVisibility(Arc<AtomicUsize>);
impl Plugin for WantsAtomicVisibility {
type Config = ();
type Input = ();
type PrepareError = Infallible;
type ApplyError = Infallible;
fn inject(&self) -> InjectSpec {
InjectSpec::none().require(AtomicVisibility::NAME)
}
fn prepare(&self, _config: ()) -> Result<(), Infallible> {
Ok(())
}
fn apply(
&self,
ctx: Context,
_prepared: &(),
) -> impl Future<Output = Result<(), Infallible>> + Send {
let applies = self.0.clone();
async move {
ctx.try_service::<AtomicVisibility>()
.expect("committed visible Service is available to the dependent");
applies.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
}
struct ProbeReset;
impl Drop for ProbeReset {
fn drop(&mut self) {
*VISIBLE_PUBLISH_PROBE.lock() = None;
}
}
#[tokio::test]
async fn visible_publish_commits_recheck_before_kick_window() {
let root = Context::new();
let applies = Arc::new(AtomicUsize::new(0));
let dependent = root
.spawn(PreparedPlugin::from_input(
WantsAtomicVisibility(applies.clone()),
(),
))
.await
.unwrap();
assert_eq!(dependent.state(), FiberState::Pending);
let reached = Arc::new(std::sync::Barrier::new(2));
let resume = Arc::new(std::sync::Barrier::new(2));
*VISIBLE_PUBLISH_PROBE.lock() = Some(Arc::new(VisiblePublishProbe {
service: AtomicVisibility::NAME,
reached: reached.clone(),
resume: resume.clone(),
}));
let _reset = ProbeReset;
let provider = std::thread::spawn({
let root = root.clone();
move || {
let _publication = root.provide(Arc::new(AtomicVisibility)).unwrap();
}
});
reached.wait();
let ready = crate::deadline::bounded(2000, dependent.ready()).await;
resume.wait();
provider.join().unwrap();
assert_eq!(
ready
.expect("ready drives the already-committed recheck")
.unwrap(),
FiberState::Active
);
assert_eq!(applies.load(Ordering::SeqCst), 1);
root.try_service::<AtomicVisibility>()
.expect("dropping the publication capability is inert");
}
}