git ssb

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