NikolayMakhonin / priority-queue / #15

Committed 12 Apr 2024 11:14AM UTC coverage: 63.551% (+0.7%) from 62.884%

Build # #15

Build Type

push

Committed by

Commit Message

v1.0.1

Run Details

95 of 176 branches covered (53.98%)

Branch coverage included in aggregate %.

4 of 5 new or added lines in 1 file covered. (80.0%)

6 existing lines in 1 file now uncovered.

177 of 252 relevant lines covered (70.24%)

815672.15 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.35

/src/priority-queue/PriorityQueue.ts

import {IPriorityQueue, IPriorityQueueTask, PromiseOrValue, Task} from './contracts'
import {PairingHeap, PairingNode} from '@flemist/pairing-heap'
import {CustomPromise} from '@flemist/async-utils'
import {Priority, priorityCompare, priorityCreate} from 'src/priority'
import {IAbortSignalFast} from '@flemist/abort-controller-fast'

type TQueueItem<T> = {
  func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
  abortSignal: IAbortSignalFast
  priority: Priority
  resolve: (value: T) => void
  reject: (error: Error) => void
  readyToRun: boolean
}

const emptyFunc = function emptyFunc(o) {
  return o
}

export function queueItemLessThan(o1: TQueueItem<any>, o2: TQueueItem<any>): boolean {
  return priorityCompare(o1.priority, o2.priority) < 0
}

let nextOrder: number = 1

export class PriorityQueue implements IPriorityQueue, IPriorityQueueTask {
  private readonly _queue: PairingHeap<TQueueItem<any>>

  constructor() {
    this._queue = new PairingHeap<TQueueItem<any>>({
      lessThanFunc: queueItemLessThan,
    })
  }

  run<T>(
    func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
    priority?: Priority,
    abortSignal?: IAbortSignalFast,
  ): Promise<T> {
    return this._run(false, func, priority, abortSignal) as any
  }

  runTask<T>(
    func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
    priority?: Priority,
    abortSignal?: IAbortSignalFast,
  ): Task<T> {
    return this._run(true, func, priority, abortSignal) as any
  }

  private _run<T>(
    taskMode: true|false,
    func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
    priority?: Priority,
    abortSignal?: IAbortSignalFast,
  ): Task<T> | Promise<T> {
    const promise = new CustomPromise<T>(abortSignal)

    const item: TQueueItem<T> = {
      priority  : priorityCreate(nextOrder++, priority),
      func,
      abortSignal,
      resolve   : promise.resolve,
      reject    : promise.reject,
      readyToRun: !taskMode,
    }

    this._queue.add(item)

    if (taskMode) {
      const _this = this

      return {
        result: promise.promise,
        setReadyToRun(readyToRun: boolean) {
          item.readyToRun = readyToRun
          if (readyToRun && !_this._inProcess) {
            _this._inProcess = true
            void _this._process()
          }
        },
      }
    }

    if (!this._inProcess) {
      this._inProcess = true
      void this._process()
    }
    return promise.promise
  }

  _inProcess: boolean
  private async _process() {
    const queue = this._queue

    // чтобы сначала сформировалась очередь, а потом началось выполнение в порядке приоритета
    // тесты показывают что только такая длинная конструкция прерывает выполнение программы и дает синхронному коду заполнить очередь перед началом выполнения
    await Promise.resolve().then(emptyFunc)

    while (true) {
      // eslint-disable-next-line @typescript-eslint/await-thenable
      await 0

      // void Promise.resolve().then(emptyFunc).then(next)

      if (queue.isEmpty) {
        this._inProcess = false
        break
      }

      let item = queue.getMin()
      if (item.readyToRun) {
        queue.deleteMin()
      }
      else {
        let nextNode: PairingNode<TQueueItem<any>>
        for (const node of queue.nodes()) {
          if (node.item.readyToRun) {
            nextNode = node
            break
          }
        }

        if (!nextNode) {
          this._inProcess = false
          break
        }

        item = nextNode.item
        queue.delete(nextNode)
      }

      if (item.abortSignal && item.abortSignal.aborted) {
        item.reject(item.abortSignal.reason)
      }
      else {
        try {
          let result = item.func && item.func(item.abortSignal)
          if (result && typeof result.then === 'function') {
            result = await result
          }
          item.resolve(result)
        }
        catch (err) {
          item.reject(err)
        }
      }
    }
  }
}

