Skip to content

Commit 04b7ea8

Browse files
Shubham8287Shubham8287
committed
fix db error
1 parent e731109 commit 04b7ea8

4 files changed

Lines changed: 102 additions & 111 deletions

File tree

‎crates/core/src/db/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,10 @@ pub struct Config {
3434
pub page_pool_max_size: Option<usize>,
3535
}
3636

37-
pub type MetricsRecorderQueue = spacetimedb_engine::relational_db::MetricsRecorderQueue;
37+
pub type MetricsRecorderQueue = spacetimedb_engine::MetricsRecorderQueue;
3838

3939
pub fn spawn_tx_metrics_recorder(
4040
handle: &spacetimedb_runtime::Handle,
4141
) -> (MetricsRecorderQueue, spacetimedb_runtime::AbortHandle) {
42-
spacetimedb_engine::relational_db::spawn_tx_metrics_recorder(handle)
42+
spacetimedb_engine::spawn_tx_metrics_recorder(handle)
4343
}

‎crates/engine/src/lib.rs‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,99 @@ pub mod snapshot;
99
pub mod update;
1010
pub mod util;
1111

12+
use std::sync::Arc;
13+
14+
use enum_map::EnumMap;
15+
use spacetimedb_datastore::execution_context::WorkloadType;
16+
use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
17+
use spacetimedb_datastore::traits::TxData;
1218
pub use spacetimedb_lib::identity;
1319
pub use spacetimedb_lib::Identity;
1420
pub use spacetimedb_sats::hash;
21+
use spacetimedb_schema::reducer_name::ReducerName;
22+
23+
use crate::metrics::ExecutionCounters;
24+
25+
/// A message that is processed by the [`spawn_metrics_recorder`] actor.
26+
/// We use a separate task to record metrics to avoid blocking transactions.
27+
pub struct MetricsMessage {
28+
/// The reducer the produced these metrics.
29+
reducer: Option<ReducerName>,
30+
/// Metrics from a mutable transaction.
31+
metrics_for_writer: Option<TxMetrics>,
32+
/// Metrics from a read-only transaction.
33+
/// A message may have metrics for both types of transactions,
34+
/// because metrics for a reducer and its subscription updates are recorded together.
35+
metrics_for_reader: Option<TxMetrics>,
36+
/// The row updates for an immutable transaction.
37+
/// Needed for insert and delete counters.
38+
tx_data: Option<Arc<TxData>>,
39+
/// Cached metrics counters for each workload type.
40+
counters: Arc<EnumMap<WorkloadType, ExecutionCounters>>,
41+
}
42+
43+
/// The handle used to send work to the tx metrics recorder.
44+
#[derive(Clone)]
45+
pub struct MetricsRecorderQueue {
46+
tx: spacetimedb_runtime::sync::mpsc::UnboundedSender<MetricsMessage>,
47+
}
48+
49+
impl MetricsRecorderQueue {
50+
pub fn send_metrics(
51+
&self,
52+
reducer: Option<ReducerName>,
53+
metrics_for_writer: Option<TxMetrics>,
54+
metrics_for_reader: Option<TxMetrics>,
55+
tx_data: Option<Arc<TxData>>,
56+
counters: Arc<EnumMap<WorkloadType, ExecutionCounters>>,
57+
) {
58+
if let Err(err) = self.tx.send(MetricsMessage {
59+
reducer,
60+
metrics_for_writer,
61+
metrics_for_reader,
62+
tx_data,
63+
counters,
64+
}) {
65+
log::warn!("failed to send metrics: {err}");
66+
}
67+
}
68+
}
69+
70+
fn record_metrics(
71+
MetricsMessage {
72+
reducer,
73+
metrics_for_writer,
74+
metrics_for_reader,
75+
tx_data,
76+
counters,
77+
}: MetricsMessage,
78+
) {
79+
if let Some(tx_metrics) = metrics_for_writer {
80+
tx_metrics.report(tx_data.as_deref(), reducer.as_ref(), |wl| &counters[wl]);
81+
}
82+
if let Some(tx_metrics) = metrics_for_reader {
83+
tx_metrics.report(None, reducer.as_ref(), |wl| &counters[wl]);
84+
}
85+
}
86+
87+
const TX_METRICS_RECORDING_INTERVAL: std::time::Duration = std::time::Duration::from_millis(5);
88+
89+
/// Spawns a task for recording transaction metrics.
90+
/// Returns the handle for pushing metrics to the recorder.
91+
pub fn spawn_tx_metrics_recorder(
92+
handle: &spacetimedb_runtime::Handle,
93+
) -> (MetricsRecorderQueue, spacetimedb_runtime::AbortHandle) {
94+
let handle_clone = handle.clone();
95+
let (tx, mut rx) = spacetimedb_runtime::sync::mpsc::unbounded_channel();
96+
let abort_handle = handle
97+
.spawn(async move {
98+
loop {
99+
handle_clone.sleep(TX_METRICS_RECORDING_INTERVAL).await;
100+
while let Ok(metrics) = rx.try_recv() {
101+
record_metrics(metrics);
102+
}
103+
}
104+
})
105+
.abort_handle();
106+
(MetricsRecorderQueue { tx }, abort_handle)
107+
}

‎crates/engine/src/relational_db.rs‎

