Commit 3d251f593ad3a3a4e9b46eaf068ca9dc8ec39bfa
added polling message
wanderer committed on 4/18/2017, 8:17:33 AMParent: 1635a21816ba2086ab136257355c7a5550d7a7fd
Files changed
| index.js | changed |
| kernel.js | changed |
| port.js | changed |
| portManager.js | changed |
| pollMessage.js | added |
| tests/index.js | added |
| index.js | ||
|---|---|---|
| @@ -1,5 +1,6 @@ | ||
| 1 | 1 | const Kernel = require('./index.js') |
| 2 | +const PollMessage = require('./pollMessage.js') | |
| 2 | 3 | |
| 3 | 4 | module.exports = class Hypervisor { |
| 4 | 5 | constructor (opts) { |
| 5 | 6 | this._opts = { |
| @@ -34,11 +35,11 @@ | ||
| 34 | 35 | } |
| 35 | 36 | |
| 36 | 37 | // given a port, wait untill its source contract has reached the threshold |
| 37 | 38 | // tick count |
| 38 | - async waitOnPort (port, ticks) { | |
| 39 | - let kernel = this.getVMFromPort(port) | |
| 40 | - const tickCount = await kernel.wait(ticks) | |
| 41 | - port.ticks = tickCount | |
| 42 | - return tickCount | |
| 39 | + waitOnPort (port, ticks) { | |
| 40 | + const kernel = this.getVMFromPort(port) | |
| 41 | + const message = new PollMessage(ticks) | |
| 42 | + kernel.queue(message) | |
| 43 | + return message.response() | |
| 43 | 44 | } |
| 44 | 45 | } |
| kernel.js | ||
|---|---|---|
| @@ -17,9 +17,8 @@ | ||
| 17 | 17 | this.ports = new PortManager(this) |
| 18 | 18 | this._waitingQueue = new PriorityQueue((a, b) => { |
| 19 | 19 | return a.threshold > b.threshold |
| 20 | 20 | }) |
| 21 | - | |
| 22 | 21 | this.on('result', this._runNextMessage) |
| 23 | 22 | } |
| 24 | 23 | |
| 25 | 24 | _updateState (message) { |
| @@ -33,9 +32,12 @@ | ||
| 33 | 32 | } |
| 34 | 33 | |
| 35 | 34 | queue (message) { |
| 36 | 35 | this.portManager.queue(message) |
| 37 | - if (this.state === 'idle') { | |
| 36 | + // handle system messages | |
| 37 | + if (message.isPoll) { | |
| 38 | + message.respond(this.wait(message.threshold)) | |
| 39 | + } else if (this.state === 'idle') { | |
| 38 | 40 | this._runNextMessage() |
| 39 | 41 | } |
| 40 | 42 | } |
| 41 | 43 | |
| port.js | ||
|---|---|---|
| @@ -6,9 +6,11 @@ | ||
| 6 | 6 | } |
| 7 | 7 | |
| 8 | 8 | queue (message) { |
| 9 | 9 | this.ticks = message.ticks |
| 10 | - if (this._resolve) { | |
| 10 | + if (message.isSystem) { | |
| 11 | + return | |
| 12 | + } if (this._resolve) { | |
| 11 | 13 | return this._resolve(message) |
| 12 | 14 | } else { |
| 13 | 15 | this.queue.push(message) |
| 14 | 16 | } |
| portManager.js | ||
|---|---|---|
| @@ -16,8 +16,10 @@ | ||
| 16 | 16 | } |
| 17 | 17 | }) |
| 18 | 18 | } |
| 19 | 19 | |
| 20 | + // temporaly queue message untill the ports have been mapped. Mapping the | |
| 21 | + // ports is async since the ports could just be merkle links | |
| 20 | 22 | queue (message) { |
| 21 | 23 | this._tempQueue.push(message) |
| 22 | 24 | } |
| 23 | 25 | |
| @@ -55,18 +57,21 @@ | ||
| 55 | 57 | return this._portMap.get(id) |
| 56 | 58 | } |
| 57 | 59 | |
| 58 | 60 | // waits till all ports have reached a threshold tick count |
| 59 | - async wait (ticks) { | |
| 61 | + async wait (threshold) { | |
| 62 | + // find the ports that have a smaller tick count then the threshold tick count | |
| 60 | 63 | const unkownPorts = [...this._ports].filter((id, port) => { |
| 61 | 64 | const message = port.peek() |
| 62 | - return !message || message.ticks < ticks | |
| 65 | + return !message || message.ticks < threshold | |
| 63 | 66 | }) |
| 64 | 67 | |
| 65 | - const promises = [] | |
| 66 | - for (const id in unkownPorts) { | |
| 67 | - promises.push(this.hypervisor.waitOnVM(id, ticks)) | |
| 68 | - } | |
| 68 | + const promises = unkownPorts.map(port => { | |
| 69 | + this.hypervisor.waitOnVM(port, threshold).then(ticks => { | |
| 70 | + // update the port's tick count | |
| 71 | + port.ticks = ticks | |
| 72 | + }) | |
| 73 | + }) | |
| 69 | 74 | await Promise.all(promises) |
| 70 | 75 | } |
| 71 | 76 | |
| 72 | 77 | async getNextMessage (ticks) { |
| pollMessage.js | ||
|---|---|---|
| @@ -1,0 +1,26 @@ | ||
| 1 | +const Message = require('./') | |
| 2 | +module.exports = class PollMessage extends Message { | |
| 3 | + constructor (threshold) { | |
| 4 | + super() | |
| 5 | + this.threshold = threshold | |
| 6 | + this.promise = new Promise((resolve, reject) => { | |
| 7 | + this._resolve = resolve | |
| 8 | + }) | |
| 9 | + } | |
| 10 | + | |
| 11 | + get isSystem () { | |
| 12 | + return true | |
| 13 | + } | |
| 14 | + | |
| 15 | + get isPoll () { | |
| 16 | + return true | |
| 17 | + } | |
| 18 | + | |
| 19 | + response () { | |
| 20 | + return this.promise | |
| 21 | + } | |
| 22 | + | |
| 23 | + respond (tickPromise) { | |
| 24 | + this._resolve(tickPromise) | |
| 25 | + } | |
| 26 | +} | |
| tests/index.js |
|---|
Built with git-ssb-web