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
      result = {
60
        exception: true,
61
        exceptionError: e
62
      }
63
    }
64
65
    if (result.exception) {
66
      // revert to the old state
67
      clearObject(this._opts.state)
68
      Object.assign(this._opts.state, oldState)
69
    }
70
71
    this._updateVmState('result', result)
72
    return result
73
  }
74
75
  // returns a promise that resolves once the kernel hits the threshould tick
76
  // count
77
  async wait (threshold) {
78
    if (this.vmState !== 'running' && threshold > this.ticks) {
79
      // the cotract is at idle so wait
80
      return this.ports.wait(threshold)
81
    } else {
82
      return new Promise((resolve, reject) => {
83
        if (threshold <= this.ticks) {
84
          resolve(this.ticks)
85
        } else {
86
          this._waitingQueue.add({
87
            threshold: threshold,
88
            resolve: resolve
89
          })
90
        }
91
      })
92
    }
93
  }
94
95
  _updateTickCount (count) {
96
    this.ticks = count
97
    while (this._waitingQueue.peek().threshold <= count) {
98
      this._waitingQueue.poll().resolve(count)
99
    }
100
  }
101
102
  createPort () {
103
    const nonce = new BN(this.nonce)
104
    nonce.iaddn(1)
105
    this.nonce = nonce.toArrayLike(Buffer)
106
    return {
107
      id: {
108
        '/': {
109
          nonce: this.nonce,
110
          parent: this.id
111
        }
112
      },
113
      link: {
114
        '/': {}
115
      }
116
    }
117
  }
118
119
  async send (port, message) {
120
    message._ticks = this.ticks
121
    return this._opts.hypervisor.send(port, message)
122
  }
123
}
124
125
function clearObject (myObject) {
126
  for (var member in myObject) {
127
    delete myObject[member]
128
  }
129
}
130