git ssb

1
const binarySearchInsert = require('binary-search-insert')
2
const LockMap = require('lockmap')
3
4
module.exports = class Scheduler {
5
  /**
6
   * The Sceduler manages the run cycle of the containes and figures out which
7
   * order they should run in
8
   */
9
  constructor () {
10
    this._waits = []
11
    this._running = new Set()
12
    this._loadingInstances = new LockMap()
13
    this.instances = new Map()
14
  }
15
16
  /**
17
   * locks the scheduler from clearing waits untill the lock is resolved
18
   * @param {string} id
19
   * @return {function} the resolve function to call once it to unlock
20
   */
21
  lock (id) {
22
    return this._loadingInstances.lock(id)
23
  }
24
25
  /**
26
   * updates an instance with a new tick count
27
   * @param {Object} instance - a container instance
28
   */
29
  update (instance) {
30
    this._update(instance)
31
    this._running.add(instance.id)
32
    this._checkWaits()
33
  }
34
35
  _update (instance) {
36
    // sorts the container instance map by tick count
37
    this.instances.delete(instance.id)
38
    const instanceArray = [...this.instances]
39
    binarySearchInsert(instanceArray, comparator, [instance.id, instance])
40
    this.instances = new Map(instanceArray)
41
42
    function comparator (a, b) {
43
      return a[1].ticks - b[1].ticks
44
    }
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)
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 () {
94
    const nextValue = this.instances.values().next().value
95
    return nextValue ? nextValue.ticks : 0
96
  }
97
98
  // checks outstanding waits to see if they can be resolved
99
  _checkWaits () {
100
    // if there are no running containers
101
    if (!this.instances.size) {
102
      // clear any remanding waits
103
      this._waits.forEach(wait => wait.resolve())
104
      this._waits = []
105
    } else {
106
      // find the old container and see if to can resolve any of the waits
107
      const least = this.leastNumberOfTicks()
108
      for (const index in this._waits) {
109
        const wait = this._waits[index]
110
        if (wait.ticks <= least) {
111
          wait.resolve()
112
          this._running.add(wait.id)
113
        } else {
114
          this._waits.splice(0, index)
115
          break
116
        }
117
      }
118
      if (!this._running.size) {
119
        // if there are no containers running find the oldest wait and update
120
        // the oldest containers to it ticks
121
        const oldest = this._waits[0].ticks
122
        for (let instance of this.instances) {
123
          instance = instance[1]
124
          if (instance.ticks > oldest) {
125
            break
126
          } else {
127
            instance.ticks = oldest
128
            this._update(instance)
129
          }
130
        }
131
        return this._checkWaits()
132
      }
133
    }
134
  }
135
}
136