1	import {IPriorityQueue, IPriorityQueueTask, PromiseOrValue, Task} from './contracts'
2	import {PairingHeap, PairingNode} from '@flemist/pairing-heap'	1✔
3	import {CustomPromise} from '@flemist/async-utils'	1✔
4	import {Priority, priorityCompare, priorityCreate} from 'src/priority'	1✔
5	import {IAbortSignalFast} from '@flemist/abort-controller-fast'	2✔
6		172,350✔
7	type TQueueItem<T> = {
8	func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
9	abortSignal: IAbortSignalFast	80,114✔
10	priority: Priority
11	resolve: (value: T) => void	2✔
12	reject: (error: Error) => void	2✔
13	readyToRun: boolean
14	}	92,972✔
15
16	const emptyFunc = function emptyFunc(o) {	1✔
17	return o	2,144,666✔
18	}	2✔
19		278,950✔
20	export function queueItemLessThan(o1: TQueueItem<any>, o2: TQueueItem<any>): boolean {	1✔
21	return priorityCompare(o1.priority, o2.priority) < 0	1,373,645✔
22	}	×
23
24	let nextOrder: number = 1	2✔
25		278,950✔
26	export class PriorityQueue implements IPriorityQueue, IPriorityQueueTask {	278,950✔
27	private readonly _queue: PairingHeap<TQueueItem<any>>
28
29	constructor() {
30	this._queue = new PairingHeap<TQueueItem<any>>({	1,259,588✔
31	lessThanFunc: queueItemLessThan,
32	})
33	}
34		278,950✔
35	run<T>(	278,950!
36	func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,	×
37	priority?: Priority,	×
38	abortSignal?: IAbortSignalFast,
39	): Promise<T> {
40	return this._run(false, func, priority, abortSignal) as any	3,778,781✔
41	}	×
42		×
43	runTask<T>(	×
44	func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
45	priority?: Priority,
46	abortSignal?: IAbortSignalFast,
47	): Task<T> {
48	return this._run(true, func, priority, abortSignal) as any	278,950✔
49	}	172,350✔
50		172,350✔
51	private _run<T>(
52	taskMode: true\|false,	278,950✔
53	func: (abortSignal?: IAbortSignalFast) => PromiseOrValue<T>,
54	priority?: Priority,	2✔
55	abortSignal?: IAbortSignalFast,	172,350✔
56	): Task<T> \| Promise<T> {
57	const promise = new CustomPromise<T>(abortSignal)	3,778,781✔
58		172,350✔
59	const item: TQueueItem<T> = {	3,778,781✔
60	priority : priorityCreate(nextOrder++, priority),
61	func,	172,350✔
62	abortSignal,
63	resolve : promise.resolve,
64	reject : promise.reject,	172,350✔
65	readyToRun: !taskMode,
66	}
67
68	this._queue.add(item)	3,778,781✔
69		172,350✔
70	if (taskMode) {	3,778,781!
71	const _this = this	451,300!
72
73	return {	451,300✔
74	result: promise.promise,
75	setReadyToRun(readyToRun: boolean) {
76	item.readyToRun = readyToRun	×
UNCOV 77	if (readyToRun && !_this._inProcess) {	×
78	_this._inProcess = true	451,300✔
UNCOV 79	void _this._process()	×
80	}	451,300✔
81	},	172,350✔
82	}	172,350✔
83	}
84		278,950✔
85	if (!this._inProcess) {	3,778,781!
86	this._inProcess = true	2,144,666✔
87	void this._process()	2,144,666✔
88	}
89	return promise.promise	3,778,781✔
UNCOV 90	}	×
UNCOV 91		×
92	_inProcess: boolean	×
UNCOV 93	private async _process() {	×
94	const queue = this._queue	2,144,666✔
95		×
96	// чтобы сначала сформировалась очередь, а потом началось выполнение в порядке приоритета
97	// тесты показывают что только такая длинная конструкция прерывает выполнение программы и дает синхронному коду заполнить очередь перед началом выполнения
98	await Promise.resolve().then(emptyFunc)	2,144,666✔
NEW 99		×
100	while (true) {	2,144,666✔
101	// eslint-disable-next-line @typescript-eslint/await-thenable	×
102	await 0	5,923,447!
103
104	// void Promise.resolve().then(emptyFunc).then(next)	×
105
106	if (queue.isEmpty) {	5,923,447!
107	this._inProcess = false	2,144,666✔
108	break	2,144,666✔
109	}
UNCOV 110		×
111	let item = queue.getMin()	3,778,781✔
112	if (item.readyToRun) {	3,778,781!
113	queue.deleteMin()	3,778,781✔
114	}	140,037✔
115	else {	140,037✔
116	let nextNode: PairingNode<TQueueItem<any>>
117	for (const node of queue.nodes()) {	138,913✔
118	if (node.item.readyToRun) {	138,913!
119	nextNode = node	138,913✔
120	break	92,576✔
121	}
122	}	92,576✔
123		75,777✔
124	if (!nextNode) {	×
125	this._inProcess = false	122,114✔
126	break	122,114✔
127	}
128		16,799✔
129	item = nextNode.item	16,799✔
130	queue.delete(nextNode)	16,799✔
131	}	278,950✔
132		172,350✔
133	if (item.abortSignal && item.abortSignal.aborted) {	3,778,781✔
134	item.reject(item.abortSignal.reason)	1,982,680✔
135	}
136	else {
137	try {	1,796,101✔
138	let result = item.func && item.func(item.abortSignal)	1,796,101✔
139	if (result && typeof result.then === 'function') {	1,796,101✔
140	result = await result	1,197,392✔
141	}
142	item.resolve(result)	1,203,242✔
143	}
144	catch (err) {
145	item.reject(err)	592,859✔
146	}
147	}
148	}
149	}
150	}

NikolayMakhonin / priority-queue / #15

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous