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