gpui_kit_grid/store/
queue.rs

1//! The write queue between producers and the ingest: operations wait here
2//! in order until the ingest takes them all. With [`WriteCoalescing::ByKey`]
3//! an upsert of a key whose upsert still waits takes that one's place, so
4//! the queue holds at most one entry per pending key between barriers,
5//! however fast producers write.
6
7use std::{hash::BuildHasher as _, mem, sync::Arc};
8
9use rustc_hash::{FxBuildHasher, FxHashMap};
10
11use super::{ChangeSet, Ingest, Op};
12use crate::{
13    highlight::{ChangeDirection, RowChange, Stamp},
14    model::{ColumnIx, ColumnSet, GridColumns as _, GridRow},
15};
16
17/// How a store queues writes that arrive while its ingest is busy.
18///
19/// For the same writes in a batch, both publish the same rows, change
20/// highlights and their directions, fading removals, pinned rows and
21/// progress, and the same [`ChangeSet`], except that
22/// [`ChangeSet::updated`] names a row once for writes merged into one entry
23/// instead of once per write. They differ in what a stream costs: the
24/// queue's memory and the ingest's work grow with the number of keys
25/// written ([`Self::ByKey`]) or with the number of writes
26/// ([`Self::KeepEveryWrite`]) between two batches. Neither pushes back on
27/// producers: a stream of new keys, or an ingest that falls behind, still
28/// grows the queue.
29#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
30#[non_exhaustive]
31pub enum WriteCoalescing {
32    /// An upsert of a key whose upsert still waits for the ingest takes that
33    /// one's place in the queue, so a stream that outruns the ingest queues
34    /// one entry per key, not per write. The writer diffs the new row
35    /// against the waiting one ([`GridRow::changed_columns`] and
36    /// [`GridRow::change_direction`]), so the ingest highlights each change
37    /// as if it had applied every write. `remove`, `mark_changed`,
38    /// `replace_all`, `clear` and the end of a reload are barriers: an
39    /// upsert after one never merges into a write before it.
40    #[default]
41    ByKey,
42    /// Every write waits in the queue in turn: under a stream faster than
43    /// the ingest the queue grows with the event rate.
44    KeepEveryWrite,
45}
46
47/// One entry of the queue.
48pub(super) enum Queued<R: GridRow> {
49    /// An operation as written.
50    Op(Op<R>),
51    /// Upserts of one key merged into the place of the first: an index into
52    /// [`TakenWrites::merges`].
53    Merged(u32),
54}
55
56/// Upserts of one key merged into one entry: the first as written, the
57/// latest row, and what the writes after the first changed.
58pub(super) struct MergedUpsert<R: GridRow> {
59    first: Arc<R>,
60    /// The first was an `upsert`, not an `upsert_silent`.
61    highlight: bool,
62    latest: Arc<R>,
63    changes: MergedChanges,
64}
65
66/// What the writes after the first of a merged entry changed, each against
67/// the one before it, as the ingest would have found applying them in turn.
68#[derive(Default)]
69struct MergedChanges {
70    /// Columns any of them changed ([`ChangeSet::columns`]).
71    columns: ColumnSet,
72    /// Columns a highlighted one changed: stamped `Updated`.
73    stamped: ColumnSet,
74    /// Of `stamped`, the columns whose last change rose, and fell.
75    up: ColumnSet,
76    down: ColumnSet,
77    /// One of them moved the row to another parent.
78    parent: bool,
79}
80
81impl MergedChanges {
82    /// Records a replacement of `previous` by `row`.
83    fn record<R: GridRow>(&mut self, row: &R, previous: &R, highlight: bool) {
84        let mut columns = ColumnSet::EMPTY;
85        row.changed_columns(previous, &mut columns);
86        self.parent |= row.parent_key() != previous.parent_key();
87        self.columns.union_with(columns);
88        if !highlight {
89            return;
90        }
91        let all = R::Column::all();
92        // Only the row's own columns are stamped (`SparseStamps::written`).
93        let columns = columns.intersection(ColumnSet::first(all.len()));
94        self.stamped.union_with(columns);
95        for column in columns.iter() {
96            let direction = row.change_direction(previous, all[column.get()]);
97            self.up.remove(column);
98            self.down.remove(column);
99            match direction {
100                Some(ChangeDirection::Up) => self.up.insert(column),
101                Some(ChangeDirection::Down) => self.down.insert(column),
102                None => {}
103            }
104        }
105    }
106
107    fn direction(&self, column: ColumnIx) -> Option<ChangeDirection> {
108        if self.up.contains(column) {
109            Some(ChangeDirection::Up)
110        } else if self.down.contains(column) {
111            Some(ChangeDirection::Down)
112        } else {
113            None
114        }
115    }
116}
117
118/// The queue's counters ([`StoreStats`](super::StoreStats)).
119#[derive(Clone, Copy, Debug, Default)]
120pub(super) struct QueueCounters {
121    pub(super) writes: u64,
122    pub(super) coalesced: u64,
123    pub(super) applied: u64,
124    pub(super) pending: usize,
125    pub(super) peak_pending: usize,
126}
127
128/// The operations waiting for the ingest, in order.
129pub(super) struct WriteQueue<R: GridRow> {
130    coalescing: WriteCoalescing,
131    entries: Vec<Queued<R>>,
132    merges: Vec<MergedUpsert<R>>,
133    /// The entry of the latest waiting upsert of each key, by the key's
134    /// hash: the key is compared with the entry's row, so a collision only
135    /// keeps two entries apart. No key is cloned.
136    upserts: FxHashMap<u64, u32>,
137    /// Entries before this one are behind a barrier: nothing merges into
138    /// them.
139    floor: u32,
140    /// Writes in `entries`, merged ones and a reload's included.
141    queued: u64,
142    counters: QueueCounters,
143}
144
145/// The entries the ingest took, with the merged upserts they refer to.
146pub(super) struct TakenWrites<R: GridRow> {
147    entries: Vec<Queued<R>>,
148    merges: Vec<MergedUpsert<R>>,
149}
150
151impl<R: GridRow> WriteQueue<R> {
152    pub(super) fn new(coalescing: WriteCoalescing) -> Self {
153        Self {
154            coalescing,
155            entries: Vec::new(),
156            merges: Vec::new(),
157            upserts: FxHashMap::default(),
158            floor: 0,
159            queued: 0,
160            counters: QueueCounters::default(),
161        }
162    }
163
164    pub(super) fn counters(&self) -> QueueCounters {
165        QueueCounters {
166            pending: self.entries.len(),
167            ..self.counters
168        }
169    }
170
171    /// Counts a write a running reload buffers instead.
172    pub(super) fn count_buffered(&mut self) {
173        self.counters.writes += 1;
174    }
175
176    /// Queues a written operation, merging an upsert into the waiting
177    /// upsert of its key.
178    pub(super) fn push(&mut self, op: Op<R>) {
179        self.counters.writes += 1;
180        self.queued += 1;
181        match op {
182            Op::Upsert(row) => self.push_upsert(row, true),
183            Op::UpsertSilent(row) => self.push_upsert(row, false),
184            Op::Remove(key) => {
185                self.forget(&key);
186                self.push_entry(Op::Remove(key));
187            }
188            Op::MarkChanged(key, columns) => {
189                self.forget(&key);
190                self.push_entry(Op::MarkChanged(key, columns));
191            }
192            op @ (Op::SetPinnedRows { .. } | Op::SetProgress(_)) => self.push_entry(op),
193            op => {
194                self.barrier();
195                self.push_entry(op);
196            }
197        }
198    }
199
200    /// Queues the end of a reload: a barrier, then the reload with the
201    /// operations it buffered (written already, counted now).
202    pub(super) fn push_finished_reload(&mut self, snapshot: Vec<Arc<R>>, live: Vec<Op<R>>) {
203        self.queued += live.len() as u64;
204        self.barrier();
205        self.push_entry(Op::FinishReload { snapshot, live });
206    }
207
208    /// Queues what a cancelled reload buffered, as written: a barrier
209    /// first, so nothing written before merges past them.
210    pub(super) fn push_cancelled_reload(&mut self, live: Vec<Op<R>>) {
211        self.queued += live.len() as u64;
212        self.barrier();
213        self.entries.extend(live.into_iter().map(Queued::Op));
214        self.note_depth();
215    }
216
217    /// Takes every entry for the ingest.
218    pub(super) fn take(&mut self) -> TakenWrites<R> {
219        self.upserts.clear();
220        self.floor = 0;
221        self.counters.applied += mem::take(&mut self.queued);
222        TakenWrites {
223            entries: mem::take(&mut self.entries),
224            merges: mem::take(&mut self.merges),
225        }
226    }
227
228    fn push_entry(&mut self, op: Op<R>) {
229        self.entries.push(Queued::Op(op));
230        self.note_depth();
231    }
232
233    fn note_depth(&mut self) {
234        self.counters.peak_pending = self.counters.peak_pending.max(self.entries.len());
235    }
236
237    /// Nothing written so far merges with what comes next.
238    fn barrier(&mut self) {
239        self.floor = self.entries.len() as u32;
240    }
241
242    /// No later upsert of `key` merges into an earlier one.
243    fn forget(&mut self, key: &R::Key) {
244        if self.coalescing == WriteCoalescing::ByKey {
245            self.upserts.remove(&FxBuildHasher.hash_one(key));
246        }
247    }
248
249    fn push_upsert(&mut self, row: Arc<R>, highlight: bool) {
250        let hash = match self.coalescing {
251            WriteCoalescing::ByKey => FxBuildHasher.hash_one(row.key()),
252            WriteCoalescing::KeepEveryWrite => {
253                self.push_entry(upsert(row, highlight));
254                return;
255            }
256        };
257        let row = match self.upserts.get(&hash) {
258            Some(&ix) if ix >= self.floor => match self.merge(ix as usize, row, highlight) {
259                Ok(()) => {
260                    self.counters.coalesced += 1;
261                    return;
262                }
263                Err(row) => row,
264            },
265            _ => row,
266        };
267        self.upserts.insert(hash, self.entries.len() as u32);
268        self.push_entry(upsert(row, highlight));
269    }
270
271    /// Merges `row` into the upsert at `ix` if it is one of the same key;
272    /// gives the row back otherwise.
273    fn merge(&mut self, ix: usize, row: Arc<R>, highlight: bool) -> Result<(), Arc<R>> {
274        let (first, first_highlight) = match &self.entries[ix] {
275            Queued::Merged(merge_ix) => {
276                let merge = &mut self.merges[*merge_ix as usize];
277                if merge.latest.key() != row.key() {
278                    return Err(row);
279                }
280                // The same row again changes nothing, as in the ingest.
281                if !Arc::ptr_eq(&merge.latest, &row) {
282                    merge.changes.record(&*row, &*merge.latest, highlight);
283                    merge.latest = row;
284                }
285                return Ok(());
286            }
287            Queued::Op(Op::Upsert(first)) => (first, true),
288            Queued::Op(Op::UpsertSilent(first)) => (first, false),
289            Queued::Op(_) => return Err(row),
290        };
291        if first.key() != row.key() {
292            return Err(row);
293        }
294        if Arc::ptr_eq(first, &row) {
295            return Ok(());
296        }
297        let mut changes = MergedChanges::default();
298        changes.record(&*row, &**first, highlight);
299        let first = first.clone();
300        self.entries[ix] = Queued::Merged(self.merges.len() as u32);
301        self.merges.push(MergedUpsert {
302            first,
303            highlight: first_highlight,
304            latest: row,
305            changes,
306        });
307        Ok(())
308    }
309}
310
311fn upsert<R: GridRow>(row: Arc<R>, highlight: bool) -> Op<R> {
312    if highlight {
313        Op::Upsert(row)
314    } else {
315        Op::UpsertSilent(row)
316    }
317}
318
319impl<R: GridRow> Ingest<R> {
320    /// Applies the entries the ingest took, in order: each merged upsert in
321    /// the place of its first write.
322    pub(super) fn apply_writes(
323        &mut self,
324        writes: TakenWrites<R>,
325        now: web_time::Instant,
326    ) -> ChangeSet {
327        let now_ms = self.now_ms(now);
328        let mut changes = ChangeSet::default();
329        let TakenWrites { entries, merges } = writes;
330        for entry in entries {
331            match entry {
332                Queued::Op(op) => self.apply_op(op, now_ms, &mut changes),
333                Queued::Merged(ix) => self.apply_merged(&merges[ix as usize], now_ms, &mut changes),
334            }
335        }
336        self.finish_batch(now_ms, changes)
337    }
338
339    /// The first write as written, against the stored row or as an insert;
340    /// then the latest row in its place with what the later writes changed,
341    /// each against the one before: the stamps, directions and changes of
342    /// applying them in turn.
343    fn apply_merged(&mut self, merge: &MergedUpsert<R>, now_ms: u32, changes: &mut ChangeSet) {
344        self.upsert(merge.first.clone(), now_ms, merge.highlight, changes);
345        let Some(&id) = self.index.get(merge.latest.key()) else {
346            return;
347        };
348        let ix = id.ix();
349        self.slot_mut(ix).row = Some(merge.latest.clone());
350        let merged = &merge.changes;
351        changes.parent |= merged.parent;
352        changes.updated.push(id);
353        changes.columns.union_with(merged.columns);
354        if !merged.stamped.is_empty() {
355            self.stamp(ix, merged.stamped, now_ms, |column| {
356                Stamp::with_direction(now_ms, RowChange::Updated, merged.direction(column))
357            });
358        }
359    }
360}
gpui_kit_grid/store/queue.rs

gpui_kit_grid/store/
queue.rs
