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
      console.log('idle')
21
      while (!this._waitingQueue.isEmpty()) {
22
        console.log('clering ')
23
        this._waitingQueue.poll().resolve()
24
      }
25
    })
26
  }
27
28
  start () {
29
    return this.ports.start()
30
  }
31
32
  _updateVmState (vmState, message) {
33
    this.vmState = vmState
34
    this.emit(vmState, message)
35
  }
36
37
  queue (message) {
38
    this.ports.queue(message)
39
    if (this.vmState === 'idle') {
40
      this._runNextMessage()
41
    }
42
  }
43
44
  _runNextMessage () {
45
    this._updateVmState('running')
46
    this.ports.getNextMessage(this.ticks).then(message => {
47
      if (message) {
48
        this.run(message)
49
      } else {
50
        this._updateVmState('idle', message)
51
      }
52
    })
53
  }
54
55
  /**
56
   * run the kernels code with a given enviroment
57
   * The Kernel Stores all of its state in the Environment. The Interface is used
58
   * to by the VM to retrive infromation from the Environment.
59
   */
60
  async run (message) {
61
    // shallow copy
62
    const oldState = Object.assign({}, this._opts.state)
63
    let result
64
    try {
65
      result = await this.vm.run(message) || {}
66
    } catch (e) {
67
      console.log(e)
68
      result = {
69
        exception: true,
70
        exceptionError: e
71
      }
72
      clearObject(this._opts.state)
73
      Object.assign(this._opts.state, oldState)
74
    }
75
76
    this.emit('result', 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
    return new Promise((resolve, reject) => {
84
      if (this.vmState === 'idle' || 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
  incrementTicks (count) {
96
    this.ticks += count
97
    while (!this._waitingQueue.isEmpty()) {
98
      const waiter = this._waitingQueue.peek()
99
      if (waiter.threshold > this.ticks) {
100
        break
101
      } else {
102
        this._waitingQueue.poll().resolve(this.ticks)
103
      }
104
    }
105
  }
106
107
  async createPort (manager, type, name, payload) {
108
    // incerment the nonce
109
    const nonce = new BN(this._opts.state.nonce)
110
    nonce.iaddn(1)
111
    this.state.nonce = nonce.toArrayLike(Buffer)
112
113
    const parentID = await this._opts.hypervisor.generateID({
114
      id: this._opts.id
115
    })
116
117
    let port = this._opts.hypervisor.createPort(type, payload, {
118
      nonce: this.nonce,
119
      parent: parentID
120
    })
121
    await manager.set(name, port)
122
    return port
123
  }
124
125
  async send (port, message) {
126
    message._ticks = this.ticks
127
    const portObject = await this.ports.get(port)
128
    portObject.hasSent = true
129
    return this._opts.hypervisor.send(port, message)
130
  }
131
}
132
133
function clearObject (myObject) {
134
  for (var member in myObject) {
135
    delete myObject[member]
136
  }
137
}
138