git ssb

1
const Tree = require('merkle-radix-tree')
2
const Graph = require('ipld-graph-builder')
3
const chunk = require('chunk')
4
const Kernel = require('./kernel.js')
5
const Scheduler = require('./scheduler.js')
6
const DFSchecker = require('./dfsChecker.js')
7
8
module.exports = class Hypervisor {
9
  /**
10
   * The Hypervisor manages the container instances by instantiating them and
11
   * destorying them when possible. It also facilitates localating Containers
12
   * @param {Graph} dag an instance of [ipfs.dag](https://github.com/ipfs/interface-ipfs-core/tree/master/API/dag#dag-api)
13
   * @param {object} state - the starting state
14
   */
15
  constructor (dag, state = {'/': Tree.emptyTreeState}) {
16
    this.graph = new Graph(dag)
17
    this.tree = new Tree({
18
      graph: this.graph,
19
      root: state
20
    })
21
    this.scheduler = new Scheduler()
22
    this.state = state
23
    this._containerTypes = {}
24
    this._nodesToCheck = new Set()
25
26
    this.ROOT_ID = 'zdpuAm6aTdLVMUuiZypxkwtA7sKm7BWERy8MPbaCrFsmiyzxr'
27
    this.CREATION_ID = 0
28
    this.ROUTING_ID = 1
29
    this.MAX_DATA_BYTES = 65533
30
  }
31
32
  /**
33
   * add a potaintail node in the state graph to check for garbage collection
34
   * @param {string} id
35
   */
36
  addNodeToCheck (id) {
37
    this._nodesToCheck.add(id)
38
  }
39
40
  /**
41
   * given a port, this finds the corridsponeding endpoint port of the channel
42
   * @param {object} port
43
   * @returns {Promise}
44
   */
45
  async getDestPort (port) {
46
    if (port.destPort) {
47
      return port.destPort
48
    } else {
49
      const containerState = await this.tree.get(port.destId)
50
      return this.graph.get(containerState, `ports/${port.destName}`)
51
    }
52
  }
53
54
  async send (port, message) {
55
    if (port.destID === this.ROUTING_ID) {
56
      port.destID = message.data.id
57
      const destPort = this.getDestPort(port.destName)
58
      if (destPort.destID === this.ROUTING_ID) {
59
        message._data = message.data.payload
60
        this.send(port, message)
61
      } else {
62
        throw new Error('invalid routing port')
63
      }
64
    } else if (port.destId) {
65
      const id = port.destId
66
      const instance = await this.getInstance(id)
67
      instance.queue(port.destName, message)
68
    } else {
69
      // port is unbound
70
      port.destPort.messages.push(message)
71
    }
72
  }
73
74
  // loads an instance of a container from the state
75
  async _loadInstance (id, state) {
76
    if (!state) {
77
      state = await this.tree.get(id)
78
    }
79
    const container = this._containerTypes[state.type]
80
    let code
81
82
    // checks if the code stored in the state is an array and that the elements
83
    // are merkle link
84
    if (state.code && state.code[0]['/']) {
85
      await this.graph.tree(state.code, 1)
86
      code = state.code.map(a => a['/']).reduce((a, b) => a + b)
87
    } else {
88
      code = state.code
89
    }
90
91
    // create a new kernel instance
92
    const kernel = new Kernel({
93
      hypervisor: this,
94
      state: state,
95
      code: code,
96
      container: container,
97
      id: id
98
    })
99
100
    // save the newly created instance
101
    this.scheduler.update(kernel)
102
    return kernel
103
  }
104
105
  // get a hash from a POJO
106
  _getHashFromObj (obj) {
107
    return this.graph.flush(obj).then(obj => obj['/'])
108
  }
109
110
  /**
111
   * gets an existsing container instances
112
   * @param {string} id - the containers ID
113
   * @returns {Promise}
114
   */
115
  async getInstance (id) {
116
    let instance = this.scheduler.getInstance(id)
117
    if (instance) {
118
      return instance
119
    } else {
120
      const resolve = this.scheduler.lock(id)
121
      const instance = await this._loadInstance(id)
122
      await instance.startup()
123
      resolve(instance)
124
      return instance
125
    }
126
  }
127
128
  /**
129
   * creates an new container instances and save it in the state
130
   * @param {string} type - the type of container to create
131
   * @param {*} code
132
   * @param {array} entryPorts
133
   * @param {object} id
134
   * @param {object} id.nonce
135
   * @param {object} id.parent
136
   * @returns {Promise}
137
   */
138
  async createInstance (message, id = {nonce: 0, parent: null}) {
139
    const idHash = await this._getHashFromObj(id)
140
    const state = {
141
      nonce: [0],
142
      ports: {},
143
      type: message.data.type
144
    }
145
146
    if (message.data.code && message.data.code.length) {
147
      state.code = message.data.code
148
    }
149
150
    // create the container instance
151
    const instance = await this._loadInstance(idHash, state)
152
153
    // send the intialization message
154
    await instance.create(message)
155
156
    if (Object.keys(instance.ports.ports).length || instance.id === this.ROOT_ID) {
157
      if (state.code && state.code.length > this.MAX_DATA_BYTES) {
158
        state.code = chunk(state.code, this.MAX_DATA_BYTES).map(chk => {
159
          return {
160
            '/': chk
161
          }
162
        })
163
      }
164
      // save the container in the state
165
      await this.tree.set(idHash, state)
166
    } else {
167
      this.scheduler.done(idHash)
168
    }
169
170
    return instance
171
  }
172
173
  createChannel () {
174
    const port1 = {
175
      messages: []
176
    }
177
178
    const port2 = {
179
      messages: [],
180
      destPort: port1
181
    }
182
183
    port1.destPort = port2
184
    return [port1, port2]
185
  }
186
187
  getCreationPort () {
188
    return {
189
      destID: this.CREATION_ID
190
    }
191
  }
192
193
  getRoutingPort () {
194
    return {
195
      destID: this.ROUTING_ID
196
    }
197
  }
198
199
  copyNativePort () {}
200
  /**
201
   * creates a state root starting from a given container and a given number of
202
   * ticks
203
   * @param {Number} ticks the number of ticks at which to create the state root
204
   * @returns {Promise}
205
   */
206
  async createStateRoot (ticks) {
207
    await this.scheduler.wait(ticks)
208
209
    const unlinked = await DFSchecker(this.tree, this._nodesToCheck, (id) => {
210
      return this.ROOT_ID === id
211
    })
212
    for (const id of unlinked) {
213
      await this.tree.delete(id)
214
    }
215
    // console.log(JSON.stringify(this.state, null, 2))
216
217
    return this.graph.flush(this.state)
218
  }
219
220
  /**
221
   * regirsters a container with the hypervisor
222
   * @param {Class} Constructor - a Class for instantiating the container
223
   * @param {*} args - any args that the contructor takes
224
   * @param {interger} typeId - the container's type identification ID
225
   */
226
  registerContainer (Constructor, args, typeId = Constructor.typeId) {
227
    this._containerTypes[typeId] = {
228
      Constructor: Constructor,
229
      args: args
230
    }
231
  }
232
}
233