git ssb

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