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.vmState = 'idle'
11
    this.ticks = 0
12
    this.ports = new PortManager(this)
13
    this.vm = new opts.VM(this)
14
    this._waitingQueue = new PriorityQueue((a, b) => {
15
      return a.threshold > b.threshold
16
    })
17
    this.on('result', this._runNextMessage)
18
  }
19
20
  start () {
21
    return this.ports.start()
22
  }
23
24
  _updateVmState (state, message) {
25
    this.vmState = state
26
    this.emit(this.vmState, message)
27
  }
28
29
  queue (message) {
30
    this.ports.queue(message)
31
    if (this.vmState === 'idle') {
32
      this._runNextMessage()
33
    }
34
  }
35
36
  _runNextMessage () {
37
    this.ports.getNextMessage(this.ticks).then(message => {
38
      if (message) {
39
        this.run(message)
40
      } else {
41
        this._updateVmState('idle', message)
42
      }
43
    })
44
  }
45
46
  /**
47
   * run the kernels code with a given enviroment
48
   * The Kernel Stores all of its state in the Environment. The Interface is used
49
   * to by the VM to retrive infromation from the Environment.
50
   */
51
  async run (message) {
52
    this._updateVmState('running', message)
53
    // shallow copy
54
    const oldState = Object.assign({}, this._opts.state)
55
    let result
56
    try {
57
      result = await this.vm.run(message) || {}
58
    } catch (e) {
59
      console.log(e)
60
      result = {
61
        exception: true,
62
        exceptionError: e
63
      }
64
    }
65
66
    if (result.exception) {
67
      // revert to the old state
68
      clearObject(this._opts.state)
69
      Object.assign(this._opts.state, oldState)
70
    }
71
72
    this._updateVmState('result', result)
73
    return result
74
  }
75
76
  // returns a promise that resolves once the kernel hits the threshould tick
77
  // count
78
  async wait (threshold) {
79
    if (this.vmState !== 'running' && threshold > this.ticks) {
80
      // the cotract is at idle so wait
81
      return this.ports.wait(threshold)
82
    } else {
83
      return new Promise((resolve, reject) => {
84
        if (threshold <= this.ticks) {
85
          resolve(this.ticks)
86
        } else {
87
          this._waitingQueue.add({
88
            threshold: threshold,
89
            resolve: resolve
90
          })
91
        }
92
      })
93
    }
94
  }
95
96
  _updateTickCount (count) {
97
    this.ticks = count
98
    while (this._waitingQueue.peek().threshold <= count) {
99
      this._waitingQueue.poll().resolve(count)
100
    }
101
  }
102
103
  async createPort (manager, type, name, payload) {
104
    // incerment the nonce
105
    const nonce = new BN(this.nonce)
106
    nonce.iaddn(1)
107
    this.nonce = nonce.toArrayLike(Buffer)
108
109
    const parentID = await this._opts.hypervisor.generateID({
110
      id: this._opts.id
111
    })
112
113
    const port = this._opts.hypervisor.createPort(type, payload, {
114
      nonce: this.nonce,
115
      parent: parentID
116
    })
117
    manager.set(name, port)
118
    return port
119
  }
120
121
  async send (port, message) {
122
    message._ticks = this.ticks
123
    return this._opts.hypervisor.send(port, message)
124
  }
125
}
126
127
function clearObject (myObject) {
128
  for (var member in myObject) {
129
    delete myObject[member]
130
  }
131
}
132