Lines changed: 2 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -2,111 +2,8 @@ use crate::durability::{request_durability, spawn_close as spawn_durability_clos
22
use crate::error::{DBError, RestoreSnapshotError};
33
use crate::metrics::ExecutionCounters;
44
use crate::metrics::ENGINE_METRICS;
5-
6-
/// Whether SpacetimeDB is run in memory, or persists objects and
7-
/// a message log to disk.
8-
#[derive(Clone, Copy)]
9-
pub enum Storage {
10-
/// The object store is in memory, and no message log is kept.
11-
Memory,
12-
13-
/// The object store is persisted to disk, and a message log is kept.
14-
Disk,
15-
}
16-
17-
/// Internal database config parameters
18-
#[derive(Clone, Copy)]
19-
pub struct Config {
20-
/// Specifies the object storage model.
21-
pub storage: Storage,
22-
/// Specifies the page pool max size in bytes.
23-
pub page_pool_max_size: Option<usize>,
24-
}
25-
26-
/// A message that is processed by the [`spawn_metrics_recorder`] actor.
27-
/// We use a separate task to record metrics to avoid blocking transactions.
28-
pub struct MetricsMessage {
29-
/// The reducer the produced these metrics.
30-
reducer: Option<ReducerName>,
31-
/// Metrics from a mutable transaction.
32-
metrics_for_writer: Option<TxMetrics>,
33-
/// Metrics from a read-only transaction.
34-
/// A message may have metrics for both types of transactions,
35-
/// because metrics for a reducer and its subscription updates are recorded together.
36-
metrics_for_reader: Option<TxMetrics>,
37-
/// The row updates for an immutable transaction.
38-
/// Needed for insert and delete counters.
39-
tx_data: Option<Arc<TxData>>,
40-
/// Cached metrics counters for each workload type.
41-
counters: Arc<EnumMap<WorkloadType, ExecutionCounters>>,
42-
}
43-
44-
/// The handle used to send work to the tx metrics recorder.
45-
#[derive(Clone)]
46-
pub struct MetricsRecorderQueue {
47-
tx: spacetimedb_runtime::sync::mpsc::UnboundedSender<MetricsMessage>,
48-
}
49-
50-
impl MetricsRecorderQueue {
51-
pub fn send_metrics(
52-
&self,
53-
reducer: Option<ReducerName>,
54-
metrics_for_writer: Option<TxMetrics>,
55-
metrics_for_reader: Option<TxMetrics>,
56-
tx_data: Option<Arc<TxData>>,
57-
counters: Arc<EnumMap<WorkloadType, ExecutionCounters>>,
58-
) {
59-
if let Err(err) = self.tx.send(MetricsMessage {
60-
reducer,
61-
metrics_for_writer,
62-
metrics_for_reader,
63-
tx_data,
64-
counters,
65-
}) {
66-
log::warn!("failed to send metrics: {err}");
67-
}
68-
}
69-
}
70-
71-
fn record_metrics(
72-
MetricsMessage {
73-
reducer,
74-
metrics_for_writer,
75-
metrics_for_reader,
76-
tx_data,
77-
counters,
78-
}: MetricsMessage,
79-
) {
80-
if let Some(tx_metrics) = metrics_for_writer {
81-
tx_metrics.report(tx_data.as_deref(), reducer.as_ref(), |wl| &counters[wl]);
82-
}
83-
if let Some(tx_metrics) = metrics_for_reader {
84-
tx_metrics.report(None, reducer.as_ref(), |wl| &counters[wl]);
85-
}
86-
}
87-
88-
const TX_METRICS_RECORDING_INTERVAL: std::time::Duration = std::time::Duration::from_millis(5);
89-
90-
/// Spawns a task for recording transaction metrics.
91-
/// Returns the handle for pushing metrics to the recorder.
92-
pub fn spawn_tx_metrics_recorder(
93-
handle: &spacetimedb_runtime::Handle,
94-
) -> (MetricsRecorderQueue, spacetimedb_runtime::AbortHandle) {
95-
let handle_clone = handle.clone();
96-
let (tx, mut rx) = spacetimedb_runtime::sync::mpsc::unbounded_channel();
97-
let abort_handle = handle
98-
.spawn(async move {
99-
loop {
100-
handle_clone.sleep(TX_METRICS_RECORDING_INTERVAL).await;
101-
while let Ok(metrics) = rx.try_recv() {
102-
record_metrics(metrics);
103-
}
104-
}
105-
})
106-
.abort_handle();
107-
(MetricsRecorderQueue { tx }, abort_handle)
108-
}
1095
use crate::util::asyncify;
6+
use crate::MetricsRecorderQueue;
1107
use anyhow::{anyhow, Context};
1118
use enum_map::EnumMap;
1129
use spacetimedb_commitlog::repo::OnNewSegmentFn;
@@ -1935,6 +1832,7 @@ fn default_row_count_fn(db: Identity) -> RowCountFn {
19351832
pub mod tests_utils {
19361833
use crate::snapshot;
19371834
use crate::snapshot::SnapshotWorker;
1835+
use crate::MetricsRecorderQueue;
19381836

19391837
use super::*;
19401838
use core::ops::Deref;

‎crates/runtime/src/lib.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
#[cfg(all(feature = "tokio", feature = "simulation"))]
2-
compile_error!(
3-
"spacetimedb-runtime requires exactly one runtime backend: enable either `tokio` or `simulation`, not both"
4-
);
5-
1+
//#[cfg(all(feature = "tokio", feature = "simulation"))]
2+
//compile_error!(
3+
// "spacetimedb-runtime requires exactly one runtime backend: enable either `tokio` or `simulation`, not both"
4+
//);
5+
//
66
#[cfg(not(any(feature = "tokio", feature = "simulation")))]
77
compile_error!("spacetimedb-runtime requires exactly one runtime backend: enable either `tokio` or `simulation`");
88

0 commit comments

Comments
 (0)