trace/database/Procedure.ts

/*
 * Copyright (C) 2022 Huawei Device Co., Ltd.
 * Licensed 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 {query} from "./SqlLite.js";

class ProcedureThread extends Worker {
    busy: boolean = false;
    isCancelled: boolean = false;
    id: number = -1;
    taskMap: any = {};
    name: string | undefined

    uuid(): string {
        // @ts-ignore
        return ([1e7] + -1e3 + -4e3 + -8e3 + -1e11).replace(/[018]/g, c => (c ^ crypto.getRandomValues(new Uint8Array(1))[0] & 15 >> c / 4).toString(16));
    }

    queryFunc(type: string, args: any, transfer: any, handler: Function) {
        this.busy = true;
        let id = this.uuid();
        this.taskMap[id] = handler
        let pam = {
            id: id,
            type: type,
            params: args,
        }
        if (transfer) {
            try {
                if (Array.isArray(transfer)) {
                    if (transfer.length > 0) {
                        this.postMessage(pam, [...transfer]);
                    } else {
                        this.postMessage(pam);
                    }
                } else {
                    this.postMessage(pam, [transfer]);
                }
            } catch (e: any) {
            }
        } else {
            this.postMessage(pam);
        }
    }

    cancel() {
        this.isCancelled = true;
        this.terminate();
    }
}

class ProcedurePool {
    static cpuCount = Math.floor((window.navigator.hardwareConcurrency || 4) / 2);
    maxThreadNumber: number = 1
    works: Array<ProcedureThread> = []
    timelineChange: ((a: any) => void) | undefined | null = null;
    cpusLen = ProcedurePool.build('cpu', 8);
    freqLen = ProcedurePool.build('freq', 2);
    processLen = ProcedurePool.build('process', 8);
    logicDataLen = ProcedurePool.build('logic', 2);
    names = [...this.cpusLen, ...this.processLen, ...this.freqLen];
    logicDataHandles = [...this.logicDataLen]

    onComplete: Function | undefined;//任务完成回调

    constructor(threadBuild: (() => ProcedureThread) | undefined = undefined) {
        this.init(threadBuild);
    }

    static build(name: string, len: number) {
        return [...Array(len).keys()].map(it => `${name}${it}`)
    }

    init(threadBuild: (() => ProcedureThread) | undefined = undefined) {
        this.maxThreadNumber = this.names.length
        for (let i = 0; i < this.maxThreadNumber; i++) {
            this.newThread();
        }
        for (let j = 0; j < this.logicDataHandles.length; j++) {
            this.logicDataThread()
        }
    }

    newThread() {
        let thread: ProcedureThread = new ProcedureThread("trace/database/ui-worker/ProcedureWorker.js", {type: "module"})
        thread.name = this.names[this.works.length]
        thread.onmessage = (event: MessageEvent) => {
            thread.busy = false;
            if ((event.data.type as string) == "timeline-range-changed") {
                this.timelineChange && this.timelineChange(event.data.results);
                thread.busy = false;
                return;
            }
            if (Reflect.has(thread.taskMap, event.data.id)) {
                if (event.data) {
                    let fun = thread.taskMap[event.data.id];
                    if (fun) {
                        fun(event.data.results, event.data.hover);
                    }
                    Reflect.deleteProperty(thread.taskMap, event.data.id)
                }
            }
            if (this.isIdle() && this.onComplete) {
                this.onComplete();
            }
        }
        thread.onmessageerror = e => {
        }
        thread.onerror = e => {
        }
        thread.id = this.works.length
        thread.busy = false
        this.works?.push(thread)
        return thread;
    }

    logicDataThread(){
        let thread: ProcedureThread = new ProcedureThread("trace/database/logic-worker/ProcedureLogicWorker.js", {type: "module"})
        thread.name = this.logicDataHandles[this.works.length - this.names.length]
        thread.onmessage = (event: MessageEvent) => {
            thread.busy = false;
            if(event.data.isQuery){
                query(event.data.type,event.data.sql,event.data.args,"exec-buf").then((res:any)=>{
                    thread.postMessage({
                        type:event.data.type,
                        params:{
                            list:res
                        },
                        id:event.data.id
                    })
                })
                return
            }
            if(event.data.isSending){
                if (Reflect.has(thread.taskMap, event.data.id)) {
                    if (event.data) {
                        let fun = thread.taskMap[event.data.id];
                        if (fun) {
                            fun(event.data.results, event.data.hover);
                        }
                        return;
                    }
                }
            }
            if (Reflect.has(thread.taskMap, event.data.id)) {
                if (event.data) {
                    let fun = thread.taskMap[event.data.id];
                    if (fun) {
                        fun(event.data.results, event.data.hover);
                    }
                    Reflect.deleteProperty(thread.taskMap, event.data.id)
                }
            }
            if (this.isIdle() && this.onComplete) {
                this.onComplete();
            }
        }
        thread.onmessageerror = e => {
        }
        thread.onerror = e => {
        }
        thread.id = this.works.length
        thread.busy = false
        this.works?.push(thread)
        return thread;
    }

    close = () => {
        for (let i = 0; i < this.works.length; i++) {
            let thread = this.works[i];
            thread.terminate();
        }
        this.works.length = 0;
    }

    clearCache = () => {
        for (let i = 0; i < this.works.length; i++) {
            let thread = this.works[i];
            thread.queryFunc("clear", {}, undefined, () => {
            })
        }
    }

    submitWithName(name: string, type: string, args: any, transfer: any, handler: Function): ProcedureThread | undefined {
        let noBusyThreads = this.works.filter(it => it.name === name);
        let thread: ProcedureThread | undefined
        if (noBusyThreads.length > 0) { //取第一个空闲的线程进行任务
            thread = noBusyThreads[0];
            thread!.queryFunc(type, args, transfer, handler)
        }
        return thread;
    }

    submitWithNamePromise(name: string, type: string, args: any, transfer: any): Promise<any> {
        return new Promise((resolve, reject) => {
            let noBusyThreads = this.works.filter(it => it.name === name);
            let thread: ProcedureThread | undefined
            if (noBusyThreads.length > 0) { //取第一个空闲的线程进行任务
                thread = noBusyThreads[0];
                thread!.queryFunc(type, args, transfer, (res: any, hover: any) => {
                    resolve({
                        res: res,
                        hover: hover,
                    })
                });
            }
        })
    }

    isIdle() {
        return this.works.every(it => !it.busy);
    }
}


export const procedurePool = new ProcedurePool()