1- use std:: sync:: Arc ;
1+ pub mod persistence {
2+ pub use spacetimedb_engine:: persistence:: * ;
3+ }
24
3- use enum_map:: EnumMap ;
4- use spacetimedb_schema:: reducer_name:: ReducerName ;
5- use tokio:: sync:: mpsc;
6- use tokio:: time:: MissedTickBehavior ;
5+ pub mod relational_db {
6+ pub use spacetimedb_engine:: relational_db:: * ;
7+ }
78
8- use crate :: subscription:: ExecutionCounters ;
9- use spacetimedb_datastore:: execution_context:: WorkloadType ;
10- use spacetimedb_datastore:: { locking_tx_datastore:: datastore:: TxMetrics , traits:: TxData } ;
9+ pub mod sql {
10+ pub mod ast {
11+ pub use spacetimedb_engine:: sql:: ast:: * ;
12+ }
1113
12- mod durability;
13- pub mod persistence;
14- pub mod relational_db;
15- pub mod snapshot;
16- pub mod update;
14+ pub mod rls {
15+ pub use spacetimedb_engine:: sql:: rls:: * ;
16+ }
17+ }
18+
19+ pub mod snapshot {
20+ pub use spacetimedb_engine:: snapshot:: * ;
21+ }
22+
23+ pub mod update {
24+ pub use spacetimedb_engine:: update:: * ;
25+ }
1726
1827/// Whether SpacetimeDB is run in memory, or persists objects and
1928/// a message log to disk.
@@ -35,111 +44,10 @@ pub struct Config {
3544pub page_pool_max_size : Option < usize > ,
3645}
3746
38- /// A message that is processed by the [`spawn_metrics_recorder`] actor.
39- /// We use a separate task to record metrics to avoid blocking transactions.
40- pub struct MetricsMessage {
41- /// The reducer the produced these metrics.
42- reducer : Option < ReducerName > ,
43- /// Metrics from a mutable transaction.
44- metrics_for_writer : Option < TxMetrics > ,
45- /// Metrics from a read-only transaction.
46- /// A message may have metrics for both types of transactions,
47- /// because metrics for a reducer and its subscription updates are recorded together.
48- metrics_for_reader : Option < TxMetrics > ,
49- /// The row updates for an immutable transaction.
50- /// Needed for insert and delete counters.
51- tx_data : Option < Arc < TxData > > ,
52- /// Cached metrics counters for each workload type.
53- counters : Arc < EnumMap < WorkloadType , ExecutionCounters > > ,
54- }
55-
56- /// The handle used to send work to the tx metrics recorder.
57- #[ derive( Clone ) ]
58- pub struct MetricsRecorderQueue {
59- tx : mpsc:: UnboundedSender < MetricsMessage > ,
60- }
61-
62- impl MetricsRecorderQueue {
63- pub fn send_metrics (
64- & self ,
65- reducer : Option < ReducerName > ,
66- metrics_for_writer : Option < TxMetrics > ,
67- metrics_for_reader : Option < TxMetrics > ,
68- tx_data : Option < Arc < TxData > > ,
69- counters : Arc < EnumMap < WorkloadType , ExecutionCounters > > ,
70- ) {
71- if let Err ( err) = self . tx . send ( MetricsMessage {
72- reducer,
73- metrics_for_writer,
74- metrics_for_reader,
75- tx_data,
76- counters,
77- } ) {
78- log:: warn!( "failed to send metrics: {err}" ) ;
79- }
80- }
81- }
82-
83- fn record_metrics (
84- MetricsMessage {
85- reducer,
86- metrics_for_writer,
87- metrics_for_reader,
88- tx_data,
89- counters,
90- } : MetricsMessage ,
91- ) {
92- if let Some ( tx_metrics) = metrics_for_writer {
93- tx_metrics. report (
94- // If row updates are present,
95- // they will always belong to the writer transaction.
96- tx_data. as_deref ( ) ,
97- reducer. as_ref ( ) ,
98- |wl| & counters[ wl] ,
99- ) ;
100- }
101- if let Some ( tx_metrics) = metrics_for_reader {
102- tx_metrics. report (
103- // If row updates are present,
104- // they will never belong to the reader transaction.
105- // Passing row updates here will most likely panic.
106- None ,
107- reducer. as_ref ( ) ,
108- |wl| & counters[ wl] ,
109- ) ;
110- }
111- }
112-
113- /// The metrics recorder is a side channel that the main database thread forwards metrics to.
114- /// While we want to avoid unnecessary compute on the critical path, communicating with other
115- /// threads is not free, and for this case in particular waking a parked task is not free.
116- ///
117- /// Previously, each tx would send its metrics to the recorder task. As soon as the recorder
118- /// task `recv`d a message, it would update the counters and gauges, and immediately wait for
119- /// the next tx's message. This meant that the tx would need to be more expensive than the
120- /// recording of its metrics in order for the recorder task not to be parked on `recv` when
121- /// the tx would `send` its metrics. This would obviously never be the case, and so each `send`
122- /// would incur the overhead of waking the task.
123- ///
124- /// To mitigate this, we now record metrics, for potentially many transactions, periodically
125- /// every 5ms.
126- const TX_METRICS_RECORDING_INTERVAL : std:: time:: Duration = std:: time:: Duration :: from_millis ( 5 ) ;
127-
128- /// Spawns a task for recording transaction metrics.
129- /// Returns the handle for pushing metrics to the recorder.
130- pub fn spawn_tx_metrics_recorder ( ) -> ( MetricsRecorderQueue , tokio:: task:: AbortHandle ) {
131- let ( tx, mut rx) = mpsc:: unbounded_channel ( ) ;
132- let abort_handle = tokio:: spawn ( async move {
133- let mut interval = tokio:: time:: interval ( TX_METRICS_RECORDING_INTERVAL ) ;
134- interval. set_missed_tick_behavior ( MissedTickBehavior :: Skip ) ;
47+ pub type MetricsRecorderQueue = spacetimedb_engine:: MetricsRecorderQueue ;
13548
136- loop {
137- interval. tick ( ) . await ;
138- while let Ok ( metrics) = rx. try_recv ( ) {
139- record_metrics ( metrics) ;
140- }
141- }
142- } )
143- . abort_handle ( ) ;
144- ( MetricsRecorderQueue { tx } , abort_handle)
49+ pub fn spawn_tx_metrics_recorder (
50+ handle : & spacetimedb_runtime:: Handle ,
51+ ) -> ( MetricsRecorderQueue , spacetimedb_runtime:: AbortHandle ) {
52+ spacetimedb_engine:: spawn_tx_metrics_recorder ( handle)
14553}
0 commit comments