git ssb

1
const binarySearchInsert = require('binary-search-insert')
2
const SortedMap = require('sortedmap')
3
const LockMap = require('lockmap')
4
5
module.exports = class Scheduler {
6
  /**
7
   * The Scheduler manages the run cycle of the containers and figures out which
8
   * order they should run in
9
   */
10
  constructor () {
11
    this._waits = []
12
    this._running = new Set()
13
    this._loadingInstances = new LockMap()
14
    this.instances = new SortedMap(comparator)
15
    this.systemServices = new Map()
16
17
    function comparator (a, b) {
18
      return a.ticks - b.ticks
19
    }
20
  }
21
22
  /**
23
   * locks the scheduler from clearing waits untill the lock is resolved
24
   * @param {string} id
25
   * @return {function} the resolve function to call once it to unlock
26
   */
27
  lock (id) {
28
    return this._loadingInstances.lock(id)
29
  }
30
31
  /**
32
   * updates an instance with a new tick count
33
   * @param {Object} instance - a container instance
34
   */
35
  update (instance) {
36
    this._waits = this._waits.filter(wait => wait.id !== instance.id)
37
    this._update(instance)
38
    this._running.add(instance.id)
39
    this._checkWaits()
40
  }
41
42
  _update (instance) {
43
    this.instances.delete(instance.id)
44
    this.instances.set(instance.id, instance)
45
  }
46
47
  /**
48
   * returns a container
49
   * @param {String} id
50
   * @return {Object}
51
   */
52
  getInstance (id) {
53
    return this.instances.get(id) || this._loadingInstances.getLock(id) || this.systemServices.get(id)
54
  }
55
56
  /**
57
   * deletes an instance from the scheduler
58
   * @param {string} id - the containers id
59
   */
60
  done (id) {
61
    this._running.delete(id)
62
    this.instances.delete(id)
63
    this._checkWaits()
64
  }
65
66
  /**
67
   * returns a promise that resolves once all containers have reached the given
68
   * number of ticks
69
   * @param {interger} ticks - the number of ticks to wait
70
   * @param {string} id - optional id of the container that is waiting
71
   * @return {Promise}
72
   */
73
  wait (ticks = Infinity, id) {
74
    this._running.delete(id)
75
    return new Promise((resolve, reject) => {
76
      binarySearchInsert(this._waits, comparator, {
77
        ticks: ticks,
78
        resolve: resolve,
79
        id: id
80
      })
81
      this._checkWaits()
82
    })
83
84
    function comparator (a, b) {
85
      return a.ticks - b.ticks
86
    }
87
  }
88
89
  /**
90
   * returns the oldest container's ticks
91
   * @return {integer}
92
   */
93
  leastNumberOfTicks (exculde) {
94
    for (const instance of this.instances) {
95
      if (instance[1].id !== exculde) {
96
        return instance[1].id
97
      }
98
    }
99
    return 0
100
  }
101
102
  // checks outstanding waits to see if they can be resolved
103
  _checkWaits () {
104
    // if there are no running containers
105
    if (!this.instances.size) {
106
      // clear any remanding waits
107
      this._waits.forEach(wait => wait.resolve())
108
      this._waits = []
109
    } else {
110
      // find the old container and see if to can resolve any of the waits
111
      while (this._waits[0]) {
112
        const wait = this._waits[0]
113
        const least = this.leastNumberOfTicks(wait.id)
114
        if (wait.ticks <= least) {
115
          this._waits.shift()
116
          wait.resolve()
117
          this._running.add(wait.id)
118
        } else {
119
          break
120
        }
121
      }
122
123
      if (!this._running.size) {
124
        // if there are no containers running find the oldest wait and update
125
        // the oldest containers to it ticks
126
        const oldest = this._waits[0].ticks
127
        for (let instance of this.instances) {
128
          instance = instance[1]
129
          if (instance.ticks > oldest) {
130
            break
131
          } else {
132
            instance.ticks = oldest
133
            this._update(instance)
134
          }
135
        }
136
        return this._checkWaits()
137
      }
138
    }
139
  }
140
}
141