Files: e08047273f81b45b7107bc0bc4fa2557d9f79eb0 / kernel.js
4831 bytesRaw
1 | const Message = require('primea-message') |
2 | const BN = require('bn.js') |
3 | const PortManager = require('./portManager.js') |
4 | const DeleteMessage = require('./deleteMessage') |
5 | |
6 | module.exports = class Kernel { |
7 | /** |
8 | * the Kernel manages the varous message passing functions and provides |
9 | * an interface for the containers to use |
10 | * @param {Object} opts |
11 | * @param {Object} opts.id - the UUID of the Kernel |
12 | * @param {Object} opts.state - the state of the container |
13 | * @param {Object} opts.hypervisor |
14 | * @param {Object} opts.container - the container constuctor and argments |
15 | */ |
16 | constructor (opts) { |
17 | this.state = opts.state |
18 | this.hypervisor = opts.hypervisor |
19 | this.id = opts.id |
20 | this.container = new opts.container.Constructor(this, opts.container.args) |
21 | this.timeout = 0 |
22 | |
23 | this.ticks = 0 |
24 | this.containerState = 'idle' |
25 | |
26 | // create the port manager |
27 | this.ports = new PortManager(Object.assign({ |
28 | kernel: this |
29 | }, opts)) |
30 | } |
31 | |
32 | /** |
33 | * adds a message to this containers message queue |
34 | * @param {string} portName |
35 | * @param {object} message |
36 | */ |
37 | queue (portName, message) { |
38 | this.ports.queue(portName, message) |
39 | return this._startMessageLoop() |
40 | } |
41 | |
42 | async initialize (message) { |
43 | await this.run(message, 'initialize') |
44 | return this._startMessageLoop() |
45 | } |
46 | |
47 | // waits for the next message |
48 | async _startMessageLoop () { |
49 | // this ensure we only every have one loop running at a time |
50 | if (this.containerState !== 'running') { |
51 | this.containerState = 'running' |
52 | |
53 | while (1) { |
54 | const message = await this.ports.getNextMessage(this.timeout) |
55 | if (!message) break |
56 | |
57 | // dequqe message |
58 | message.fromPort.messages.shift() |
59 | // if the message we recived had more ticks then we currently have the |
60 | // update it |
61 | if (message._fromTicks > this.ticks) { |
62 | this.ticks = message._fromTicks |
63 | this.hypervisor.scheduler.update(this) |
64 | } |
65 | // run the next message |
66 | await this.run(message) |
67 | } |
68 | |
69 | this.containerState = 'idle' |
70 | this.container.onIdle() |
71 | } |
72 | } |
73 | |
74 | shutdown () { |
75 | this.hypervisor.scheduler.done(this.id) |
76 | if (this.container.shutdown) { |
77 | this.container.shutdown() |
78 | } |
79 | } |
80 | |
81 | /** |
82 | * run the kernels code with a given enviroment |
83 | * @param {object} message - the message to run |
84 | * @param {boolean} init - whether or not to run the intialization routine |
85 | * @returns {Promise} |
86 | */ |
87 | async run (message, method = 'run') { |
88 | if (message.constructor === DeleteMessage) { |
89 | this.ports._delete(message.fromName) |
90 | } else { |
91 | const responsePort = message.responsePort |
92 | delete message.responsePort |
93 | |
94 | this.ports.addReceivedPorts(message) |
95 | let result |
96 | try { |
97 | result = await this.container[method](message) |
98 | } catch (e) { |
99 | result = { |
100 | exception: true, |
101 | exceptionError: e |
102 | } |
103 | } |
104 | |
105 | if (responsePort) { |
106 | this.send(responsePort, new Message({ |
107 | data: result |
108 | })) |
109 | } |
110 | } |
111 | |
112 | this.ports.clearUnboundedPorts() |
113 | } |
114 | |
115 | getResponsePort (message) { |
116 | if (message.responsePort) { |
117 | return message.responsePort.destPort |
118 | } else { |
119 | const [portRef1, portRef2] = this.ports.createChannel() |
120 | message.responsePort = portRef2 |
121 | this.ports._unboundPorts.delete(portRef2) |
122 | return portRef1 |
123 | } |
124 | } |
125 | |
126 | /** |
127 | * updates the number of ticks that the container has run |
128 | * @param {Number} count - the number of ticks to add |
129 | */ |
130 | incrementTicks (count) { |
131 | this.ticks += count |
132 | this.hypervisor.scheduler.update(this) |
133 | } |
134 | |
135 | /** |
136 | * creates a new message |
137 | * @param {*} data |
138 | */ |
139 | createMessage (opts) { |
140 | const message = new Message(opts) |
141 | this.ports.checkSendingPorts(message) |
142 | return message |
143 | } |
144 | |
145 | /** |
146 | * creates a new container. Returning a port to it. |
147 | * @param {String} type |
148 | * @param {*} data - the data to populate the initail state with |
149 | * @returns {Object} |
150 | */ |
151 | createInstance (type, message) { |
152 | let nonce = this.state.nonce |
153 | |
154 | const id = { |
155 | nonce: nonce, |
156 | parent: this.id |
157 | } |
158 | |
159 | // incerment the nonce |
160 | nonce = new BN(nonce) |
161 | nonce.iaddn(1) |
162 | this.state.nonce = nonce.toArray() |
163 | this.ports.removeSentPorts(message) |
164 | |
165 | return this.hypervisor.createInstance(type, message, id) |
166 | } |
167 | |
168 | /** |
169 | * sends a message to a given port |
170 | * @param {Object} portRef - the port |
171 | * @param {Message} message - the message |
172 | */ |
173 | send (port, message) { |
174 | message._hops++ |
175 | // set the port that the message came from |
176 | message._fromTicks = this.ticks |
177 | this.ports.removeSentPorts(message) |
178 | |
179 | // if (this.currentMessage !== message && !message.responsePort) { |
180 | // this.currentMessage._addSubMessage(message) |
181 | // } |
182 | return this.hypervisor.send(port, message) |
183 | } |
184 | } |
185 |
Built with git-ssb-web