| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import {assert, isArray} from 'zrender/src/core/util'; |
| import SeriesModel from '../model/Series'; |
| import { Pipeline } from './Scheduler'; |
| import { Payload } from '../util/types'; |
| import List from '../data/List'; |
| |
| |
| export interface TaskContext { |
| outputData?: List; |
| data?: List; |
| payload?: Payload; |
| model?: SeriesModel; |
| }; |
| |
| export type TaskResetCallback<Ctx extends TaskContext> = ( |
| this: Task<Ctx>, context: Ctx |
| ) => TaskResetCallbackReturn<Ctx>; |
| export type TaskResetCallbackReturn<Ctx extends TaskContext> = |
| void |
| | (TaskProgressCallback<Ctx> | TaskProgressCallback<Ctx>[]) |
| | { |
| forceFirstProgress?: boolean |
| progress: TaskProgressCallback<Ctx> | TaskProgressCallback<Ctx>[] |
| }; |
| export type TaskProgressCallback<Ctx extends TaskContext> = ( |
| this: Task<Ctx>, params: TaskProgressParams, context: Ctx |
| ) => void; |
| export type TaskProgressParams = { |
| start: number, end: number, count: number, next?: TaskDataIteratorNext |
| }; |
| export type TaskPlanCallback<Ctx extends TaskContext> = ( |
| this: Task<Ctx>, context: Ctx |
| ) => TaskPlanCallbackReturn; |
| export type TaskPlanCallbackReturn = 'reset' | false | null | undefined; |
| export type TaskCountCallback<Ctx extends TaskContext> = ( |
| this: Task<Ctx>, context: Ctx |
| ) => number; |
| export type TaskOnDirtyCallback<Ctx extends TaskContext> = ( |
| this: Task<Ctx>, context: Ctx |
| ) => void; |
| |
| type TaskDataIteratorNext = () => number; |
| type TaskDataIterator = { |
| reset: (s: number, e: number, sStep: number, sCount: number) => void, |
| next?: TaskDataIteratorNext |
| }; |
| |
| type TaskDefineParam<Ctx extends TaskContext> = { |
| reset?: TaskResetCallback<Ctx>, |
| // Returns 'reset' indicate reset immediately |
| plan?: TaskPlanCallback<Ctx>, |
| // count is used to determine data task. |
| count?: TaskCountCallback<Ctx>, |
| onDirty?: TaskOnDirtyCallback<Ctx> |
| }; |
| export type PerformArgs = { |
| step?: number, |
| skip?: boolean, |
| modBy?: number, |
| modDataCount?: number |
| }; |
| |
| /** |
| * @param {Object} define |
| * @return See the return of `createTask`. |
| */ |
| export function createTask<Ctx extends TaskContext>( |
| define: TaskDefineParam<Ctx> |
| ): Task<Ctx> { |
| return new Task<Ctx>(define); |
| } |
| |
| export class Task<Ctx extends TaskContext> { |
| |
| private _reset: TaskResetCallback<Ctx>; |
| private _plan: TaskPlanCallback<Ctx>; |
| private _count: TaskCountCallback<Ctx>; |
| private _onDirty: TaskOnDirtyCallback<Ctx>; |
| private _progress: TaskProgressCallback<Ctx> | TaskProgressCallback<Ctx>[]; |
| private _callingProgress: TaskProgressCallback<Ctx>; |
| |
| private _dirty: boolean; |
| private _modBy: number; |
| private _modDataCount: number; |
| private _upstream: Task<Ctx>; |
| private _downstream: Task<Ctx>; |
| private _dueEnd: number; |
| private _outputDueEnd: number; |
| private _settedOutputEnd: number; |
| private _dueIndex: number; |
| private _disposed: boolean; |
| |
| // Injected in schedular |
| __pipeline: Pipeline; |
| __idxInPipeline: number; |
| __block: boolean; |
| |
| // Context must be specified implicitly, to |
| // avoid miss update context when model changed. |
| context: Ctx; |
| |
| constructor(define: TaskDefineParam<Ctx>) { |
| define = define || {}; |
| |
| this._reset = define.reset; |
| this._plan = define.plan; |
| this._count = define.count; |
| this._onDirty = define.onDirty; |
| |
| this._dirty = true; |
| } |
| |
| /** |
| * @param step Specified step. |
| * @param skip Skip customer perform call. |
| * @param modBy Sampling window size. |
| * @param modDataCount Sampling count. |
| * @return whether unfinished. |
| */ |
| perform(performArgs?: PerformArgs): boolean { |
| const upTask = this._upstream; |
| const skip = performArgs && performArgs.skip; |
| |
| // TODO some refactor. |
| // Pull data. Must pull data each time, because context.data |
| // may be updated by Series.setData. |
| if (this._dirty && upTask) { |
| const context = this.context; |
| context.data = context.outputData = upTask.context.outputData; |
| } |
| |
| if (this.__pipeline) { |
| this.__pipeline.currentTask = this; |
| } |
| |
| let planResult; |
| if (this._plan && !skip) { |
| planResult = this._plan(this.context); |
| } |
| |
| // Support sharding by mod, which changes the render sequence and makes the rendered graphic |
| // elements uniformed distributed when progress, especially when moving or zooming. |
| const lastModBy = normalizeModBy(this._modBy); |
| const lastModDataCount = this._modDataCount || 0; |
| const modBy = normalizeModBy(performArgs && performArgs.modBy); |
| const modDataCount = performArgs && performArgs.modDataCount || 0; |
| if (lastModBy !== modBy || lastModDataCount !== modDataCount) { |
| planResult = 'reset'; |
| } |
| |
| function normalizeModBy(val: number) { |
| !(val >= 1) && (val = 1); // jshint ignore:line |
| return val; |
| } |
| |
| let forceFirstProgress; |
| if (this._dirty || planResult === 'reset') { |
| this._dirty = false; |
| forceFirstProgress = this._doReset(skip); |
| } |
| |
| this._modBy = modBy; |
| this._modDataCount = modDataCount; |
| |
| const step = performArgs && performArgs.step; |
| |
| if (upTask) { |
| if (__DEV__) { |
| assert(upTask._outputDueEnd != null); |
| } |
| this._dueEnd = upTask._outputDueEnd; |
| } |
| // DataTask or overallTask |
| else { |
| if (__DEV__) { |
| assert(!this._progress || this._count); |
| } |
| this._dueEnd = this._count ? this._count(this.context) : Infinity; |
| } |
| |
| // Note: Stubs, that its host overall task let it has progress, has progress. |
| // If no progress, pass index from upstream to downstream each time plan called. |
| if (this._progress) { |
| const start = this._dueIndex; |
| const end = Math.min( |
| step != null ? this._dueIndex + step : Infinity, |
| this._dueEnd |
| ); |
| |
| if (!skip && (forceFirstProgress || start < end)) { |
| const progress = this._progress; |
| if (isArray(progress)) { |
| for (let i = 0; i < progress.length; i++) { |
| this._doProgress(progress[i], start, end, modBy, modDataCount); |
| } |
| } |
| else { |
| this._doProgress(progress, start, end, modBy, modDataCount); |
| } |
| } |
| |
| this._dueIndex = end; |
| // If no `outputDueEnd`, assume that output data and |
| // input data is the same, so use `dueIndex` as `outputDueEnd`. |
| const outputDueEnd = this._settedOutputEnd != null |
| ? this._settedOutputEnd : end; |
| |
| if (__DEV__) { |
| // ??? Can not rollback. |
| assert(outputDueEnd >= this._outputDueEnd); |
| } |
| |
| this._outputDueEnd = outputDueEnd; |
| } |
| else { |
| // (1) Some overall task has no progress. |
| // (2) Stubs, that its host overall task do not let it has progress, has no progress. |
| // This should always be performed so it can be passed to downstream. |
| this._dueIndex = this._outputDueEnd = this._settedOutputEnd != null |
| ? this._settedOutputEnd : this._dueEnd; |
| } |
| |
| return this.unfinished(); |
| } |
| |
| dirty(): void { |
| this._dirty = true; |
| this._onDirty && this._onDirty(this.context); |
| } |
| |
| private _doProgress( |
| progress: TaskProgressCallback<Ctx>, |
| start: number, |
| end: number, |
| modBy: number, |
| modDataCount: number |
| ): void { |
| iterator.reset(start, end, modBy, modDataCount); |
| this._callingProgress = progress; |
| this._callingProgress({ |
| start: start, end: end, count: end - start, next: iterator.next |
| }, this.context); |
| } |
| |
| private _doReset(skip: boolean): boolean { |
| this._dueIndex = this._outputDueEnd = this._dueEnd = 0; |
| this._settedOutputEnd = null; |
| |
| let progress: TaskResetCallbackReturn<Ctx>; |
| let forceFirstProgress: boolean; |
| |
| if (!skip && this._reset) { |
| progress = this._reset(this.context); |
| if (progress && (progress as any).progress) { |
| forceFirstProgress = (progress as any).forceFirstProgress; |
| progress = (progress as any).progress; |
| } |
| // To simplify no progress checking, array must has item. |
| if (isArray(progress) && !progress.length) { |
| progress = null; |
| } |
| } |
| |
| this._progress = progress as TaskProgressCallback<Ctx>; |
| this._modBy = this._modDataCount = null; |
| |
| const downstream = this._downstream; |
| downstream && downstream.dirty(); |
| |
| return forceFirstProgress; |
| } |
| |
| unfinished(): boolean { |
| return this._progress && this._dueIndex < this._dueEnd; |
| } |
| |
| /** |
| * @param downTask The downstream task. |
| * @return The downstream task. |
| */ |
| pipe(downTask: Task<Ctx>): void { |
| if (__DEV__) { |
| assert(downTask && !downTask._disposed && downTask !== this); |
| } |
| |
| // If already downstream, do not dirty downTask. |
| if (this._downstream !== downTask || this._dirty) { |
| this._downstream = downTask; |
| downTask._upstream = this; |
| downTask.dirty(); |
| } |
| } |
| |
| dispose(): void { |
| if (this._disposed) { |
| return; |
| } |
| |
| this._upstream && (this._upstream._downstream = null); |
| this._downstream && (this._downstream._upstream = null); |
| |
| this._dirty = false; |
| this._disposed = true; |
| } |
| |
| getUpstream(): Task<Ctx> { |
| return this._upstream; |
| } |
| |
| getDownstream(): Task<Ctx> { |
| return this._downstream; |
| } |
| |
| setOutputEnd(end: number): void { |
| // This only happend in dataTask, dataZoom, map, currently. |
| // where dataZoom do not set end each time, but only set |
| // when reset. So we should record the setted end, in case |
| // that the stub of dataZoom perform again and earse the |
| // setted end by upstream. |
| this._outputDueEnd = this._settedOutputEnd = end; |
| } |
| |
| } |
| |
| const iterator: TaskDataIterator = (function () { |
| |
| let end: number; |
| let current: number; |
| let modBy: number; |
| let modDataCount: number; |
| let winCount: number; |
| |
| const it: TaskDataIterator = { |
| reset: function (s: number, e: number, sStep: number, sCount: number): void { |
| current = s; |
| end = e; |
| |
| modBy = sStep; |
| modDataCount = sCount; |
| winCount = Math.ceil(modDataCount / modBy); |
| |
| it.next = (modBy > 1 && modDataCount > 0) ? modNext : sequentialNext; |
| } |
| }; |
| |
| return it; |
| |
| function sequentialNext(): number { |
| return current < end ? current++ : null; |
| } |
| |
| function modNext(): number { |
| const dataIndex = (current % winCount) * modBy + Math.ceil(current / winCount); |
| const result = current >= end |
| ? null |
| : dataIndex < modDataCount |
| ? dataIndex |
| // If modDataCount is smaller than data.count() (consider `appendData` case), |
| // Use normal linear rendering mode. |
| : current; |
| current++; |
| return result; |
| } |
| })(); |
| |
| |
| |
| /////////////////////////////////////////////////////////// |
| // For stream debug (Should be commented out after used!) |
| // @usage: printTask(this, 'begin'); |
| // @usage: printTask(this, null, {someExtraProp}); |
| // @usage: Use `__idxInPipeline` as conditional breakpiont. |
| // |
| // window.printTask = function (task: any, prefix: string, extra: { [key: string]: unknown }): void { |
| // window.ecTaskUID == null && (window.ecTaskUID = 0); |
| // task.uidDebug == null && (task.uidDebug = `task_${window.ecTaskUID++}`); |
| // task.agent && task.agent.uidDebug == null && (task.agent.uidDebug = `task_${window.ecTaskUID++}`); |
| // let props = []; |
| // if (task.__pipeline) { |
| // let val = `${task.__idxInPipeline}/${task.__pipeline.tail.__idxInPipeline} ${task.agent ? '(stub)' : ''}`; |
| // props.push({text: '__idxInPipeline/total', value: val}); |
| // } else { |
| // let stubCount = 0; |
| // task.agentStubMap.each(() => stubCount++); |
| // props.push({text: 'idx', value: `overall (stubs: ${stubCount})`}); |
| // } |
| // props.push({text: 'uid', value: task.uidDebug}); |
| // if (task.__pipeline) { |
| // props.push({text: 'pipelineId', value: task.__pipeline.id}); |
| // task.agent && props.push( |
| // {text: 'stubFor', value: task.agent.uidDebug} |
| // ); |
| // } |
| // props.push( |
| // {text: 'dirty', value: task._dirty}, |
| // {text: 'dueIndex', value: task._dueIndex}, |
| // {text: 'dueEnd', value: task._dueEnd}, |
| // {text: 'outputDueEnd', value: task._outputDueEnd} |
| // ); |
| // if (extra) { |
| // Object.keys(extra).forEach(key => { |
| // props.push({text: key, value: extra[key]}); |
| // }); |
| // } |
| // let args = ['color: blue']; |
| // let msg = `%c[${prefix || 'T'}] %c` + props.map(item => ( |
| // args.push('color: green', 'color: red'), |
| // `${item.text}: %c${item.value}` |
| // )).join('%c, '); |
| // console.log.apply(console, [msg].concat(args)); |
| // // console.log(this); |
| // }; |
| // window.printPipeline = function (task: any, prefix: string) { |
| // const pipeline = task.__pipeline; |
| // let currTask = pipeline.head; |
| // while (currTask) { |
| // window.printTask(currTask, prefix); |
| // currTask = currTask._downstream; |
| // } |
| // }; |
| // window.showChain = function (chainHeadTask) { |
| // var chain = []; |
| // var task = chainHeadTask; |
| // while (task) { |
| // chain.push({ |
| // task: task, |
| // up: task._upstream, |
| // down: task._downstream, |
| // idxInPipeline: task.__idxInPipeline |
| // }); |
| // task = task._downstream; |
| // } |
| // return chain; |
| // }; |
| // window.findTaskInChain = function (task, chainHeadTask) { |
| // let chain = window.showChain(chainHeadTask); |
| // let result = []; |
| // for (let i = 0; i < chain.length; i++) { |
| // let chainItem = chain[i]; |
| // if (chainItem.task === task) { |
| // result.push(i); |
| // } |
| // } |
| // return result; |
| // }; |
| // window.printChainAEachInChainB = function (chainHeadTaskA, chainHeadTaskB) { |
| // let chainA = window.showChain(chainHeadTaskA); |
| // for (let i = 0; i < chainA.length; i++) { |
| // console.log('chainAIdx:', i, 'inChainB:', window.findTaskInChain(chainA[i].task, chainHeadTaskB)); |
| // } |
| // }; |