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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
// Copyright 2023 Developers of the reconcile project.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::hash::Hash;
use crate::bounds::{Key, Value};
use crate::clock::Timestamp;
use crate::entry::Entry;
use super::ReplicatedMap;
impl<K: Key + Hash, V: Value> ReplicatedMap<K, V> {
/// Mutate the value for `k` in place, then propagate like [`insert`](ReplicatedMap::insert).
///
/// The callback sees `Some(&mut V)` for a live key, `None` for an absent or tombstoned one; a
/// mutated entry is re-stamped and broadcast. Holds the write lock for the whole
/// read-modify-write, so it is atomic against the reconciliation loop.
///
/// # Deadlock
///
/// `callback` runs while the map **write** lock is held. Calling any read or write method
/// (`get`, `insert`, `for_each`, another `get_mut`, …) from `callback` self-deadlocks — see
/// [`get`](Self::get)'s `# Deadlock` section.
///
/// # Panics
///
/// See [`insert`](Self::insert) — the broadcast requires an ambient Tokio runtime (only when
/// the callback mutates a live entry).
pub fn get_mut<F: FnOnce(Option<&mut V>)>(&self, k: &K, callback: F) {
// Mint the timestamp before taking the map lock, matching the lock order of `insert`
// (clock, then map → projection).
let now = self.engine.clock_now();
let mut updated: Option<Entry<Timestamp, V>> = None;
let mut guard = self.engine.map.write();
guard.with_mut(k, |maybe_entry| {
if let Some(entry) = maybe_entry {
callback(entry.value_mut());
entry.stamp = now;
updated = Some(entry.clone());
} else {
callback(None);
}
});
// The mutation bypassed `insert`: refresh the projection (lock order map → projection).
if let Some(entry) = guard.get(k) {
let projected = entry.project();
self.engine.projection.write().insert(k.clone(), projected);
}
drop(guard);
if let Some(value) = updated {
self.engine.broadcast_update(k.clone(), value);
}
}
/// Mutate `k` in place **only when live**, re-stamping and broadcasting; returns whether it
/// was. Atomic against the reconciliation loop.
///
/// The shared core of [`update`](Self::update) and [`upsert`](Self::upsert).
fn mutate_live<F: FnOnce(&mut V)>(&self, k: &K, callback: F) -> bool {
// Mint the timestamp before taking the map lock, matching the lock order of `insert`.
let now = self.engine.clock_now();
let mut updated: Option<Entry<Timestamp, V>> = None;
let mut guard = self.engine.map.write();
guard.with_mut(k, |maybe_entry| {
if let Some(entry) = maybe_entry {
if let Some(value) = entry.value_mut() {
callback(value);
entry.stamp = now;
updated = Some(entry.clone());
}
}
});
if updated.is_some() {
if let Some(entry) = guard.get(k) {
let projected = entry.project();
self.engine.projection.write().insert(k.clone(), projected);
}
}
drop(guard);
if let Some(value) = updated {
self.engine.broadcast_update(k.clone(), value);
true
} else {
false
}
}
/// Atomically mutate the live value for `k`, then re-stamp and broadcast; returns whether the
/// key was live. The race-free replacement for a `get`-then-`insert`.
///
/// # Deadlock
///
/// `f` runs while the map write lock is held — same hazard as
/// [`get_mut`](Self::get_mut)'s `# Deadlock` section.
///
/// # Panics
///
/// See [`insert`](Self::insert) — the broadcast requires an ambient Tokio runtime (only when
/// `k` is live).
///
/// ```
/// # use std::sync::Arc;
/// use reconcile::{replicated_map::Config, InMemoryNetwork, ReplicatedMap};
///
/// # #[tokio::main]
/// # async fn main() {
/// let network = InMemoryNetwork::new();
/// let transport = Arc::new(network.bind("127.0.0.1:8305".parse().unwrap()));
/// let store = ReplicatedMap::<String, i32>::new_with_transport(
/// Config::default().with_insecure_no_key(),
/// transport,
/// );
///
/// // Absent: no race-free `get`-then-`insert` needed, `update` just reports it and does nothing.
/// assert!(!store.update(&"a".to_string(), |v| *v += 1));
///
/// store.insert("a".to_string(), 1);
/// assert!(store.update(&"a".to_string(), |v| *v += 1)); // atomic against a concurrent writer
/// assert_eq!(store.get_cloned(&"a".to_string()), Some(2));
/// # }
/// ```
#[must_use]
pub fn update<F: FnOnce(&mut V)>(&self, k: &K, f: F) -> bool {
self.mutate_live(k, f)
}
/// Update the live value for `k` with `f`, or insert `default` if it is absent or tombstoned.
///
/// The update branch is atomic; the insert branch behaves like [`insert`](Self::insert).
///
/// # Deadlock
///
/// `f` runs while the map write lock is held on the update branch — same hazard as
/// [`get_mut`](Self::get_mut)'s `# Deadlock` section.
///
/// # Panics
///
/// See [`insert`](Self::insert) — the broadcast requires an ambient Tokio runtime.
///
/// ```
/// # use std::sync::Arc;
/// use reconcile::{replicated_map::Config, InMemoryNetwork, ReplicatedMap};
///
/// # #[tokio::main]
/// # async fn main() {
/// let network = InMemoryNetwork::new();
/// let transport = Arc::new(network.bind("127.0.0.1:8306".parse().unwrap()));
/// let store = ReplicatedMap::<String, i32>::new_with_transport(
/// Config::default().with_insecure_no_key(),
/// transport,
/// );
///
/// // Absent: the default is inserted as-is, `f` never runs.
/// store.upsert("a".to_string(), 1, |v| *v += 100);
/// assert_eq!(store.get_cloned(&"a".to_string()), Some(1));
///
/// // Live: `f` runs against the existing value, `default` is discarded.
/// store.upsert("a".to_string(), 1, |v| *v += 100);
/// assert_eq!(store.get_cloned(&"a".to_string()), Some(101));
/// # }
/// ```
pub fn upsert<F: FnOnce(&mut V)>(&self, k: K, default: V, f: F) {
if !self.mutate_live(&k, f) {
self.insert(k, default);
}
}
/// Return the live value for `k`, inserting (and broadcasting) `f()` first if it is
/// absent/tombstoned. Under last-write-wins, two nodes racing to insert converge by timestamp
/// order; this node returns the value it observed/created.
///
/// # Panics
///
/// See [`insert`](Self::insert) — the broadcast requires an ambient Tokio runtime (only when
/// `k` is absent/tombstoned).
///
/// ```
/// # use std::sync::Arc;
/// use reconcile::{replicated_map::Config, InMemoryNetwork, ReplicatedMap};
///
/// # #[tokio::main]
/// # async fn main() {
/// let network = InMemoryNetwork::new();
/// let transport = Arc::new(network.bind("127.0.0.1:8307".parse().unwrap()));
/// let store = ReplicatedMap::<String, i32>::new_with_transport(
/// Config::default().with_insecure_no_key(),
/// transport,
/// );
///
/// // Absent: `f` runs, its result is both inserted and returned.
/// assert_eq!(store.get_or_insert_with(&"a".to_string(), || 1), 1);
/// // Live: `f` never runs, the existing value is returned instead.
/// assert_eq!(store.get_or_insert_with(&"a".to_string(), || 999), 1);
/// # }
/// ```
pub fn get_or_insert_with<F: FnOnce() -> V>(&self, k: &K, f: F) -> V {
if let Some(value) = self.get(k) {
return value.clone();
}
let value = f();
self.insert(k.clone(), value.clone());
value
}
}