git ssb

1
const Graph = require('ipld-graph-builder')
2
const Message = require('primea-message')
3
const Kernel = require('./kernel.js')
4
const Scheduler = require('./scheduler.js')
5
const DFSchecker = require('./dfsChecker.js')
6
const chunk = require('chunk')
7
const flatten = require('flatten')
8
9
const ROOT_ID = 'zdpuAm6aTdLVMUuiZypxkwtA7sKm7BWERy8MPbaCrFsmiyzxr'
10
11
module.exports = class Hypervisor {
12
  /**
13
   * The Hypervisor manages the container instances by instantiating them and
14
   * destorying them when possible. It also facilitates localating Containers
15
   * @param {Graph} dag an instance of [ipfs.dag](https://github.com/ipfs/interface-ipfs-core/tree/master/API/dag#dag-api)
16
   * @param {object} state - the starting state
17
   */
18
  constructor (dag, state = {}) {
19
    this.graph = new Graph(dag)
20
    this.scheduler = new Scheduler()
21
    this.state = state
22
    this._containerTypes = {}
23
    this._nodesToCheck = new Set()
24
25
    this.MAX_DATA_BYTES = 6553
26
  }
27
28
  /**
29
   * add a potaintail node in the state graph to check for garbage collection
30
   * @param {string} id
31
   */
32
  addNodeToCheck (id) {
33
    this._nodesToCheck.add(id)
34
  }
35
36
  /**
37
   * given a port, this finds the corridsponeding endpoint port of the channel
38
   * @param {object} port
39
   * @returns {Promise}
40
   */
41
  getDestPort (port) {
42
    if (port.destPort) {
43
      return port.destPort
44
    } else {
45
      return this.graph.get(this.state, `${port.destId}/ports/${port.destName}`)
46
    }
47
  }
48
49
  async send (port, message) {
50
    if (port.destId) {
51
      const id = port.destId
52
      const instance = await this.getInstance(id)
53
      return instance.queue(port.destName, message)
54
    } else {
55
      // port is unbound
56
      port.destPort.messages.push(message)
57
    }
58
  }
59
60
  // loads an instance of a container from the state
61
  async _loadInstance (id) {
62
    const state = await this.graph.get(this.state, id)
63
    const container = this._containerTypes[state.type]
64
65
    // if (state.code && Array.isArray(state.code[0])) {
66
    //   state.code = flatten(state.code)
67
    // }
68
69
    // create a new kernel instance
70
    const kernel = new Kernel({
71
      hypervisor: this,
72
      state: state,
73
      container: container,
74
      id: id
75
    })
76
77
    // save the newly created instance
78
    this.scheduler.update(kernel)
79
    return kernel
80
  }
81
82
  // get a hash from a POJO
83
  _getHashFromObj (obj) {
84
    return this.graph.flush(obj).then(obj => obj['/'])
85
  }
86
87
  /**
88
   * gets an existsing container instances
89
   * @param {string} id - the containers ID
90
   * @returns {Promise}
91
   */
92
  async getInstance (id) {
93
    let instance = this.scheduler.getInstance(id)
94
    if (instance) {
95
      return instance
96
    } else {
97
      const resolve = this.scheduler.getLock(id)
98
      const instance = await this._loadInstance(id)
99
      resolve(instance)
100
      return instance
101
    }
102
  }
103
104
  /**
105
   * creates an new container instances and save it in the state
106
   * @param {string} type - the type of container to create
107
   * @param {*} code
108
   * @param {array} entryPorts
109
   * @param {object} id
110
   * @param {object} id.nonce
111
   * @param {object} id.parent
112
   * @returns {Promise}
113
   */
114
  async createInstance (type, message = new Message(), id = {nonce: 0, parent: null}) {
115
    // create a lock to prevent the scheduler from reloving waits before the
116
    // new container is loaded
117
    const resolve = this.scheduler.getLock(id)
118
    const idHash = await this._getHashFromObj(id)
119
    // const code = message.data.byteLength ? message.data : undefined
120
    const state = {
121
      nonce: [0],
122
      ports: {},
123
      type: type
124
    }
125
126
    // save the container in the state
127
    await this.graph.set(this.state, idHash, state)
128
    // create the container instance
129
    const instance = await this._loadInstance(idHash)
130
    resolve(instance)
131
    // send the intialization message
132
    await instance.initialize(message)
133
134
    if (state.code && state.code.length > this.MAX_DATA_BYTES) {
135
      state.code = chunk(state.code, this.MAX_DATA_BYTES).map(chk => {
136
        return {
137
          '/': chk
138
        }
139
      })
140
    } else {
141
      console.log(state.code)
142
    }
143
144
    return instance
145
  }
146
147
  /**
148
   * creates a state root starting from a given container and a given number of
149
   * ticks
150
   * @param {Number} ticks the number of ticks at which to create the state root
151
   * @returns {Promise}
152
   */
153
  async createStateRoot (ticks) {
154
    await this.scheduler.wait(ticks)
155
    const unlinked = await DFSchecker(this.graph, this.state, ROOT_ID, this._nodesToCheck)
156
    unlinked.forEach(id => {
157
      delete this.state[id]
158
    })
159
    console.log(this.state)
160
    return this.graph.flush(this.state)
161
  }
162
163
  /**
164
   * regirsters a container with the hypervisor
165
   * @param {String} type - the name of the type
166
   * @param {Class} Constructor - a Class for instantiating the container
167
   * @param {*} args - any args that the contructor takes
168
   */
169
  registerContainer (type, Constructor, args) {
170
    Constructor.type = type
171
    this._containerTypes[type] = {
172
      Constructor: Constructor,
173
      args: args
174
    }
175
  }
176
}
177