fix: treat batches as a linked list (#18205)

This is another attempt to address some of the tricky edge cases that
arise when async batches resolve out of order. The idea is this:

Essentially, when a batch resolves:

1. we find the latest batch it shares changes with
2. if none exists (either because this is the earliest batch, or because
it is independent of any earlier batches):
    - we commit it: effects are flushed, `oncommit` callbacks are run
- we restart any async work in later batches that depends on both the
committed values and the later batch's changes
3. otherwise, we don't commit the batch. instead:
- we merge the changes from the later batch onto the earlier batch. in
some cases this may mean restarting async work on the earlier batch, but
we avoid doing so unnecessarily.
    - we then `#process()` the earlier batch. if it resolves, goto 1

This feels like it ought to work. There are still two failing tests,
which I'm currently looking into.

Notable changes:

- Instead of having a `batches` set, we have a linked list. When a batch
resolves, this makes it easy to find the batch that it should be merged
into
- The `#is_deferred` logic now takes account of skipped effects —
there's no need to wait for a promise inside a falsy `if` block (this
accounts for the change in the `async-inner-after-outer` test)
- We no longer care about `#blockers`

I would like to believe that this approach will allow us to simplify and
delete some code, for example the `rebase` logic, though that remains to
be seen. It also feels like some version of #18035 would be helpful.

- closes #18189
- closes #18162
- fixes https://github.com/sveltejs/kit/issues/15431

### Before submitting the PR, please make sure you do the following

- [x] It's really useful if your PR references an issue where it is
discussed ahead of time. In many cases, features are absent for a
reason. For large changes, please create an RFC:
https://github.com/sveltejs/rfcs
- [x] Prefix your PR title with `feat:`, `fix:`, `chore:`, or `docs:`.
- [x] This message body should clearly illustrate what problems it
solves.
- [x] Ideally, include a test that fails without this PR but passes with
it.
- [x] If this PR changes code within `packages/svelte/src`, add a
changeset (`npx changeset`).

### Tests and linting

- [x] Run the tests with `pnpm test` and lint the project with `pnpm
lint`

---------

Co-authored-by: Simon H <5968653+dummdidumm@users.noreply.github.com>
Co-authored-by: Simon Holthausen <simon.holthausen@vercel.com>
pull/18217/head
Rich Harris 2 months ago committed by GitHub
parent 1688c84be9
commit 5122936edb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -0,0 +1,5 @@
---
"svelte": patch
---
fix: replacing async 'blocking' strategy with 'merging'

