Files: 4cf54e8cb5dc0e00ff0b10ad85fc54a4049d984f / kernel.js
3292 bytesRaw
1 | const PriorityQueue = require('fastpriorityqueue') |
2 | const EventEmitter = require('events') |
3 | const BN = require('bn.js') |
4 | const PortManager = require('./portManager.js') |
5 | |
6 | module.exports = class Kernel extends EventEmitter { |
7 | constructor (opts) { |
8 | super() |
9 | this._opts = opts |
10 | this.state = opts.parentPort.link['/'] |
11 | this.vmState = 'idle' |
12 | this.ticks = 0 |
13 | this.ports = new PortManager(this) |
14 | this.vm = new opts.VM(this) |
15 | this._waitingQueue = new PriorityQueue((a, b) => { |
16 | return a.threshold > b.threshold |
17 | }) |
18 | this.on('result', this._runNextMessage) |
19 | this.on('idle', () => { |
20 | while (!this._waitingQueue.isEmpty()) { |
21 | this._waitingQueue.poll().resolve() |
22 | } |
23 | }) |
24 | } |
25 | |
26 | start () { |
27 | return this.ports.start() |
28 | } |
29 | |
30 | _updateVmState (vmState, message) { |
31 | this.vmState = vmState |
32 | this.emit(vmState, message) |
33 | } |
34 | |
35 | queue (message) { |
36 | this.ports.queue(message) |
37 | if (this.vmState === 'idle') { |
38 | this._updateVmState('running') |
39 | this._runNextMessage() |
40 | } |
41 | } |
42 | |
43 | _runNextMessage () { |
44 | this.ports.getNextMessage(this.ticks).then(message => { |
45 | if (message) { |
46 | this.run(message) |
47 | } else { |
48 | this._updateVmState('idle', message) |
49 | } |
50 | }) |
51 | } |
52 | |
53 | /** |
54 | * run the kernels code with a given enviroment |
55 | * The Kernel Stores all of its state in the Environment. The Interface is used |
56 | * to by the VM to retrive infromation from the Environment. |
57 | */ |
58 | async run (message) { |
59 | // shallow copy |
60 | const oldState = Object.assign({}, this._opts.state) |
61 | let result |
62 | try { |
63 | result = await this.vm.run(message) || {} |
64 | } catch (e) { |
65 | result = { |
66 | exception: true, |
67 | exceptionError: e |
68 | } |
69 | clearObject(this._opts.state) |
70 | Object.assign(this._opts.state, oldState) |
71 | } |
72 | |
73 | this.emit('result', result) |
74 | return result |
75 | } |
76 | |
77 | // returns a promise that resolves once the kernel hits the threshould tick |
78 | // count |
79 | async wait (threshold) { |
80 | return new Promise((resolve, reject) => { |
81 | if (this.vmState === 'idle' || threshold <= this.ticks) { |
82 | resolve(this.ticks) |
83 | } else { |
84 | this._waitingQueue.add({ |
85 | threshold: threshold, |
86 | resolve: resolve |
87 | }) |
88 | } |
89 | }) |
90 | } |
91 | |
92 | incrementTicks (count) { |
93 | this.ticks += count |
94 | while (!this._waitingQueue.isEmpty()) { |
95 | const waiter = this._waitingQueue.peek() |
96 | if (waiter.threshold > this.ticks) { |
97 | break |
98 | } else { |
99 | this._waitingQueue.poll().resolve(this.ticks) |
100 | } |
101 | } |
102 | } |
103 | |
104 | async createPort (manager, type, name) { |
105 | // incerment the nonce |
106 | const nonce = new BN(this.state.nonce) |
107 | nonce.iaddn(1) |
108 | this.state.nonce = nonce.toArray() |
109 | |
110 | let portRef = this._opts.hypervisor.createPort(type, { |
111 | nonce: this.state.nonce, |
112 | parent: this._opts.parentPort.id |
113 | }) |
114 | await manager.set(name, portRef) |
115 | return portRef |
116 | } |
117 | |
118 | getPort (manager, name) { |
119 | return manager.getRef(name) |
120 | } |
121 | |
122 | async send (portRef, message) { |
123 | message._ticks = this.ticks |
124 | const portInstance = await this.ports.get(portRef) |
125 | portInstance.hasSent = true |
126 | return this._opts.hypervisor.send(portRef, message) |
127 | } |
128 | } |
129 | |
130 | function clearObject (myObject) { |
131 | for (var member in myObject) { |
132 | delete myObject[member] |
133 | } |
134 | } |
135 |
Built with git-ssb-web