Commit 710c6eda4be5c2e63702e608a3b03af9c647ee2e
all tests passing agina
wanderer committed on 6/27/2017, 6:32:31 PMParent: 864bc27090fe788a4bf5a86c00246ca6e4391dd0
Files changed
| exoInterface.js | changed |
| index.js | changed |
| portManager.js | changed |
| scheduler.js | changed |
| tests/index.js | changed |
| exoInterface.js | ||
|---|---|---|
| @@ -47,29 +47,51 @@ | ||
| 47 | 47 | // waits for the next message |
| 48 | 48 | async _runNextMessage () { |
| 49 | 49 | // check if the ports are saturated, if so we don't have to wait on the |
| 50 | 50 | // scheduler |
| 51 | - if (!this.ports.isSaturated()) { | |
| 52 | - await this.hypervisor.scheduler.wait(this.ticks, this.id) | |
| 53 | - } | |
| 51 | + try { | |
| 52 | + let message = this.ports.peekNextMessage() | |
| 53 | + let saturated = this.ports.isSaturated() | |
| 54 | + let oldestTime = this.hypervisor.scheduler.smallest() | |
| 54 | 55 | |
| 55 | - let message = this.ports.peekNextMessage() | |
| 56 | - if (message) { | |
| 57 | - if (this.ticks < message._fromTicks) { | |
| 56 | + // this.hypervisor.scheduler.print() | |
| 57 | + while (!saturated && | |
| 58 | + !((message && oldestTime >= message._fromTicks) || | |
| 59 | + (!message && (oldestTime === this.ticks || !this.hypervisor.scheduler._running.size)))) { | |
| 60 | + const ticksToWait = message ? message._fromTicks : this.ticks | |
| 61 | + | |
| 62 | + await Promise.race([ | |
| 63 | + this.hypervisor.scheduler.wait(ticksToWait, this.id).then(m => { | |
| 64 | + // this.hypervisor.scheduler.print() | |
| 65 | + message = this.ports.peekNextMessage() | |
| 66 | + }), | |
| 67 | + this.ports.olderMessage(message).then(m => { | |
| 68 | + message = m | |
| 69 | + }), | |
| 70 | + this.ports.whenSaturated().then(() => { | |
| 71 | + saturated = true | |
| 72 | + }) | |
| 73 | + ]) | |
| 74 | + | |
| 75 | + oldestTime = this.hypervisor.scheduler.smallest() | |
| 76 | + saturated = this.ports.isSaturated() | |
| 77 | + } | |
| 78 | + | |
| 79 | + if (!message) { | |
| 80 | + // if no more messages then shut down | |
| 81 | + this.hypervisor.scheduler.done(this) | |
| 82 | + return | |
| 83 | + } | |
| 84 | + | |
| 85 | + message.fromPort.messages.shift() | |
| 86 | + if (message._fromTicks > this.ticks) { | |
| 58 | 87 | this.ticks = message._fromTicks |
| 59 | - // check for tie messages | |
| 60 | - this.hypervisor.scheduler.update(this) | |
| 61 | - if (!this.ports.isSaturated()) { | |
| 62 | - await this.hypervisor.scheduler.wait(this.ticks, this.id) | |
| 63 | - message = this.ports.peekNextMessage() | |
| 64 | - } | |
| 65 | 88 | } |
| 66 | - message.fromPort.messages.shift() | |
| 89 | + this.hypervisor.scheduler.update(this) | |
| 67 | 90 | // run the next message |
| 68 | 91 | this.run(message) |
| 69 | - } else { | |
| 70 | - // if no more messages then shut down | |
| 71 | - this.hypervisor.scheduler.done(this) | |
| 92 | + } catch (e) { | |
| 93 | + console.log(e) | |
| 72 | 94 | } |
| 73 | 95 | } |
| 74 | 96 | |
| 75 | 97 | /** |
| index.js | ||
|---|---|---|
| @@ -61,10 +61,10 @@ | ||
| 61 | 61 | id: id |
| 62 | 62 | }) |
| 63 | 63 | |
| 64 | 64 | // save the newly created instance |
| 65 | + this.scheduler.releaseLock(lock) | |
| 65 | 66 | this.scheduler.update(exoInterface) |
| 66 | - this.scheduler.releaseLock(lock) | |
| 67 | 67 | return exoInterface |
| 68 | 68 | } |
| 69 | 69 | |
| 70 | 70 | /** |
| portManager.js | ||
|---|---|---|
| @@ -36,8 +36,14 @@ | ||
| 36 | 36 | Object.assign(this, opts) |
| 37 | 37 | this.ports = this.state.ports |
| 38 | 38 | this._unboundPorts = new Set() |
| 39 | 39 | this._waitingPorts = {} |
| 40 | + this._saturationPromise = new Promise((resolve, reject) => { | |
| 41 | + this._saturationResolve = resolve | |
| 42 | + }) | |
| 43 | + this._oldestMessagePromise = new Promise((resolve, reject) => { | |
| 44 | + this._oldestMessageResolve = resolve | |
| 45 | + }) | |
| 40 | 46 | } |
| 41 | 47 | |
| 42 | 48 | /** |
| 43 | 49 | * binds a port to a name |
| @@ -129,10 +135,25 @@ | ||
| 129 | 135 | * @param {Message} message |
| 130 | 136 | */ |
| 131 | 137 | queue (name, message) { |
| 132 | 138 | if (name) { |
| 133 | - this.ports[name].messages.push(message) | |
| 139 | + const port = this.ports[name] | |
| 140 | + if (port.messages.push(message) === 1 && message._fromTicks < this._messageTickThreshold) { | |
| 141 | + message._fromPort = port | |
| 142 | + message.fromName = name | |
| 143 | + this._oldestMessageResolve(message) | |
| 144 | + this._oldestMessagePromise = new Promise((resolve, reject) => { | |
| 145 | + this._oldestMessageResolve = resolve | |
| 146 | + }) | |
| 147 | + this._messageTickThreshold = Infinity | |
| 148 | + } | |
| 134 | 149 | } |
| 150 | + if (this.isSaturated()) { | |
| 151 | + this._saturationResolve() | |
| 152 | + this._saturationPromise = new Promise((resolve, reject) => { | |
| 153 | + this._saturationResolve = resolve | |
| 154 | + }) | |
| 155 | + } | |
| 135 | 156 | } |
| 136 | 157 | |
| 137 | 158 | /** |
| 138 | 159 | * gets a port given it's name |
| @@ -199,8 +220,9 @@ | ||
| 199 | 220 | if (names.length) { |
| 200 | 221 | const portName = names.reduce(messageArbiter.bind(this)) |
| 201 | 222 | const port = this.ports[portName] |
| 202 | 223 | const message = port.messages[0] |
| 224 | + | |
| 203 | 225 | if (message) { |
| 204 | 226 | message._fromPort = port |
| 205 | 227 | message.fromName = portName |
| 206 | 228 | return message |
| @@ -212,7 +234,17 @@ | ||
| 212 | 234 | * tests wether or not all the ports have a message |
| 213 | 235 | * @returns {boolean} |
| 214 | 236 | */ |
| 215 | 237 | isSaturated () { |
| 216 | - return Object.keys(this.ports).every(name => this.ports[name].messages.length) | |
| 238 | + const keys = Object.keys(this.ports) | |
| 239 | + return keys.length ? keys.every(name => this.ports[name].messages.length) : 0 | |
| 217 | 240 | } |
| 241 | + | |
| 242 | + whenSaturated () { | |
| 243 | + return this._saturationPromise | |
| 244 | + } | |
| 245 | + | |
| 246 | + olderMessage (message) { | |
| 247 | + this._messageTickThreshold = message ? message._fromTicks : 0 | |
| 248 | + return this._oldestMessagePromise | |
| 249 | + } | |
| 218 | 250 | } |
| scheduler.js | ||
|---|---|---|
| @@ -3,11 +3,16 @@ | ||
| 3 | 3 | const comparator = function (a, b) { |
| 4 | 4 | return a.ticks - b.ticks |
| 5 | 5 | } |
| 6 | 6 | |
| 7 | +const instancesComparator = function (a, b) { | |
| 8 | + return a[1].ticks - b[1].ticks | |
| 9 | +} | |
| 10 | + | |
| 7 | 11 | module.exports = class Scheduler { |
| 8 | 12 | constructor () { |
| 9 | 13 | this._waits = [] |
| 14 | + this._running = new Set() | |
| 10 | 15 | this.instances = new Map() |
| 11 | 16 | this.locks = new Set() |
| 12 | 17 | } |
| 13 | 18 | |
| @@ -21,58 +26,81 @@ | ||
| 21 | 26 | this.locks.delete(id) |
| 22 | 27 | } |
| 23 | 28 | |
| 24 | 29 | update (instance) { |
| 30 | + this._update(instance) | |
| 31 | + this._checkWaits() | |
| 32 | + } | |
| 33 | + | |
| 34 | + _update (instance) { | |
| 35 | + this._running.add(instance.id) | |
| 25 | 36 | this.instances.delete(instance.id) |
| 26 | 37 | const instanceArray = [...this.instances] |
| 27 | - binarySearchInsert(instanceArray, comparator, [instance.id, { | |
| 28 | - ticks: instance.ticks, | |
| 29 | - instance: instance | |
| 30 | - }]) | |
| 38 | + // console.log(instanceArray) | |
| 39 | + binarySearchInsert(instanceArray, instancesComparator, [instance.id, instance]) | |
| 31 | 40 | this.instances = new Map(instanceArray) |
| 32 | - this._checkWaits() | |
| 33 | 41 | } |
| 34 | 42 | |
| 35 | 43 | getInstance (id) { |
| 36 | - const item = this.instances.get(id) | |
| 37 | - if (item) { | |
| 38 | - return item.instance | |
| 39 | - } | |
| 44 | + return this.instances.get(id) | |
| 40 | 45 | } |
| 41 | 46 | |
| 42 | 47 | done (instance) { |
| 48 | + this._running.delete(instance.id) | |
| 43 | 49 | this.instances.delete(instance.id) |
| 44 | 50 | this._checkWaits() |
| 45 | 51 | } |
| 46 | 52 | |
| 47 | - wait (ticks = Infinity) { | |
| 53 | + wait (ticks = Infinity, id) { | |
| 54 | + this._running.delete(id) | |
| 48 | 55 | if (!this.locks.size && ticks <= this.smallest()) { |
| 49 | - return | |
| 56 | + return Promise.resolve() | |
| 50 | 57 | } else { |
| 51 | 58 | return new Promise((resolve, reject) => { |
| 52 | 59 | binarySearchInsert(this._waits, comparator, { |
| 53 | 60 | ticks: ticks, |
| 54 | 61 | resolve: resolve |
| 55 | 62 | }) |
| 63 | + this._checkWaits() | |
| 56 | 64 | }) |
| 57 | 65 | } |
| 58 | 66 | } |
| 59 | 67 | |
| 60 | 68 | smallest () { |
| 61 | - return [...this.instances][0][1].ticks | |
| 69 | + return this.instances.size ? [...this.instances][0][1].ticks : 0 | |
| 62 | 70 | } |
| 63 | 71 | |
| 64 | 72 | _checkWaits () { |
| 65 | 73 | if (!this.locks.size) { |
| 74 | + // if there are no running containers | |
| 66 | 75 | if (!this.isRunning()) { |
| 67 | 76 | // clear any remanding waits |
| 68 | 77 | this._waits.forEach(wait => wait.resolve()) |
| 69 | 78 | this._waits = [] |
| 79 | + } else if (!this._running.size) { | |
| 80 | + const smallest = this._waits[0].ticks | |
| 81 | + const toUpdate = [] | |
| 82 | + for (let instance of this.instances) { | |
| 83 | + instance = instance[1] | |
| 84 | + const ticks = instance.ticks | |
| 85 | + if (ticks > smallest) { | |
| 86 | + break | |
| 87 | + } else { | |
| 88 | + toUpdate.push(instance) | |
| 89 | + } | |
| 90 | + } | |
| 91 | + toUpdate.forEach(instance => { | |
| 92 | + instance.ticks = smallest | |
| 93 | + this._update(instance) | |
| 94 | + }) | |
| 95 | + this._checkWaits() | |
| 70 | 96 | } else { |
| 71 | 97 | const smallest = this.smallest() |
| 72 | 98 | for (const index in this._waits) { |
| 73 | 99 | const wait = this._waits[index] |
| 74 | 100 | if (wait.ticks <= smallest) { |
| 101 | + // this.print() | |
| 102 | + // console.log('resolve', wait.ticks) | |
| 75 | 103 | wait.resolve() |
| 76 | 104 | } else { |
| 77 | 105 | this._waits.splice(0, index) |
| 78 | 106 | break |
| tests/index.js | ||
|---|---|---|
| @@ -176,19 +176,21 @@ | ||
| 176 | 176 | t.plan(2) |
| 177 | 177 | let runs = 0 |
| 178 | 178 | |
| 179 | 179 | class Root extends BaseContainer { |
| 180 | - run (m) { | |
| 180 | + async run (m) { | |
| 181 | 181 | if (!runs) { |
| 182 | 182 | runs++ |
| 183 | 183 | const one = this.exInterface.ports.create('first') |
| 184 | 184 | const two = this.exInterface.ports.create('second') |
| 185 | 185 | |
| 186 | 186 | this.exInterface.ports.bind('one', one) |
| 187 | 187 | this.exInterface.ports.bind('two', two) |
| 188 | 188 | |
| 189 | - this.exInterface.send(one, this.exInterface.createMessage()) | |
| 190 | - this.exInterface.send(two, this.exInterface.createMessage()) | |
| 189 | + await Promise.all([ | |
| 190 | + this.exInterface.send(one, this.exInterface.createMessage()), | |
| 191 | + this.exInterface.send(two, this.exInterface.createMessage()) | |
| 192 | + ]) | |
| 191 | 193 | } else if (runs === 1) { |
| 192 | 194 | runs++ |
| 193 | 195 | t.equals(m.data, 'second', 'should recived the second message') |
| 194 | 196 | } else if (runs === 2) { |
| @@ -199,16 +201,20 @@ | ||
| 199 | 201 | |
| 200 | 202 | class First extends BaseContainer { |
| 201 | 203 | run (m) { |
| 202 | 204 | this.exInterface.incrementTicks(2) |
| 203 | - this.exInterface.send(m.fromPort, this.exInterface.createMessage({data: 'first'})) | |
| 205 | + return this.exInterface.send(m.fromPort, this.exInterface.createMessage({ | |
| 206 | + data: 'first' | |
| 207 | + })) | |
| 204 | 208 | } |
| 205 | 209 | } |
| 206 | 210 | |
| 207 | 211 | class Second extends BaseContainer { |
| 208 | 212 | run (m) { |
| 209 | 213 | this.exInterface.incrementTicks(1) |
| 210 | - this.exInterface.send(m.fromPort, this.exInterface.createMessage({data: 'second'})) | |
| 214 | + return this.exInterface.send(m.fromPort, this.exInterface.createMessage({ | |
| 215 | + data: 'second' | |
| 216 | + })) | |
| 211 | 217 | } |
| 212 | 218 | } |
| 213 | 219 | |
| 214 | 220 | const hypervisor = new Hypervisor(node.dag) |
Built with git-ssb-web