git ssb

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