@ -1,8 +1,8 @@
/** @import { TemplateNode } from '#client' */
import { hydrate_node, hydrating, set_hydrate_node, set_hydrating } from '../hydration.js';
import { create_text, get_first_child, get_next_sibling } from '../operations.js';
import { block } from '../../reactivity/effects.js';
import { COMMENT_NODE, EFFECT_PRESERVED, HEAD_EFFECT } from '#client/constants';
import { block, branch } from '../../reactivity/effects.js';
import { COMMENT_NODE, HEAD_EFFECT } from '#client/constants';
/**
* @param {string} hash
@ -49,9 +49,10 @@ export function head(hash, render_fn) {
}
try {
// normally a branch is the child of a block and would have the EFFECT_PRESERVED flag,
// but since head blocks don't necessarily only have direct branch children we add it on the block itself
block(() => render_fn(anchor), HEAD_EFFECT | EFFECT_PRESERVED);
block(() => {
var e = branch(() => render_fn(anchor));
e.f |= HEAD_EFFECT;
});
} finally {
if (was_hydrating) {
set_hydrating(true);

@ -41,8 +41,11 @@ import { legacy_is_updating_store } from './store.js';
import { invariant } from '../../shared/dev.js';
import { log_effect_tree } from '../dev/debug.js';
/** @type {Set<Batch>} */
const batches = new Set();
/** @type {Batch | null} */
let first_batch = null;
/** @type {Batch | null} */
let last_batch = null;
/** @type {Batch | null} */
export let current_batch = null;
@ -94,9 +97,20 @@ let uid = 1;
export class Batch {
id = uid++;
/** True as soon as `#process()` was called */
/** True as soon as `#process` was called */
#started = false;
linked = true;
/** @type {Batch | null} */
#prev = null;
/** @type {Batch | null} */
#next = null;
/** @type {Map<Effect, ReturnType<typeof deferred<any>>>} */
async_deriveds = new Map();
/**
* The current values of any signals that are updated in this batch.
* Tuple format: [value, is_derived] (note: is_derived is false for deriveds, too, if they were overridden via assignment)
@ -199,33 +213,24 @@ export class Batch {
#decrement_queued = false;
/** @type {Set<Batch>} */
#blockers = new Set();
#is_deferred() {
return this.is_fork || this.#blocking_pending.size > 0;
}
#is_blocked() {
for (const batch of this.#blockers) {
for (const effect of batch.#blocking_pending.keys()) {
if (this.unblocked.has(effect)) continue;
if (this.is_fork) return true;
var skipped = false;
var e = effect;
for (const effect of this.#blocking_pending.keys()) {
var e = effect;
var skipped = false;
while (e.parent !== null) {
if (this.#skipped_branches.has(e)) {
skipped = true;
break;
}
e = e.parent;
while (e.parent !== null) {
if (this.#skipped_branches.has(e)) {
skipped = true;
break;
}
if (!skipped) {
return true;
}
e = e.parent;
}
if (!skipped) {
return true;
}
}
@ -271,7 +276,7 @@ export class Batch {
this.#started = true;
if (flush_count++ > 1000) {
batches.delete(this);
this.#unlink();
infinite_loop_guard();
}
@ -337,7 +342,8 @@ export class Batch {
collected_effects = null;
legacy_updates = null;
if (this.#is_deferred() || this.#is_blocked()) {
// if the batch has outstanding pending work, stash effects and bail
if (this.#is_deferred()) {
this.#defer_effects(render_effects);
this.#defer_effects(effects);
@ -352,6 +358,13 @@ export class Batch {
return;
}
const earlier_batch = this.#find_earlier_batch();
if (earlier_batch) {
earlier_batch.#merge(this);
return;
}
// clear effects. Those that are still needed will be rescheduled through unskipping the skipped branches.
this.#dirty_effects.clear();
this.#maybe_dirty_effects.clear();
@ -369,11 +382,15 @@ export class Batch {
var next_batch = /** @type {Batch | null} */ (/** @type {unknown} */ (current_batch));
if (this.linked && this.#pending === 0) {
this.#unlink();
}
// Order matters here - we need to commit and THEN continue flushing new batches, not the other way around,
// else we could start flushing a new batch and then, if it has pending work, rebase it right afterwards, which is wrong.
// In sync mode flushSync can cause #commit to wrongfully think that there needs to be a rebase, so we only do it in async mode
// TODO fix the underlying cause, otherwise this will likely regress when non-async mode is removed
if (async_mode_flag && this.#pending === 0) {
if (async_mode_flag && !this.linked) {
this.#commit();
// Rebases can activate other batches or null it out, therefore restore the new one here
current_batch = next_batch;
@ -383,8 +400,12 @@ export class Batch {
// causing an effect and therefore a root to be scheduled again. We need to traverse the current batch
// once more in that case - most of the time this will just clean up dirty branches.
if (this.#roots.length > 0) {
const batch = (next_batch ??= this);
batches.add(batch);
if (next_batch === null) {
next_batch = this;
this.#link();
}
const batch = next_batch;
batch.#roots.push(...this.#roots.filter((r) => !batch.#roots.includes(r)));
}
@ -445,6 +466,82 @@ export class Batch {
}
}
#find_earlier_batch() {
var batch = this.#prev;
while (batch !== null) {
if (!batch.is_fork) {
// if the batches are connected, break
for (const [value, [, is_derived]] of this.current) {
if (batch.current.has(value) && !is_derived) {
return batch;
}
}
}
batch = batch.#prev;
}
return null;
}
/**
* @param {Batch} batch
*/
#merge(batch) {
for (const [source, value] of batch.current) {
if (!this.previous.has(source) && batch.previous.has(source)) {
this.previous.set(source, batch.previous.get(source));
}
this.current.set(source, value);
}
for (const [effect, deferred] of batch.async_deriveds) {
const d = this.async_deriveds.get(effect);
if (d) deferred.promise.then(d.resolve);
}
/**
* mark all effects that depend on `batch.current`, except the
* async effects that we just resolved (TODO unless they depend
* on values in this batch that are NOT in the later batch?).
* Through this we also will populate the correct #skipped_branches,
* oncommit callbacks etc, so we don't need to merge them separately.
* @param {Value} value
*/
const mark = (value) => {
var reactions = value.reactions;
if (reactions === null) return;
for (const reaction of reactions) {
var flags = reaction.f;
if ((flags & DERIVED) !== 0) {
mark(/** @type {Derived} */ (reaction));
} else {
var effect = /** @type {Effect} */ (reaction);
if (flags & (ASYNC | BLOCK_EFFECT) && !this.async_deriveds.has(effect)) {
this.#maybe_dirty_effects.delete(effect);
set_signal_status(effect, DIRTY);
this.schedule(effect);
}
}
}
};
for (const source of this.current.keys()) {
mark(source);
}
this.oncommit(() => batch.discard());
batch.#unlink();
current_batch = this;
this.#process();
}
/**
* @param {Effect[]} effects
*/
@ -521,7 +618,7 @@ export class Batch {
this.#discard_callbacks.clear();
this.#fork_commit_callbacks.clear();
batches.delete(this);
this.#unlink();
}
/**
@ -532,13 +629,13 @@ export class Batch {
}
#commit() {
batches.delete(this);
this.#unlink();
// If there are other pending batches, they now need to be 'rebased' —
// in other words, we re-run block/async effects with the newly
// committed state, unless the batch in question has a more
// recent value for a given source
for (const batch of batches) {
for (let batch = first_batch; batch !== null; batch = batch.#next) {
var is_earlier = batch.id < this.id;
/** @type {Source[]} */
@ -561,6 +658,15 @@ export class Batch {
sources.push(source);
}
if (is_earlier) {
// TODO do we need to restart these in some cases, instead of
// immediately resolving them? Likely not because of how this.apply() works.
for (const [effect, deferred] of this.async_deriveds) {
const d = batch.async_deriveds.get(effect);
if (d) deferred.promise.then(d.resolve);
}
}
if (!batch.#started) continue;
// Re-run async/block effects that depend on distinct values changed in both batches
@ -638,17 +744,6 @@ export class Batch {
batch.deactivate();
}
}
for (const batch of batches) {
if (batch.#blockers.has(this)) {
batch.#blockers.delete(this);
if (batch.#blockers.size === 0 && !batch.#is_deferred()) {
batch.activate();
batch.#process();
}
}
}
}
/**
@ -687,7 +782,7 @@ export class Batch {
queue_micro_task(() => {
this.#decrement_queued = false;
if (batches.has(this)) {
if (this.linked) {
this.flush();
}
});
@ -737,7 +832,7 @@ export class Batch {
static ensure() {
if (current_batch === null) {
const batch = (current_batch = new Batch());
batches.add(batch);
batch.#link();
if (!is_processing && !is_flushing_sync) {
queue_micro_task(() => {
@ -752,7 +847,7 @@ export class Batch {
}
apply() {
if (!async_mode_flag || (!this.is_fork && batches.size === 1)) {
if (!async_mode_flag || (!this.is_fork && this.#prev === null && this.#next === null)) {
batch_values = null;
return;
}
@ -764,28 +859,33 @@ export class Batch {
batch_values.set(source, value);
}
// ...and undo changes belonging to other batches unless they block this one
for (const batch of batches) {
// ...and undo changes belonging to other batches unless they intersect
for (let batch = first_batch; batch !== null; batch = batch.#next) {
if (batch === this || batch.is_fork) continue;
// A batch is blocked on an earlier batch if it overlaps with the earlier batch's changes but is not a superset
// If two batches intersect, the latter batch will be merged into the earlier batch,
// and we should treat them as a single set of changes
var intersects = false;
var differs = false;
if (batch.id < this.id) {
for (const [source, [, is_derived]] of batch.current) {
// Derived values don't partake in the blocking mechanism, because a derived could
// Derived values don't partake in the intersection mechanism, because a derived could
// be triggered in one batch already but not the other one yet, causing a false-positive
if (is_derived) continue;
intersects ||= this.current.has(source);
differs ||= !this.current.has(source);
if (this.current.has(source)) {
intersects = true;
break;
}
}
}
if (intersects && differs) {
this.#blockers.add(batch);
} else {
// Since the latter batch merges into the earlier (if it resolves before the earlier one),
// we treat the earlier values as "already applied". This way we don't need to rerun async
// effects of the earlier batch in case they are merged.
// As a result you can think of batch_values as having the latest values of all intersecting
// batches up until this batch.
if (!intersects) {
for (const [source, previous] of batch.previous) {
if (!batch_values.has(source)) {
batch_values.set(source, previous);
@ -851,6 +951,36 @@ export class Batch {
this.#roots.push(e);
}
#link() {
if (last_batch === null) {
first_batch = last_batch = this;
} else {
last_batch.#next = this;
this.#prev = last_batch;
}
last_batch = this;
}
#unlink() {
var prev = this.#prev;
var next = this.#next;
if (prev === null) {
first_batch = next;
} else {
prev.#next = next;
}
if (next === null) {
last_batch = prev;
} else {
next.#prev = prev;
}
this.linked = false;
}
}
// TODO Svelte@6 think about removing the callback argument.
@ -1234,7 +1364,7 @@ export function fork(fn) {
return;
}
if (!batches.has(batch)) {
if (!batch.linked) {
e.fork_discarded();
}
@ -1280,7 +1410,7 @@ export function fork(fn) {
source.wv = increment_write_version();
}
if (!committed && batches.has(batch)) {
if (!committed && batch.linked) {
batch.discard();
}
}
@ -1291,5 +1421,5 @@ export function fork(fn) {
* Forcibly remove all current batches, to prevent cross-talk between tests
*/
export function clear() {
batches.clear();
first_batch = last_batch = null;
}

@ -100,7 +100,7 @@ export function derived(fn) {
return signal;
}
const OBSOLETE = {};
export const OBSOLETE = Symbol('obsolete');
/**
* @template V
@ -125,8 +125,8 @@ export function async_derived(fn, label, location) {
// only suspend in async deriveds created on initialisation
var should_suspend = !active_reaction;
/** @type {Map<Batch, ReturnType<typeof deferred<V>>>} */
var deferreds = new Map();
/** @type {Set<ReturnType<typeof deferred<V>>>} */
var deferreds = new Set();
async_effect(() => {
var effect = /** @type {Effect} */ (active_effect);
@ -188,7 +188,7 @@ export function async_derived(fn, label, location) {
}
if (/** @type {Boundary} */ (parent.b).is_rendered()) {
deferreds.get(batch)?.reject(OBSOLETE);
batch.async_deriveds.get(effect)?.reject(OBSOLETE);
} else {
// While the boundary is still showing pending, a new run supersedes all older in-flight runs
// for this async expression. Cancel eagerly so resolution cannot commit stale values.
@ -197,7 +197,8 @@ export function async_derived(fn, label, location) {
}
}
deferreds.set(batch, d);
deferreds.add(d);
batch.async_deriveds.set(effect, d);
}
/**
@ -210,7 +211,7 @@ export function async_derived(fn, label, location) {
}
decrement_pending?.();
deferreds.delete(batch);
deferreds.delete(d);
if (error === OBSOLETE) return;
@ -228,18 +229,6 @@ export function async_derived(fn, label, location) {
internal_set(signal, value);
// All prior async derived runs are now stale
for (const [b, d] of deferreds) {
if (b.id < batch.id) {
// Don't delete + resolve directly, instead only do that once
// the current batch commits. This way we avoid tearing when
// `b` is rendering through the early resolve while `batch` is
// still pending.
batch.unblocked.add(effect);
batch.oncommit(() => d.resolve(value));
}
}
if (DEV && location !== undefined) {
recent_async_deriveds.add(signal);
@ -259,7 +248,7 @@ export function async_derived(fn, label, location) {
});
teardown(() => {
for (const d of deferreds.values()) {
for (const d of deferreds) {
d.reject(OBSOLETE);
}
});

@ -34,18 +34,6 @@ export default test({
shift?.click();
await tick();
assert.htmlEqual(
target.innerHTML,
`
<p>true</p>
<button>toggle</button>
<button>shift</button>
`
);
shift?.click();
await tick();
assert.htmlEqual(
target.innerHTML,
`

@ -0,0 +1,20 @@
import { tick } from 'svelte';
import { test } from '../../test';
export default test({
async test({ assert, target }) {
const [increment, pop] = target.querySelectorAll('button');
increment.click();
await tick();
increment.click();
await tick();
pop.click();
await tick();
assert.htmlEqual(target.innerHTML, `<button>increment</button> <button>pop</button> 2 2 1`);
pop.click();
await tick();
assert.htmlEqual(target.innerHTML, `<button>increment</button> <button>pop</button> 2 2 1`);
}
});

@ -0,0 +1,26 @@
<script>
let count = $state(0);
let other = $state(0);
const queue = [];
function push(v) {
return new Promise((r,e) => queue.push(() => v === 1 ? e(v) : r(v)));
}
</script>
<button onclick={() => {
if (count === 0) {
other++;
count++;
} else {
count++
}
}}>increment</button>
<button onclick={() => queue.pop()?.()}>pop</button>
{#if count > 0}
<svelte:boundary>
{await push(count)} {count} {other}
{#snippet failed()}boom{/snippet}
</svelte:boundary>
{/if}

@ -0,0 +1,20 @@
import { tick } from 'svelte';
import { test } from '../../test';
export default test({
async test({ assert, target }) {
const [increment, pop] = target.querySelectorAll('button');
increment.click();
await tick();
increment.click();
await tick();
pop.click();
await tick();
assert.htmlEqual(target.innerHTML, '<button>increment</button><button>pop</button> 2 2 1'); // showing nothing here yet would also be ok
pop.click();
await tick();
assert.htmlEqual(target.innerHTML, '<button>increment</button><button>pop</button> 2 2 1');
}
});

@ -0,0 +1,23 @@
<script>
let count = $state(0);
let other = $state(0);
const queue = [];
function push(v) {
if (v === 0) return v;
return new Promise((resolve) => queue.push(() => resolve(v)));
}
</script>
<button onclick={() => {
if (count === 0) other++;
count++;
}}>increment</button>
<button onclick={() => queue.pop()?.()}>pop</button>
{#if count > 0}
{await push(count)} {count} {other}
{/if}
Loading…
Cancel
Save