import { Task, TaskStatus, Timestamp } from "./Task" import { pullAt, findIndex, defaults } from "lodash" import { performance } from 'perf_hooks' const PriorityQueue = require("fastpriorityqueue") /** * returns the current relative time, since the start of the webpage / script, in ms * @returns the current relative time in ms */ export function getCurrentTimestamp(): number { return Number(typeof window === 'undefined' ? performance.now() : window.performance.now()) } export interface SchedulerConfig { numOfTicksBeforeReschedule: number, // Number of preemption points before task yields meanTimeToReschedule: number, // Miliseconds before rescheduling is done meanTimeToYield: number, // Miliseconds before we return to the main loop debug: boolean, policy: string, // Fixed Priority (FP) || Earliest Deadline First (EDF) onRequestAnimationFrame: boolean } export const DEFAULT_SCHEDULER_CONFIG: Readonly = Object.freeze({ numOfTicksBeforeReschedule: 300, meanTimeToReschedule: 1, meanTimeToYield: 5, debug: false, policy: 'FP', // If there are no animation frames, it does not make sense to // request them. This normally means that we do not run in a browser. onRequestAnimationFrame: !(typeof requestAnimationFrame === 'undefined') }) export enum SchedulerMessage { TerminateTask, Continue } export class TaskState { /** * the enveloped task structure */ task: Task lastMessage: SchedulerMessage | Promise | undefined /** * the generator function of the task */ iterator: IterableIterator> lastAsyncRespose: any | undefined /** * the time at which the task was enqueued, for accouting */ enqueuedTime: Timestamp /** * the time the task has run, for accounting */ elapsedTime: number = 0 /** * the number of ticks that this task was executed */ elapsedTicks: number = 0 /** * return value of the task after it has finished */ retval: any = null /** * whether or not the task has finished running */ completed = false parent?: TaskState /** * The absolute deadline of this task instance */ absoluteDeadline: Timestamp get status(): TaskStatus { return this.task.status } set status(s: TaskStatus) { this.task.status = s } get priority(): number { return this.task.priority } get isRunnable(): boolean { return this.task.status === TaskStatus.Runnable } get isWaiting(): boolean { return this.task.status == TaskStatus.Waiting } /** * call the generator function of the task again */ next(input: any = undefined): SchedulerMessage | Promise { if (this.completed) { return SchedulerMessage.TerminateTask } let n = this.iterator.next(input || this.lastAsyncRespose) if ( n.done == true ) { this.completed = true; } else { this.completed = false; } this.lastMessage = n.value this.lastAsyncRespose = undefined return this.lastMessage! } /** * set up the structure * @param task the task to envelop * @param parent the parent structure */ constructor(task: Task, parent?: TaskState) { this.task = task this.iterator = task.run() this.enqueuedTime = getCurrentTimestamp() this.parent = parent if (parent) { this.absoluteDeadline = parent.absoluteDeadline } else { this.absoluteDeadline = task.creationTime + task.deadline } } /** * update the times of the task and call onTermined() */ terminate() { this.task.retval = this.retval this.task.enqueuedTime = this.enqueuedTime this.task.elapsedTime = this.elapsedTime this.task.endTime = getCurrentTimestamp() this.task.onTerminated() } /** * convert task to a human readable string */ toString() { return `${this.task.name} (priority: ${this.task.priority}, deadline: ${this.absoluteDeadline}): ${this.task.status} (${this.elapsedTime}ms)` } } export class TimeoutState { /** * Holds information an alarm state */ timeout: number callback: () => void // Called from scheduler loop js_timeout: any js_callback: () => void // called via Javascript event loop } /** * abstract scheduler class, interface for concrete deadline and fixed priority schedulers */ export abstract class Scheduler { /** * the currently running task */ currentTask?: TaskState /** * number of pauses of the scheduler, may be used nestedly */ private pauses: boolean = true get paused(): boolean { return this.pauses } /** * mostly constant configuration options for the scheduler */ config: SchedulerConfig /** * How long was the scheduler active on the executor */ schedulerTime: number = 0 /** * Number of preemption points that are visited before thread must actually call yield */ RTJS_BUDGET: number /** * This flag forces the scheduler to reschedule immediately, * before finishing the current timeslice. */ protected forceReschedule: boolean printTimes: boolean /** * Timeout queue to manage more precise timeouts */ private timeoutQueue: PriorityQueue = new PriorityQueue((a: TimeoutState, b: TimeoutState) => { return b.timeout > a.timeout }) /** * Flag indicates a non-preemptive critical section */ NPCS: boolean /** * constructor, setting up a scheduler object * @param config set to DEFAULT_SCHEDULER_CONFIG, may use any other values */ constructor(config: Object = DEFAULT_SCHEDULER_CONFIG) { this.config = defaults({}, config, DEFAULT_SCHEDULER_CONFIG) } /** * print messages to stdout if the ''debug'' flag was set */ protected debug(msg: any, ...optionalParams: any[]) { if (this.config.debug) { console.error(msg, ...optionalParams) } } /** * pauses the scheduler, may be used nestedly */ pause() { this.pauses = true this.debug("> Pausing Scheduler (pauses %i)", this.pauses) } /** * resumes the scheduler after pausing, may be used nestedly */ resume() { if (this.pauses) { this.pauses = false this.debug("> Resuming Scheduler (pauses %i)", this.pauses) this.repostMe(); } } /** * re-introduces the scheduler itself to the event loop */ repostMe() { //if ( this.config.onRequestAnimationFrame ) { // requestAnimationFrame ( this.executeRound.bind( this ) ) //} else { if (!this.pauses) { setImmediate(this.executeRound.bind(this)) } //} } /** * Add a single-short or periodic alarm function. * @param fn - The function to be callled after the delay has passed * @param delay - The deay in miliseconds * @param periodic - Whether the alarm is enqueued again * */ addAlarm(fn: () => void, delay: number, periodic: boolean = false) { let scheduler = this let ts = new TimeoutState() ts.timeout = 0 function startTimeout() { // If we were triggered too late for the next activation, // we use the current time as the next starting point let now = getCurrentTimestamp() if (now > (ts.timeout + delay)) { ts.timeout = now } ts.timeout += delay // trigger at the exact next timeout time scheduler.timeoutQueue.add(ts) ts.js_timeout = setTimeout(ts.js_callback, ts.timeout - now) } // Setup the Javascript Timeout ts.js_callback = () => { scheduler.timeoutQueue.removeOne(t => t == ts) if (periodic) { startTimeout() } fn() } // Setup the bAUTOSAR timeout. This is called from the scheduler ts.callback = () => { clearTimeout(ts.js_timeout) if (periodic) { startTimeout() } fn() } // Start the timeout for the first time startTimeout() return ts } /** * add a task to the queue of the scheduler, sends a scheduler-message to event queue after it is finished, if no task is running atm * @param task the task object to enqueue * @param parent optional parameter, scheduler-structure for the parent task if any */ addTask(task: Task, parent?: TaskState) { this.debug("> Adding Task %s with priority %i and deadline %i", task.name, task.priority, task.deadline) task.scheduler = this this.enqueueTask(new TaskState(task, parent)) this.resume() /*if (this.currentTask === undefined && !this.paused) { this.repostMe() }*/ } /** * Used by tasks to spawn new tasks, sets the currently running one as the parent * @param task the child task */ spawnTask(task: Task) { if (this.currentTask) { this.currentTask.status = TaskStatus.Waiting task.parent = this.currentTask.task } this.addTask(task, this.currentTask) } /** * enqueue a task into the queue of the scheduler (implementation dependent) */ abstract enqueueTask(task: TaskState): void /** * @returns the number of elements in the task queue */ abstract get numberOfTasks(): number /** * @returns the next task to schedule */ protected abstract getNextTask(): TaskState | undefined /** * empties the task queue */ abstract emptyTaskQueue(): void /** * determines whether a reschedule needs to take place */ protected abstract needsReschedule(): boolean /** * swaps the currently running task with a new one, uses needsReschedule() */ protected prepareNextTask() { if (this.paused) { return } if (this.NPCS && this.currentTask) { return } // Obey the non-preemptive critical section // if the currently running task is not runnable or not completed it is re-enqueued for the scheduler to consider if (this.currentTask && (!this.currentTask.isRunnable || !this.currentTask.completed)) { this.debug("> Reenqueuing Task %s", this.currentTask.toString()) this.enqueueTask(this.currentTask) } // determine next task this.currentTask = this.getNextTask() // Rescheduling done, clear flag this.forceReschedule = false // all tasks are executed correctly if (!this.currentTask) { if (this.numberOfTasks === 0) { this.debug("> All out of tasks, calling isDone handler") //this.isDone() this.pause() } return } this.debug("> Reschedule to: %s", this.currentTask.toString()) } /** * Execute the current thread for a single timelice */ protected executeTimelice() { if (!this.currentTask) { return } let __step_start: number = getCurrentTimestamp() let __start: number let __end: number = __step_start let taskProgress: number = 0 do { if (this.paused || this.currentTask.status !== TaskStatus.Runnable) { break } /* If a component requests that we reschedule immediately, we will obey! */ if (this.forceReschedule) { break; } /** We run the current task for one budget (RTJS_BUDGET). The task can hit * introduced preemption points that often, before it actually calls yield */ if (this.currentTask.task.budget > -1) { this.RTJS_BUDGET = this.currentTask.task.budget; } else { this.RTJS_BUDGET = this.config.numOfTicksBeforeReschedule } __start = getCurrentTimestamp() let msg = this.currentTask.next() /* Execute! */ __end = getCurrentTimestamp() /* We account for the elapsed time */ taskProgress += __end - __start this.currentTask.elapsedTime += __end - __start this.currentTask.elapsedTicks += 1 /* Process the message from the thread. */ if (msg instanceof Promise) { let taskStatus = this.currentTask taskStatus.status = TaskStatus.Waiting msg.then(result => { taskStatus.lastAsyncRespose = result taskStatus.status = TaskStatus.Runnable if (!this.currentTask) { this.repostMe(); } }) break } else if (msg === SchedulerMessage.TerminateTask || this.currentTask.completed) { if (!(msg === SchedulerMessage.TerminateTask)) { this.currentTask.retval = msg; } this.terminateTask() break } } while ((__end - __step_start) < this.config.meanTimeToReschedule); // Poll the timeout queue do { let el = this.timeoutQueue.peek() if (!el) { break } // Next timeout is more than 1 milisecond away if ((el.timeout - __end) > 1) { break } // FIXME: Timeout granularity this.timeoutQueue.poll() // remove el el.callback() } while (true) this.debug("Scheduler Timeslice: %d ms/%d ms", (__end - __step_start), taskProgress) } /** * terminates the currently running task */ private terminateTask() { if (!this.currentTask) return; this.debug( "> TerminateTask: Task %s with priority %i (time: %i ms)", this.currentTask.task.name, this.currentTask.priority, this.currentTask.elapsedTime ) if (this.currentTask.parent) { this.currentTask.parent.lastAsyncRespose = this.currentTask.lastMessage this.currentTask.parent.status = TaskStatus.Runnable } this.currentTask.terminate() this.forceReschedule = true this.NPCS = false this.currentTask = undefined } /** * if the scheduler is not paused, and there are still tasks left, run * the scheduler for meanTimeToYield miliseconds */ private executeRound() { let __round_start = getCurrentTimestamp() let __round_end = __round_start do { if (this.paused) { break } if (this.forceReschedule || this.needsReschedule()) { this.prepareNextTask() } if (!this.currentTask) { break } if (this.currentTask.status == TaskStatus.Runnable) { this.executeTimelice() } else { break; } __round_end = getCurrentTimestamp() } while ((__round_end - __round_start) < this.config.meanTimeToYield) this.debug("Scheduler Round: %d ms"); this.schedulerTime += __round_end - __round_start // Yield to the mainloop this.repostMe() } }