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
106
107
108
109
110
|
//! The worker that keeps the document file (`document::file`) up to date: it looks at
//! the node (ready, the document's epoch and heads, this node's actors), and the file
//! says what is due. While a ready node rejoins, the file follows the document before
//! the rejoin, which takes the node's writes meanwhile (`state::StaleView`): a restart
//! then resumes on it, and rejoins again.
use std::sync::Arc;
use crate::document::Document;
use crate::document::file::Due;
use crate::document::replica::Epoch;
use crate::locks::RwLockExt;
use crate::state::AppState;
/// Keeps the document file up to date. Runs for as long as the node does.
pub(crate) async fn run(state: AppState) {
let Some(file) = state.document_file.clone() else {
return;
};
let mut tick = tokio::time::interval(file.append_every());
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tick.tick().await;
flush(&state).await;
}
}
/// Writes what is due now (`DocumentFile::due`). Also called at shutdown.
pub(crate) async fn flush(state: &AppState) {
let Some(file) = state.document_file.clone() else {
return;
};
if state.remake.is_rejoining() {
if let Some((epoch, stale)) = state.stale_document() {
flush_before_rejoin(&file, epoch, stale).await;
}
return;
}
if !state.is_ready().await {
return;
}
let own = state.remake.own_actors();
let (epoch, heads) = {
let replica = state.doc.read().await;
(replica.epoch, replica.heads().to_vec())
};
match file.due(epoch, own.len(), &heads) {
Due::Rewrite(reason) => {
let (epoch, copy) = {
let replica = state.doc.read().await;
(replica.epoch, replica.doc.clone())
};
file.rewrite(epoch, copy, own, reason).await;
}
Due::Append { epoch, since } => {
let (heads, increment) = {
let replica = state.doc.read().await;
// Switched since: the next flush rewrites the file instead.
if replica.epoch != epoch {
return;
}
let heads = replica.heads().to_vec();
(heads, crate::document::save_since(&replica.doc, &since))
};
file.append(since, heads, increment).await;
}
Due::Nothing => {}
}
}
/// [`flush`] of the document before a rejoin (`epoch`'s), while it takes this node's
/// writes. Only appends: a rewrite (another epoch, or the increments outgrew the full
/// save) waits for the end of the rejoin, so the file keeps the actors it had.
async fn flush_before_rejoin(
file: &Arc<crate::document::file::DocumentFile>,
epoch: Epoch,
stale: Arc<std::sync::RwLock<Document>>,
) {
let heads = {
let stale = Arc::clone(&stale);
// Sorted, as the live document's (`Replica::heads`) the file compares with.
tokio::task::spawn_blocking(move || {
let mut heads = stale.read_or_recover().get_heads();
heads.sort_unstable();
heads
})
.await
.expect("document read task panicked")
};
let Some(actors) = file.actors() else {
return;
};
if let Due::Append {
epoch: written,
since,
} = file.due(epoch, actors, &heads)
&& written == epoch
{
let since_then = since.clone();
let (heads, increment) = tokio::task::spawn_blocking(move || {
let doc = stale.read_or_recover();
let mut heads = doc.get_heads();
heads.sort_unstable();
(heads, crate::document::save_since(&doc, &since_then))
})
.await
.expect("document read task panicked");
file.append(since, heads, increment).await;
}
}
|
