@ -14,7 +14,6 @@ import {
MAYBE _DIRTY ,
DERIVED ,
EAGER _EFFECT ,
HEAD _EFFECT ,
ERROR _VALUE ,
MANAGED _EFFECT ,
REACTION _RAN
@ -23,6 +22,7 @@ import { async_mode_flag } from '../../flags/index.js';
import { deferred , define _property , includes } from '../../shared/utils.js' ;
import {
active _effect ,
active _reaction ,
get ,
increment _write _version ,
is _dirty ,
@ -37,6 +37,7 @@ import { eager_effect, unlink_effect } from './effects.js';
import { defer _effect } from './utils.js' ;
import { UNINITIALIZED } from '../../../constants.js' ;
import { set _signal _status } from './status.js' ;
import { legacy _is _updating _store } from './store.js' ;
/** @type {Set<Batch>} */
const batches = new Set ( ) ;
@ -45,8 +46,7 @@ const batches = new Set();
export let current _batch = null ;
/ * *
* This is needed to avoid overwriting inputs in non - async mode
* TODO 6.0 remove this , as non - async mode will go away
* This is needed to avoid overwriting inputs
* @ type { Batch | null }
* /
export let previous _batch = null ;
@ -59,14 +59,11 @@ export let previous_batch = null;
* /
export let batch _values = null ;
// TODO this should really be a property of `batch`
/** @type {Effect[]} */
let queued _root _effects = [ ] ;
/** @type {Effect | null} */
let last _scheduled _effect = null ;
export let is _flushing _sync = false ;
let is _processing = false ;
/ * *
* During traversal , this is an array . Newly created effects are ( if not immediately
@ -76,7 +73,24 @@ export let is_flushing_sync = false;
* /
export let collected _effects = null ;
/ * *
* An array of effects that are marked during traversal as a result of a ` set `
* ( not ` internal_set ` ) call . These will be added to the next batch and
* trigger another ` batch.process() `
* @ type { Effect [ ] | null }
* @ deprecated when we get rid of legacy mode and stores , we can get rid of this
* /
export let legacy _updates = null ;
var flush _count = 0 ;
var source _stacks = DEV ? new Set ( ) : null ;
let uid = 1 ;
export class Batch {
// for debugging. TODO remove once async is stable
id = uid ++ ;
/ * *
* The current values of any sources that are updated in this batch
* They keys of this map are identical to ` this.#previous `
@ -121,6 +135,12 @@ export class Batch {
* /
# deferred = null ;
/ * *
* The root effects that need to be flushed
* @ type { Effect [ ] }
* /
# roots = [ ] ;
/ * *
* Deferred effects ( which run after async work has completed ) that are DIRTY
* @ type { Set < Effect > }
@ -172,22 +192,23 @@ export class Batch {
for ( var e of tracked . d ) {
set _signal _status ( e , DIRTY ) ;
schedule _effect ( e ) ;
this . schedule ( e ) ;
}
for ( e of tracked . m ) {
set _signal _status ( e , MAYBE _DIRTY ) ;
schedule _effect ( e ) ;
this . schedule ( e ) ;
}
}
}
/ * *
*
* @ param { Effect [ ] } root _effects
* /
process ( root _effects ) {
queued _root _effects = [ ] ;
# process ( ) {
if ( flush _count ++ > 1000 ) {
infinite _loop _guard ( ) ;
}
const roots = this . # roots ;
this . # roots = [ ] ;
this . apply ( ) ;
@ -197,16 +218,28 @@ export class Batch {
/** @type {Effect[]} */
var render _effects = [ ] ;
for ( const root of root _effects ) {
this . # traverse _effect _tree ( root , effects , render _effects ) ;
// Note: #traverse_effect_tree runs block effects eagerly, which can schedule effects,
// which means queued_root_effects now may be filled again.
/ * *
* @ type { Effect [ ] }
* @ deprecated when we get rid of legacy mode and stores , we can get rid of this
* /
var updates = ( legacy _updates = [ ] ) ;
for ( const root of roots ) {
this . # traverse ( root , effects , render _effects ) ;
}
// any writes should take effect in a subsequent batch
current _batch = null ;
// Helpful for debugging reactivity loss that has to do with branches being skipped:
// log_inconsistent_branches(root);
if ( updates . length > 0 ) {
var batch = Batch . ensure ( ) ;
for ( const e of updates ) {
batch . schedule ( e ) ;
}
}
collected _effects = null ;
legacy _updates = null ;
if ( this . # is _deferred ( ) ) {
this . # defer _effects ( render _effects ) ;
@ -216,32 +249,39 @@ export class Batch {
reset _branch ( e , t ) ;
}
} else {
// If sources are written to, then work needs to happen in a separate batch, else prior sources would be mixed with
// newly updated sources, which could lead to infinite loops when effects run over and over again.
previous _batch = this ;
current _batch = null ;
// clear effects. Those that are still needed will be rescheduled through unskipping the skipped branches.
this . # dirty _effects . clear ( ) ;
this . # maybe _dirty _effects . clear ( ) ;
// append/remove branches
for ( const fn of this . # commit _callbacks ) fn ( this ) ;
this . # commit _callbacks . clear ( ) ;
previous _batch = this ;
flush _queued _effects ( render _effects ) ;
flush _queued _effects ( effects ) ;
previous _batch = null ;
if ( this . # pending === 0 ) {
this . # commit ( ) ;
}
flush _queued _effects ( render _effects ) ;
flush _queued _effects ( effects ) ;
this . # deferred ? . resolve ( ) ;
}
// Clear effects. Those that are still needed will be rescheduled through unskipping the skipped branches.
this . # dirty _effects . clear ( ) ;
this . # maybe _dirty _effects . clear ( ) ;
var next _batch = /** @type {Batch | null} */ ( /** @type {unknown} */ ( current _batch ) ) ;
previous _batch = null ;
if ( next _batch !== null ) {
batches . add ( next _batch ) ;
this . # deferred ? . resolve ( ) ;
}
if ( DEV ) {
for ( const source of this . current . keys ( ) ) {
/** @type {Set<Source>} */ ( source _stacks ) . add ( source ) ;
}
}
batch _values = null ;
next _batch . # process ( ) ;
}
}
/ * *
@ -251,7 +291,7 @@ export class Batch {
* @ param { Effect [ ] } effects
* @ param { Effect [ ] } render _effects
* /
# traverse _effect _tree ( root , effects , render _effects ) {
# traverse ( root , effects , render _effects ) {
root . f ^= CLEAN ;
var effect = root . first ;
@ -325,32 +365,54 @@ export class Batch {
activate ( ) {
current _batch = this ;
this . apply ( ) ;
}
deactivate ( ) {
// If we're not the current batch, don't deactivate,
// else we could create zombie batches that are never flushed
if ( current _batch !== this ) return ;
current _batch = null ;
batch _values = null ;
}
flush ( ) {
if ( queued _root _effects . length > 0 ) {
var source _stacks = DEV ? new Set ( ) : null ;
try {
is _processing = true ;
current _batch = this ;
flush _effects ( ) ;
} else if ( this . # pending === 0 && ! this . is _fork ) {
// append/remove branches
for ( const fn of this . # commit _callbacks ) fn ( this ) ;
this . # commit _callbacks . clear ( ) ;
this . # commit ( ) ;
this . # deferred ? . resolve ( ) ;
}
// we only reschedule previously-deferred effects if we expect
// to be able to run them after processing the batch
if ( ! this . # is _deferred ( ) ) {
for ( const e of this . # dirty _effects ) {
this . # maybe _dirty _effects . delete ( e ) ;
set _signal _status ( e , DIRTY ) ;
this . schedule ( e ) ;
}
for ( const e of this . # maybe _dirty _effects ) {
set _signal _status ( e , MAYBE _DIRTY ) ;
this . schedule ( e ) ;
}
}
this . # process ( ) ;
} finally {
flush _count = 0 ;
last _scheduled _effect = null ;
collected _effects = null ;
legacy _updates = null ;
is _processing = false ;
current _batch = null ;
batch _values = null ;
old _values . clear ( ) ;
this . deactivate ( ) ;
if ( DEV ) {
for ( const source of /** @type {Set<Source>} */ ( source _stacks ) ) {
source . updated = null ;
}
}
}
}
discard ( ) {
@ -401,9 +463,7 @@ export class Batch {
// Re-run async/block effects that depend on distinct values changed in both batches
const others = [ ... batch . current . keys ( ) ] . filter ( ( s ) => ! this . current . has ( s ) ) ;
if ( others . length > 0 ) {
// Avoid running queued root effects on the wrong branch
var prev _queued _root _effects = queued _root _effects ;
queued _root _effects = [ ] ;
batch . activate ( ) ;
/** @type {Set<Value>} */
const marked = new Set ( ) ;
@ -413,20 +473,17 @@ export class Batch {
mark _effects ( source , others , marked , checked ) ;
}
if ( queued _root _effects . length > 0 ) {
current _batch = batch ;
if ( batch . # roots . length > 0 ) {
batch . apply ( ) ;
for ( const root of queued_root _effec ts) {
batch . # traverse _effect _tree ( root , [ ] , [ ] ) ;
for ( const root of batch. # roo ts) {
batch . # traverse ( root , [ ] , [ ] ) ;
}
// TODO do we need to do anything with the dummy effect arrays?
batch . deactivate ( ) ;
}
queued_root _effects = prev _queued _root _effects ;
batch. deactivate ( ) ;
}
}
@ -448,46 +505,22 @@ export class Batch {
}
/ * *
*
* @ param { boolean } blocking
* @ param { boolean } skip - whether to skip updates ( because this is triggered by a stale reaction )
* /
decrement ( blocking {
decrement ( blocking , skip ) {
this . # pending -= 1 ;
if ( blocking ) this . # blocking _pending -= 1 ;
if ( this . # decrement _queued ) return ;
if ( this . # decrement _queued || skip ) return ;
this . # decrement _queued = true ;
queue _micro _task ( ( ) => {
this . # decrement _queued = false ;
if ( ! this . # is _deferred ( ) ) {
// we only reschedule previously-deferred effects if we expect
// to be able to run them after processing the batch
this . revive ( ) ;
} else if ( queued _root _effects . length > 0 ) {
// if other effects are scheduled, process the batch _without_
// rescheduling the previously-deferred effects
this . flush ( ) ;
}
this . flush ( ) ;
} ) ;
}
revive ( ) {
for ( const e of this . # dirty _effects ) {
this . # maybe _dirty _effects . delete ( e ) ;
set _signal _status ( e , DIRTY ) ;
schedule _effect ( e ) ;
}
for ( const e of this . # maybe _dirty _effects ) {
set _signal _status ( e , MAYBE _DIRTY ) ;
schedule _effect ( e ) ;
}
this . flush ( ) ;
}
/** @param {(batch: Batch) => void} fn */
oncommit ( fn ) {
this . # commit _callbacks . add ( fn ) ;
@ -505,17 +538,20 @@ export class Batch {
static ensure ( ) {
if ( current _batch === null ) {
const batch = ( current _batch = new Batch ( ) ) ;
batches . add ( current _batch ) ;
if ( ! is _flushing _sync ) {
queue _micro _task ( ( ) => {
if ( current _batch !== batch ) {
// a flushSync happened in the meantime
return ;
}
if ( ! is _processing ) {
batches . add ( current _batch ) ;
if ( ! is _flushing _sync ) {
queue _micro _task ( ( ) => {
if ( current _batch !== batch ) {
// a flushSync happened in the meantime
return ;
}
batch . flush ( ) ;
} ) ;
batch . flush ( ) ;
} ) ;
}
}
}
@ -559,6 +595,63 @@ export class Batch {
}
}
}
/ * *
*
* @ param { Effect } effect
* /
schedule ( effect ) {
last _scheduled _effect = effect ;
// defer render effects inside a pending boundary
// TODO the `REACTION_RAN` check is only necessary because of legacy `$:` effects AFAICT — we can remove later
if (
effect . b ? . is _pending &&
( effect . f & ( EFFECT | RENDER _EFFECT | MANAGED _EFFECT ) ) !== 0 &&
( effect . f & REACTION _RAN ) === 0
) {
effect . b . defer _effect ( effect ) ;
return ;
}
var e = effect ;
while ( e . parent !== null ) {
e = e . parent ;
var flags = e . f ;
// if the effect is being scheduled because a parent (each/await/etc) block
// updated an internal source, or because a branch is being unskipped,
// bail out or we'll cause a second flush
if ( collected _effects !== null && e === active _effect ) {
if ( async _mode _flag ) return ;
// in sync mode, render effects run during traversal. in an extreme edge case
// — namely that we're setting a value inside a derived read during traversal —
// they can be made dirty after they have already been visited, in which
// case we shouldn't bail out. we also shouldn't bail out if we're
// updating a store inside a `$:`, since this might invalidate
// effects that were already visited
if (
( active _reaction === null || ( active _reaction . f & DERIVED ) === 0 ) &&
! legacy _is _updating _store
) {
return ;
}
}
if ( ( flags & ( ROOT _EFFECT | BRANCH _EFFECT ) ) !== 0 ) {
if ( ( flags & CLEAN ) === 0 ) {
// branch is already dirty, bail
return ;
}
e . f ^= CLEAN ;
}
}
this . # roots . push ( e ) ;
}
}
/ * *
@ -576,8 +669,8 @@ export function flushSync(fn) {
var result ;
if ( fn ) {
if ( current _batch !== null ) {
flush_effects ( ) ;
if ( current _batch !== null && ! current _batch . is _fork ) {
current_batch . flush ( ) ;
}
result = fn ( ) ;
@ -586,87 +679,42 @@ export function flushSync(fn) {
while ( true ) {
flush _tasks ( ) ;
if ( queued _root _effects . length === 0 ) {
current _batch ? . flush ( ) ;
// we need to check again, in case we just updated an `$effect.pending()`
if ( queued _root _effects . length === 0 ) {
// this would be reset in `flush_effects()` but since we are early returning here,
// we need to reset it here as well in case the first time there's 0 queued root effects
last _scheduled _effect = null ;
return /** @type {T} */ ( result ) ;
}
if ( current _batch === null ) {
return /** @type {T} */ ( result ) ;
}
flush_effects ( ) ;
current _batch . flush ( ) ;
}
} finally {
is _flushing _sync = was _flushing _sync ;
}
}
function flush _effects ( ) {
var source _stacks = DEV ? new Set ( ) : null ;
try {
var flush _count = 0 ;
while ( queued _root _effects . length > 0 ) {
var batch = Batch . ensure ( ) ;
if ( flush _count ++ > 1000 ) {
if ( DEV ) {
var updates = new Map ( ) ;
for ( const source of batch . current . keys ( ) ) {
for ( const [ stack , update ] of source . updated ? ? [ ] ) {
var entry = updates . get ( stack ) ;
if ( ! entry ) {
entry = { error : update . error , count : 0 } ;
updates . set ( stack , entry ) ;
}
function infinite _loop _guard ( ) {
if ( DEV ) {
var updates = new Map ( ) ;
entry . count += update . count ;
}
}
for ( const source of /** @type {Batch} */ ( current _batch ) . current . keys ( ) ) {
for ( const [ stack , update ] of source . updated ? ? [ ] ) {
var entry = updates . get ( stack ) ;
for ( const update of updates . values ( ) ) {
if ( update . error ) {
// eslint-disable-next-line no-console
console . error ( update . error ) ;
}
}
if ( ! entry ) {
entry = { error : update . error , count : 0 } ;
updates . set ( stack , entry ) ;
}
infinite _loop _guard ( ) ;
}
batch . process ( queued _root _effects ) ;
old _values . clear ( ) ;
if ( DEV ) {
for ( const source of batch . current . keys ( ) ) {
/** @type {Set<Source>} */ ( source _stacks ) . add ( source ) ;
}
entry . count += update . count ;
}
}
} finally {
queued _root _effects = [ ] ;
last _scheduled _effect = null ;
collected _effects = null ;
if ( DEV ) {
for ( const source of /** @type {Set<Source>} */ ( source _stacks ) ) {
source . updated = null ;
for ( const update of updates . values ( ) ) {
if ( update . error ) {
// eslint-disable-next-line no-console
console . error ( update . error ) ;
}
}
}
}
function infinite _loop _guard ( ) {
try {
e . effect _update _depth _exceeded ( ) ;
} catch ( error ) {
@ -836,52 +884,11 @@ function depends_on(reaction, sources, checked) {
}
/ * *
* @ param { Effect } signal
* @ param { Effect } effect
* @ returns { void }
* /
export function schedule _effect ( signal ) {
var effect = ( last _scheduled _effect = signal ) ;
var boundary = effect . b ;
// defer render effects inside a pending boundary
// TODO the `REACTION_RAN` check is only necessary because of legacy `$:` effects AFAICT — we can remove later
if (
boundary ? . is _pending &&
( signal . f & ( EFFECT | RENDER _EFFECT | MANAGED _EFFECT ) ) !== 0 &&
( signal . f & REACTION _RAN ) === 0
) {
boundary . defer _effect ( signal ) ;
return ;
}
while ( effect . parent !== null ) {
effect = effect . parent ;
var flags = effect . f ;
// if the effect is being scheduled because a parent (each/await/etc) block
// updated an internal source, or because a branch is being unskipped,
// bail out or we'll cause a second flush
if ( collected _effects !== null && effect === active _effect ) {
// in sync mode, render effects run during traversal. in an extreme edge case
// they can be made dirty after they have already been visited, in which
// case we shouldn't bail out
if ( async _mode _flag || ( signal . f & RENDER _EFFECT ) === 0 ) {
return ;
}
}
if ( ( flags & ( ROOT _EFFECT | BRANCH _EFFECT ) ) !== 0 ) {
if ( ( flags & CLEAN ) === 0 ) {
// branch is already dirty, bail
return ;
}
effect . f ^= CLEAN ;
}
}
queued _root _effects . push ( effect ) ;
export function schedule _effect ( effect ) {
/** @type {Batch} */ ( current _batch ) . schedule ( effect ) ;
}
/** @type {Source<number>[]} */
@ -1058,7 +1065,7 @@ export function fork(fn) {
flush _eager _effects ( ) ;
} ) ;
batch . revive ( ) ;
batch . flush ( ) ;
await settled ;
} ,
discard : ( ) => {
Rich Harris
2 days ago