git ssb

1
const PortManager = require('./portManager.js')
2
const Message = require('primea-message')
3
4
module.exports = class ExoInterface {
5
  /**
6
   * the ExoInterface manages the varous message passing functions and provides
7
   * an interface for the containers to use
8
   * @param {Object} opts
9
   * @param {Object} opts.id
10
   * @param {Object} opts.state
11
   * @param {Object} opts.hypervisor
12
   * @param {Object} opts.Container
13
   */
14
  constructor (opts) {
15
    this.state = opts.state
16
    this.hypervisor = opts.hypervisor
17
    this.id = opts.id
18
    this.container = new opts.container.Constructor(this, opts.container.args)
19
20
    this.ticks = 0
21
    this.containerState = 'idle'
22
23
    // create the port manager
24
    this.ports = new PortManager(Object.assign({
25
      exInterface: this
26
    }, opts))
27
  }
28
29
  /**
30
   * adds a message to this containers message queue
31
   * @param {string} portName
32
   * @param {object} message
33
   */
34
  queue (portName, message) {
35
    message._hops++
36
    this.ports.queue(portName, message)
37
    if (this.containerState !== 'running') {
38
      this.containerState = 'running'
39
      if (portName) {
40
        this._runNextMessage()
41
      } else {
42
        this.run(message, true)
43
      }
44
    }
45
  }
46
47
  // waits for the next message
48
  async _runNextMessage () {
49
    // check if the ports are saturated, if so we don't have to wait on the
50
    // scheduler
51
    if (!this.ports.isSaturated()) {
52
      await this.hypervisor.scheduler.wait(this.ticks, this.id)
53
    }
54
55
    let message = this.ports.peekNextMessage()
56
    if (message) {
57
      if (this.ticks < message._fromTicks) {
58
        this.ticks = message._fromTicks
59
          // check for tie messages
60
        this.hypervisor.scheduler.update(this)
61
        if (!this.ports.isSaturated()) {
62
          await this.hypervisor.scheduler.wait(this.ticks, this.id)
63
          message = this.ports.peekNextMessage()
64
        }
65
      }
66
      message.fromPort.messages.shift()
67
      // run the next message
68
      this.run(message)
69
    } else {
70
      // if no more messages then shut down
71
      this.hypervisor.scheduler.done(this)
72
    }
73
  }
74
75
  /**
76
   * run the kernels code with a given enviroment
77
   * The Kernel Stores all of its state in the Environment. The Interface is used
78
   * to by the VM to retrive infromation from the Environment.
79
   * @returns {Promise}
80
   */
81
  async run (message, init = false) {
82
    let result
83
    message.ports.forEach(port => this.ports._unboundPorts.add(port))
84
    if (message.data === 'delete') {
85
      this.ports._delete(message.fromName)
86
    } else {
87
      const method = init ? 'initailize' : 'run'
88
89
      try {
90
        result = await this.container[method](message) || {}
91
      } catch (e) {
92
        result = {
93
          exception: true,
94
          exceptionError: e
95
        }
96
      }
97
    }
98
    this.ports.clearUnboundedPorts()
99
    // message.response(result)
100
    this._runNextMessage()
101
    return result
102
  }
103
104
  /**
105
   * updates the number of ticks that the container has run
106
   * @param {Number} count - the number of ticks to add
107
   */
108
  incrementTicks (count) {
109
    this.ticks += count
110
    this.hypervisor.scheduler.update(this)
111
  }
112
113
  /**
114
   * creates a new message
115
   * @param {*} data
116
   */
117
  createMessage (opts) {
118
    const message = new Message(opts)
119
    for (const port of message.ports) {
120
      if (this.ports.isBound(port)) {
121
        throw new Error('message must not contain bound ports')
122
      }
123
    }
124
    return message
125
  }
126
127
  /**
128
   * sends a message to a given port
129
   * @param {Object} portRef - the port
130
   * @param {Message} message - the message
131
   */
132
  async send (port, message) {
133
    // set the port that the message came from
134
    message._fromTicks = this.ticks
135
    message.ports.forEach(port => this.ports._unboundPorts.delete(port))
136
137
    // if (this.currentMessage !== message && !message.responsePort) {
138
    //   this.currentMessage._addSubMessage(message)
139
    // }
140
141
    if (port.destId) {
142
      const id = port.destId
143
      const instance = await this.hypervisor.getInstance(id)
144
      instance.queue(port.destName, message)
145
    } else {
146
      port.destPort.messages.push(message)
147
    }
148
  }
149
}
150