git ssb

0+

wanderer🌟 / js-primea-hypervisor



Commit c77c7618fb3abd31a3d1b797af55b9f21a318173

Browse Files

add setTimeout to throttle actor shutdown

Signed-off-by: wanderer <mjbecze@gmail.com>
wanderer committed on 1/4/2018, 8:39:54 AM
Parent: 818c7e2b325f1dac05933b54857443ddadc56c93

Files changed

actor.jschanged
inbox.jschanged
scheduler.jschanged
tests/index.jschanged
actor.jsView
@@ -1,7 +1,6 @@
11 const Pipe = require('buffer-pipe')
22 const Cap = require('primea-capability')
3-const Message = require('primea-message')
43 const leb128 = require('leb128').unsigned
54 const Inbox = require('./inbox.js')
65
76 module.exports = class Actor {
@@ -64,21 +63,20 @@
6463
6564 // waits for the next message
6665 async _startMessageLoop () {
6766 // this ensure we only every have one loop running at a time
68- while (1) {
69- const message = await this.inbox.nextMessage(0, true)
70- if (!message) break
71-
72- // run the next message
67+ while (!this.inbox.isEmpty) {
68+ const message = await this.inbox.nextMessage(0)
7369 await this.runMessage(message)
7470 }
7571 this.running = false
7672 // wait for state ops to finish
7773 await this.state.done()
78- if (!this.running) {
79- this.container.onIdle()
80- }
74+ setTimeout(() => {
75+ if (!this.running) {
76+ this.container.onIdle()
77+ }
78+ }, 0)
8179 }
8280
8381 serializeMetaData () {
8482 return Actor.serializeMetaData(this.type, this.nonce)
inbox.jsView
@@ -29,8 +29,12 @@
2929 this._oldestMessageResolve = resolve
3030 })
3131 }
3232
33+ get isEmpty () {
34+ return !this._queue.length
35+ }
36+
3337 /**
3438 * queues a message
3539 * @param {Message} message
3640 */
@@ -69,25 +73,16 @@
6973 * Waits for the the next message if any
7074 * @param {Integer} timeout
7175 * @returns {Promise}
7276 */
73- nextMessage (timeout, getCurrent = false) {
77+ async nextMessage (timeout) {
7478 if (!this._gettingNextMessage) {
75- this._gettingNextMessage = this._nextMessage(timeout)
76- this._gettingNextMessage.then(() => {
77- this._gettingNextMessage = false
78- })
79- } else if (!getCurrent) {
79+ this._gettingNextMessage = true
80+ } else {
8081 throw new Error('already waiting for next message')
8182 }
82- return this._gettingNextMessage
83- }
8483
85- async _nextMessage (timeout) {
8684 let message = this._getOldestMessage()
87- if (message === undefined && timeout === 0) {
88- return
89- }
9085
9186 timeout += this.actor.ticks
9287 let oldestTime = this.hypervisor.scheduler.leastNumberOfTicks(this.actor.id)
9388
@@ -122,8 +117,9 @@
122117 })
123118 ])
124119 oldestTime = this.hypervisor.scheduler.leastNumberOfTicks(this.actor.id)
125120 }
121+ this._gettingNextMessage = false
126122 return this._deQueueMessage()
127123 }
128124
129125 // returns a promise that resolve when a message older then the given message
scheduler.jsView
@@ -12,9 +12,8 @@
1212 */
1313 constructor () {
1414 this._waits = []
1515 this._running = new Set()
16- this._checkingWaits = false
1716 this.instances = new SortedMap(comparator)
1817 }
1918
2019 /**
@@ -109,10 +108,8 @@
109108 // if there are no instances, clear any remaining waits
110109 if (!this.instances.size) {
111110 this._waits.forEach(wait => wait.resolve())
112111 this._waits = []
113- this._checkingWaits = false
114-
115112 return
116113 }
117114
118115 // find the old container, see if any of the waits can be resolved
@@ -139,12 +136,8 @@
139136 }
140137 instance.ticks = oldest
141138 this._update(instance)
142139 }
143- this._checkingWaits = false
144-
145140 return this._checkWaits()
146141 }
147-
148- this._checkingWaits = false
149142 }
150143 }
tests/index.jsView
@@ -436,56 +436,8 @@
436436 const stateRoot = await hypervisor.createStateRoot()
437437 t.deepEquals(stateRoot, expectedState, 'expected root!')
438438 })
439439
440-tape('return while waiting for tag', async t => {
441- t.plan(4)
442- const expectedState = {
443- '/': Buffer.from('b8eb399087a990e30373e954b627a9512c9af40b', 'hex')
444- }
445-
446- const tree = new RadixTree({
447- db: db
448- })
449-
450- class testVMContainerA extends BaseContainer {
451- async onMessage (m) {
452- if (m.tag === 1) {
453- t.true(m, 'should recive second message')
454- } else {
455- t.true(m, 'should recive first message')
456- const rCap = this.actor.mintCap(1)
457- const message = new Message({caps: [rCap]})
458- this.actor.send(m.caps[0], message)
459- this.actor.inbox.nextTaggedMessage([1], 44)
460- }
461- }
462- }
463-
464- class testVMContainerB extends BaseContainer {
465- onMessage (m) {
466- t.true(m, 'should recive a message')
467- this.actor.send(m.caps[0], new Message())
468- }
469-
470- static get typeId () {
471- return 8
472- }
473- }
474-
475- const hypervisor = new Hypervisor(tree)
476- hypervisor.registerContainer(testVMContainerA)
477- hypervisor.registerContainer(testVMContainerB)
478-
479- let capA = await hypervisor.createActor(testVMContainerA.typeId, new Message())
480- let capB = await hypervisor.createActor(testVMContainerB.typeId, new Message())
481-
482- hypervisor.send(capA, new Message({caps: [capB]}))
483-
484- const stateRoot = await hypervisor.createStateRoot()
485- t.deepEquals(stateRoot, expectedState, 'expected root!')
486-})
487-
488440 tape('trying to listen for caps more then once', async t => {
489441 t.plan(4)
490442 const expectedState = {
491443 '/': Buffer.from('b8eb399087a990e30373e954b627a9512c9af40b', 'hex')
@@ -497,9 +449,8 @@
497449
498450 class testVMContainerA extends BaseContainer {
499451 async onMessage (m) {
500452 t.true(m, 'should recive first message')
501- const rCap = this.actor.mintCap(1)
502453 const message = new Message({data: 'first'})
503454 this.actor.send(m.caps[0], message)
504455 const promise = this.actor.inbox.nextTaggedMessage([1], 44)
505456 try {

Built with git-ssb-web