diff --git a/doc/developer/design/20261105_rendezvous_protocol/README.md b/doc/developer/design/20261105_rendezvous_protocol/README.md new file mode 100644 index 0000000000000..85beead2fc21e --- /dev/null +++ b/doc/developer/design/20261105_rendezvous_protocol/README.md @@ -0,0 +1,175 @@ +# Rendezvous protocol model + +`RendezvousProtocol.lean` is a Veil model of the rendezvous publication primitive, the shared-trace layer on top of it, and the compute lifecycle that drives both across the maintenance and interactive runtimes, reconciliation included. +It checks three rules that together let the registry keep identifying an incarnation by `(GlobalId, as_of)`: a publication point accepts one writer in its life, the registry never hands out a slot whose writer closed or dropped, and the two runtimes synchronize their reconciliations with a barrier. +The safety clauses state that an import reads one writer's chain without missing updates, and that every reader is bound to its own incarnation. +SMT proves the clauses inductive for all instance sizes, and an explicit-state check confirms them on a small instance. +Deleting each guard the design relies on in turn reaches a violation, except the single-attach refusal, a backstop that is inert in the model, and the physical-meet bound, which needs a larger instance than the committed one (see the mutation table). + +## Running + +Veil pins its own Lean toolchain (`leanprover/lean4:v4.32.0`) and a no-Mathlib `lean-smt`, which differ from the toolchain and dependencies of `misc/lean`. +No CI job builds this file, so it can drift from the code it models. +The checked run used Veil at commit `0a20c4c7154bdb020bda4404e5051e4298a06d92`. + +```sh +git clone https://github.com/verse-lab/veil ~/dev/repos/veil +curl -sSfL https://raw.githubusercontent.com/leanprover/elan/master/elan-init.sh | sh -s -- -y --default-toolchain none +cd ~/dev/repos/veil && lake build +mkdir -p Examples/Materialize +cp /RendezvousProtocol.lean Examples/Materialize/ +lake build Examples.Materialize.RendezvousProtocol +``` + +`lake env lean` cannot load cvc5's native library, so the file has to be built as a module of the Veil project. +The SMT check and the explicit-state check run in one build of about seven minutes, and either can be commented out to run the other alone. +SMT carries the proof for all sizes, and the explicit-state run is only a cross-check. +Its cost is elaboration and compilation of the checker, not exploration: alone, the committed instance builds in 4 min 52 s at 4.06 GB peak RSS, and the same instance with two epochs, 7,011 states, still takes 4 min 13 s at 3.99 GB, measured with other builds running. +A mutation run, which keeps only the safety clauses, takes about 3 min at under 3 GB. + +A mutation is checked on a copy of the file. +The design switches (`barrier`, `barrier_first`, `epoch_match`, `single_attach`, `fresh_slot`) are flipped in both the `assumption [design]` line and the `#model_check` theory. +The other mutations delete the named guard. +The explicit-state checker stops at the first violated clause, and a supporting invariant usually breaks first, so a mutation run deletes every `invariant` line and keeps the `safety` lines. +To see a specific clause fail, delete the other `safety` lines too. + +## What is modeled + +The model tracks one index id with one worker per runtime and only the `oks` arrangement. +A `writer` is one incarnation of the index on the maintenance runtime, a `point` is one `Rendezvous`, an `imp` is one interactive import with its subscription, its reader, and the trace it hands downstream, and an `epoch` is one controller connection. +The collection itself is abstract: its content at each time is identical across incarnations, so an incarnation differs only in its `upper`, its span boundaries, and its compaction frontiers. +A downstream read through the import's acknowledged frontier `f` is correct when the chain its trace reads covers `f`, has a span boundary at `f`, delivered every item without a gap, and is not logically compacted past the read's `as_of`. +Holds are one registration per holder. +The importer's queue is the importer's own view of the point, so every precondition the importer acts on reads its queue and never the live point. + +Reconciliation is modeled per runtime as four phases. +A runtime that sees a newer connection while serving starts reconciling it, possibly skipping connections in between. +In the drop phase it drops every incarnation, peer bundle, and import its history replaces, then arrives at the barrier. +It waits until the other runtime arrived at the same connection or a later one, creates the connection's incarnation, peer bundle, and imports, and serves. +The model also lets a runtime waiting at the barrier begin a newer connection, which the code cannot do, since a waiting worker reads no command. +This over-approximates the code's behavior, so the safety results carry over, but the model says nothing about how the code's wait ends. +Each connection's history installs one incarnation, `goal e`, which either keeps the previous connection's or names one never seen before. +A `stray` holder may clone a slot's `Arc` at any time and drop it later, standing for a peek in flight or any other reference the lifecycle does not account for. + +Abstractions to keep in mind when reading the results: + +* Items are reduced to the frontier through which a subscriber received them, plus a flag for a gap. +* The downstream consumer of an import's trace is folded into the import: it reads through the acknowledged frontier and holds physically there. +* Every incarnation has the same `as_of` in the explicit-state check, the worst case for the registry key. The SMT check leaves `as_of` free. +* Both runtimes decide their drops from the same history of a connection. A connection only one runtime finishes reconciling, as the design section describes, is not modeled. +* Liveness is not checked, so neither is the claim that the barrier wait ends. The design section argues it. +* `errs`, the logging dataflow's extra points, and the publisher's reserved peer holds are left out. The peer's hold is registered when it binds. +* A live writer never publishes the empty frontier: `m_seal` requires a time below `top`, and `no_empty_upper` and `no_empty_queue` state that no point and no queue ever show it. + The code allows it. A writer whose collection is complete, such as an index over a constant collection, seals its `upper` to the empty frontier while it lives, and the import completes when it receives it. + What holds in the code is the narrower rule that a closed writer never publishes the empty frontier, see `Attachment::close`. +* Each writer backs one point, which it attaches fresh, with an empty chain compacted to its `as_of`. + In the code a trace can back several points, and attach to a further point with a non-empty chain and advanced frontiers: an index that re-exports another index's arrangement publishes the other index's trace under its own id. + Such a re-export point lives as long as the backing trace, not as long as the re-exporting index, so it does not freeze when that index drops, and the fresh-slot rule does not apply to it. + The model's `closed_dropped` invariant, which equates a dropped incarnation with a frozen point, does not hold for these points. + A recreated re-export gets a fresh slot only because reconciliation drops every holder of the old slot first. A `stray` holder of it, which the model allows, would hand the old, unfrozen slot to the next incarnation. +* Registering an import's reader and subscribing it are one action, `x_create`. + In the code they take separate acquisitions of the point's lock, and the writer can publish and merge between them. The import's physical hold, taken at registration, keeps the chain's boundaries in that window. +* The writer reads the holds' meet, publishes it and applies it to its trace in one action, `m_sync`. In the code the point advertises the new `since` before the trace applies it, a window that refuses readers the trace could still serve. + +## Lifecycle facts + +The facts below are about code the later changes of this stack add. They name the code by symbol. + +* Both runtimes receive every non-peek command, each at its own rate (`Multiplexer::send`). +* The transport serves one controller connection at a time and finishes the previous one before it builds the client of the next (`mz_service::transport::serve`). + The process records each connection when the multiplexing client forwards its `Hello`, before either runtime receives a command of it (`ConnectionOrder::record`, called from clusterd's `ProcessComputeClient::send`), so the recorded order is the order the process accepted connections in. +* A connection's position comes from one sequencer per runtime and travels with its commands. + The controller sends `Hello` and `UpdateConfiguration` to every process and every other command to process 0 only (`mz_compute_client::service`), so a process other than 0 can record a connection after its workers received the connection's commands, or never. + Global worker 0 of each runtime runs in process 0 and looks the position up in that process's order when it sequences the connection's first command (`mz_compute::command_channel::render`), and the command channel delivers it with every command to every worker of the runtime (`UnifiedCommand::Compute`). + Both runtimes' worker 0 read the one order of process 0, so the two workers of a pair agree on every position, and a worker arrives at the barrier with the delivered position and never looks it up (`Worker::await_reconcile_barrier`). + The model hands a runtime the connection it begins to reconcile with its place in the order, as an action parameter of `m_begin` and `i_begin`, which is this delivery, so the state space does not change. +* Each runtime reconciles on its own (`Worker::reconcile`). + It discards peeks and reduces the command history before it compares it with the new connection's commands, so two runtimes with the same history make the same decisions. +* Reconciliation runs the compactions and drops of existing dataflows, then waits at the barrier, then runs the rest (`Worker::reconcile`). +* The drop phase does not wait on another worker. + An `AllowCompaction` moves this worker's trace holds or drops the collection, and `drop_collection` drops the dataflow with `drop_dataflow`. + Timely's `drop_dataflow` removes the dataflow from this worker's map and drops it (`src/worker.rs:739` in timely 0.32.0), a dropped capability updates a worker-local change batch (`src/dataflow/operators/capability.rs:172`), and a dropped channel pushes its id onto a worker-local queue (`src/allocator/canary.rs:19` in timely_communication 0.32.0). +* A worker waiting at the barrier steps its Timely worker and reads no command (`Worker::await_reconcile_barrier`). + The workers of its runtime that already passed are therefore not held up, and since passing is the only way out, every worker of a runtime executes the same commands in the same order. +* The barrier admits a runtime once the paired worker of the other runtime arrived at the same connection or a later one (`ReconcileBarrier::passable`). The design section says why a later one counts. +* The barrier is per worker pair, as the registry is (`ArrangementSharingRegistry::per_worker`). A slot is only ever touched by the two workers of its pair. +* The maintenance runtime publishes an index into `get_or_create(id, as_of)` (`ArrangementSharingRegistry::publish`), and the interactive runtime binds its peer bundle to the same call (`ArrangementSharingRegistry::peer_bundle`). +* `get_or_create` returns the slot in the map when it is alive, its `as_of` is equal, and it is not frozen, and replaces the map entry otherwise. + A slot is frozen once its traces closed or dropped. +* A slot is alive while the publisher's token, a peer bundle, or an importing dataflow's tokens hold it (`PublishToken`). +* `Rendezvous::attach` panics on a point that ever had a writer, and a point freezes when its writer closes or drops (`Attachment::close`, `Attachment`'s `Drop`). +* `drop_dataflow` drops an index's writer, and the trace box closes its attachments so the empty seal stays out of the point (`TraceBox::close`). +* The import's trace reads whatever chain its point currently publishes (`SharedReader`). With one writer per point, that is the chain the import subscribed to. + +Whether the controller ever chooses an equal `as_of` for a replaced incarnation is not settled by this code. +The model checks the worst case, where every incarnation has the same `as_of`. + +## The design + +### One writer per point + +A publication point accepts exactly one writer attach in its life. +A second attach would replace the chain with one that can regress below what the point's subscribers acknowledged, and a subscriber cannot follow a regression. +The earlier design cut such subscribers with `Event::Detached`, and a live import panicked on it, but a downstream read could miss updates before the import observed the cut. +With one writer per point there is nothing to cut: a point's chain only grows, and freezes when its writer closes or drops. +The fence that kept a stale writer from publishing goes too, since a point has no second writer, and a closed writer still publishes nothing while its holds keep adjusting. + +### A fresh slot for a frozen one + +The registry holds one slot per `GlobalId`, and an incarnation is the `GlobalId` together with the `as_of` stored in its slot. +When the slot under an id is frozen, `get_or_create` creates a fresh slot for the new incarnation instead of returning the frozen one, whose points would refuse the new writer. +Readers of the frozen slot keep its final view. +This is the local backstop against a stray `Arc` keeping an old slot alive past the barrier: whoever binds the id after the old writer dropped gets the fresh slot. +It does not cover a re-export's point, which does not freeze when the re-exporting index drops. + +### The reconciliation barrier + +Without coordination, the interactive runtime can reconcile a connection that recreates an index at the same `as_of` before the maintenance runtime dropped the old incarnation. +`get_or_create` then returns the old incarnation's live slot, the new peer binds to it, and once the old writer drops, imports through that peer read a frozen chain that never advances. +The barrier closes this: each worker arrives once its drops for a connection are done, and creates nothing until the paired worker of the other runtime arrived at the same connection or a later one. + +Scope: one barrier per worker pair, created with the pair's registry. +A per-process barrier is not required: a slot is shared by exactly the two workers of its pair, and a worker of one runtime never waits on a worker of its own runtime at the barrier. +`Solo` runtimes skip it. + +Where it blocks: between the drop phase and the create phase of `Worker::reconcile`, on every reconciliation, including the first connection, whose drop phase is empty. +After reconciliation, ids are unique, so the live command stream needs no barrier. + +A worker waiting at the barrier cannot leave it by skipping the connection's creates, because the other workers of its runtime may already have passed, and timely requires every worker of a runtime to build the same dataflows in the same order. +So the barrier counts the other runtime's arrival at the same connection or any later one, for two reasons. +First, the barrier keeps only each side's latest arrival. +The other runtime can arrive at a connection, pass, serve, and arrive at the next connection before the waiting worker rechecks, which overwrites the arrival the waiting worker needs. +Second, a runtime may never finish reconciling a connection the other runtime finished. +Each runtime's channel adapter stops forwarding a connection's queued commands once a response send to the connection fails (`spawn_channel_adapter`), and the two runtimes notice the dropped connection independently. +One runtime can then receive the connection's `InitializationComplete` and arrive, while the other moves on to the next connection without arriving. +A runtime that arrived at a later connection has dropped everything its later history replaces, and creates nothing until the waiting runtime arrives at that later connection too, which the waiting runtime does only after it finished the earlier connection. +Of two waiting workers, the one at the earlier connection passes, and the controller's last connection reaches both runtimes, so the wait ends. +Shutdown needs no abort: a panic on either runtime aborts the process (the panic hook clusterd's `main` installs), and workers do not exit otherwise. + +## Results + +SMT proves the conjunction of 16 safety clauses and 48 supporting invariants inductive across all 24 actions, `(64 + 1) x (24 + 1) = 1625` checks. +The explicit-state checker explores 961,101 states of the instance with three times, three epochs, two incarnations, two points, and one import without a violation. +That is the smallest instance in which every mutation below reaches its counterexample, except the physical-meet row: three epochs let a runtime skip a connection, two points let the registry choose between a frozen and a fresh slot, and three times let a writer seal once before it drops. + +## Mutation table + +Each row deletes one guard from the committed model, or flips one design switch, and runs the explicit-state check over the safety clauses. +The failing clauses are the first violation the checker reports, which it reaches. + +| Guard deleted or switch flipped | Failing clauses | +| --- | --- | +| No barrier (`barrier = false`) | `peer_own_writer`, `peer_holds_own_incarnation` | +| Barrier after the creates (`barrier_first = false`) | `peer_own_writer`, `peer_holds_own_incarnation` | +| Barrier counts any arrival of the other runtime, not the same or a later connection (`epoch_match = false`) | `peer_own_writer`, `peer_holds_own_incarnation` | +| Reuse a closed slot: no single attach and no fresh slot | `peer_own_writer`, and with only the read clauses kept, `ack_covered`: the new writer's chain regresses below what an import acknowledged | +| No fresh slot for a frozen one (`fresh_slot = false`) | `no_panic`: the new writer's attach is refused | +| No single-attach refusal alone (`single_attach = false`) | inert, the fresh-slot rule never hands a frozen slot to a writer (961,101 states) | +| Seed under the registering lock | `queue_current` | +| Close before the drop seal | `no_empty_upper` | +| Merge only below the applied physical frontier | `ack_on_boundary` | +| Physical meet bounded by the holds | inert with three times (1,005,837 states), `ack_on_boundary` with four times and two epochs | +| Release on read completion | `holds_balance` | + +The single-attach refusal is inert because the model's only attach path goes through the registry, which the fresh-slot rule already guards. +It stays in the code as the backstop for any path that does not. diff --git a/doc/developer/design/20261105_rendezvous_protocol/RendezvousProtocol.lean b/doc/developer/design/20261105_rendezvous_protocol/RendezvousProtocol.lean new file mode 100644 index 0000000000000..7ba127d09c916 --- /dev/null +++ b/doc/developer/design/20261105_rendezvous_protocol/RendezvousProtocol.lean @@ -0,0 +1,653 @@ +-- Copyright Materialize, Inc. and contributors. All rights reserved. +-- +-- Use of this software is governed by the Business Source License +-- included in the LICENSE file at the root of this repository. +-- +-- As of the Change Date specified in that file, in accordance with +-- the Business Source License, use of this software will be governed +-- by the Apache License, Version 2.0. + +module + +public import Veil + +/- +A model of the rendezvous publication protocol and the compute lifecycle that drives it. + +See `README.md` next to this file for how to run it, the lifecycle facts it assumes with their +citations, and what the checks found. + +Scope. One index id, one worker of each runtime, the `oks` arrangement only. + +* `writer` is one incarnation of the index on the maintenance runtime: one trace, its writer, and + its attachment to a point. +* `point` is one `Rendezvous`, the publication point of one registry slot. +* `imp` is one import on the interactive runtime: the subscription, the reader that holds the + point, and the trace it hands downstream, which reads the point's published chain. +* The interactive runtime's peer bundle of incarnation `w` is indexed by `w` itself. +* `epoch` is one controller connection, ordered as the process accepted them. Both runtimes + receive the same command history for it, abstracted to `goal e`, the incarnation of the index + that the history installs. A runtime may never see a connection the other one reconciles. + +The collection is abstract. Its content at each time is fixed and identical across incarnations, +so an incarnation is characterized by its `upper`, its span boundaries (`w_cut`), and its +compaction frontiers. A downstream read through frontier `f` is correct exactly when the chain the +trace reads covers `f`, has a boundary at `f`, and is logically compacted no further than the +read's `as_of`. + +Times and epochs are totally ordered. `zero` is the minimum time, `top` stands for the empty +frontier, and `ezero` is the epoch both runtimes start in, before any connection. + +The registry keys a slot by the dataflow's `as_of`, and the explicit-state check gives every +incarnation the same `as_of`, the worst case. Design switches, all `true` in the committed design +and flipped one at a time by the mutation runs the README lists: + +* `barrier`: a runtime creates dataflows only after the other runtime finished its drops. +* `barrier_first`: the barrier sits between the drops and the creates, not after the creates. +* `epoch_match`: the barrier admits a runtime once the other one finished the drops of the same + or a later connection, not of any connection. +* `single_attach`: `Rendezvous::attach` refuses a point that was ever attached, and panics. +* `fresh_slot`: the registry does not hand out a slot whose writer closed. +-/ + +veil module RendezvousProtocol + +-- The initializer and `x_create` assign more fields than the default depth elaborates. +set_option maxRecDepth 8192 +-- Without it, `#gen_spec` warns that it cannot synthesize the model checker's `LocalRProp` +-- instance for the invariants within the default budget. +set_option synthInstance.maxHeartbeats 400000 + +type time +type epoch +type writer +type point +type imp + +enum phase = { serving, dropping, waiting, creating } + +instantiate tot : TotalOrder time +instantiate etot : TotalOrder epoch + +immutable individual zero : time +immutable individual top : time +immutable individual ezero : epoch +immutable individual barrier : Bool +immutable individual barrier_first : Bool +immutable individual epoch_match : Bool +immutable individual single_attach : Bool +immutable individual fresh_slot : Bool +-- The `as_of` of each incarnation's dataflow, as the controller chose it. +immutable function wasof : writer → time + +-- The controller: its current connection, the connections it opened, and the incarnation each +-- connection's history installs. +individual ctl : epoch +relation opened : epoch → Bool +function goal : epoch → writer +-- Whether a connection's history named the incarnation. A replaced incarnation never returns. +relation assigned : writer → Bool + +-- Reconciliation, per runtime: the connection it reconciles or serves, its phase, and the latest +-- connection whose drops it finished, which is what the barrier publishes. +individual m_ep : epoch +individual m_ph : phase +individual m_arr : epoch +individual i_ep : epoch +individual i_ph : phase +individual i_arr : epoch + +-- Maintenance runtime, per incarnation. +relation m_live : writer → Bool +relation m_dropped : writer → Bool +relation w_closed : writer → Bool +-- The connection whose create phase rendered the incarnation. +function w_ep : writer → epoch +function w_pt : writer → point +function w_upper : writer → time +relation w_cut : writer → time → Bool +function w_since : writer → time +-- The physical compaction frontier the spine applied, read from the holds at its last sync. +function w_phys : writer → time + +-- Publication points. +relation p_used : point → Bool +-- The slot the registry map returns for the id. +relation p_inmap : point → Bool +function p_asof : point → time +-- Whether a writer ever attached. +relation p_attached : point → Bool +-- Whether the attached writer closed or dropped. +relation p_closed : point → Bool +-- The writer that attached last, the only one unless `single_attach` is off. +function p_cur : point → writer +function p_upper : point → time +relation p_cut : point → time → Bool +function p_since : point → time +-- The point's hold accumulations, one registration per holder. +relation acc : point → imp → Bool +relation acc_peer : point → writer → Bool +-- A holder of the slot's `Arc` the lifecycle does not account for, such as a peek in flight. +relation stray : point → Bool + +-- Interactive runtime: the peer bundle of each incarnation. +relation i_peer : writer → Bool +relation i_peer_done : writer → Bool +function peer_ep : writer → epoch +function peer_pt : writer → point +function peer_l : writer → time + +-- Interactive runtime: imports. +relation x_live : imp → Bool +relation x_dropped : imp → Bool +-- The import's capabilities are empty: its read is over. +relation x_done : imp → Bool +-- The incarnation whose peer bundle the import was built from. +function x_inc : imp → writer +function x_ep : imp → epoch +function x_pt : imp → point +-- The frontier the import's stream acknowledged. +function x_ack : imp → time +-- The latest `Upper` in the import's queue. +function x_q : imp → time +-- The read's `as_of`, which is also the logical hold of the trace handed downstream. +function x_l : imp → time +function x_ph : imp → time +-- The items delivered to the import cover the chain through here. +function x_deliv : imp → time +-- An item arrived whose lower was not where the delivered items ended. +relation x_hole : imp → Bool +-- Whether the subscription was seeded from a writer's view, and which one. +relation x_seeded : imp → Bool +function x_sub : imp → writer +function x_until : imp → time +-- The importer's own, possibly stale, observation of the published upper. +function x_snap : imp → time + +-- A panic on either runtime halts the process. +individual panicked : Bool + +#gen_state + +theory ghost relation lt (x y : time) := tot.le x y ∧ x ≠ y +theory ghost relation elt (x y : epoch) := etot.le x y ∧ x ≠ y + +-- A slot is alive while a holder of its `Arc` exists: the publisher's `PublishToken`, the peer +-- bundle, an importing dataflow's tokens, or a stray holder. +ghost relation referenced (q : point) := + (∃ w, m_live w ∧ w_pt w = q) ∨ (∃ w, i_peer w ∧ peer_pt w = q) ∨ (∃ x, x_live x ∧ x_pt x = q) + ∨ stray q + +-- `get_or_create` returns the slot in the map when it is alive, its `as_of` matches, and, under +-- the fresh-slot rule, its writer has not closed. +ghost relation reusable (q : point) (w : writer) := + p_inmap q ∧ referenced q ∧ p_asof q = wasof w ∧ (fresh_slot → ¬ p_closed q) + +assumption ∀ t, tot.le zero t +assumption ∀ t, tot.le t top +assumption zero ≠ top +assumption ∀ e, etot.le ezero e +assumption ∀ w, wasof w ≠ top +assumption [design] barrier = true ∧ barrier_first = true ∧ epoch_match = true ∧ single_attach = true ∧ fresh_slot = true + +after_init { + ctl := ezero + opened E := decide (E = ezero) + assigned U := false + m_ep := ezero + m_ph := serving + m_arr := ezero + i_ep := ezero + i_ph := serving + i_arr := ezero + m_live U := false + m_dropped U := false + w_closed U := false + w_cut U T := false + p_used P := false + p_inmap P := false + p_attached P := false + p_closed P := false + p_cut P T := false + acc P X := false + acc_peer P U := false + stray P := false + i_peer U := false + i_peer_done U := false + x_live X := false + x_dropped X := false + x_done X := false + x_hole X := false + x_seeded X := false + panicked := false +} + +-- A new controller connection. Its history keeps the current incarnation or names a new one. +action reconnect (e : epoch) (w : writer) { + require ¬ panicked + require elt ctl e + require w = goal ctl ∨ ¬ assigned w + opened e := true + goal e := w + assigned w := true + ctl := e +} + +-- A runtime sees the nonce of a newer connection, while it serves the current one or waits at its +-- barrier, and starts reconciling. Its drop and create phases issue no receive, so a nonce change +-- reaches it only in these two phases. +action m_begin (e : epoch) { + require ¬ panicked + require m_ph = serving ∨ m_ph = waiting + require opened e ∧ elt m_ep e + m_ep := e + m_ph := dropping +} + +action i_begin (e : epoch) { + require ¬ panicked + require i_ph = serving ∨ i_ph = waiting + require opened e ∧ elt i_ep e + i_ep := e + i_ph := dropping +} + +-- A runtime finished the drops its history asks for and publishes its arrival. +action m_arrive { + require ¬ panicked + require m_ph = dropping + require ∀ U, m_live U → U = goal m_ep + m_arr := m_ep + m_ph := waiting +} + +action i_arrive { + require ¬ panicked + require i_ph = dropping + require ∀ U, i_peer U → U = goal i_ep + require ∀ X, x_live X → x_inc X = goal i_ep + i_arr := i_ep + i_ph := waiting +} + +action m_pass { + require ¬ panicked + require m_ph = waiting + -- The barrier, given the other runtime's phase and latest arrival. A later arrival counts, as + -- the other runtime may never see this connection. + require barrier → (epoch_match → etot.le m_ep i_arr) ∧ (¬ epoch_match → i_ph ≠ dropping) + m_ph := creating +} + +action i_pass { + require ¬ panicked + require i_ph = waiting + require barrier → (epoch_match → etot.le i_ep m_arr) ∧ (¬ epoch_match → m_ph ≠ dropping) + i_ph := creating +} + +action m_finish { + require ¬ panicked + require m_ph = creating + m_ph := serving +} + +action i_finish { + require ¬ panicked + require i_ph = creating + i_ph := serving +} + +-- `get_or_create` found no reusable slot: a fresh, unattached point replaces the map entry. +procedure new_slot (p : point) (w : writer) { + p_used p := true + p_inmap Q := decide (Q = p) + p_asof p := wasof w + p_attached p := false + p_closed p := false + p_upper p := zero + p_cut p T := decide (T = zero) + p_since p := zero +} + +-- `Rendezvous::attach` of a freshly rendered trace, which is empty and compacted to its `as_of`. +procedure attach (w : writer) (p : point) { + if single_attach ∧ p_attached p then + -- The point had a writer: `attach` refuses it. + panicked := true + else + m_live w := true + w_ep w := m_ep + w_pt w := p + w_closed w := false + w_upper w := zero + w_cut w T := decide (T = zero) + w_since w := wasof w + w_phys w := zero + -- Seed live subscribers with the new view. + x_q X := if x_live X ∧ x_pt X = p then zero else x_q X + x_sub X := if x_live X ∧ x_pt X = p then w else x_sub X + x_seeded X := decide (x_seeded X ∨ (x_live X ∧ x_pt X = p)) + p_attached p := true + p_closed p := false + p_cur p := w + p_upper p := zero + p_cut p T := decide (T = zero) + p_since p := wasof w +} + +-- Maintenance renders the incarnation its history installs and publishes it into the slot +-- `get_or_create` returns, here an existing one. +action m_create_reuse (w : writer) (p : point) { + require ¬ panicked + -- Creates follow the barrier, or precede it with `barrier_first` off. + require m_ph = creating ∨ (¬ barrier_first ∧ m_ph = waiting) + require w = goal m_ep + require ¬ m_live w ∧ ¬ m_dropped w + require reusable p w + attach w p +} + +action m_create_fresh (w : writer) (p : point) { + require ¬ panicked + require m_ph = creating ∨ (¬ barrier_first ∧ m_ph = waiting) + require w = goal m_ep + require ¬ m_live w ∧ ¬ m_dropped w + require ¬ p_used p + require ∀ Q, ¬ reusable Q w + new_slot p w + attach w p +} + +-- The writer inserts a span `[w_upper w, t)` and `publish_chain` mirrors it. +action m_seal (w : writer) (t : time) { + require ¬ panicked + require m_live w + require lt (w_upper w) t + require t ≠ top + let lo := w_upper w + w_upper w := t + w_cut w t := true + if ¬ w_closed w then + p_upper (w_pt w) := t + p_cut (w_pt w) T := w_cut w T + x_hole X := decide (x_hole X ∨ (x_live X ∧ x_pt X = w_pt w ∧ x_deliv X ≠ lo)) + x_deliv X := if x_live X ∧ x_pt X = w_pt w ∧ x_deliv X = lo then t else x_deliv X + x_q X := if x_live X ∧ x_pt X = w_pt w then t else x_q X +} + +-- A merge removes the boundary at `c`, which the applied physical frontier has passed. +action m_merge (w : writer) (c : time) { + require ¬ panicked + require m_live w + require w_cut w c + require lt c (w_phys w) + w_cut w c := false + if ¬ w_closed w then + p_cut (w_pt w) T := w_cut w T +} + +-- `sync_holds`: read the meets under the lock, publish the logical one, apply both to the spine. +-- Each frontier either moves to a value the holds allow or stays, as the joins with the spine's +-- current frontiers make it. +action m_sync (w : writer) (ph : time) (s : time) { + require ¬ panicked + require m_live w + require tot.le ph (w_upper w) + require ph = w_phys w ∨ (tot.le (w_phys w) ph ∧ ∀ X, acc (w_pt w) X → tot.le ph (x_ph X)) + require s ≠ top + require s = w_since w ∨ (tot.le (w_since w) s ∧ (∀ X, acc (w_pt w) X → tot.le s (x_l X)) + ∧ (∀ V, acc_peer (w_pt w) V → tot.le s (peer_l V))) + w_phys w := ph + w_since w := s + if ¬ w_closed w then + p_since (w_pt w) := s +} + +-- Reconciliation drops an incarnation its history replaces: the writer drops, sealing its spine to +-- the empty frontier, and the trace box closes the point first so the seal stays out of it. +action m_drop (w : writer) { + require ¬ panicked + require m_ph = dropping + require m_live w + require w ≠ goal m_ep + w_closed w := true + p_closed (w_pt w) := true + let lo := w_upper w + w_upper w := top + w_cut w top := true + if ¬ w_closed w then + p_upper (w_pt w) := top + p_cut (w_pt w) T := w_cut w T + x_hole X := decide (x_hole X ∨ (x_live X ∧ x_pt X = w_pt w ∧ x_deliv X ≠ lo)) + x_deliv X := if x_live X ∧ x_pt X = w_pt w ∧ x_deliv X = lo then top else x_deliv X + x_q X := if x_live X ∧ x_pt X = w_pt w then top else x_q X + m_live w := false + m_dropped w := true +} + +-- `peer_bundle`: the interactive runtime records the incarnation its history installs and holds +-- its slot logically at the join of `as_of` and the published `since`. +procedure bind_peer (w : writer) (p : point) { + i_peer w := true + peer_ep w := i_ep + peer_pt w := p + peer_l w := if tot.le (wasof w) (p_since p) then p_since p else wasof w + acc_peer p w := true +} + +action i_peer_reuse (w : writer) (p : point) { + require ¬ panicked + require i_ph = creating ∨ (¬ barrier_first ∧ i_ph = waiting) + require w = goal i_ep + require ¬ i_peer w ∧ ¬ i_peer_done w + require reusable p w + bind_peer w p +} + +action i_peer_fresh (w : writer) (p : point) { + require ¬ panicked + require i_ph = creating ∨ (¬ barrier_first ∧ i_ph = waiting) + require w = goal i_ep + require ¬ i_peer w ∧ ¬ i_peer_done w + require ¬ p_used p + require ∀ Q, ¬ reusable Q w + new_slot p w + bind_peer w p +} + +action i_peer_drop (w : writer) { + require ¬ panicked + require i_ph = dropping + require i_peer w + require w ≠ goal i_ep + i_peer w := false + i_peer_done w := true + acc_peer P w := false +} + +-- Something clones a slot's `Arc` outside the lifecycle, or drops that clone. +action stray_take (p : point) { + require ¬ panicked + require p_used p ∧ referenced p + stray p := true +} + +action stray_drop (p : point) { + require ¬ panicked + stray p := false +} + +-- The importer reads the published upper outside the lock. Only the stale-seed mutation uses it. +action x_observe (x : imp) (w : writer) { + require ¬ panicked + require i_peer w + x_snap x := p_upper (peer_pt w) +} + +-- An interactive dataflow imports the index through peer bundle `w`: `fresh_reader`, then +-- `subscribe`, seeded with the published chain and upper. +action x_create (x : imp) (w : writer) (u : time) { + require ¬ panicked + require i_ph = creating ∨ i_ph = serving ∨ (¬ barrier_first ∧ i_ph = waiting) + require i_peer w + require ¬ x_live x ∧ ¬ x_dropped x + require lt (peer_l w) u + -- Seed and registration happen under one lock acquisition. + require x_snap x = p_upper (peer_pt w) + -- `fresh_reader` expects `reader_at` to accept the peer's hold. + if ¬ tot.le (p_since (peer_pt w)) (peer_l w) then + panicked := true + x_live x := true + x_inc x := w + x_ep x := i_ep + x_pt x := peer_pt w + x_l x := peer_l w + x_ph x := p_upper (peer_pt w) + acc (peer_pt w) x := true + x_ack x := zero + x_q x := x_snap x + x_deliv x := x_snap x + x_done x := false + x_hole x := false + x_until x := u + x_seeded x := p_attached (peer_pt w) + x_sub x := p_cur (peer_pt w) +} + +-- The import operator drains its queue, acknowledging the latest `Upper`. +action x_drain (x : imp) { + require ¬ panicked + require x_live x ∧ ¬ x_done x + x_ack x := x_q x + x_ph x := x_q x + if x_q x = top ∨ tot.le (x_until x) (x_q x) then + x_done x := true + acc (x_pt x) x := false +} + +-- `drop_dataflow` of the importing dataflow, in any phase but the barrier wait. +action x_drop (x : imp) { + require ¬ panicked + require i_ph ≠ waiting + require x_live x + x_live x := false + x_dropped x := true + acc P x := false +} + +safety [no_panic] ¬ panicked + +-- A downstream read through the acknowledged frontier misses updates when the chain does not cover +-- it, panics in `chain_through` when a span straddles it, double counts or misses items when +-- delivery had a hole, and reads collapsed times when the chain is compacted past the read's +-- `as_of`. +safety [ack_covered] x_live X ∧ ¬ x_done X → tot.le (x_ack X) (p_upper (x_pt X)) +safety [ack_on_boundary] x_live X ∧ ¬ x_done X ∧ x_ack X ≠ zero → p_cut (x_pt X) (x_ack X) +safety [no_hole] x_live X ∧ ¬ x_done X → ¬ x_hole X +safety [since_below_read] x_live X ∧ ¬ x_done X → tot.le (p_since (x_pt X)) (x_l X) + +-- A live import reads the chain of the writer whose view seeded it. +safety [reads_own_chain] x_live X ∧ x_seeded X → p_cur (x_pt X) = x_sub X + +-- A peer and an import are bound to their own incarnation's point, or to one no writer attached +-- yet, never to another incarnation's. +safety [peer_own_writer] i_peer U ∧ p_attached (peer_pt U) → p_cur (peer_pt U) = U +safety [imp_own_writer] x_live X ∧ p_attached (x_pt X) → p_cur (x_pt X) = x_inc X + +-- Old readers never hold back a later incarnation's trace. +safety [holds_own_incarnation] acc P X ∧ m_live U ∧ w_pt U = P → x_inc X = U +safety [peer_holds_own_incarnation] acc_peer P V ∧ m_live U ∧ w_pt U = P → V = U + +-- No reader sees the empty frontier, since a dropped writer says nothing about later times. +safety [no_empty_upper] p_used P → p_upper P ≠ top +safety [no_empty_queue] x_live X → x_q X ≠ top + +-- Holds balance: a holder is registered exactly while it is live. +safety [holds_balance] acc P X ↔ (x_live X ∧ ¬ x_done X ∧ x_pt X = P) +safety [peer_holds_balance] acc_peer P U ↔ (i_peer U ∧ peer_pt U = P) + +-- The writer compacts no further than any live hold on its point. +safety [since_below_holds] acc P X ∧ m_live U ∧ w_pt U = P → tot.le (w_since U) (x_l X) + +-- A subscriber has seen every upper its point published. +safety [queue_current] x_live X → x_q X = p_upper (x_pt X) + +-- Supporting invariants. They make the clauses above inductive for the committed design. +-- Reconciliation bookkeeping. +invariant [opened_le_ctl] opened E → etot.le E ctl +invariant [opened_eps] opened ctl ∧ opened m_ep ∧ opened i_ep ∧ opened m_arr ∧ opened i_arr +invariant [arr_le_ep] etot.le m_arr m_ep ∧ etot.le i_arr i_ep +invariant [m_arr_ep] m_ph ≠ dropping → m_arr = m_ep +invariant [i_arr_ep] i_ph ≠ dropping → i_arr = i_ep +invariant [m_reconciling] m_ph ≠ serving → elt ezero m_ep +invariant [i_reconciling] i_ph ≠ serving → elt ezero i_ep +invariant [goal_assigned] opened E ∧ elt ezero E → assigned (goal E) +-- A replaced incarnation never returns, so the connections that install one are contiguous. +invariant [goal_convex] + opened E1 ∧ opened E2 ∧ opened E3 ∧ elt ezero E1 ∧ etot.le E1 E2 ∧ etot.le E2 E3 ∧ + goal E1 = goal E3 → goal E2 = goal E1 +-- What each runtime holds is the incarnation of the latest connection whose drops it finished. +invariant [m_live_goal] m_live U → U = goal m_arr +invariant [i_peer_goal] i_peer U → U = goal i_arr +invariant [x_live_goal] x_live X → x_inc X = goal i_arr +-- The barrier: a runtime past it is no later than the other runtime's latest arrival, and so is +-- every connection whose create phase built something still live. +invariant [m_passed] (m_ph = creating ∨ m_ph = serving) → etot.le m_ep i_arr +invariant [i_passed] (i_ph = creating ∨ i_ph = serving) → etot.le i_ep m_arr +invariant [w_ep_bound] + m_live U → goal (w_ep U) = U ∧ opened (w_ep U) ∧ elt ezero (w_ep U) ∧ etot.le (w_ep U) m_arr ∧ + etot.le (w_ep U) i_arr +invariant [peer_ep_bound] + i_peer U → goal (peer_ep U) = U ∧ opened (peer_ep U) ∧ elt ezero (peer_ep U) ∧ + etot.le (peer_ep U) i_arr ∧ etot.le (peer_ep U) m_arr +invariant [x_ep_bound] + x_live X → goal (x_ep X) = x_inc X ∧ opened (x_ep X) ∧ elt ezero (x_ep X) ∧ + etot.le (x_ep X) i_arr ∧ etot.le (x_ep X) m_arr +-- Consequences: a runtime past the barrier sees only its connection's incarnation on the other +-- runtime. +invariant [i_passed_m_live] (i_ph = creating ∨ i_ph = serving) ∧ m_live U → U = goal i_ep +invariant [m_passed_i_peer] (m_ph = creating ∨ m_ph = serving) ∧ i_peer U → U = goal m_ep +invariant [m_passed_x_live] + (m_ph = creating ∨ m_ph = serving) ∧ x_live X → x_inc X = goal m_ep +invariant [live_assigned] (m_live U ∨ m_dropped U ∨ i_peer U ∨ i_peer_done U) → assigned U +-- Lifecycle of one incarnation. +invariant [live_open] m_live U → ¬ w_closed U ∧ ¬ m_dropped U +invariant [writer_used] (m_live U ∨ m_dropped U) → p_used (w_pt U) ∧ p_attached (w_pt U) +invariant [peer_used] i_peer U → p_used (peer_pt U) +invariant [imp_used] x_live X → p_used (x_pt X) +invariant [imp_peer] x_live X → (i_peer (x_inc X) ∨ i_peer_done (x_inc X)) ∧ x_pt X = peer_pt (x_inc X) +invariant [unused_clean] ¬ p_used P → ¬ p_attached P ∧ ¬ p_closed P ∧ ¬ acc P X ∧ ¬ acc_peer P U ∧ ¬ stray P +invariant [attached_cur] p_attached P → (m_live (p_cur P) ∨ m_dropped (p_cur P)) ∧ w_pt (p_cur P) = P +invariant [closed_dropped] p_attached P → (p_closed P ↔ m_dropped (p_cur P)) +invariant [closed_attached] p_closed P → p_attached P +invariant [one_writer] (m_live U1 ∨ m_dropped U1) ∧ (m_live U2 ∨ m_dropped U2) ∧ w_pt U1 = w_pt U2 → U1 = U2 +invariant [peer_key] i_peer U → tot.le (wasof U) (peer_l U) +invariant [imp_key] x_live X → tot.le (wasof (x_inc X)) (x_l X) +-- The point mirrors its live writer. +invariant [mirror_upper] m_live U → p_upper (w_pt U) = w_upper U +invariant [mirror_cut] m_live U → (p_cut (w_pt U) T ↔ w_cut U T) +invariant [mirror_since] m_live U → p_since (w_pt U) = w_since U +invariant [upper_cut] p_used P → p_cut P (p_upper P) +invariant [w_upper_cut] m_live U → w_cut U (w_upper U) +invariant [phys_le_upper] m_live U → tot.le (w_phys U) (w_upper U) +invariant [phys_le_holds] acc P X ∧ m_live U ∧ w_pt U = P → tot.le (w_phys U) (x_ph X) +invariant [ack_le_q] x_live X → tot.le (x_ack X) (x_q X) +invariant [ph_is_ack] x_live X ∧ x_ack X ≠ zero → x_ph X = x_ack X +invariant [ph_le_upper] x_live X → tot.le (x_ph X) (p_upper (x_pt X)) +invariant [deliv_current] x_live X → x_deliv X = p_upper (x_pt X) +invariant [peer_since] i_peer U → tot.le (p_since (peer_pt U)) (peer_l U) +invariant [seeded_attached] x_live X → (x_seeded X ↔ p_attached (x_pt X)) +invariant [sub_is_cur] x_live X ∧ x_seeded X → x_sub X = p_cur (x_pt X) +invariant [unattached_empty] p_used P ∧ ¬ p_attached P → p_upper P = zero ∧ p_since P = zero + +#time #gen_spec + +#check_invariants + +-- Explicit-state cross-check. Every incarnation has the same `as_of`, the case in which the +-- registry finds an earlier incarnation's slot under the new incarnation's key. Three epochs are +-- the start and two connections, enough for a runtime to skip one. +#model_check { time := Fin 3, epoch := Fin 3, writer := Fin 2, point := Fin 2, imp := Fin 1 } { zero := 0, top := 2, ezero := 0, barrier := true, barrier_first := true, epoch_match := true, single_attach := true, fresh_slot := true, wasof := fun _ => 0 } + +end RendezvousProtocol diff --git a/src/timely-util/src/lib.rs b/src/timely-util/src/lib.rs index a17665b51d065..c0a5514877185 100644 --- a/src/timely-util/src/lib.rs +++ b/src/timely-util/src/lib.rs @@ -33,6 +33,7 @@ pub mod panic; pub mod pool_config; pub mod probe; pub mod reclock; +pub mod rendezvous; pub mod replay; pub mod scope_label; pub mod temporal; diff --git a/src/timely-util/src/rendezvous.rs b/src/timely-util/src/rendezvous.rs new file mode 100644 index 0000000000000..e609188fcf9c8 --- /dev/null +++ b/src/timely-util/src/rendezvous.rs @@ -0,0 +1,585 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! A meeting point between one writer thread and readers on any thread. +//! +//! A [`Rendezvous`] carries a view a writer publishes, [`Published`]: a list of items, the `upper` +//! they cover, and writer-defined metadata. Next to it sit `N` hold accumulations, one per hold +//! kind, into which every [`Reader`] registers a frontier per kind. The writer reads the meet of +//! each kind to decide what it may discard, and a reader that moves a meet wakes the writer. +//! Subscribers receive the view as a seed and then every item the writer reports as arrived. +//! +//! A writer [attaches](Rendezvous::attach) to a point and publishes through the returned +//! [`Attachment`]. A writer may attach to several points, but a point accepts one writer in its +//! life. Once that writer closes or drops, the view is final and the point is +//! [frozen](Rendezvous::is_frozen): its readers keep the view as last published, and a later +//! writer needs a point of its own. The writer keeps adjusting holds after it closes, so the holds +//! it registered still release. +//! +//! Everything a point owns sits behind one lock. Critical sections are frontier arithmetic, vector +//! swaps, the seed a new subscription clones, which must be taken under the lock that registers +//! the subscription for the seed and the deltas to leave no gap, and the closures callers pass to +//! [`Rendezvous::read`], [`Rendezvous::reader_with`] and the metadata updates. The items publishing +//! pushes to queues are cloned, and the items it replaces dropped, outside the lock. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, Weak}; + +use itertools::Itertools; +use timely::PartialOrder; +use timely::progress::frontier::{AntichainRef, MutableAntichain}; +use timely::progress::{Antichain, Timestamp}; +use timely::scheduling::activate::SyncActivator; + +/// The view a writer publishes on a [`Rendezvous`]. +pub struct Published { + /// The writer's latest list of items, replaced wholesale on every publish. + pub items: Vec, + /// The frontier `items` covers. Starts at the minimum time, and only the writer moves it. + pub upper: Antichain, + /// Writer-defined metadata, published together with `items`. + pub meta: M, +} + +/// One entry of a [`Subscription`]'s queue. +pub enum Event { + /// An item, in the order the writer published it. + Item(I), + /// The `upper` the view had once every preceding item was in it. + Upper(Antichain), +} + +/// The queue and wakeup for one subscriber. Its own lock, so a subscriber draining it never +/// contends with the writer publishing. +struct Queue { + events: Mutex>>, + activator: SyncActivator, +} + +impl Queue { + fn push(&self, events: impl IntoIterator>) { + let mut queue = self.events.lock().expect("subscription queue poisoned"); + queue.extend(events); + } + + fn activate(&self) { + let _ = self.activator.activate(); + } +} + +struct State { + published: Published, + /// The holds of every reader, plus whatever the writer registered, per kind. + holds: [MutableAntichain; N], + /// Held weakly: the subscriber owns the only strong reference, and its entry is pruned once + /// it drops. + queues: Vec>>, + /// Wakes the writer. + writer_activator: Option, + /// Where the point is in its one writer's life. + writer: WriterState, +} + +/// The life of a point's one writer. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum WriterState { + /// No writer attached yet. + Unattached, + /// The writer attached and has neither closed nor dropped. + Attached, + /// The writer closed or dropped. The view is final. + Frozen, +} + +impl State { + /// The live subscriber queues, dropping entries whose subscriber has gone. Returns strong + /// references so the caller can push and activate after releasing the lock. + fn live_queues(&mut self) -> Vec>> { + let mut live = Vec::with_capacity(self.queues.len()); + self.queues.retain(|weak| match weak.upgrade() { + Some(queue) => { + live.push(queue); + true + } + None => false, + }); + live + } + + fn wake_writer(&self) { + if let Some(activator) = &self.writer_activator { + let _ = activator.activate(); + } + } +} + +/// A point one writer publishes a view to and readers on any thread hold. +/// +/// Created unattached, with no items, `upper` at the minimum time, and the given metadata. Readers +/// and subscribers may register before a writer attaches. Subscribers are seeded from the writer's +/// view at attachment. +pub struct Rendezvous { + state: Mutex>, + /// Whether a reader moved a meet since the writer last took the flag. Outside the lock, so the + /// writer can skip reading the holds without taking it. + holds_moved: AtomicBool, +} + +impl Rendezvous { + fn lock(&self) -> MutexGuard<'_, State> { + self.state.lock().expect("rendezvous poisoned") + } + + /// Calls `f` on the published view, under the point's lock. + pub fn read(&self, f: impl FnOnce(&Published) -> R) -> R { + f(&self.lock().published) + } +} + +impl Rendezvous { + /// An unattached point publishing `meta`. + pub fn with_meta(meta: M) -> Self { + Rendezvous { + state: Mutex::new(State { + published: Published { + items: Vec::new(), + upper: Antichain::from_elem(T::minimum()), + meta, + }, + holds: std::array::from_fn(|_| MutableAntichain::new()), + queues: Vec::new(), + writer_activator: None, + writer: WriterState::Unattached, + }), + holds_moved: AtomicBool::new(false), + } + } + + /// Whether this point's writer attached and has since closed or dropped. The view of a frozen + /// point never changes again, and the point refuses another writer. + pub fn is_frozen(&self) -> bool { + self.lock().writer == WriterState::Frozen + } + + /// The meet of every hold of `kind` on this point. Empty when nothing holds. + pub fn holds(&self, kind: usize) -> Antichain { + self.lock().holds[kind].frontier().to_owned() + } + + /// A reader holding the frontiers `holds` computes from the published view, or the error it + /// returns. + /// + /// Reads the view and registers the holds under one lock acquisition, so the writer cannot + /// publish between the two. + pub fn reader_with( + self: &Arc, + holds: impl FnOnce(&Published) -> Result<[Antichain; N], E>, + ) -> Result, E> { + let mut state = self.lock(); + let holds = holds(&state.published)?; + for (accumulated, hold) in state.holds.iter_mut().zip_eq(holds.iter()) { + adjust(accumulated, &[], hold); + } + drop(state); + Ok(Reader { + point: Arc::clone(self), + holds, + }) + } +} + +impl Rendezvous { + /// A subscription seeded with the published items and `upper`, and the seeded `upper`. + /// + /// Seeding and registering happen under one lock acquisition. An item the writer publishes as + /// arrived is therefore either in the seed or delivered afterwards, never both and never + /// neither. + pub fn subscribe(&self, activator: SyncActivator) -> (Subscription, Antichain) { + let mut state = self.lock(); + let mut events = VecDeque::new(); + events.extend(state.published.items.iter().cloned().map(Event::Item)); + let seed = state.published.upper.clone(); + events.push_back(Event::Upper(seed.clone())); + let queue = Arc::new(Queue { + events: Mutex::new(events), + activator, + }); + state.queues.push(Arc::downgrade(&queue)); + (Subscription { queue }, seed) + } + + /// Attaches the point's writer and publishes its view. + /// + /// Replaces the items and `upper`, applies `meta` to the metadata, adds `holds` to the + /// accumulations, and installs the writer wakeup `activator`. `activator` must schedule the + /// code that reads the holds: a reader moving its hold from another thread fires it. Live + /// subscribers, all registered before any writer attached, are seeded with the view. + /// + /// # Panics + /// + /// Panics if a writer attached to this point before, even one that has since dropped. Its + /// subscribers acknowledged its `upper`, and a second writer's view can regress below it, + /// which a subscriber cannot follow. + pub fn attach( + self: &Arc, + activator: SyncActivator, + items: Vec, + upper: Antichain, + meta: impl FnOnce(&mut M), + holds: [Vec<(T, i64)>; N], + ) -> Attachment { + let mut next = items.clone(); + let live = { + let mut state = self.lock(); + if state.writer != WriterState::Unattached { + // Released first, so the panic does not poison the point for its readers. + drop(state); + panic!("rendezvous: attach to a point that already had a writer"); + } + state.writer = WriterState::Attached; + state.writer_activator = Some(activator); + std::mem::swap(&mut state.published.items, &mut next); + state.published.upper = upper.clone(); + meta(&mut state.published.meta); + for (accumulated, updates) in state.holds.iter_mut().zip_eq(holds) { + accumulated.update_iter(updates); + } + state.live_queues() + }; + drop(next); + for queue in live { + queue.push( + items + .iter() + .cloned() + .map(Event::Item) + .chain(std::iter::once(Event::Upper(upper.clone()))), + ); + queue.activate(); + } + Attachment { + point: Arc::clone(self), + closed: false, + } + } +} + +/// Replaces the elements of `lower` with those of `upper` in `accumulated`, and returns whether +/// that moved the frontier. +/// +/// The writer reads the holds only when it runs, so a change that leaves the frontier in place +/// gives it nothing to do and needs no wake. +pub(crate) fn adjust( + accumulated: &mut MutableAntichain, + lower: &[T], + upper: &[T], +) -> bool { + // The drained changes are consolidated, so they are empty exactly when the frontier is. + let updates = upper + .iter() + .cloned() + .map(|time| (time, 1)) + .chain(lower.iter().cloned().map(|time| (time, -1))); + accumulated.update_iter(updates).next().is_some() +} + +/// A writer's attachment to one [`Rendezvous`]. +pub struct Attachment { + point: Arc>, + /// Whether the writer has closed. See [`Attachment::close`]. + closed: bool, +} + +impl Attachment { + /// The point this attachment publishes to. + pub fn point(&self) -> &Arc> { + &self.point + } + + /// Whether anything besides this attachment references the point. Minting a reader or a + /// subscription needs a reference, so an unreachable point cannot gain one. + /// + /// A [`Subscription`] does not reference the point. A writer that stops publishing to an + /// unreachable point starves a subscriber that holds no [`Reader`] of it. + pub fn reachable(&self) -> bool { + Arc::strong_count(&self.point) > 1 + } + + /// Whether a reader moved a meet of this point since the last call, which resets it. + /// + /// A reader sets the flag after it moved the meet and before it wakes the writer, so a writer + /// that reads `false` and skips the holds is woken again for any later move. + pub fn take_holds_moved(&self) -> bool { + self.point.holds_moved.swap(false, Ordering::AcqRel) + } + + /// Whether [`Attachment::close`] has been called. + pub fn is_closed(&self) -> bool { + self.closed + } + + /// Stops this attachment publishing, leaving the point's items, `upper` and metadata as last + /// published, and freezes the point. Hold changes still apply. + /// + /// NOTE: A closing writer must not publish an empty `upper`. A dropped writer says nothing + /// about the times past its last `upper`, and advancing readers to the empty frontier lets + /// their dataflows complete work on times that never happened. `drop_dataflow` reclaims a + /// dataflow without letting it finish its computation, so a writer's drop is no evidence that + /// its collection is complete. Dropping an attachment without closing it leaves the view as + /// last published too. + pub fn close(&mut self) { + self.closed = true; + self.point.lock().writer = WriterState::Frozen; + } +} + +impl Attachment { + /// The point's accumulated holds, as updates per kind, consuming the attachment. + /// + /// For a writer that registered holds on a point it alone references: no reader can hold the + /// point then, so every hold on it is one the writer registered. + pub fn into_holds(self) -> [Vec<(T, i64)>; N] { + // Another reference could mint a reader, whose holds this would hand to the writer. + mz_ore::soft_assert_or_log!( + Arc::strong_count(&self.point) == 1, + "into_holds: the point is referenced elsewhere" + ); + let mut state = self.point.lock(); + std::array::from_fn(|kind| state.holds[kind].updates().cloned().collect()) + } + + /// Locks the points of `attachments` together, in address order. + /// + /// Every point has one writer, so no other thread locks more than one of these points. + /// + /// Panics if two attachments name the same point, which would deadlock instead. + pub fn lock_all<'a>(attachments: &[&'a Self]) -> Locked<'a, T, I, M, N> { + let mut order: Vec = (0..attachments.len()).collect(); + order.sort_unstable_by_key(|&index| Arc::as_ptr(&attachments[index].point)); + assert!( + order + .windows(2) + .all(|pair| !Arc::ptr_eq(&attachments[pair[0]].point, &attachments[pair[1]].point)), + "lock_all: a point is named twice" + ); + let mut slots: Vec>> = + (0..attachments.len()).map(|_| None).collect(); + for index in order { + let attachment = attachments[index]; + let state = attachment.point.lock(); + slots[index] = Some(Guard { + publishing: !attachment.closed, + state, + }); + } + Locked { + guards: slots + .into_iter() + .map(|slot| slot.expect("every point is locked")) + .collect(), + } + } +} + +impl Attachment { + /// Publishes a view, unless this attachment is closed. Returns whether it published and that + /// moved `upper`. + /// + /// Replaces the items and `upper` and applies `meta` to the metadata. When `arrived` is given, + /// it is an item `items` contains that no earlier publish did, and every live subscriber + /// receives it followed by `upper`. Without it, a moved `upper` reaches every live subscriber + /// on its own. `upper` must not regress below the published one. + pub fn publish( + &mut self, + items: Vec, + upper: Antichain, + meta: impl FnOnce(&mut M), + arrived: Option<&I>, + ) -> bool { + if self.closed { + return false; + } + let mut next = items; + let (advanced, live, regressed_from) = { + let mut state = self.point.lock(); + let regressed_from = (!PartialOrder::less_equal(&state.published.upper, &upper)) + .then(|| state.published.upper.clone()); + std::mem::swap(&mut state.published.items, &mut next); + let advanced = state.published.upper != upper; + state.published.upper = upper.clone(); + meta(&mut state.published.meta); + let live = if arrived.is_some() || advanced { + state.live_queues() + } else { + Vec::new() + }; + (advanced, live, regressed_from) + }; + // A subscriber that acknowledged the published `upper` cannot follow a regression. Checked + // outside the lock, so the panic does not poison the point for its readers. + mz_ore::soft_assert_or_log!( + regressed_from.is_none(), + "rendezvous: publish regresses upper from {regressed_from:?} to {upper:?}", + ); + // The previous items drop here, outside the lock. + drop(next); + for queue in &live { + match arrived { + Some(item) => queue.push([Event::Item(item.clone()), Event::Upper(upper.clone())]), + None => queue.push([Event::Upper(upper.clone())]), + } + queue.activate(); + } + advanced + } +} + +impl Drop for Attachment { + fn drop(&mut self) { + if let Ok(mut state) = self.point.state.lock() { + state.writer_activator = None; + state.writer = WriterState::Frozen; + } + } +} + +/// The points of several attachments, locked together by [`Attachment::lock_all`]. +/// +/// Indices are positions in the slice passed to `lock_all`. +pub struct Locked<'a, T, I, M, const N: usize> { + guards: Vec>, +} + +/// One point locked by [`Attachment::lock_all`]. +struct Guard<'a, T, I, M, const N: usize> { + /// Whether the attachment may change the published view, which it may until it closes. + publishing: bool, + state: MutexGuard<'a, State>, +} + +impl<'a, T: Timestamp, I, M, const N: usize> Locked<'a, T, I, M, N> { + /// Replaces `lower` with `upper` in the holds of `kind` on the point at `index`, and returns + /// whether that moved the meet. Applies whether or not the attachment has closed. + /// + /// A moved meet wakes nobody: the caller is the point's writer, and reads the meet under these + /// locks. + pub fn adjust(&mut self, index: usize, kind: usize, lower: &[T], upper: &[T]) -> bool { + adjust(&mut self.guards[index].state.holds[kind], lower, upper) + } + + /// The meet of the holds of `kind` across every locked point. + pub fn meet(&self, kind: usize) -> Antichain { + let mut meet = Antichain::new(); + for guard in &self.guards { + for time in guard.state.holds[kind].frontier().iter() { + meet.insert_ref(time); + } + } + meet + } + + /// Applies `f` to the metadata of every point where the attachment has not closed. + pub fn update_open(&mut self, mut f: impl FnMut(&mut M)) { + for guard in &mut self.guards { + if guard.publishing { + f(&mut guard.state.published.meta); + } + } + } +} + +/// A `Send` handle holding one frontier per hold kind on a [`Rendezvous`]. +/// +/// Each clone registers its own holds, so two owners cannot release each other's. Dropping a +/// reader releases its holds and wakes the writer if that moved a meet. +pub struct Reader { + point: Arc>, + /// Kept locally so changes adjust the accumulation by a delta, and so `hold` can borrow. + holds: [Antichain; N], +} + +impl Reader { + /// The point this reader holds. + pub fn point(&self) -> &Arc> { + &self.point + } + + /// This reader's hold of `kind`. + pub fn hold(&self, kind: usize) -> AntichainRef<'_, T> { + self.holds[kind].borrow() + } + + /// Replaces this reader's hold of `kind` with `frontier`, waking the writer if that moved the + /// meet. + /// + /// A hold below one the writer already applied brings back nothing the writer discarded. + pub fn set_hold(&mut self, kind: usize, frontier: Antichain) { + let previous = std::mem::replace(&mut self.holds[kind], frontier); + let mut state = self.point.lock(); + if adjust(&mut state.holds[kind], &previous, &self.holds[kind]) { + self.point.holds_moved.store(true, Ordering::Release); + state.wake_writer(); + } + } +} + +impl Clone for Reader { + fn clone(&self) -> Self { + { + let mut state = self.point.lock(); + for (accumulated, hold) in state.holds.iter_mut().zip_eq(self.holds.iter()) { + adjust(accumulated, &[], hold); + } + } + Reader { + point: Arc::clone(&self.point), + holds: self.holds.clone(), + } + } +} + +impl Drop for Reader { + fn drop(&mut self) { + if let Ok(mut state) = self.point.state.lock() { + let mut moved = false; + for (accumulated, hold) in state.holds.iter_mut().zip_eq(self.holds.iter()) { + moved |= adjust(accumulated, hold, &[]); + } + if moved { + self.point.holds_moved.store(true, Ordering::Release); + state.wake_writer(); + } + } + } +} + +/// A subscriber's queue on a [`Rendezvous`], from [`Rendezvous::subscribe`]. +/// +/// Dropping it unregisters the subscriber. It does not keep the point reachable, see +/// [`Attachment::reachable`]. +pub struct Subscription { + queue: Arc>, +} + +impl Subscription { + /// Takes every event queued so far, in order. Takes only the subscription's own lock, never + /// the point's. + pub fn drain(&self) -> Vec> { + let mut events = self + .queue + .events + .lock() + .expect("subscription queue poisoned"); + events.drain(..).collect() + } +} + +#[cfg(test)] +mod tests; diff --git a/src/timely-util/src/rendezvous/tests.rs b/src/timely-util/src/rendezvous/tests.rs new file mode 100644 index 0000000000000..eb1dfe35dfcac --- /dev/null +++ b/src/timely-util/src/rendezvous/tests.rs @@ -0,0 +1,811 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use std::cell::RefCell; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; + +use itertools::Itertools; +use proptest::prelude::*; +use proptest::strategy::Union; +use timely::progress::Antichain; +use timely::progress::frontier::MutableAntichain; +use timely::scheduling::activate::{Activations, SyncActivator}; + +use super::{Attachment, Event, Reader, Rendezvous, Subscription, adjust}; + +/// Items are `u64`s, the metadata counts the publishes that reached the point, and there are two +/// hold kinds. +type Point = Rendezvous; + +fn at(t: u64) -> Antichain { + Antichain::from_elem(t) +} + +fn frontier(t: Option) -> Antichain { + t.map_or_else(Antichain::new, at) +} + +/// Records the activations of paths `[0]`, `[1]`, ... so a test can see which writer was woken. +struct Wakes(Activations); + +impl Wakes { + fn new() -> Self { + Wakes(Activations::new(None)) + } + + fn activator(&self, path: usize) -> SyncActivator { + SyncActivator::new(vec![path], self.0.sync()) + } + + /// The paths activated since the last call. + fn take(&mut self) -> BTreeSet { + self.0.advance(); + let woken = RefCell::new(BTreeSet::new()); + self.0.map_active(|path| { + woken.borrow_mut().insert(path[0]); + }); + woken.into_inner() + } +} + +/// A toy writer publishing the log `0..len` with `upper` at `len`. +fn publish_log(attachment: &mut Attachment, len: u64) -> bool { + let arrived = len.checked_sub(1); + attachment.publish( + (0..len).collect(), + at(len), + |count| *count += 1, + arrived.as_ref(), + ) +} + +fn upper(point: &Point) -> Antichain { + point.read(|published| published.upper.clone()) +} + +/// The accumulated hold updates of `kind`, consolidated. +fn hold_updates(point: &Point, kind: usize) -> Vec<(u64, i64)> { + let mut state = point.lock(); + state.holds[kind].updates().cloned().collect() +} + +#[mz_ore::test] +fn a_hold_at_or_above_the_meet_does_not_move_it() { + let mut holds = MutableAntichain::new(); + assert!( + adjust(&mut holds, &Antichain::new(), &at(5)), + "the first hold sets the meet" + ); + assert!( + !adjust(&mut holds, &Antichain::new(), &at(7)), + "a second hold above the meet leaves it at 5" + ); + assert!( + !adjust(&mut holds, &at(7), &at(9)), + "advancing the hold above the meet leaves it at 5" + ); + assert!( + !adjust(&mut holds, &at(9), &Antichain::new()), + "releasing the hold above the meet leaves it at 5" + ); +} + +#[mz_ore::test] +fn moving_the_lowest_hold_moves_the_meet() { + let mut holds = MutableAntichain::new(); + adjust(&mut holds, &Antichain::new(), &at(5)); + adjust(&mut holds, &Antichain::new(), &at(7)); + assert!(adjust(&mut holds, &at(5), &at(6)), "the meet advances to 6"); + assert!( + adjust(&mut holds, &at(6), &Antichain::new()), + "the meet advances to 7" + ); + assert!( + adjust(&mut holds, &at(7), &Antichain::new()), + "the meet becomes empty" + ); +} + +#[mz_ore::test] +fn a_reader_wakes_the_writer_only_when_it_moves_a_meet() { + let mut wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let mut low = point + .reader_with(|_| Ok::<_, ()>([at(2), at(2)])) + .expect("infallible"); + let mut high = point + .reader_with(|_| Ok::<_, ()>([at(5), at(5)])) + .expect("infallible"); + assert_eq!(wakes.take(), BTreeSet::new(), "registering wakes nobody"); + + high.set_hold(0, at(6)); + assert_eq!(wakes.take(), BTreeSet::new(), "the meet stays at 2"); + assert!(!writer.take_holds_moved()); + low.set_hold(0, at(3)); + assert_eq!(wakes.take(), BTreeSet::from([0]), "the meet moved to 3"); + assert!(writer.take_holds_moved()); + assert!(!writer.take_holds_moved(), "taking the flag resets it"); + assert_eq!(point.holds(0), at(3)); + assert_eq!(point.holds(1), at(2), "kinds are independent"); + + drop(high); + assert_eq!(wakes.take(), BTreeSet::new(), "dropping above the meet"); + assert!(!writer.take_holds_moved()); + drop(low); + assert_eq!(wakes.take(), BTreeSet::from([0]), "dropping the last hold"); + assert!(writer.take_holds_moved()); + assert!(point.holds(0).is_empty()); + assert!(point.holds(1).is_empty()); +} + +#[mz_ore::test] +fn reader_with_refuses_without_registering() { + let point = Arc::new(Point::with_meta(0)); + let refused = point.reader_with(|published| Err::<[Antichain; 2], _>(published.meta)); + assert_eq!(refused.err(), Some(0)); + assert!(point.holds(0).is_empty()); + assert!(point.holds(1).is_empty()); +} + +#[mz_ore::test] +fn dropping_the_writer_clears_the_wakeup() { + let mut wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let mut reader = point + .reader_with(|_| Ok::<_, ()>([at(1), at(1)])) + .expect("infallible"); + drop(writer); + reader.set_hold(0, at(2)); + assert_eq!(wakes.take(), BTreeSet::new()); +} + +/// Writer drop leaves readers at the last published `upper`, whether the writer closes before it +/// drops, drops without closing, or tries to publish the empty frontier after closing. +#[mz_ore::test] +fn writer_drop_freezes_readers_at_the_last_published_upper() { + for close_first in [false, true] { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let (subscription, _) = point.subscribe(wakes.activator(1)); + assert!(publish_log(&mut writer, 4)); + if close_first { + writer.close(); + assert!(writer.is_closed()); + assert!( + !writer.publish(Vec::new(), Antichain::new(), |_| {}, None), + "a closed writer does not publish" + ); + } + drop(writer); + assert_eq!(upper(&point), at(4)); + let uppers: Vec<_> = subscription + .drain() + .into_iter() + .filter_map(|event| match event { + Event::Upper(upper) => Some(upper), + Event::Item(_) => None, + }) + .collect(); + assert_eq!(uppers.last(), Some(&at(4))); + assert!( + uppers.iter().all(|upper| !upper.is_empty()), + "no subscriber sees the empty frontier" + ); + let reader = point + .reader_with(|published| Ok::<_, ()>([published.upper.clone(), Antichain::new()])) + .expect("infallible"); + assert_eq!(reader.hold(0), at(4).borrow()); + } +} + +#[derive(Clone, Debug)] +enum HoldOp { + NewReader(Option, Option), + SetHold(prop::sample::Index, usize, Option), + CloneReader(prop::sample::Index), + DropReader(prop::sample::Index), + Attach(Option, Option), + WriterSet(prop::sample::Index, usize, Option), + DropWriter(prop::sample::Index), +} + +fn hold_op() -> impl Strategy { + let time = || prop::option::weighted(0.9, 0..8u64); + Union::new(vec![ + (time(), time()) + .prop_map(|(a, b)| HoldOp::NewReader(a, b)) + .boxed(), + (any::(), 0..2usize, time()) + .prop_map(|(i, k, t)| HoldOp::SetHold(i, k, t)) + .boxed(), + any::() + .prop_map(HoldOp::CloneReader) + .boxed(), + any::() + .prop_map(HoldOp::DropReader) + .boxed(), + (time(), time()) + .prop_map(|(a, b)| HoldOp::Attach(a, b)) + .boxed(), + (any::(), 0..2usize, time()) + .prop_map(|(i, k, t)| HoldOp::WriterSet(i, k, t)) + .boxed(), + any::() + .prop_map(HoldOp::DropWriter) + .boxed(), + ]) +} + +/// The reference accumulation: the count of every time held, per kind. +#[derive(Default)] +struct Model([BTreeMap; 2]); + +impl Model { + fn update(&mut self, kind: usize, lower: &Antichain, upper: &Antichain) { + for time in upper.iter() { + *self.0[kind].entry(*time).or_default() += 1; + } + for time in lower.iter() { + *self.0[kind].entry(*time).or_default() -= 1; + } + self.0[kind].retain(|_, count| *count != 0); + } + + fn meet(&self, kind: usize) -> Antichain { + frontier(self.0[kind].keys().next().copied()) + } +} + +/// A toy writer that registers holds of its own at attachment and releases them before dropping. +struct ToyWriter { + attachment: Attachment, + holds: [Antichain; 2], +} + +impl ToyWriter { + fn set(&mut self, kind: usize, next: Antichain) { + let previous = std::mem::replace(&mut self.holds[kind], next); + let mut locked = Attachment::lock_all(&[&self.attachment]); + locked.adjust(0, kind, &previous, &self.holds[kind]); + } +} + +proptest! { + /// After every operation the point's meets equal the reference model's, and once every reader + /// and writer has dropped every accumulation is empty. + #[mz_ore::test] + #[cfg_attr(miri, ignore)] + fn holds_balance(ops in prop::collection::vec(hold_op(), 0..64)) { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut model = Model::default(); + let mut readers: Vec> = Vec::new(); + let mut writers: Vec = Vec::new(); + // The point accepts one writer in its life, so later `Attach`es are skipped. + let mut attached = false; + for op in ops { + match op { + HoldOp::NewReader(a, b) => { + let holds = [frontier(a), frontier(b)]; + for (kind, hold) in holds.iter().enumerate() { + model.update(kind, &Antichain::new(), hold); + } + readers.push( + point.reader_with(|_| Ok::<_, ()>(holds)).expect("infallible"), + ); + } + HoldOp::SetHold(index, kind, t) => { + if readers.is_empty() { + continue; + } + let index = index.index(readers.len()); + let reader = &mut readers[index]; + let next = frontier(t); + model.update(kind, &reader.hold(kind).to_owned(), &next); + reader.set_hold(kind, next); + } + HoldOp::CloneReader(index) => { + if readers.is_empty() { + continue; + } + let clone = readers[index.index(readers.len())].clone(); + for kind in 0..2 { + model.update(kind, &Antichain::new(), &clone.hold(kind).to_owned()); + } + readers.push(clone); + } + HoldOp::DropReader(index) => { + if readers.is_empty() { + continue; + } + let reader = readers.swap_remove(index.index(readers.len())); + for kind in 0..2 { + model.update(kind, &reader.hold(kind).to_owned(), &Antichain::new()); + } + } + HoldOp::Attach(a, b) => { + if std::mem::replace(&mut attached, true) { + continue; + } + let holds = [frontier(a), frontier(b)]; + for (kind, hold) in holds.iter().enumerate() { + model.update(kind, &Antichain::new(), hold); + } + let updates = holds.clone().map(|hold| { + hold.into_iter().map(|time| (time, 1)).collect() + }); + let attachment = point.attach( + wakes.activator(writers.len()), + Vec::new(), + at(0), + |_| {}, + updates, + ); + writers.push(ToyWriter { attachment, holds }); + } + HoldOp::WriterSet(index, kind, t) => { + if writers.is_empty() { + continue; + } + let index = index.index(writers.len()); + let writer = &mut writers[index]; + let next = frontier(t); + model.update(kind, &writer.holds[kind], &next); + writer.set(kind, next); + } + HoldOp::DropWriter(index) => { + if writers.is_empty() { + continue; + } + let mut writer = writers.swap_remove(index.index(writers.len())); + for kind in 0..2 { + model.update(kind, &writer.holds[kind], &Antichain::new()); + writer.set(kind, Antichain::new()); + } + } + } + for kind in 0..2 { + prop_assert_eq!(point.holds(kind), model.meet(kind)); + } + } + for mut writer in writers.drain(..) { + for kind in 0..2 { + writer.set(kind, Antichain::new()); + } + } + readers.clear(); + for kind in 0..2 { + prop_assert!(hold_updates(&point, kind).is_empty()); + } + } +} + +#[derive(Clone, Debug)] +enum UpperOp { + /// The writer publishes the next item to both points. + Publish, + /// The writer closes both attachments. + Close, + /// The writer drops both attachments. + Drop, +} + +fn upper_op() -> impl Strategy { + Union::new_weighted(vec![ + (6, Just(UpperOp::Publish).boxed()), + (1, Just(UpperOp::Close).boxed()), + (1, Just(UpperOp::Drop).boxed()), + ]) +} + +proptest! { + /// With a writer publishing a growing log to two points, closing and dropping, the `upper` of + /// every point only moves forward. + #[mz_ore::test] + #[cfg_attr(miri, ignore)] + fn upper_is_monotone_at_every_point(ops in prop::collection::vec(upper_op(), 0..64)) { + let wakes = Wakes::new(); + let points = [Arc::new(Point::with_meta(0)), Arc::new(Point::with_meta(0))]; + let mut attachments: Vec> = points + .iter() + .map(|point| { + point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ) + }) + .collect(); + let mut len = 0; + let mut previous: Vec> = points.iter().map(|p| upper(p)).collect(); + for op in ops { + match op { + UpperOp::Publish => { + len += 1; + for attachment in attachments.iter_mut() { + publish_log(attachment, len); + } + } + UpperOp::Close => { + for attachment in attachments.iter_mut() { + attachment.close(); + } + } + UpperOp::Drop => attachments.clear(), + } + for (point, previous) in points.iter().zip_eq(previous.iter_mut()) { + let now = upper(point); + prop_assert!( + timely::PartialOrder::less_equal(previous, &now), + "upper regressed from {:?} to {:?}", + previous, + now + ); + *previous = now; + } + } + } +} + +/// Checks that `events` is a seed `Item(0..s), Upper(s)` followed by `Item(k), Upper(k + 1)` for +/// every `k` in `s..len`. +fn assert_seed_then_deltas(events: Vec>, len: u64) { + let mut items = Vec::new(); + let mut last_upper = None; + for event in events { + match event { + Event::Item(item) => items.push(item), + Event::Upper(upper) => { + assert_eq!( + upper, + at(u64::try_from(items.len()).expect("fits")), + "an upper covers exactly the items before it" + ); + last_upper = Some(upper); + } + } + } + assert_eq!( + items, + (0..len).collect::>(), + "no gap and no duplicate" + ); + assert_eq!(last_upper, Some(at(len))); +} + +#[mz_ore::test] +fn a_subscriber_sees_the_seed_and_every_later_item_once() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let (early, seed) = point.subscribe(wakes.activator(1)); + assert_eq!(seed, at(0)); + for len in 1..=3 { + publish_log(&mut writer, len); + } + let (late, seed) = point.subscribe(wakes.activator(2)); + assert_eq!(seed, at(3)); + for len in 4..=5 { + publish_log(&mut writer, len); + } + // A publish without an arrived item, as after merge work, enqueues nothing. + writer.publish((0..5).collect(), at(5), |_| {}, None); + assert_seed_then_deltas(early.drain(), 5); + assert_seed_then_deltas(late.drain(), 5); +} + +#[mz_ore::test] +fn a_subscriber_registered_before_the_writer_attaches_is_seeded() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let (subscription, seed) = point.subscribe(wakes.activator(1)); + assert_eq!(seed, at(0)); + let mut writer = point.attach( + wakes.activator(0), + (0..3).collect(), + at(3), + |_| {}, + [Vec::new(), Vec::new()], + ); + publish_log(&mut writer, 4); + // The empty seed's `Upper(0)`, then the attachment's seed, then the delta. + let events = subscription.drain(); + assert!(matches!(events.first(), Some(Event::Upper(upper)) if *upper == at(0))); + assert_seed_then_deltas(events.into_iter().skip(1).collect(), 4); +} + +#[mz_ore::test] +#[cfg_attr(miri, ignore)] +fn subscribers_registered_while_the_writer_publishes_see_no_gap_and_no_duplicate() { + const LEN: u64 = 2000; + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let publisher = std::thread::spawn(move || { + for len in 1..=LEN { + publish_log(&mut writer, len); + } + writer + }); + // Bounded, since each subscription buffers up to the whole log. + let mut subscriptions: Vec> = Vec::new(); + while !publisher.is_finished() && subscriptions.len() < 128 { + subscriptions.push(point.subscribe(wakes.activator(1)).0); + std::thread::sleep(std::time::Duration::from_micros(20)); + } + let _writer = publisher.join().expect("publisher panicked"); + assert!(!subscriptions.is_empty()); + for subscription in subscriptions { + assert_seed_then_deltas(subscription.drain(), LEN); + } +} + +#[mz_ore::test] +fn a_closed_writer_publishes_no_metadata_but_adjusts_holds() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [vec![(1, 1)], Vec::new()], + ); + writer.close(); + { + let mut locked = Attachment::lock_all(&[&writer]); + locked.update_open(|count| *count = 99); + assert!( + locked.adjust(0, 0, &[1], &[2]), + "a closed writer's holds move" + ); + } + assert_eq!( + point.read(|published| published.meta), + 0, + "a closed writer leaves the metadata as last published" + ); + assert_eq!(point.holds(0), at(2)); +} + +#[mz_ore::test] +fn the_writers_own_adjust_wakes_nobody() { + let mut wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [vec![(1, 1)], Vec::new()], + ); + let (_subscription, _) = point.subscribe(wakes.activator(1)); + wakes.take(); + { + let mut locked = Attachment::lock_all(&[&writer]); + assert!(locked.adjust(0, 0, &[1], &[2]), "the meet moved"); + } + assert_eq!(wakes.take(), BTreeSet::new()); +} + +#[mz_ore::test] +fn a_publish_without_an_arrived_item_still_reports_a_moved_upper() { + let mut wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + vec![0], + at(1), + |_| {}, + [Vec::new(), Vec::new()], + ); + let (subscription, _) = point.subscribe(wakes.activator(1)); + subscription.drain(); + wakes.take(); + + assert!( + !writer.publish(vec![0], at(1), |_| {}, None), + "an unchanged upper" + ); + assert!( + subscription.drain().is_empty(), + "nothing moved, nothing queued" + ); + assert_eq!(wakes.take(), BTreeSet::new()); + + assert!(writer.publish(vec![0], at(3), |_| {}, None)); + assert!( + matches!(subscription.drain().as_slice(), [Event::Upper(upper)] if *upper == at(3)), + "the moved upper reaches the subscriber" + ); + assert_eq!(wakes.take(), BTreeSet::from([1])); +} + +#[mz_ore::test] +#[should_panic(expected = "publish regresses upper")] +fn publishing_a_regressing_upper_is_refused_by_a_soft_assert() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + publish_log(&mut writer, 3); + writer.publish(Vec::new(), at(1), |_| {}, None); +} + +/// Attaches a second writer to a point whose first writer is still attached, has closed, or has +/// dropped, as `end` names. +fn attach_twice(end: &str) { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let mut first = Some(point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + )); + match end { + "closed" => first.as_mut().expect("attached").close(), + "dropped" => first = None, + _ => {} + } + let _second = point.attach( + wakes.activator(1), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + drop(first); +} + +#[mz_ore::test] +#[should_panic(expected = "attach to a point that already had a writer")] +fn a_point_refuses_a_second_writer_while_the_first_is_attached() { + attach_twice("attached"); +} + +#[mz_ore::test] +#[should_panic(expected = "attach to a point that already had a writer")] +fn a_point_refuses_a_second_writer_after_the_first_closed() { + attach_twice("closed"); +} + +#[mz_ore::test] +#[should_panic(expected = "attach to a point that already had a writer")] +fn a_point_refuses_a_second_writer_after_the_first_dropped() { + attach_twice("dropped"); +} + +#[mz_ore::test] +fn a_point_freezes_when_its_writer_closes_or_drops() { + let wakes = Wakes::new(); + for close in [false, true] { + let point = Arc::new(Point::with_meta(0)); + assert!(!point.is_frozen(), "an unattached point is not frozen"); + let mut writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + assert!(!point.is_frozen(), "an attached point is not frozen"); + if close { + writer.close(); + assert!(point.is_frozen(), "closing freezes the point"); + } + drop(writer); + assert!(point.is_frozen(), "dropping freezes the point"); + } +} + +#[mz_ore::test] +fn into_holds_returns_the_writers_holds() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [vec![(3, 1), (5, 1)], vec![(4, 1)]], + ); + { + let mut locked = Attachment::lock_all(&[&writer]); + locked.adjust(0, 0, &[3], &[6]); + } + drop(point); + let consolidated = writer.into_holds().map(|updates| { + let mut counts = BTreeMap::::new(); + for (time, diff) in updates { + *counts.entry(time).or_default() += diff; + } + counts + .into_iter() + .filter(|(_, count)| *count != 0) + .collect::>() + }); + assert_eq!(consolidated, [vec![(5, 1), (6, 1)], vec![(4, 1)]]); +} + +#[mz_ore::test] +#[should_panic(expected = "into_holds: the point is referenced elsewhere")] +fn into_holds_on_a_point_others_reference_is_refused_by_a_soft_assert() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let _reader = point + .reader_with(|_| Ok::<_, ()>([at(1), at(1)])) + .expect("infallible"); + drop(point); + let _ = writer.into_holds(); +} + +#[mz_ore::test] +#[should_panic(expected = "lock_all: a point is named twice")] +fn lock_all_refuses_a_point_named_twice() { + let wakes = Wakes::new(); + let point = Arc::new(Point::with_meta(0)); + let writer = point.attach( + wakes.activator(0), + Vec::new(), + at(0), + |_| {}, + [Vec::new(), Vec::new()], + ); + let _ = Attachment::lock_all(&[&writer, &writer]); +}