Files: 5663a9c6ffec1a589222e24ac20e696eb4ff1f6a / scheduler.js
3616 bytesRaw
1 | const binarySearchInsert = require('binary-search-insert') |
2 | const SortedMap = require('sortedmap') |
3 | |
4 | function comparator (a, b) { |
5 | return a.ticks - b.ticks |
6 | } |
7 | |
8 | module.exports = class Scheduler { |
9 | /** |
10 | * The Scheduler manages the actor instances and tracks how many "ticks" they |
11 | * have ran. |
12 | */ |
13 | constructor () { |
14 | this._waits = [] |
15 | this._running = new Set() |
16 | this.instances = new SortedMap(comparator) |
17 | } |
18 | |
19 | /** |
20 | * locks the scheduler from clearing waits untill the lock is resolved |
21 | * @param {string} id |
22 | * @return {function} the resolve function to call once it to unlock |
23 | */ |
24 | lock (id) { |
25 | id = id.toString('hex') |
26 | let r |
27 | const p = new Promise((resolve, reject) => { |
28 | r = resolve |
29 | }) |
30 | p.ticks = 0 |
31 | this.instances.set(id, p) |
32 | this._running.add(id) |
33 | return r |
34 | } |
35 | |
36 | /** |
37 | * updates an instance with a new tick count |
38 | * @param {Object} instance - an actor instance |
39 | */ |
40 | update (instance) { |
41 | this._update(instance) |
42 | this._running.add(instance.id.toString('hex')) |
43 | this._checkWaits() |
44 | } |
45 | |
46 | _update (instance) { |
47 | this.instances.delete(instance.id.toString('hex')) |
48 | this.instances.set(instance.id.toString('hex'), instance) |
49 | } |
50 | |
51 | /** |
52 | * returns an Actor instance |
53 | * @param {String} id |
54 | * @return {Object} |
55 | */ |
56 | getInstance (id) { |
57 | id = id.toString('hex') |
58 | return this.instances.get(id) |
59 | } |
60 | |
61 | /** |
62 | * deletes an instance from the scheduler |
63 | * @param {String} id - the containers id |
64 | */ |
65 | done (id) { |
66 | id = id.toString('hex') |
67 | this._running.delete(id) |
68 | this.instances.delete(id) |
69 | this._checkWaits() |
70 | } |
71 | |
72 | /** |
73 | * returns a promise that resolves once all containers have reached the given |
74 | * number of ticks |
75 | * @param {interger} ticks - the number of ticks to wait |
76 | * @param {string} id - optional id of the container that is waiting |
77 | * @return {Promise} |
78 | */ |
79 | wait (ticks, id) { |
80 | if (id) { |
81 | id = id.toString('hex') |
82 | this._running.delete(id) |
83 | } |
84 | |
85 | return new Promise((resolve, reject) => { |
86 | binarySearchInsert(this._waits, comparator, { |
87 | ticks: ticks, |
88 | resolve: resolve, |
89 | id: id |
90 | }) |
91 | this._checkWaits() |
92 | }) |
93 | } |
94 | |
95 | /** |
96 | * returns the oldest container's ticks |
97 | * @return {integer} |
98 | */ |
99 | leastNumberOfTicks (exclude) { |
100 | let ticks = Infinity |
101 | for (const instance of this.instances) { |
102 | ticks = instance[1].ticks |
103 | if (instance[1].id !== exclude) { |
104 | return ticks |
105 | } |
106 | } |
107 | |
108 | return ticks |
109 | } |
110 | |
111 | // checks outstanding waits to see if they can be resolved |
112 | _checkWaits () { |
113 | // if there are no instances, clear any remaining waits |
114 | if (!this.instances.size) { |
115 | // console.log('here', this._waits) |
116 | this._waits.forEach(wait => wait.resolve()) |
117 | this._waits = [] |
118 | return |
119 | } |
120 | |
121 | // find the old container, see if any of the waits can be resolved |
122 | while (this._waits[0]) { |
123 | const wait = this._waits[0] |
124 | const least = this.leastNumberOfTicks(wait.id) |
125 | if (wait.ticks <= least) { |
126 | this._waits.shift() |
127 | wait.resolve() |
128 | this._running.add(wait.id) |
129 | } else { |
130 | break |
131 | } |
132 | } |
133 | |
134 | // if there are no containers running find the oldest wait |
135 | // and update the oldest containers to its ticks |
136 | if (!this._running.size && this._waits.length) { |
137 | const oldest = this._waits[0].ticks |
138 | for (let instance of this.instances) { |
139 | instance = instance[1] |
140 | if (instance.ticks > oldest) { |
141 | break |
142 | } |
143 | instance.ticks = oldest |
144 | this._update(instance) |
145 | } |
146 | return this._checkWaits() |
147 | } |
148 | } |
149 | } |
150 |
Built with git-ssb-web