commonware_runtime/tokio/
runtime.rs

1#[cfg(not(feature = "iouring-network"))]
2use crate::network::tokio::{Config as TokioNetworkConfig, Network as TokioNetwork};
3#[cfg(feature = "iouring-storage")]
4use crate::storage::iouring::{Config as IoUringConfig, Storage as IoUringStorage};
5#[cfg(not(feature = "iouring-storage"))]
6use crate::storage::tokio::{Config as TokioStorageConfig, Storage as TokioStorage};
7#[cfg(feature = "external")]
8use crate::Pacer;
9use crate::{
10    child_label,
11    network::metered::Network as MeteredNetwork,
12    prefixed_name,
13    process::metered::Metrics as MeteredProcess,
14    signal::Signal,
15    storage::metered::Storage as MeteredStorage,
16    telemetry::metrics::{
17        add_attribute, raw, task::Label, validate_label, CounterFamily, GaugeFamily, Metric,
18        Register, Registered, Registry,
19    },
20    utils::{self, signal::Stopper, supervision::Tree, Panicker},
21    BufferPool, BufferPoolConfig, Clock, Error, Execution, Handle, Name, SinkOf, Spawner as _,
22    StreamOf, Supervisor as _, METRICS_PREFIX,
23};
24#[cfg(feature = "iouring-network")]
25use crate::{
26    iouring,
27    network::iouring::{Config as IoUringNetworkConfig, Network as IoUringNetwork},
28};
29use commonware_macros::{select, stability};
30#[stability(BETA)]
31use commonware_parallel::Rayon;
32use commonware_utils::{sync::Mutex, sys_rng, NZUsize};
33use governor::clock::{Clock as GClock, ReasonablyRealtime};
34use rand_core::{Rng, TryCryptoRng, TryRng};
35#[stability(BETA)]
36use rayon::ThreadPoolBuilder;
37use std::{
38    convert::Infallible,
39    env,
40    future::Future,
41    net::{IpAddr, SocketAddr},
42    num::NonZeroUsize,
43    path::PathBuf,
44    sync::Arc,
45    time::{Duration, SystemTime},
46};
47use tokio::runtime::{Builder, Runtime};
48
49#[cfg(feature = "iouring-network")]
50cfg_if::cfg_if! {
51    if #[cfg(test)] {
52        // Use a smaller ring in tests to reduce `io_uring_setup` failures
53        // under parallel test load due to mlock/resource limits.
54        const IOURING_NETWORK_SIZE: u32 = 128;
55    } else {
56        const IOURING_NETWORK_SIZE: u32 = 1024;
57    }
58}
59
60#[derive(Debug)]
61struct Metrics {
62    tasks_spawned: CounterFamily<Label>,
63    tasks_running: GaugeFamily<Label>,
64}
65
66impl Metrics {
67    pub fn init(registry: &mut impl Register) -> Self {
68        Self {
69            tasks_spawned: registry.register(
70                "tasks_spawned",
71                "Total number of tasks spawned",
72                raw::Family::default(),
73            ),
74            tasks_running: registry.register(
75                "tasks_running",
76                "Number of tasks currently running",
77                raw::Family::default(),
78            ),
79        }
80    }
81}
82
83#[derive(Clone, Debug)]
84pub struct NetworkConfig {
85    /// If Some, explicitly sets TCP_NODELAY on the socket.
86    /// Otherwise uses system default.
87    ///
88    /// Defaults to `Some(true)`.
89    tcp_nodelay: Option<bool>,
90
91    /// Whether to set `SO_LINGER` to zero on the socket.
92    ///
93    /// When enabled, causes an immediate RST on close, avoiding
94    /// `TIME_WAIT` state. This is useful in adversarial environments to
95    /// reclaim socket resources immediately when closing connections to
96    /// misbehaving peers.
97    ///
98    /// Defaults to `true`.
99    zero_linger: bool,
100
101    /// Read/write timeout for network operations.
102    ///
103    /// Bounds the full `Sink::send` and `Stream::recv` calls rather than each
104    /// individual socket syscall. Larger
105    /// batched writes may therefore require a larger timeout.
106    read_write_timeout: Duration,
107}
108
109impl Default for NetworkConfig {
110    fn default() -> Self {
111        Self {
112            tcp_nodelay: Some(true),
113            zero_linger: true,
114            read_write_timeout: Duration::from_secs(60),
115        }
116    }
117}
118
119/// Configuration for the `tokio` runtime.
120#[derive(Clone)]
121pub struct Config {
122    /// Number of threads to use for handling async tasks.
123    ///
124    /// Worker threads are always active (waiting for work).
125    ///
126    /// Tokio sets the default value to the number of logical CPUs.
127    worker_threads: usize,
128
129    /// Number of scheduler ticks between global queue polls.
130    ///
131    /// When unset, Tokio uses its default behavior for the multi-thread
132    /// scheduler. Smaller values reduce the delay before tasks woken from
133    /// outside a worker, such as io_uring completion notifications, are polled
134    /// from the global queue again.
135    global_queue_interval: Option<u32>,
136
137    /// Maximum number of threads to use for blocking tasks.
138    ///
139    /// Unlike worker threads, blocking threads are created as needed and
140    /// exit if left idle for too long.
141    ///
142    /// Tokio sets the default value to 512 to avoid hanging on lower-level
143    /// operations that require blocking (like `fs` and writing to `Stdout`).
144    max_blocking_threads: usize,
145
146    /// Stack size to use for runtime-owned threads.
147    ///
148    /// Defaults to the system stack size when the current platform exposes it,
149    /// and otherwise falls back to Rust's default spawned-thread stack size.
150    ///
151    /// See [utils::thread::system_thread_stack_size].
152    thread_stack_size: usize,
153
154    /// Whether or not to catch panics.
155    catch_panics: bool,
156
157    /// Base directory for all storage operations.
158    storage_directory: PathBuf,
159
160    /// Maximum buffer size for operations on blobs.
161    ///
162    /// Tokio sets the default value to 2MB.
163    maximum_buffer_size: usize,
164
165    /// Network configuration.
166    network_cfg: NetworkConfig,
167
168    /// Explicit buffer pool configuration for network I/O, if provided.
169    network_buffer_pool_cfg: Option<BufferPoolConfig>,
170
171    /// Explicit buffer pool configuration for storage I/O, if provided.
172    storage_buffer_pool_cfg: Option<BufferPoolConfig>,
173}
174
175impl Config {
176    /// Returns a new [Config] with default values.
177    pub fn new() -> Self {
178        let rng = sys_rng().next_u64();
179        let storage_directory = env::temp_dir().join(format!("commonware_tokio_runtime_{rng}"));
180        Self {
181            worker_threads: 2,
182            global_queue_interval: None,
183            max_blocking_threads: 512,
184            thread_stack_size: utils::thread::system_thread_stack_size(),
185            catch_panics: false,
186            storage_directory,
187            maximum_buffer_size: 2 * 1024 * 1024, // 2 MB
188            network_cfg: NetworkConfig::default(),
189            network_buffer_pool_cfg: None,
190            storage_buffer_pool_cfg: None,
191        }
192    }
193
194    // Setters
195    /// See [Config]
196    pub const fn with_worker_threads(mut self, n: usize) -> Self {
197        self.worker_threads = n;
198        self
199    }
200    /// See [Config]
201    pub const fn with_global_queue_interval(mut self, n: u32) -> Self {
202        self.global_queue_interval = Some(n);
203        self
204    }
205    /// See [Config]
206    pub const fn with_max_blocking_threads(mut self, n: usize) -> Self {
207        self.max_blocking_threads = n;
208        self
209    }
210    /// See [Config]
211    pub const fn with_thread_stack_size(mut self, n: usize) -> Self {
212        self.thread_stack_size = n;
213        self
214    }
215    /// See [Config]
216    pub const fn with_catch_panics(mut self, b: bool) -> Self {
217        self.catch_panics = b;
218        self
219    }
220    /// See [Config]
221    pub const fn with_read_write_timeout(mut self, d: Duration) -> Self {
222        self.network_cfg.read_write_timeout = d;
223        self
224    }
225    /// See [Config]
226    pub const fn with_tcp_nodelay(mut self, n: Option<bool>) -> Self {
227        self.network_cfg.tcp_nodelay = n;
228        self
229    }
230    /// See [Config]
231    pub const fn with_zero_linger(mut self, l: bool) -> Self {
232        self.network_cfg.zero_linger = l;
233        self
234    }
235    /// See [Config]
236    pub fn with_storage_directory(mut self, p: impl Into<PathBuf>) -> Self {
237        self.storage_directory = p.into();
238        self
239    }
240    /// See [Config]
241    pub const fn with_maximum_buffer_size(mut self, n: usize) -> Self {
242        self.maximum_buffer_size = n;
243        self
244    }
245    /// See [Config]
246    pub const fn with_network_buffer_pool_config(mut self, cfg: BufferPoolConfig) -> Self {
247        self.network_buffer_pool_cfg = Some(cfg);
248        self
249    }
250    /// See [Config]
251    pub const fn with_storage_buffer_pool_config(mut self, cfg: BufferPoolConfig) -> Self {
252        self.storage_buffer_pool_cfg = Some(cfg);
253        self
254    }
255
256    // Getters
257    /// See [Config]
258    pub const fn worker_threads(&self) -> usize {
259        self.worker_threads
260    }
261    /// See [Config]
262    pub const fn global_queue_interval(&self) -> Option<u32> {
263        self.global_queue_interval
264    }
265    /// See [Config]
266    pub const fn max_blocking_threads(&self) -> usize {
267        self.max_blocking_threads
268    }
269    /// See [Config]
270    pub const fn thread_stack_size(&self) -> usize {
271        self.thread_stack_size
272    }
273    /// See [Config]
274    pub const fn catch_panics(&self) -> bool {
275        self.catch_panics
276    }
277    /// See [Config]
278    pub const fn read_write_timeout(&self) -> Duration {
279        self.network_cfg.read_write_timeout
280    }
281    /// See [Config]
282    pub const fn tcp_nodelay(&self) -> Option<bool> {
283        self.network_cfg.tcp_nodelay
284    }
285    /// See [Config]
286    pub const fn zero_linger(&self) -> bool {
287        self.network_cfg.zero_linger
288    }
289    /// See [Config]
290    pub const fn storage_directory(&self) -> &PathBuf {
291        &self.storage_directory
292    }
293    /// See [Config]
294    pub const fn maximum_buffer_size(&self) -> usize {
295        self.maximum_buffer_size
296    }
297
298    /// Returns the network buffer pool config, deriving pool parallelism from
299    /// `worker_threads` if not explicitly configured.
300    fn resolved_network_buffer_pool_config(&self) -> BufferPoolConfig {
301        self.network_buffer_pool_cfg.clone().unwrap_or_else(|| {
302            BufferPoolConfig::for_network().with_parallelism(NZUsize!(self.worker_threads))
303        })
304    }
305
306    /// Returns the storage buffer pool config, deriving pool parallelism from
307    /// `worker_threads` if not explicitly configured.
308    fn resolved_storage_buffer_pool_config(&self) -> BufferPoolConfig {
309        self.storage_buffer_pool_cfg.clone().unwrap_or_else(|| {
310            BufferPoolConfig::for_storage().with_parallelism(NZUsize!(self.worker_threads))
311        })
312    }
313}
314
315impl Default for Config {
316    fn default() -> Self {
317        Self::new()
318    }
319}
320
321/// Runtime based on [Tokio](https://tokio.rs).
322pub struct Executor {
323    registry: Registry,
324    metrics: Arc<Metrics>,
325    runtime: Runtime,
326    shutdown: Mutex<Stopper>,
327    panicker: Panicker,
328    thread_stack_size: usize,
329}
330
331/// Implementation of [crate::Runner] for the `tokio` runtime.
332pub struct Runner {
333    cfg: Config,
334}
335
336impl Default for Runner {
337    fn default() -> Self {
338        Self::new(Config::default())
339    }
340}
341
342impl Runner {
343    /// Initialize a new `tokio` runtime with the given number of threads.
344    pub const fn new(cfg: Config) -> Self {
345        Self { cfg }
346    }
347}
348
349impl crate::Runner for Runner {
350    type Context = Context;
351
352    fn start<F, Fut>(self, f: F) -> Fut::Output
353    where
354        F: FnOnce(Self::Context) -> Fut,
355        Fut: Future,
356    {
357        // Create a new registry
358        let mut registry = Registry::new();
359        let mut runtime_registry = registry.sub_registry(METRICS_PREFIX);
360
361        // Initialize runtime
362        let metrics = Arc::new(Metrics::init(&mut runtime_registry));
363        let mut builder = Builder::new_multi_thread();
364        builder
365            .worker_threads(self.cfg.worker_threads)
366            .max_blocking_threads(self.cfg.max_blocking_threads)
367            .thread_stack_size(self.cfg.thread_stack_size)
368            .enable_all();
369        if let Some(global_queue_interval) = self.cfg.global_queue_interval {
370            builder.global_queue_interval(global_queue_interval);
371        }
372        let runtime = builder.build().expect("failed to create Tokio runtime");
373
374        // Initialize panicker
375        let (panicker, panicked) = Panicker::new(self.cfg.catch_panics);
376
377        // Collect process metrics.
378        //
379        // We prefer to collect process metrics outside of `Context` because
380        // we are using `runtime_registry` rather than the one provided by `Context`.
381        let process = MeteredProcess::init(&mut runtime_registry);
382        runtime.spawn(process.collect(tokio::time::sleep));
383
384        // Initialize buffer pools
385        let network_buffer_pool = BufferPool::new(
386            self.cfg.resolved_network_buffer_pool_config(),
387            &mut runtime_registry.sub_registry("network_buffer_pool"),
388        );
389        let storage_buffer_pool = BufferPool::new(
390            self.cfg.resolved_storage_buffer_pool_config(),
391            &mut runtime_registry.sub_registry("storage_buffer_pool"),
392        );
393
394        // Make any storage a prior process left in the page cache crash-durable before we open it,
395        // so the data read during init is durable.
396        if let Err(e) = crate::storage::sync(&self.cfg.storage_directory) {
397            panic!(
398                "failed to sync storage filesystem at startup ({}): {e}",
399                self.cfg.storage_directory.display()
400            );
401        }
402
403        // Initialize storage
404        cfg_if::cfg_if! {
405            if #[cfg(feature = "iouring-storage")] {
406                let mut iouring_registry = runtime_registry.sub_registry("iouring_storage");
407                let storage = MeteredStorage::new(
408                    IoUringStorage::start(
409                        IoUringConfig {
410                            storage_directory: self.cfg.storage_directory.clone(),
411                            iouring_config: Default::default(),
412                            thread_stack_size: self.cfg.thread_stack_size,
413                        },
414                        &mut iouring_registry,
415                        storage_buffer_pool.clone(),
416                    ),
417                    &mut runtime_registry,
418                );
419            } else {
420                let storage = MeteredStorage::new(
421                    TokioStorage::new(
422                        TokioStorageConfig::new(
423                            self.cfg.storage_directory.clone(),
424                            self.cfg.maximum_buffer_size,
425                        ),
426                        storage_buffer_pool.clone(),
427                    ),
428                    &mut runtime_registry,
429                );
430            }
431        }
432
433        // Initialize network
434        cfg_if::cfg_if! {
435            if #[cfg(feature = "iouring-network")] {
436                let mut iouring_registry = runtime_registry.sub_registry("iouring_network");
437                let config = IoUringNetworkConfig {
438                    tcp_nodelay: self.cfg.network_cfg.tcp_nodelay,
439                    zero_linger: self.cfg.network_cfg.zero_linger,
440                    read_write_timeout: self.cfg.network_cfg.read_write_timeout,
441                    iouring_config: iouring::Config {
442                        // TODO (#1045): make `IOURING_NETWORK_SIZE` configurable
443                        size: IOURING_NETWORK_SIZE,
444                        max_request_timeout: self.cfg.network_cfg.read_write_timeout,
445                        shutdown_timeout: Some(self.cfg.network_cfg.read_write_timeout),
446                        ..Default::default()
447                    },
448                    thread_stack_size: self.cfg.thread_stack_size,
449                    ..Default::default()
450                };
451                let network = MeteredNetwork::new(
452                    IoUringNetwork::start(
453                        config,
454                        &mut iouring_registry,
455                        network_buffer_pool.clone(),
456                    )
457                    .unwrap(),
458                    &mut runtime_registry,
459                );
460            } else {
461                let config = TokioNetworkConfig::default()
462                    .with_read_timeout(self.cfg.network_cfg.read_write_timeout)
463                    .with_write_timeout(self.cfg.network_cfg.read_write_timeout)
464                    .with_tcp_nodelay(self.cfg.network_cfg.tcp_nodelay)
465                    .with_zero_linger(self.cfg.network_cfg.zero_linger);
466                let network = MeteredNetwork::new(
467                    TokioNetwork::new(config, network_buffer_pool.clone()),
468                    &mut runtime_registry,
469                );
470            }
471        }
472
473        // Initialize executor
474        let executor = Arc::new(Executor {
475            registry,
476            metrics,
477            runtime,
478            shutdown: Mutex::new(Stopper::default()),
479            panicker,
480            thread_stack_size: self.cfg.thread_stack_size,
481        });
482
483        // Get metrics
484        let label = Label::root();
485        executor.metrics.tasks_spawned.get_or_create(&label).inc();
486        let gauge = executor.metrics.tasks_running.get_or_create(&label).clone();
487
488        // Run the future
489        let context = Context {
490            storage,
491            name: label.name(),
492            attributes: Vec::new(),
493            executor: executor.clone(),
494            network,
495            network_buffer_pool,
496            storage_buffer_pool,
497            tree: Tree::root(),
498            execution: Execution::default(),
499        };
500        let output = executor.runtime.block_on(panicked.interrupt(f(context)));
501        gauge.dec();
502
503        output
504    }
505}
506
507cfg_if::cfg_if! {
508    if #[cfg(feature = "iouring-storage")] {
509        type Storage = MeteredStorage<IoUringStorage>;
510    } else {
511        type Storage = MeteredStorage<TokioStorage>;
512    }
513}
514
515cfg_if::cfg_if! {
516    if #[cfg(feature = "iouring-network")] {
517        type Network = MeteredNetwork<IoUringNetwork>;
518    } else {
519        type Network = MeteredNetwork<TokioNetwork>;
520    }
521}
522
523/// Implementation of [crate::Spawner], [crate::Clock],
524/// [crate::Network], and [crate::Storage] for the `tokio`
525/// runtime.
526pub struct Context {
527    name: String,
528    attributes: Vec<(String, String)>,
529    executor: Arc<Executor>,
530    storage: Storage,
531    network: Network,
532    network_buffer_pool: BufferPool,
533    storage_buffer_pool: BufferPool,
534    tree: Arc<Tree>,
535    execution: Execution,
536}
537
538impl Context {
539    /// Access the [Metrics] of the runtime.
540    fn metrics(&self) -> &Metrics {
541        &self.executor.metrics
542    }
543}
544
545impl crate::Spawner for Context {
546    fn dedicated(mut self) -> Self {
547        self.execution = Execution::Dedicated;
548        self
549    }
550
551    fn shared(mut self, blocking: bool) -> Self {
552        self.execution = Execution::Shared(blocking);
553        self
554    }
555
556    fn spawn<F, Fut, T>(mut self, f: F) -> Handle<T>
557    where
558        F: FnOnce(Self) -> Fut + Send + 'static,
559        Fut: Future<Output = T> + Send + 'static,
560        T: Send + 'static,
561    {
562        // Get metrics
563        let (_, metric) = spawn_metrics!(self);
564
565        // Track supervision before resetting configuration
566        let parent = Arc::clone(&self.tree);
567        let past = self.execution;
568        self.execution = Execution::default();
569        let (child, aborted) = Tree::child(&parent);
570        if aborted {
571            return Handle::closed(metric);
572        }
573        self.tree = child;
574
575        // Spawn the task
576        let executor = self.executor.clone();
577        let future = f(self);
578        let (f, handle) = Handle::init(
579            future,
580            metric,
581            executor.panicker.clone(),
582            Arc::clone(&parent),
583        );
584
585        if matches!(past, Execution::Dedicated) {
586            utils::thread::spawn(executor.thread_stack_size, {
587                // Ensure the task can access the tokio runtime
588                let handle = executor.runtime.handle().clone();
589                move || {
590                    handle.block_on(f);
591                }
592            });
593        } else if matches!(past, Execution::Shared(true)) {
594            executor.runtime.spawn_blocking({
595                // Ensure the task can access the tokio runtime
596                let handle = executor.runtime.handle().clone();
597                move || {
598                    handle.block_on(f);
599                }
600            });
601        } else {
602            executor.runtime.spawn(f);
603        }
604
605        // Register the task on the parent
606        if let Some(aborter) = handle.aborter() {
607            parent.register(aborter);
608        }
609
610        handle
611    }
612
613    async fn stop(self, value: i32, timeout: Option<Duration>) -> Result<(), Error> {
614        let stop_resolved = {
615            let mut shutdown = self.executor.shutdown.lock();
616            shutdown.stop(value)
617        };
618
619        // Wait for all tasks to complete or the timeout to fire
620        let timeout_future = timeout.map_or_else(
621            || futures::future::Either::Right(futures::future::pending()),
622            |duration| futures::future::Either::Left(self.sleep(duration)),
623        );
624        select! {
625            result = stop_resolved => {
626                result.map_err(|_| Error::Closed)?;
627                Ok(())
628            },
629            _ = timeout_future => Err(Error::Timeout),
630        }
631    }
632
633    fn stopped(&self) -> Signal {
634        self.executor.shutdown.lock().stopped()
635    }
636}
637
638#[stability(BETA)]
639impl crate::Strategizer for Context {
640    fn strategy(&self, parallelism: NonZeroUsize) -> Rayon {
641        let pool = ThreadPoolBuilder::new()
642            .num_threads(parallelism.get())
643            .spawn_handler(move |thread| {
644                // Tasks spawned in a thread pool are expected to run longer than any single
645                // task and thus should be provisioned as a dedicated thread.
646                self.child("rayon_thread")
647                    .dedicated()
648                    .spawn(move |_| async move { thread.run() });
649                Ok(())
650            })
651            .build()
652            .expect("failed to create Tokio Rayon thread pool");
653        Rayon::with_pool(Arc::new(pool))
654    }
655}
656
657impl crate::Supervisor for Context {
658    fn child(&self, label: &'static str) -> Self {
659        let (tree, _) = Tree::child(&self.tree);
660        Self {
661            name: child_label(&self.name, label),
662            attributes: self.attributes.clone(),
663            executor: self.executor.clone(),
664            storage: self.storage.clone(),
665            network: self.network.clone(),
666            network_buffer_pool: self.network_buffer_pool.clone(),
667            storage_buffer_pool: self.storage_buffer_pool.clone(),
668            tree,
669            execution: Execution::default(),
670        }
671    }
672
673    fn with_attribute(mut self, key: &'static str, value: impl std::fmt::Display) -> Self {
674        // Validate label format (must match [a-zA-Z][a-zA-Z0-9_]*)
675        validate_label(key);
676
677        // Add the attribute to the list of attributes
678        add_attribute(&mut self.attributes, key, value);
679        self
680    }
681
682    fn name(&self) -> Name {
683        Name {
684            label: self.name.clone(),
685            attributes: self.attributes.clone(),
686        }
687    }
688}
689
690impl crate::Metrics for Context {
691    fn register<N: Into<String>, H: Into<String>, M: Metric>(
692        &self,
693        name: N,
694        help: H,
695        metric: M,
696    ) -> Registered<M> {
697        let name = name.into();
698        let help = help.into();
699        let metric = Arc::new(metric);
700        self.executor.registry.register(
701            prefixed_name(&self.name, &name),
702            help,
703            self.attributes.clone(),
704            metric,
705        )
706    }
707
708    fn encode(&self) -> String {
709        self.executor.registry.encode()
710    }
711}
712
713impl Clock for Context {
714    fn current(&self) -> SystemTime {
715        SystemTime::now()
716    }
717
718    fn sleep(&self, duration: Duration) -> impl Future<Output = ()> + Send + 'static {
719        tokio::time::sleep(duration)
720    }
721
722    fn sleep_until(&self, deadline: SystemTime) -> impl Future<Output = ()> + Send + 'static {
723        let duration_until_deadline = deadline.duration_since(self.current()).unwrap_or_default();
724        tokio::time::sleep(duration_until_deadline)
725    }
726}
727
728#[cfg(feature = "external")]
729impl Pacer for Context {
730    fn pace<'a, F, T>(
731        &'a self,
732        _latency: Duration,
733        future: F,
734    ) -> impl Future<Output = T> + Send + 'a
735    where
736        F: Future<Output = T> + Send + 'a,
737        T: Send + 'a,
738    {
739        // Execute the future immediately
740        future
741    }
742}
743
744impl GClock for Context {
745    type Instant = SystemTime;
746
747    fn now(&self) -> Self::Instant {
748        self.current()
749    }
750}
751
752impl ReasonablyRealtime for Context {}
753
754impl crate::Network for Context {
755    type Listener = <Network as crate::Network>::Listener;
756
757    async fn bind(&self, socket: SocketAddr) -> Result<Self::Listener, Error> {
758        self.network.bind(socket).await
759    }
760
761    async fn dial(&self, socket: SocketAddr) -> Result<(SinkOf<Self>, StreamOf<Self>), Error> {
762        self.network.dial(socket).await
763    }
764}
765
766impl crate::Resolver for Context {
767    async fn resolve(&self, host: &str) -> Result<Vec<IpAddr>, Error> {
768        // Uses the host's DNS configuration (e.g. /etc/resolv.conf on Unix,
769        // registry on Windows). This delegates to the system's libc resolver.
770        //
771        // The `:0` port is required by lookup_host's API but is not used
772        // for DNS resolution.
773        let addrs = tokio::net::lookup_host(format!("{host}:0"))
774            .await
775            .map_err(|e| Error::ResolveFailed(e.to_string()))?;
776        Ok(addrs.map(|addr| addr.ip()).collect())
777    }
778}
779
780impl TryRng for Context {
781    type Error = Infallible;
782
783    fn try_next_u32(&mut self) -> Result<u32, Self::Error> {
784        Ok(sys_rng().next_u32())
785    }
786
787    fn try_next_u64(&mut self) -> Result<u64, Self::Error> {
788        Ok(sys_rng().next_u64())
789    }
790
791    fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), Self::Error> {
792        sys_rng().fill_bytes(dest);
793        Ok(())
794    }
795}
796
797impl TryCryptoRng for Context {}
798
799impl crate::Storage for Context {
800    type Blob = <Storage as crate::Storage>::Blob;
801
802    async fn open_versioned(
803        &self,
804        partition: &str,
805        name: &[u8],
806        versions: std::ops::RangeInclusive<u16>,
807    ) -> Result<(Self::Blob, u64, u16), Error> {
808        self.storage.open_versioned(partition, name, versions).await
809    }
810
811    async fn remove(&self, partition: &str, name: Option<&[u8]>) -> Result<(), Error> {
812        self.storage.remove(partition, name).await
813    }
814
815    async fn scan(&self, partition: &str) -> Result<Vec<Vec<u8>>, Error> {
816        self.storage.scan(partition).await
817    }
818}
819
820impl crate::BufferPooler for Context {
821    fn network_buffer_pool(&self) -> &BufferPool {
822        &self.network_buffer_pool
823    }
824
825    fn storage_buffer_pool(&self) -> &BufferPool {
826        &self.storage_buffer_pool
827    }
828}
829
830#[cfg(test)]
831mod tests {
832    use super::*;
833
834    #[test]
835    fn test_worker_threads_updates_default_buffer_pool_parallelism() {
836        let cfg = Config::new().with_worker_threads(8);
837
838        assert_eq!(cfg.worker_threads, 8);
839        let network = cfg.resolved_network_buffer_pool_config();
840        assert_eq!(network.parallelism, NZUsize!(8));
841        assert_eq!(
842            network.thread_cache_config,
843            BufferPoolConfig::for_network().thread_cache_config
844        );
845
846        let storage = cfg.resolved_storage_buffer_pool_config();
847        assert_eq!(storage.parallelism, NZUsize!(8));
848        assert_eq!(
849            storage.thread_cache_config,
850            BufferPoolConfig::for_storage().thread_cache_config
851        );
852    }
853
854    #[test]
855    fn test_default_thread_stack_size_uses_system_default() {
856        let cfg = Config::new();
857        assert_eq!(
858            cfg.thread_stack_size(),
859            utils::thread::system_thread_stack_size()
860        );
861    }
862
863    #[test]
864    fn test_thread_stack_size_override() {
865        let cfg = Config::new().with_thread_stack_size(4 * 1024 * 1024);
866        assert_eq!(cfg.thread_stack_size(), 4 * 1024 * 1024);
867    }
868
869    #[test]
870    fn test_explicit_buffer_pool_configs_override_worker_threads() {
871        // Order does not matter -- explicit configs always win.
872        let cfg = Config::new()
873            .with_network_buffer_pool_config(
874                BufferPoolConfig::for_network().with_parallelism(NZUsize!(2)),
875            )
876            .with_worker_threads(8)
877            .with_storage_buffer_pool_config(
878                BufferPoolConfig::for_storage().with_thread_cache_disabled(),
879            );
880
881        let network = cfg.resolved_network_buffer_pool_config();
882        assert_eq!(network.parallelism, NZUsize!(2));
883        assert_eq!(
884            network.thread_cache_config,
885            BufferPoolConfig::for_network().thread_cache_config
886        );
887
888        let storage = cfg.resolved_storage_buffer_pool_config();
889        assert_eq!(storage.parallelism, NZUsize!(1));
890        assert_eq!(
891            storage.thread_cache_config,
892            BufferPoolConfig::for_storage()
893                .with_thread_cache_disabled()
894                .thread_cache_config
895        );
896    }
897
898    #[cfg(windows)]
899    #[test]
900    fn test_startup_flush_survives_restart() {
901        use crate::{Blob as _, Runner as _, Storage as _};
902
903        // Write and sync a blob, drop the runtime, then reopen the same storage directory in a new
904        // runtime. `sync` runs on startup and reads the blob back.
905        // Confirms the startup flush path runs and storage survives a restart.
906        let cfg = Config::new();
907        let dir = cfg.storage_directory().clone();
908        Runner::new(cfg).start(|context| async move {
909            let (blob, _) = context.open("test", b"blob").await.unwrap();
910            blob.write_at(0, vec![1u8, 2, 3, 4]).await.unwrap();
911            blob.sync().await.unwrap();
912        });
913        let reopened_len = Runner::new(Config::new().with_storage_directory(dir.clone())).start(
914            |context| async move {
915                let (_, len) = context.open("test", b"blob").await.unwrap();
916                len
917            },
918        );
919        assert_eq!(reopened_len, 4);
920        let _ = std::fs::remove_dir_all(&dir);
921    }
922}
commonware_runtime/tokio/runtime.rs

commonware_runtime/tokio/
runtime.rs
