rlink 0.6.16

High performance Stream Processing Framework
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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

use crate::core::cluster::MetadataStorageType;
use crate::core::runtime::{ClusterDescriptor, ManagerStatus};
use crate::runtime::HeartbeatItem;
use crate::storage::metadata::mem_metadata_storage::MemoryMetadataStorage;

pub mod mem_metadata_storage;

pub mod metadata_loader;
pub(crate) use metadata_loader::MetadataLoader;

pub trait TMetadataStorage {
    /// save metadata to storage
    fn save(&mut self, metadata: ClusterDescriptor) -> anyhow::Result<()>;

    /// load metadata from storage
    fn load(&self) -> anyhow::Result<ClusterDescriptor>;

    /// update coordinator status change
    fn update_coordinator_status(&self, status: ManagerStatus) -> anyhow::Result<()>;

    /// update worker status change, return the coordinator's status
    fn update_worker_status(
        &self,
        task_manager_id: String,
        heartbeat_items: Vec<HeartbeatItem>,
        worker_manager_status: ManagerStatus,
    ) -> anyhow::Result<ManagerStatus>;
}

#[derive(Clone)]
pub enum MetadataStorage {
    MemoryMetadataStorage(MemoryMetadataStorage),
}

impl MetadataStorage {
    pub fn new(mode: &MetadataStorageType) -> Self {
        match mode {
            MetadataStorageType::Memory => {
                let storage = MemoryMetadataStorage::new();
                MetadataStorage::MemoryMetadataStorage(storage)
            }
        }
    }
}

impl TMetadataStorage for MetadataStorage {
    fn save(&mut self, metadata: ClusterDescriptor) -> anyhow::Result<()> {
        match self {
            MetadataStorage::MemoryMetadataStorage(storage) => storage.save(metadata),
        }
    }

    fn load(&self) -> anyhow::Result<ClusterDescriptor> {
        match self {
            MetadataStorage::MemoryMetadataStorage(storage) => storage.load(),
        }
    }

    fn update_coordinator_status(&self, status: ManagerStatus) -> anyhow::Result<()> {
        match self {
            MetadataStorage::MemoryMetadataStorage(storage) => {
                storage.update_coordinator_status(status)
            }
        }
    }

    fn update_worker_status(
        &self,
        task_manager_id: String,
        heartbeat_items: Vec<HeartbeatItem>,
        status: ManagerStatus,
    ) -> anyhow::Result<ManagerStatus> {
        match self {
            MetadataStorage::MemoryMetadataStorage(storage) => {
                storage.update_worker_status(task_manager_id, heartbeat_items, status)
            }
        }
    }
}

pub(crate) fn loop_read_cluster_descriptor(
    metadata_storage: &MetadataStorage,
) -> ClusterDescriptor {
    loop_fn!(metadata_storage.load(), std::time::Duration::from_secs(2))
}

pub(crate) fn loop_save_cluster_descriptor(
    metadata_storage: &mut MetadataStorage,
    cluster_descriptor: ClusterDescriptor,
) {
    loop_fn!(
        metadata_storage.save(cluster_descriptor.clone()),
        std::time::Duration::from_secs(2)
    );
}

pub(crate) fn loop_update_application_status(
    metadata_storage: &mut MetadataStorage,
    coordinator_status: ManagerStatus,
) {
    loop_fn!(
        metadata_storage.update_coordinator_status(coordinator_status.clone()),
        std::time::Duration::from_secs(2)
    );
}