perf: 优化定时器

packages/core/pipeline/src/core/context.ts

@@ -1,10 +1,13 @@
 import { IStorage, MemoryStorage } from "./storage";
-
+const CONTEXT_VERSION_KEY = "contextVersion";
 export interface IContext {
+  getInt(key: string): Promise<number>;
   get(key: string): Promise<string | null>;
   set(key: string, value: string): Promise<void>;
-  getObj(key: string): Promise<any>;
-  setObj(key: string, value: any): Promise<void>;
+  getObj<T = any>(key: string): Promise<T | null>;
+  setObj<T = any>(key: string, value: T): Promise<void>;
+  updateVersion(): Promise<void>;
+  initVersion(): Promise<void>;
 }
 
 export class ContextFactory {
@@ -17,11 +20,13 @@ export class ContextFactory {
   }
 
   getContext(scope: string, namespace: string): IContext {
-    return new StorageContext(scope, namespace, this.storage);
+    const context = new StorageContext(scope, namespace, this.storage);
+    return context;
   }
 
   getMemoryContext(scope: string, namespace: string): IContext {
-    return new StorageContext(scope, namespace, this.memoryStorage);
+    const context = new StorageContext(scope, namespace, this.memoryStorage);
+    return context;
   }
 }
 
@@ -29,19 +34,46 @@ export class StorageContext implements IContext {
   storage: IStorage;
   namespace: string;
   scope: string;
+
+  _version = 0;
+  _initialVersion = 0;
   constructor(scope: string, namespace: string, storage: IStorage) {
     this.storage = storage;
     this.scope = scope;
     this.namespace = namespace;
   }
 
+  async initVersion() {
+    const version = await this.getInt(CONTEXT_VERSION_KEY);
+    this._initialVersion = version;
+    this._version = version;
+  }
+
+  async updateVersion() {
+    if (this._version === this._initialVersion) {
+      this._version++;
+    }
+
+    await this.set(CONTEXT_VERSION_KEY, this._version.toString());
+  }
+
   async get(key: string) {
-    return await this.storage.get(this.scope, this.namespace, key);
+    const version = key === CONTEXT_VERSION_KEY ? 0 : this._version;
+    return await this.storage.get(this.scope, this.namespace, version.toString(), key);
   }
   async set(key: string, value: string) {
-    return await this.storage.set(this.scope, this.namespace, key, value);
+    const version = key === CONTEXT_VERSION_KEY ? 0 : this._version;
+    return await this.storage.set(this.scope, this.namespace, version.toString(), key, value);
   }
-  async getObj(key: string): Promise<any> {
+
+  async getInt(key: string): Promise<number> {
+    const str = await this.get(key);
+    if (str) {
+      return parseInt(str);
+    }
+    return 0;
+  }
+  async getObj<T = any>(key: string): Promise<T | null> {
     const str = await this.get(key);
     if (str) {
       const store = JSON.parse(str);
@@ -50,7 +82,7 @@ export class StorageContext implements IContext {
     return null;
   }
 
-  async setObj(key: string, value: any) {
+  async setObj<T = any>(key: string, value: T) {
     await this.set(key, JSON.stringify({ value }));
   }
 }

packages/core/pipeline/src/core/executor.ts

@@ -1,7 +1,7 @@
 import { ConcurrencyStrategy, Pipeline, ResultType, Runnable, RunStrategy, Stage, Step, Task } from "../d.ts";
 import _ from "lodash";
-import { RunHistory } from "./run-history";
-import { PluginDefine, pluginRegistry } from "../plugin";
+import { RunHistory, RunnableCollection } from "./run-history";
+import { AbstractTaskPlugin, PluginDefine, pluginRegistry } from "../plugin";
 import { ContextFactory, IContext } from "./context";
 import { IStorage } from "./storage";
 import { logger } from "../utils/util.log";
@@ -18,11 +18,20 @@ export class Executor {
   accessService: IAccessService;
   contextFactory: ContextFactory;
   logger: Logger;
-  pipelineContext: IContext;
+  pipelineContext!: IContext;
+  lastStatusMap!: RunnableCollection;
   onChanged: (history: RunHistory) => void;
-  constructor(options: { userId: any; pipeline: Pipeline; storage: IStorage; onChanged: (history: RunHistory) => void; accessService: IAccessService }) {
+  constructor(options: {
+    userId: any;
+    pipeline: Pipeline;
+    storage: IStorage;
+    onChanged: (history: RunHistory) => Promise<void>;
+    accessService: IAccessService;
+  }) {
     this.pipeline = _.cloneDeep(options.pipeline);
-    this.onChanged = options.onChanged;
+    this.onChanged = async (history: RunHistory) => {
+      await options.onChanged(history);
+    };
     this.accessService = options.accessService;
     this.userId = options.userId;
     this.pipeline.userId = this.userId;
@@ -31,17 +40,25 @@ export class Executor {
     this.pipelineContext = this.contextFactory.getContext("pipeline", this.pipeline.id);
   }
 
+  async init() {
+    const lastRuntime = await this.pipelineContext.getObj(`lastRuntime`);
+    this.lastStatusMap = new RunnableCollection(lastRuntime?.pipeline);
+  }
+
   async run(runtimeId: any = 0, triggerType: string) {
     try {
+      await this.init();
       const trigger = { type: triggerType };
+      // 读取last
       this.runtime = new RunHistory(runtimeId, trigger, this.pipeline);
       this.logger.info(`pipeline.${this.pipeline.id}  start`);
       await this.runWithHistory(this.pipeline, "pipeline", async () => {
-        await this.runStages();
+        await this.runStages(this.pipeline);
       });
     } catch (e) {
       this.logger.error("pipeline 执行失败", e);
     } finally {
+      await this.pipelineContext.setObj("lastRuntime", this.runtime);
       this.logger.info(`pipeline.${this.pipeline.id}  end`);
     }
   }
@@ -50,20 +67,18 @@ export class Executor {
     runnable.runnableType = runnableType;
     this.runtime.start(runnable);
     await this.onChanged(this.runtime);
-    const contextKey = `status.${runnable.id}`;
-    const inputKey = `input.${runnable.id}`;
 
     if (runnable.strategy?.runStrategy === RunStrategy.SkipWhenSucceed) {
       //如果是成功后跳过策略
-      const lastResult = await this.pipelineContext.getObj(contextKey);
-      const lastInput = await this.pipelineContext.get(inputKey);
+      const lastNode = this.lastStatusMap.get(runnable.id);
+      const lastResult = lastNode?.status?.status;
+      const lastInput = JSON.stringify(lastNode?.status?.input);
       let inputChanged = false;
-      //TODO 参数不变
       if (runnableType === "step") {
         const step = runnable as Step;
         const input = JSON.stringify(step.input);
-        await this.pipelineContext.set(inputKey, input);
         if (input != null && lastInput !== input) {
+          //参数有变化
           inputChanged = true;
         }
       }
@@ -76,12 +91,10 @@ export class Executor {
     try {
       await run();
       this.runtime.success(runnable);
-      await this.pipelineContext.setObj(contextKey, ResultType.success);
       await this.onChanged(this.runtime);
       return ResultType.success;
     } catch (e: any) {
       this.runtime.error(runnable, e);
-      await this.pipelineContext.setObj(contextKey, ResultType.error);
       await this.onChanged(this.runtime);
       throw e;
     } finally {
@@ -89,9 +102,9 @@ export class Executor {
     }
   }
 
-  private async runStages() {
+  private async runStages(pipeline: Pipeline) {
     const resList: ResultType[] = [];
-    for (const stage of this.pipeline.stages) {
+    for (const stage of pipeline.stages) {
       const res: ResultType = await this.runWithHistory(stage, "stage", async () => {
         await this.runStage(stage);
       });
@@ -150,17 +163,12 @@ export class Executor {
   }
 
   private async runStep(step: Step) {
+    const lastStatus = this.lastStatusMap.get(step.id);
     //执行任务
-    const plugin: RegistryItem = pluginRegistry.get(step.type);
-    const context: any = {
-      logger: this.runtime.loggers[step.id],
-      accessService: this.accessService,
-      pipelineContext: this.pipelineContext,
-      userContext: this.contextFactory.getContext("user", this.userId),
-      http: request,
-    };
+    const plugin: RegistryItem<AbstractTaskPlugin> = pluginRegistry.get(step.type);
+
     // @ts-ignore
-    const instance = new plugin.target();
+    const instance: ITaskPlugin = new plugin.target();
     // @ts-ignore
     const define: PluginDefine = plugin.define;
     //从outputContext读取输入参数
@@ -173,14 +181,27 @@ export class Executor {
       }
     });
 
+    const context: any = {
+      logger: this.runtime._loggers[step.id],
+      accessService: this.accessService,
+      pipelineContext: this.pipelineContext,
+      lastStatus,
+      userContext: this.contextFactory.getContext("user", this.userId),
+      http: request,
+    };
     Decorator.inject(define.autowire, instance, context);
+
     await instance.onInstance();
     await instance.execute();
 
+    if (instance.result.clearLastStatus) {
+      this.lastStatusMap.clear();
+    }
     //输出到output context
     _.forEach(define.output, (item, key) => {
-      const contextKey = `step.${step.id}.${key}`;
-      this.runtime.context[contextKey] = instance[key];
+      step!.status!.output[key] = instance[key];
+      const stepOutputKey = `step.${step.id}.${key}`;
+      this.runtime.context[stepOutputKey] = instance[key];
     });
   }
 }

packages/core/pipeline/src/core/run-history.ts

@@ -1,4 +1,4 @@
-import { Context, HistoryResult, Pipeline, Runnable } from "../d.ts";
+import { Context, HistoryResult, Pipeline, ResultType, Runnable, RunnableMap, Stage, Step, Task } from "../d.ts";
 import _ from "lodash";
 import { buildLogger } from "../utils/util.log";
 import { Logger } from "log4js";
@@ -11,18 +11,27 @@ export type HistoryStatus = {
 export type RunTrigger = {
   type: string; // user , timer
 };
+
+export function NewRunHistory(obj: any) {
+  const history = new RunHistory(obj.id, obj.trigger, obj.pipeline);
+  history.context = obj.context;
+  history.logs = obj.logs;
+  history._loggers = obj.loggers;
+  return history;
+}
 export class RunHistory {
-  id: string;
+  id!: string;
   //运行时上下文变量
   context: Context = {};
-  pipeline: Pipeline;
+  pipeline!: Pipeline;
   logs: {
     [runnableId: string]: string[];
   } = {};
-  loggers: {
+  _loggers: {
     [runnableId: string]: Logger;
   } = {};
-  trigger: RunTrigger;
+  trigger!: RunTrigger;
+
   constructor(runtimeId: any, trigger: RunTrigger, pipeline: Pipeline) {
     this.id = runtimeId;
     this.pipeline = pipeline;
@@ -32,13 +41,16 @@ export class RunHistory {
   start(runnable: Runnable): HistoryResult {
     const now = new Date().getTime();
     this.logs[runnable.id] = [];
-    this.loggers[runnable.id] = buildLogger((text) => {
+    this._loggers[runnable.id] = buildLogger((text) => {
       this.logs[runnable.id].push(text);
     });
+    const input = (runnable as Step).input;
     const status: HistoryResult = {
-      status: "start",
+      output: {},
+      input: _.cloneDeep(input),
+      status: ResultType.start,
       startTime: now,
-      result: "start",
+      result: ResultType.start,
     };
     runnable.status = status;
     this.log(runnable, `开始执行`);
@@ -71,9 +83,9 @@ export class RunHistory {
     const now = new Date().getTime();
     const status = runnable.status;
     _.merge(status, {
-      status: "error",
+      status: ResultType.error,
       endTime: now,
-      result: "error",
+      result: ResultType.error,
       message: e.message,
     });
 
@@ -82,15 +94,65 @@ export class RunHistory {
 
   log(runnable: Runnable, text: string) {
     // @ts-ignore
-    this.loggers[runnable.id].info(`[${runnable.title}]<id:${runnable.id}> [${runnable.runnableType}]`, text);
+    this._loggers[runnable.id].info(`[${runnable.title}]<id:${runnable.id}> [${runnable.runnableType}]`, text);
   }
 
   logError(runnable: Runnable, e: Error) {
     // @ts-ignore
-    this.loggers[runnable.id].error(`[${runnable.title}]<id:${runnable.id}> [${runnable.runnableType}]`, e);
+    this._loggers[runnable.id].error(`[${runnable.title}]<id:${runnable.id}> [${runnable.runnableType}]`, e);
   }
 
   finally(runnable: Runnable) {
     //
   }
 }
+
+export class RunnableCollection {
+  private collection: RunnableMap = {};
+  private pipeline!: Pipeline;
+  constructor(pipeline?: Pipeline) {
+    if (!pipeline) {
+      return;
+    }
+    this.pipeline = pipeline;
+    const map = this.toMap(pipeline);
+    this.collection = map;
+  }
+
+  private each<T extends Runnable>(list: T[], exec: (item: Runnable) => void) {
+    list.forEach((item) => {
+      exec(item);
+      if (item.runnableType === "pipeline") {
+        // @ts-ignore
+        this.each<Stage>(item.stages, exec);
+      } else if (item.runnableType === "stage") {
+        // @ts-ignore
+        this.each<Task>(item.tasks, exec);
+      } else if (item.runnableType === "task") {
+        // @ts-ignore
+        this.each<Step>(item.steps, exec);
+      }
+    });
+  }
+  private toMap(pipeline: Pipeline) {
+    const map: RunnableMap = {};
+
+    this.each(pipeline.stages, (item) => {
+      map[item.id] = item;
+    });
+    return map;
+  }
+
+  get(id: string): Runnable | undefined {
+    return this.collection[id];
+  }
+
+  clear() {
+    if (!this.pipeline) {
+      return;
+    }
+    this.each(this.pipeline.stages, (item) => {
+      item.status = undefined;
+    });
+  }
+}

packages/core/pipeline/src/core/storage.ts

@@ -1,9 +1,12 @@
 import fs from "fs";
 import path from "path";
 
+
+
 export interface IStorage {
-  get(scope: string, namespace: string, key: string): Promise<string | null>;
-  set(scope: string, namespace: string, key: string, value: string): Promise<void>;
+  get(scope: string, namespace: string, version: string, key: string): Promise<string | null>;
+  set(scope: string, namespace: string, version: string, key: string, value: string): Promise<void>;
+  remove(scope: string, namespace: string, version: string, key: string): Promise<void>;
 }
 
 export class FileStorage implements IStorage {
@@ -20,6 +23,18 @@ export class FileStorage implements IStorage {
     }
   }
 
+  async remove(scope: string, namespace: string, version: string, key: string): Promise<void> {
+    if (key) {
+      fs.unlinkSync(this.buildPath(scope, namespace, version, key));
+    } else if (version) {
+      fs.rmdirSync(this.buildPath(scope, namespace, version));
+    } else if (namespace) {
+      fs.rmdirSync(this.buildPath(scope, namespace));
+    } else {
+      fs.rmdirSync(this.buildPath(scope));
+    }
+  }
+
   writeFile(filePath: string, value: string) {
     const dir = path.dirname(filePath);
     if (!fs.existsSync(dir)) {
@@ -36,18 +51,26 @@ export class FileStorage implements IStorage {
     return fs.readFileSync(filePath).toString();
   }
 
-  async get(scope: string, namespace: string, key: string): Promise<string | null> {
-    const path = this.buildPath(scope, namespace, key);
+  async get(scope: string, namespace: string, version: string, key: string): Promise<string | null> {
+    const path = this.buildPath(scope, namespace, version, key);
     return this.readFile(path);
   }
 
-  async set(scope: string, namespace: string, key: string, value: string): Promise<void> {
-    const path = this.buildPath(scope, namespace, key);
+  async set(scope: string, namespace: string, version: string, key: string, value: string): Promise<void> {
+    const path = this.buildPath(scope, namespace, version, key);
     this.writeFile(path, value);
   }
 
-  private buildPath(scope: string, namespace: string, key: string) {
-    return `${this.root}/${scope}/${namespace}/${key}`;
+  private buildPath(scope: string, namespace?: string, version?: string, key?: string) {
+    if (key) {
+      return `${this.root}/${scope}/${namespace}/${version}/${key}`;
+    } else if (version) {
+      return `${this.root}/${scope}/${namespace}/${version}`;
+    } else if (namespace) {
+      return `${this.root}/${scope}/${namespace}`;
+    } else {
+      return `${this.root}/${scope}`;
+    }
   }
 }
 
@@ -63,19 +86,55 @@ export class MemoryStorage implements IStorage {
     };
   } = {};
 
-  async get(scope: string, namespace: string, key: string): Promise<string | null> {
-    const context = this.context[scope];
-    if (context == null) {
+  async get(scope: string, namespace: string, version: string, key: string): Promise<string | null> {
+    const scopeContext = this.context[scope];
+    if (scopeContext == null) {
       return null;
     }
-    return context[namespace + "." + key];
+    const nsContext = scopeContext[namespace];
+    if (nsContext == null) {
+      return null;
+    }
+    const versionContext = nsContext[version];
+    if (versionContext == null) {
+      return null;
+    }
+    return versionContext[key];
   }
 
-  async set(scope: string, namespace: string, key: string, value: string): Promise<void> {
-    let context = this.context[scope];
-    if (context == null) {
-      context = context[scope];
+  async set(scope: string, namespace: string, version: string, key: string, value: string): Promise<void> {
+    let scopeContext = this.context[scope];
+    if (scopeContext == null) {
+      scopeContext = scopeContext[scope];
+    }
+    let nsContext = scopeContext[namespace];
+    if (nsContext == null) {
+      nsContext = {};
+      scopeContext[namespace] = nsContext;
+    }
+    let versionContext = nsContext[version];
+    if (versionContext == null) {
+      versionContext = {};
+      nsContext[version] = versionContext;
+    }
+    versionContext[key] = value;
+  }
+
+  async remove(scope: string, namespace = "", version = "", key = "") {
+    if (key) {
+      if (this.context[scope] && this.context[scope][namespace] && this.context[scope][namespace][version]) {
+        delete this.context[scope][namespace][version][key];
+      }
+    } else if (version) {
+      if (this.context[scope] && this.context[scope][namespace]) {
+        delete this.context[scope][namespace][version];
+      }
+    } else if (namespace) {
+      if (this.context[scope]) {
+        delete this.context[scope][namespace];
+      }
+    } else {
+      delete this.context[scope];
     }
-    context[namespace + "." + key] = value;
   }
 }

packages/core/pipeline/src/d.ts/pipeline.ts

@@ -85,9 +85,11 @@ export type Log = {
 };
 
 export enum ResultType {
-  success,
-  error,
-  skip,
+  start = "start",
+  success = "success",
+  error = "error",
+  skip = "skip",
+  none = "none",
 }
 
 export type HistoryResultGroup = {
@@ -97,15 +99,21 @@ export type HistoryResultGroup = {
   };
 };
 export type HistoryResult = {
+  input: any;
+  output: any;
   /**
    * 任务状态
    */
-  status: string;
+  status: ResultType;
   startTime: number;
   endTime?: number;
   /**
    * 处理结果
    */
-  result?: string; //success, error,skip
+  result?: ResultType; //success, error,skip
   message?: string;
 };
+
+export type RunnableMap = {
+  [id: string]: Runnable;
+};

packages/core/pipeline/src/plugin/api.ts

@@ -33,9 +33,24 @@ export type PluginDefine = Registrable & {
   };
 };
 
-export interface ITaskPlugin {
+export type ITaskPlugin = {
   onInstance(): Promise<void>;
   execute(): Promise<void>;
+  [key: string]: any;
+};
+
+export type TaskResult = {
+  clearLastStatus?: boolean;
+};
+export abstract class AbstractTaskPlugin implements ITaskPlugin {
+  result: TaskResult = {};
+  clearLastStatus() {
+    this.result.clearLastStatus = true;
+  }
+  async onInstance(): Promise<void> {
+    return;
+  }
+  abstract execute(): Promise<void>;
 }
 
 export type OutputVO = {

packages/core/pipeline/src/plugin/registry.ts

@@ -1,4 +1,4 @@
 import { Registry } from "../registry";
+import { AbstractTaskPlugin } from "./api";
 
-// @ts-ignore
-export const pluginRegistry = new Registry();
+export const pluginRegistry = new Registry<AbstractTaskPlugin>();

packages/core/pipeline/src/registry/registry.ts

@@ -4,23 +4,23 @@ export type Registrable = {
   desc?: string;
 };
 
-export type RegistryItem = {
+export type RegistryItem<T> = {
   define: Registrable;
-  target: any;
+  target: T;
 };
-export class Registry {
+export class Registry<T> {
   storage: {
-    [key: string]: RegistryItem;
+    [key: string]: RegistryItem<T>;
   } = {};
 
-  register(key: string, value: RegistryItem) {
+  register(key: string, value: RegistryItem<T>) {
     if (!key || value == null) {
       return;
     }
     this.storage[key] = value;
   }
 
-  get(name: string): RegistryItem {
+  get(name: string): RegistryItem<T> {
     if (!name) {
       throw new Error("插件名称不能为空");
     }