Files
@ 0de39654770f
Branch filter:
Location: CSY/reowolf/src/runtime2/runtime.rs
0de39654770f
7.1 KiB
application/rls-services+xml
WIP: Closing ports
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | use std::sync::{Arc, Mutex, Condvar};
use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};
use std::collections::VecDeque;
use crate::protocol::*;
use super::communication::Message;
use super::component::{CompCtx, CompPDL};
use super::store::{ComponentStore, QueueDynMpsc, QueueDynProducer};
// -----------------------------------------------------------------------------
// Component
// -----------------------------------------------------------------------------
/// Key to a component. Type system somewhat ensures that there can only be one
/// of these. Only with a key one may retrieve privately-accessible memory for
/// a component. Practically just a generational index, like `CompId` is.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub(crate) struct CompKey(pub u32);
impl CompKey {
pub(crate) fn downgrade(&self) -> CompId {
return CompId(self.0);
}
}
/// Generational ID of a component
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct CompId(pub u32);
impl CompId {
pub(crate) fn new_invalid() -> CompId {
return CompId(u32::MAX);
}
/// Upgrade component ID to component key. Unsafe because the caller needs
/// to make sure that only one component key can exist at a time (to ensure
/// a component can only be scheduled/executed by one thread).
pub(crate) unsafe fn upgrade(&self) -> CompKey {
return CompKey(self.0);
}
}
/// Private fields of a component, may only be modified by a single thread at
/// a time.
pub(crate) struct RuntimeComp {
pub public: CompPublic,
pub code: CompPDL,
pub ctx: CompCtx,
pub inbox: QueueDynMpsc<Message>
}
/// Should contain everything that is accessible in a thread-safe manner
// TODO: Do something about the `num_handles` thing. This needs to be a bit more
// "foolproof" to lighten the mental burden of using the `num_handles`
// variable.
pub(crate) struct CompPublic {
pub sleeping: AtomicBool,
pub num_handles: AtomicU32, // manually modified (!)
pub inbox: QueueDynProducer<Message>,
}
/// Handle to public part of a component. Would be nice if we could
/// automagically manage the `num_handles` counter. But when it reaches zero we
/// need to manually remove the handle from the runtime. So we just have debug
/// code to make sure this actually happens.
pub(crate) struct CompHandle {
target: *const CompPublic,
#[cfg(debug_assertions)] decremented: bool,
}
impl CompHandle {
fn new(public: &CompPublic) -> CompHandle {
let handle = CompHandle{
target: public,
#[cfg(debug_assertions)] decremented: false,
};
handle.increment_users();
return handle;
}
fn increment_users(&self) {
let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);
debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed
}
/// Returns true if the component should be destroyed
pub(crate) fn decrement_users(&mut self) -> bool {
debug_assert!(!self.decremented, "illegal to 'decrement_users' twice");
dbg_code!(self.decremented = true);
let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);
return old_count == 1;
}
}
impl Clone for CompHandle {
fn clone(&self) -> Self {
debug_assert!(!self.decremented, "illegal to clone after 'decrement_users'");
self.increment_users();
return CompHandle{
target: self.target,
#[cfg(debug_assertions)] decremented: false,
};
}
}
impl std::ops::Deref for CompHandle {
type Target = CompPublic;
fn deref(&self) -> &Self::Target {
return unsafe{ &*self.target };
}
}
impl Drop for CompHandle {
fn drop(&mut self) {
debug_assert!(self.decremented, "need call to 'decrement_users' before dropping");
}
}
// -----------------------------------------------------------------------------
// Runtime
// -----------------------------------------------------------------------------
pub type RuntimeHandle = Arc<Runtime>;
/// Memory that is maintained by "the runtime". In practice it is maintained by
/// multiple schedulers, and this serves as the common interface to that memory.
pub struct Runtime {
pub protocol: ProtocolDescription,
components: ComponentStore<RuntimeComp>,
work_queue: Mutex<VecDeque<CompKey>>,
work_condvar: Condvar,
active_elements: AtomicU32, // active components and APIs (i.e. component creators)
}
impl Runtime {
pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
assert!(num_threads > 0, "need a thread to perform work");
return Runtime{
protocol: protocol_description,
components: ComponentStore::new(128),
work_queue: Mutex::new(VecDeque::with_capacity(128)),
work_condvar: Condvar::new(),
active_elements: AtomicU32::new(0),
};
}
// Scheduling and retrieving work
pub(crate) fn take_work(&self) -> Option<CompKey> {
let mut lock = self.work_queue.lock().unwrap();
while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {
lock = self.work_condvar.wait(lock).unwrap();
}
return lock.pop_front();
}
pub(crate) fn enqueue_work(&self, key: CompKey) {
let mut lock = self.work_queue.lock().unwrap();
lock.push_back(key);
self.work_condvar.notify_one();
}
// Creating/destroying components
/// Creates a new component. Note that the public part will be properly
/// initialized, but the private fields (e.g. owned ports, peers, etc.)
/// are not.
pub(crate) fn create_pdl_component(&self, comp: CompPDL, initially_sleeping: bool) -> (CompKey, &mut RuntimeComp) {
let inbox_queue = QueueDynMpsc::new(16);
let inbox_producer = inbox_queue.producer();
let comp = RuntimeComp{
public: CompPublic{
sleeping: AtomicBool::new(initially_sleeping),
num_handles: AtomicU32::new(1), // the component itself acts like a handle
inbox: inbox_producer,
},
code: comp,
ctx: CompCtx::default(),
inbox: inbox_queue,
};
let index = self.components.create(comp);
// TODO: just do a reserve_index followed by a consume_index or something
let component = self.components.get_mut(index);
component.ctx.id = CompId(index);
return (CompKey(index), component);
}
pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {
let component = self.components.get_mut(key.0);
return component;
}
pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {
let component = self.components.get(id.0);
return CompHandle::new(&component.public);
}
pub(crate) fn destroy_component(&self, key: CompKey) {
debug_assert_eq!(self.get_component(key).public.num_handles.load(Ordering::Acquire), 0);
self.components.destroy(key.0);
}
// Interacting with components
}
|