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.state = opts.state
10
    this.entryPort = opts.entryPort
11
    this.hypervisor = opts.hypervisor
12
13
    this.vmState = 'idle'
14
    this.ticks = 0
15
    // create the port manager
16
    this.ports = new PortManager({
17
      kernel: this,
18
      hypervisor: opts.hypervisor,
19
      ports: opts.state.ports,
20
      entryPort: opts.entryPort,
21
      parentPort: opts.parentPort
22
    })
23
24
    this.vm = new opts.VM(this)
25
    this._waitingQueue = new PriorityQueue((a, b) => {
26
      return a.threshold > b.threshold
27
    })
28
    this.on('result', this._runNextMessage)
29
    this.on('idle', () => {
30
      while (!this._waitingQueue.isEmpty()) {
31
        this._waitingQueue.poll().resolve()
32
      }
33
    })
34
  }
35
36
  start () {
37
    return this.ports.start()
38
  }
39
40
  queue (message) {
41
    this.ports.queue(message)
42
    if (this.vmState !== 'running') {
43
      this._updateVmState('running')
44
      this._runNextMessage()
45
    }
46
  }
47
48
  _updateVmState (vmState, message) {
49
    this.vmState = vmState
50
    this.emit(vmState, message)
51
  }
52
53
  async _runNextMessage () {
54
    const message = await this.ports.getNextMessage()
55
    // if the vm is paused and it gets a message; save that message for use when the VM is resumed
56
    if (message && this.vmState === 'paused') {
57
      this.ports._portMap(message._fromPort).unshfit(message)
58
    } else if (!message && this.vmState !== 'paused') {
59
      // if no more messages then shut down
60
      this._updateVmState('idle')
61
    } else {
62
      // run the next message
63
      this._run(message)
64
    }
65
  }
66
67
  _updateEntryPort (entryPort) {
68
    // reset waits, update parent port
69
  }
70
71
  destroy () {
72
    // destory waits
73
  }
74
75
  pause () {
76
    this._setState('paused')
77
  }
78
79
  resume () {
80
    this._setState('running')
81
    this._runNextMessage()
82
  }
83
84
  /**
85
   * run the kernels code with a given enviroment
86
   * The Kernel Stores all of its state in the Environment. The Interface is used
87
   * to by the VM to retrive infromation from the Environment.
88
   */
89
  async _run (message) {
90
    // shallow copy
91
    const oldState = Object.assign({}, this.state)
92
    let result
93
    try {
94
      result = await this.vm.run(message) || {}
95
    } catch (e) {
96
      result = {
97
        exception: true,
98
        exceptionError: e
99
      }
100
      clearObject(this.state)
101
      Object.assign(this.state, oldState)
102
    }
103
104
    this.emit('result', result)
105
    return result
106
  }
107
108
  // returns a promise that resolves once the kernel hits the threshould tick
109
  // count
110
  async wait (threshold) {
111
    return new Promise((resolve, reject) => {
112
      if (this.vmState === 'idle' || threshold <= this.ticks) {
113
        resolve(this.ticks)
114
      } else {
115
        this._waitingQueue.add({
116
          threshold: threshold,
117
          resolve: resolve
118
        })
119
      }
120
    })
121
  }
122
123
  incrementTicks (count) {
124
    this.ticks += count
125
    while (!this._waitingQueue.isEmpty()) {
126
      const waiter = this._waitingQueue.peek()
127
      if (waiter.threshold > this.ticks) {
128
        break
129
      } else {
130
        this._waitingQueue.poll().resolve(this.ticks)
131
      }
132
    }
133
  }
134
135
  async createPort (type, name) {
136
    const VM = this.hypervisor._VMs[type]
137
    const parentId = this.entryPort ? this.entryPort.id : null
138
139
    const portRef = {
140
      'messages': [],
141
      'id': {
142
        '/': {
143
          nonce: this.state.nonce,
144
          parent: parentId
145
        }
146
      },
147
      'type': type,
148
      'link': {
149
        '/': VM.createState()
150
      }
151
    }
152
153
    // create the port instance
154
    await this.ports.set(name, portRef)
155
156
    // incerment the nonce
157
    const nonce = new BN(this.state.nonce)
158
    nonce.iaddn(1)
159
    this.state.nonce = nonce.toArray()
160
161
    return portRef
162
  }
163
164
  async send (portRef, message) {
165
    try {
166
      const portInstance = await this.ports.get(portRef)
167
      portInstance.hasSent = true
168
    } catch (e) {
169
      throw new Error('invalid port referance, which means the port that the port was either moved or destoried')
170
    }
171
    const id = await this.hypervisor.generateID(this.entryPort)
172
    message._fromPort = id
173
    message._ticks = this.ticks
174
175
    const receiverEntryPort = portRef === this.entryPort ? this.parentPort : portRef
176
    const vm = await this.hypervisor.getInstance(receiverEntryPort)
177
    vm.queue(message)
178
    if (this.vmState !== 'running') {
179
      this._updateVmState('running')
180
      this._runNextMessage()
181
    }
182
  }
183
}
184
185
function clearObject (myObject) {
186
  for (var member in myObject) {
187
    delete myObject[member]
188
  }
189
}
190