chronon-runtime 0.1.0

Chronon runtime assembly and ChrononBuilder
Documentation 
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

//! Embedded deployment: coordinator tick + worker in one process.

use std::sync::Arc;

use chronon_executor::Executor;
use chronon_scheduler::{run_coordinator_tick_loop, Scheduler};
use chronon_telemetry::TelemetrySink;
use tokio::sync::Notify;

use crate::worker::run_worker_loop;

/// Run coordinator tick and worker loops in one process until `shutdown` is notified.
///
/// Called from [`Chronon::run`](crate::Chronon::run) for [`DeploymentShape::Embedded`](crate::DeploymentShape::Embedded).
pub async fn run_embedded_loops(
    scheduler: Arc<Scheduler>,
    executor: Arc<Executor>,
    telemetry: Arc<dyn TelemetrySink>,
    shutdown: Arc<Notify>,
) {
    scheduler.init_partitions().await;

    let store = scheduler.store();
    let instance_id = scheduler.instance_id().to_string();
    let assigner = scheduler.assigner();
    let pool = chronon_scheduler::worker_pool_from_env();

    let tick_shutdown = Arc::clone(&shutdown);
    let tick_store = Arc::clone(&store);
    let tick_telemetry = Arc::clone(&telemetry);
    let tick_instance = instance_id.clone();
    let tick_assigner = Arc::clone(&assigner);
    tokio::spawn(async move {
        run_coordinator_tick_loop(
            tick_store,
            tick_telemetry,
            tick_instance,
            tick_assigner,
            tick_shutdown,
        )
        .await;
    });

    run_worker_loop(
        store,
        executor,
        telemetry,
        pool,
        instance_id,
        shutdown,
    )
    .await;
}