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 (state, message) {
31
    this.vmState = state
32
    this.emit(this.vmState, message)
33
  }
34
35
  queue (message) {
36
    this.ports.queue(message)
37
    if (this.vmState === 'idle') {
38
      this._runNextMessage()
39
    }
40
  }
41
42
  _runNextMessage () {
43
    this._updateVmState('running')
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
      console.log(e)
66
      result = {
67
        exception: true,
68
        exceptionError: e
69
      }
70
      clearObject(this._opts.state)
71
      Object.assign(this._opts.state, oldState)
72
    }
73
74
    this.emit('result', result)
75
    return result
76
  }
77
78
  // returns a promise that resolves once the kernel hits the threshould tick
79
  // count
80
  async wait (threshold) {
81
    return new Promise((resolve, reject) => {
82
      if (threshold <= this.ticks) {
83
        resolve(this.ticks)
84
      } else {
85
        this._waitingQueue.add({
86
          threshold: threshold,
87
          resolve: resolve
88
        })
89
      }
90
    })
91
  }
92
93
  incrementTicks (count) {
94
    this.ticks += count
95
    while (!this._waitingQueue.isEmpty()) {
96
      const waiter = this._waitingQueue.peek()
97
      if (waiter.threshold > this.ticks) {
98
        break
99
      } else {
100
        this._waitingQueue.poll().resolve(this.ticks)
101
      }
102
    }
103
  }
104
105
  async createPort (manager, type, name, payload) {
106
    // incerment the nonce
107
    const nonce = new BN(this._opts.state.nonce)
108
    nonce.iaddn(1)
109
    this._opts.state.nonce = nonce.toArrayLike(Buffer)
110
111
    const parentID = await this._opts.hypervisor.generateID({
112
      id: this._opts.id
113
    })
114
115
    let port = this._opts.hypervisor.createPort(type, payload, {
116
      nonce: this.nonce,
117
      parent: parentID
118
    })
119
    await manager.set(name, port)
120
    return port
121
  }
122
123
  async send (port, message) {
124
    message._ticks = this.ticks
125
    const portObject = await this.ports.get(port)
126
    portObject.hasSent = true
127
    return this._opts.hypervisor.send(port, message)
128
  }
129
}
130
131
function clearObject (myObject) {
132
  for (var member in myObject) {
133
    delete myObject[member]
134
  }
135
}
136