Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
7.4 kB · 195 lines
Rust
at sl/gitmirror
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196use std::num::NonZeroUsize;use std::sync::Arc;use std::sync::atomic::{AtomicU64, Ordering};use std::time::Duration;
use bobbin_edge_index::{ClaimIndex, CoverageWatch, EdgeStore, StreamHealth};use bobbin_ingest::{ DEFAULT_IDLE_PROMOTE_MIN_EVENTS, DEFAULT_INGEST_PARALLELISM, DisconnectSink, IngestConfig, IngestRuntime, RepoIdResolver, WarmingBuffer, WarmingShadowBuffer, run as run_ingest,};use bobbin_record_lru::NoopRecordStore;use bobbin_resolver::IdentityResolver;use bobbin_runtime::{ Clock, DEFAULT_MEM_WS_CAPACITY, MemHttpTransport, MemWsTransport, RuntimeHasher, SeededEntropy, SimClock, UnixMicros,};use bobbin_slingshot_client::SlingshotClient;use bobbin_types::search::NoopSearchSink;use tokio_util::sync::CancellationToken;use url::Url;
use crate::report::{SimOutcome, SimReport};use crate::workload::{Workload, WorkloadCtx};
#[derive(Clone, Debug)]pub struct SimConfig { pub seed: u64, pub base_unix: UnixMicros, pub max_virtual_runtime: Duration, pub parallelism: NonZeroUsize, pub hydrant_base: Url, pub slingshot_base: Url, pub mem_ws_capacity: usize, pub warming_buffer_enabled: bool,}
impl SimConfig { pub fn new(seed: u64) -> Self { Self { seed, base_unix: UnixMicros::new(1_700_000_000_000_000), max_virtual_runtime: Duration::from_secs(60), parallelism: DEFAULT_INGEST_PARALLELISM, hydrant_base: Url::parse("ws://hydrant.sim/").unwrap(), slingshot_base: Url::parse("http://slingshot.sim/").unwrap(), mem_ws_capacity: DEFAULT_MEM_WS_CAPACITY, warming_buffer_enabled: true, } }}
pub struct Sim { config: SimConfig, workload: Box<dyn Workload>,}
impl Sim { pub fn new(config: SimConfig, workload: Box<dyn Workload>) -> Self { Self { config, workload } }
pub async fn run(self) -> SimReport { let SimConfig { seed, base_unix, max_virtual_runtime, parallelism, hydrant_base, slingshot_base, mem_ws_capacity, warming_buffer_enabled, } = self.config;
let entropy = Arc::new(SeededEntropy::new(seed)); let hasher = RuntimeHasher::from_entropy(&*entropy); let clock: Arc<dyn Clock> = Arc::new(SimClock::at(base_unix));
let store = Arc::new(EdgeStore::new(hasher.clone())); let coverage = Arc::new(CoverageWatch::new()); let records = Arc::new(NoopRecordStore); let cancel = CancellationToken::new(); let consumer_too_slow_count = Arc::new(AtomicU64::new(0)); let disconnects = Arc::new(DisconnectSink::new()); let warming_shadow = Arc::new(WarmingShadowBuffer::new(hasher.clone())); let warming_buffer = Arc::new(WarmingBuffer::new(hasher.clone()));
let workload_name = self.workload.name(); let ctx = WorkloadCtx { seed, clock: clock.clone(), entropy: entropy.clone(), store: store.clone(), coverage: coverage.clone(), records: records.clone(), cancel: cancel.clone(), consumer_too_slow_count: consumer_too_slow_count.clone(), }; let hooks = self.workload.build(ctx);
let slingshot_http = MemHttpTransport::shared(hooks.slingshot.clone(), clock.clone()); let slingshot_client = SlingshotClient::new(slingshot_base, slingshot_http) .expect("slingshot base url is valid"); let resolver = Arc::new(RepoIdResolver::with_slingshot( slingshot_client, clock.clone(), hasher.clone(), ));
let mem_ws = MemWsTransport::shared_with_capacity(hooks.hydrant.clone(), mem_ws_capacity);
let ingest_runtime: IngestRuntime<NoopSearchSink> = IngestRuntime { store: store.clone(), claims: Arc::new(ClaimIndex::new()), coverage: coverage.clone(), search: Arc::new(NoopSearchSink), records: records.clone() as Arc<dyn bobbin_record_lru::RecordStore>, resolver: resolver.clone(), identity: Arc::new(IdentityResolver::detached(hasher.clone())), clock: clock.clone(), entropy: entropy.clone(), stream_health: Arc::new(StreamHealth::new()), ws: mem_ws, hydrant: None, cancel: cancel.clone(), disconnects: Some(disconnects.clone()), warming_shadow: Some(warming_shadow.clone()), warming_buffer: warming_buffer_enabled.then(|| warming_buffer.clone()), settlements: None, analytics: None, }; let ingest_config = IngestConfig { hydrant_base, start_cursor: bobbin_edge_index::HydrantCursor::new(0), parallelism, idle_promote_min_events: DEFAULT_IDLE_PROMOTE_MIN_EVENTS, }; let mut ingest_handle = tokio::spawn(async move { let _ = run_ingest(ingest_config, ingest_runtime).await; });
let script = hooks.script; let initial_report = tokio::select! { biased; _ = clock.sleep(max_virtual_runtime) => SimReport { workload: workload_name, seed, outcome: SimOutcome::TimedOut, virtual_runtime: max_virtual_runtime, virtual_clock_end: clock.now_unix_micros(), events_processed: coverage.snapshot().events_processed(), last_cursor: coverage.snapshot().last_cursor().raw(), edge_count: store.key_count() as u64, resolver_hits: 0, resolver_misses: 0, consumer_too_slow_count: consumer_too_slow_count.load(Ordering::Relaxed), disconnect_count: disconnects.count(), last_disconnect: disconnects.snapshot(), warming_shadow: warming_shadow.snapshot(), warming_buffer: warming_buffer.snapshot(), failure_reason: Some(format!( "max_virtual_runtime {max_virtual_runtime:?} exhausted", )), }, r = script => r, };
cancel.cancel(); let drain_deadline = clock.sleep(Duration::from_secs(30)); tokio::pin!(drain_deadline); tokio::select! { _ = &mut drain_deadline => { ingest_handle.abort(); let _ = ingest_handle.await; } res = &mut ingest_handle => { let _ = res; } }
let stats = resolver.stats(); SimReport { edge_count: store.key_count() as u64, events_processed: coverage.snapshot().events_processed(), last_cursor: coverage.snapshot().last_cursor().raw(), resolver_hits: stats.hits, resolver_misses: stats.miss_count(), consumer_too_slow_count: consumer_too_slow_count.load(Ordering::Relaxed), disconnect_count: disconnects.count(), last_disconnect: disconnects.snapshot(), warming_shadow: warming_shadow.snapshot(), warming_buffer: warming_buffer.snapshot(), ..initial_report } }}