refactor(scheduler): use bitwise flags for scheduler jobs + move scheduler into reactivity

related: https://github.com/vuejs/core/pull/10407

packages/reactivity/__tests__/baseWatch.spec.ts

@@ -1,8 +1,9 @@
-import type { Scheduler, SchedulerJob } from '../src/baseWatch'
 import {
   BaseWatchErrorCodes,
   EffectScope,
   type Ref,
+  type SchedulerJob,
+  type WatchScheduler,
   baseWatch,
   onEffectCleanup,
   ref,
@@ -15,10 +16,14 @@ let isFlushPending = false
 const resolvedPromise = /*#__PURE__*/ Promise.resolve() as Promise<any>
 const nextTick = (fn?: () => any) =>
   fn ? resolvedPromise.then(fn) : resolvedPromise
-const scheduler: Scheduler = job => {
-  queue.push(job)
+const scheduler: WatchScheduler = (job, effect, immediateFirstRun, hasCb) => {
+  if (immediateFirstRun) {
+    !hasCb && effect.run()
+  } else {
+    queue.push(() => job(immediateFirstRun))
     flushJobs()
   }
+}
 const flushJobs = () => {
   if (isFlushPending) return
   isFlushPending = true
@@ -214,7 +219,11 @@ describe('baseWatch', () => {
       },
     )
 
-    expect(effectCalls).toEqual([])
+    expect(effectCalls).toEqual([
+      'before effect running',
+      'effect',
+      'effect ran',
+    ])
     expect(watchCalls).toEqual([])
     await nextTick()
     expect(effectCalls).toEqual([

packages/reactivity/src/baseWatch.ts

@@ -22,7 +22,7 @@ import {
 } from './effect'
 import { isReactive, isShallow } from './reactive'
 import { type Ref, isRef } from './ref'
-import { getCurrentScope } from './effectScope'
+import { type SchedulerJob, SchedulerJobFlags } from './scheduler'
 
 // These errors were transferred from `packages/runtime-core/src/errorHandling.ts`
 // along with baseWatch to maintain code compatibility. Hence,
@@ -33,32 +33,6 @@ export enum BaseWatchErrorCodes {
   WATCH_CLEANUP,
 }
 
-// TODO move to a scheduler package
-export interface SchedulerJob extends Function {
-  id?: number
-  // TODO refactor these boolean flags to a single bitwise flag
-  pre?: boolean
-  active?: boolean
-  computed?: boolean
-  queued?: boolean
-  /**
-   * Indicates whether the effect is allowed to recursively trigger itself
-   * when managed by the scheduler.
-   *
-   * By default, a job cannot trigger itself because some built-in method calls,
-   * e.g. Array.prototype.push actually performs reads as well (#1740) which
-   * can lead to confusing infinite loops.
-   * The allowed cases are component update functions and watch callbacks.
-   * Component update functions may update child component props, which in turn
-   * trigger flush: "pre" watch callbacks that mutates state that the parent
-   * relies on (#1801). Watch callbacks doesn't track its dependencies so if it
-   * triggers itself again, it's likely intentional and it is the user's
-   * responsibility to perform recursive state mutation that eventually
-   * stabilizes (#1727).
-   */
-  allowRecurse?: boolean
-}
-
 type WatchEffect = (onCleanup: OnCleanup) => void
 type WatchSource<T = any> = Ref<T> | ComputedRef<T> | (() => T)
 type WatchCallback<V = any, OV = any> = (
@@ -254,8 +228,11 @@ export function baseWatch(
   let oldValue: any = isMultiSource
     ? new Array((source as []).length).fill(INITIAL_WATCHER_VALUE)
     : INITIAL_WATCHER_VALUE
-  const job: SchedulerJob = () => {
-    if (!effect.active || !effect.dirty) {
+  const job: SchedulerJob = (immediateFirstRun?: boolean) => {
+    if (
+      !(effect.flags & EffectFlags.ACTIVE) ||
+      (!effect.dirty && !immediateFirstRun)
+    ) {
       return
     }
     if (cb) {
@@ -310,11 +287,10 @@ export function baseWatch(
 
   // important: mark the job as a watcher callback so that scheduler knows
   // it is allowed to self-trigger (#1727)
-  job.allowRecurse = !!cb
+  if (cb) job.flags! |= SchedulerJobFlags.ALLOW_RECURSE
 
-  let effectScheduler: EffectScheduler = () => scheduler(job, effect, false)
-
-  effect = new ReactiveEffect(getter, NOOP, effectScheduler, scope)
+  effect = new ReactiveEffect(getter)
+  effect.scheduler = () => scheduler(job, effect, false, !!cb)
 
   cleanup = effect.onStop = () => {
     const cleanups = cleanupMap.get(effect)
@@ -337,13 +313,14 @@ export function baseWatch(
 
   // initial run
   if (cb) {
+    scheduler(job, effect, true, !!cb)
     if (immediate) {
-      job()
+      job(true)
     } else {
       oldValue = effect.run()
     }
   } else {
-    scheduler(job, effect, true)
+    scheduler(job, effect, true, !!cb)
   }
 
   return effect

packages/runtime-core/__tests__/scheduler.spec.ts

@@ -1,6 +1,6 @@
+import { SchedulerJobFlags } from '@vue/reactivity'
 import {
   type SchedulerJob,
-  SchedulerJobFlags,
   flushPostFlushCbs,
   flushPreFlushCbs,
   invalidateJob,

packages/runtime-core/src/components/BaseTransition.ts

@@ -14,12 +14,11 @@ import {
 } from '../vnode'
 import { warn } from '../warning'
 import { isKeepAlive } from './KeepAlive'
-import { toRaw } from '@vue/reactivity'
+import { SchedulerJobFlags, toRaw } from '@vue/reactivity'
 import { ErrorCodes, callWithAsyncErrorHandling } from '../errorHandling'
 import { PatchFlags, ShapeFlags, isArray } from '@vue/shared'
 import { onBeforeUnmount, onMounted } from '../apiLifecycle'
 import type { RendererElement } from '../renderer'
-import { SchedulerJobFlags } from '../scheduler'
 
 type Hook<T = () => void> = T | T[]
 

packages/runtime-core/src/renderer.ts

@@ -40,7 +40,6 @@ import {
 import {
   type SchedulerFactory,
   type SchedulerJob,
-  SchedulerJobFlags,
   flushPostFlushCbs,
   flushPreFlushCbs,
   invalidateJob,
@@ -50,6 +49,7 @@ import {
 import {
   EffectFlags,
   ReactiveEffect,
+  SchedulerJobFlags,
   pauseTracking,
   resetTracking,
 } from '@vue/reactivity'
@@ -289,14 +289,14 @@ export const queuePostRenderEffect = __FEATURE_SUSPENSE__
   : queuePostFlushCb
 
 export const createPostRenderScheduler: SchedulerFactory =
-  instance => (job, effect, isInit) => {
-    if (isInit) {
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (!immediateFirstRun) {
+      queuePostRenderEffect(job, instance && instance.suspense)
+    } else if (!hasCb) {
       queuePostRenderEffect(
         effect.run.bind(effect),
         instance && instance.suspense,
       )
-    } else {
-      queuePostRenderEffect(job, instance && instance.suspense)
     }
   }
 

packages/runtime-core/src/scheduler.ts

@@ -1,37 +1,14 @@
 import { ErrorCodes, callWithErrorHandling, handleError } from './errorHandling'
 import { type Awaited, NOOP, isArray } from '@vue/shared'
 import { type ComponentInternalInstance, getComponentName } from './component'
-import type { Scheduler } from '@vue/reactivity'
+import {
+  type SchedulerJob as BaseSchedulerJob,
+  EffectFlags,
+  SchedulerJobFlags,
+  type WatchScheduler,
+} from '@vue/reactivity'
 
-export enum SchedulerJobFlags {
-  QUEUED = 1 << 0,
-  PRE = 1 << 1,
-  /**
-   * Indicates whether the effect is allowed to recursively trigger itself
-   * when managed by the scheduler.
-   *
-   * By default, a job cannot trigger itself because some built-in method calls,
-   * e.g. Array.prototype.push actually performs reads as well (#1740) which
-   * can lead to confusing infinite loops.
-   * The allowed cases are component update functions and watch callbacks.
-   * Component update functions may update child component props, which in turn
-   * trigger flush: "pre" watch callbacks that mutates state that the parent
-   * relies on (#1801). Watch callbacks doesn't track its dependencies so if it
-   * triggers itself again, it's likely intentional and it is the user's
-   * responsibility to perform recursive state mutation that eventually
-   * stabilizes (#1727).
-   */
-  ALLOW_RECURSE = 1 << 2,
-  DISPOSED = 1 << 3,
-}
-
-export interface SchedulerJob extends Function {
-  id?: number
-  /**
-   * flags can technically be undefined, but it can still be used in bitwise
-   * operations just like 0.
-   */
-  flags?: SchedulerJobFlags
+export interface SchedulerJob extends BaseSchedulerJob {
   /**
    * Attached by renderer.ts when setting up a component's render effect
    * Used to obtain component information when reporting max recursive updates.
@@ -301,24 +278,25 @@ function checkRecursiveUpdates(seen: CountMap, fn: SchedulerJob) {
 
 export type SchedulerFactory = (
   instance: ComponentInternalInstance | null,
-) => Scheduler
+) => WatchScheduler
 
 export const createSyncScheduler: SchedulerFactory =
-  instance => (job, effect, isInit) => {
-    if (isInit) {
-      effect.run()
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (immediateFirstRun) {
+      effect.flags |= EffectFlags.NO_BATCH
+      if (!hasCb) effect.run()
     } else {
       job()
     }
   }
 
 export const createPreScheduler: SchedulerFactory =
-  instance => (job, effect, isInit) => {
-    if (isInit) {
-      effect.run()
-    } else {
-      job.pre = true
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (!immediateFirstRun) {
+      job.flags! |= SchedulerJobFlags.PRE
       if (instance) job.id = instance.uid
       queueJob(job)
+    } else if (!hasCb) {
+      effect.run()
     }
   }

packages/runtime-vapor/__tests__/renderWatch.spec.ts

@@ -38,11 +38,24 @@ describe('renderWatch', () => {
     renderEffect(() => {
       dummy = source.value
     })
+    expect(dummy).toBe(0)
     await nextTick()
     expect(dummy).toBe(0)
+
     source.value++
+    expect(dummy).toBe(0)
     await nextTick()
     expect(dummy).toBe(1)
+
+    source.value++
+    expect(dummy).toBe(1)
+    await nextTick()
+    expect(dummy).toBe(2)
+
+    source.value++
+    expect(dummy).toBe(2)
+    await nextTick()
+    expect(dummy).toBe(3)
   })
 
   test('watch', async () => {
@@ -53,9 +66,16 @@ describe('renderWatch', () => {
     })
     await nextTick()
     expect(dummy).toBe(undefined)
+
     source.value++
+    expect(dummy).toBe(undefined)
     await nextTick()
     expect(dummy).toBe(1)
+
+    source.value++
+    expect(dummy).toBe(1)
+    await nextTick()
+    expect(dummy).toBe(2)
   })
 
   test('should run with the scheduling order', async () => {
@@ -136,6 +156,28 @@ describe('renderWatch', () => {
       'post 1',
       'updated 1',
     ])
+    calls.length = 0
+
+    // Update
+    changeRender()
+    change()
+
+    expect(calls).toEqual(['sync cleanup 1', 'sync 2'])
+    calls.length = 0
+
+    await nextTick()
+    expect(calls).toEqual([
+      'pre cleanup 1',
+      'pre 2',
+      'beforeUpdate 2',
+      'renderEffect cleanup 1',
+      'renderEffect 2',
+      'renderWatch cleanup 1',
+      'renderWatch 2',
+      'post cleanup 1',
+      'post 2',
+      'updated 2',
+    ])
   })
 
   test('errors should include the execution location with beforeUpdate hook', async () => {

packages/runtime-vapor/src/scheduler.ts

@@ -1,4 +1,9 @@
-import type { Scheduler, SchedulerJob } from '@vue/reactivity'
+import {
+  EffectFlags,
+  type SchedulerJob,
+  SchedulerJobFlags,
+  type WatchScheduler,
+} from '@vue/reactivity'
 import type { ComponentInternalInstance } from './component'
 import { isArray } from '@vue/shared'
 
@@ -28,19 +33,21 @@ const resolvedPromise = /*#__PURE__*/ Promise.resolve() as Promise<any>
 let currentFlushPromise: Promise<void> | null = null
 
 function queueJob(job: SchedulerJob) {
-  if (!job.queued) {
+  if (!(job.flags! & SchedulerJobFlags.QUEUED)) {
     if (job.id == null) {
       queue.push(job)
-    } else {
+    } else if (
       // fast path when the job id is larger than the tail
-      if (!job.pre && job.id >= (queue[queue.length - 1]?.id || 0)) {
+      !(job.flags! & SchedulerJobFlags.PRE) &&
+      job.id >= (queue[queue.length - 1]?.id || 0)
+    ) {
       queue.push(job)
     } else {
       queue.splice(findInsertionIndex(job.id), 0, job)
     }
-    }
-    if (!job.allowRecurse) {
-      job.queued = true
+
+    if (!(job.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+      job.flags! |= SchedulerJobFlags.QUEUED
     }
     queueFlush()
   }
@@ -48,10 +55,10 @@ function queueJob(job: SchedulerJob) {
 
 export function queuePostRenderEffect(cb: SchedulerJobs) {
   if (!isArray(cb)) {
-    if (!cb.queued) {
+    if (!(cb.flags! & SchedulerJobFlags.QUEUED)) {
       pendingPostFlushCbs.push(cb)
-      if (!cb.allowRecurse) {
-        cb.queued = true
+      if (!(cb.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+        cb.flags! |= SchedulerJobFlags.QUEUED
       }
     }
   } else {
@@ -92,7 +99,7 @@ export function flushPostFlushCbs() {
     postFlushIndex++
   ) {
     activePostFlushCbs[postFlushIndex]()
-    activePostFlushCbs[postFlushIndex].queued = false
+    activePostFlushCbs[postFlushIndex].flags! &= ~SchedulerJobFlags.QUEUED
   }
   activePostFlushCbs = null
   postFlushIndex = 0
@@ -114,8 +121,8 @@ function flushJobs() {
 
   try {
     for (let i = 0; i < queue!.length; i++) {
-      queue![i]()
-      queue![i].queued = false
+      queue[i]()
+      queue[i].flags! &= ~SchedulerJobFlags.QUEUED
     }
   } finally {
     flushIndex = 0
@@ -154,7 +161,10 @@ function findInsertionIndex(id: number) {
     const middle = (start + end) >>> 1
     const middleJob = queue[middle]
     const middleJobId = getId(middleJob)
-    if (middleJobId < id || (middleJobId === id && middleJob.pre)) {
+    if (
+      middleJobId < id ||
+      (middleJobId === id && middleJob.flags! & SchedulerJobFlags.PRE)
+    ) {
       start = middle + 1
     } else {
       end = middle
@@ -170,52 +180,54 @@ const getId = (job: SchedulerJob): number =>
 const comparator = (a: SchedulerJob, b: SchedulerJob): number => {
   const diff = getId(a) - getId(b)
   if (diff === 0) {
-    if (a.pre && !b.pre) return -1
-    if (b.pre && !a.pre) return 1
+    const isAPre = a.flags! & SchedulerJobFlags.PRE
+    const isBPre = b.flags! & SchedulerJobFlags.PRE
+    if (isAPre && !isBPre) return -1
+    if (isBPre && !isAPre) return 1
   }
   return diff
 }
 
 export type SchedulerFactory = (
   instance: ComponentInternalInstance | null,
-) => Scheduler
+) => WatchScheduler
 
 export const createVaporSyncScheduler: SchedulerFactory =
-  () => (job, effect, isInit) => {
-    if (isInit) {
-      effect.run()
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (immediateFirstRun) {
+      effect.flags |= EffectFlags.NO_BATCH
+      if (!hasCb) effect.run()
     } else {
       job()
     }
   }
 
 export const createVaporPreScheduler: SchedulerFactory =
-  instance => (job, effect, isInit) => {
-    if (isInit) {
-      effect.run()
-    } else {
-      job.pre = true
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (!immediateFirstRun) {
+      job.flags! |= SchedulerJobFlags.PRE
       if (instance) job.id = instance.uid
       queueJob(job)
+    } else if (!hasCb) {
+      effect.run()
     }
   }
 
 export const createVaporRenderingScheduler: SchedulerFactory =
-  instance => (job, effect, isInit) => {
-    if (isInit) {
-      effect.run()
-    } else {
-      job.pre = false
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (!immediateFirstRun) {
       if (instance) job.id = instance.uid
       queueJob(job)
+    } else if (!hasCb) {
+      effect.run()
     }
   }
 
 export const createVaporPostScheduler: SchedulerFactory =
-  () => (job, effect, isInit) => {
-    if (isInit) {
-      queuePostRenderEffect(effect.run.bind(effect))
-    } else {
+  instance => (job, effect, immediateFirstRun, hasCb) => {
+    if (!immediateFirstRun) {
       queuePostRenderEffect(job)
+    } else if (!hasCb) {
+      queuePostRenderEffect(effect.run.bind(effect))
     }
   }