all files / lib/spdy-transport/protocol/base/ scheduler.js

97.27% Statements 107/110
88.24% Branches 30/34
100% Functions 14/14
100% Lines 106/106
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205                                              572×     572×   572×   572× 572× 572× 572×       10×     45×         5409× 5409× 5409×     5409× 3669× 3669× 3669×   3669× 3669×     1740× 1740× 1740×     1740× 1727×   13×     1740×   1740×   1740×     12288× 4339×     7949× 2153×   5796×   5796× 5796× 5796× 5796×         5796×   5796×       5848× 5848× 5848× 5848× 3669× 3669× 3669× 4461× 4461×   3669×     3669× 529×     5848×     5848× 5848× 5848× 4133×     1715× 1715×   1740× 1740× 1740×   1740× 1740×   1740× 1727× 1727× 11×   1727×     1740× 1740× 1882× 1882×   1740×     1740× 1567×   1740×     1715×     52×     52×     52×     1740× 1740× 1740×     1740×     1740×     1740×    
'use strict'
 
var transport = require('../../../spdy-transport')
var utils = transport.utils
 
var assert = require('assert')
var util = require('util')
var debug = require('debug')('spdy:scheduler')
var Readable = require('readable-stream').Readable
 
/*
 * We create following structure in `pending`:
 * [ [ id = 0 ], [ id = 1 ], [ id = 2 ], [ id = 0 ] ]
 *     chunks      chunks      chunks      chunks
 *     chunks                  chunks
 *     chunks
 *
 * Then on the `.tick()` pass we pick one chunks from each item and remove the
 * item if it is empty:
 *
 * [ [ id = 0 ], [ id = 2 ] ]
 *     chunks      chunks
 *     chunks
 *
 * Writing out: chunks for 0, chunks for 1, chunks for 2, chunks for 0
 *
 * This way data is interleaved between the different streams.
 */
 
function Scheduler (options) {
  Readable.call(this)
 
  // Pretty big window by default
  this.window = 0.25
 
  Iif (options && options.window) { this.window = options.window }
 
  this.sync = []
  this.list = []
  this.count = 0
  this.pendingTick = false
}
util.inherits(Scheduler, Readable)
module.exports = Scheduler
 
// Just for testing, really
Scheduler.create = function create (options) {
  return new Scheduler(options)
}
 
function insertCompare (a, b) {
  return a.priority === b.priority
    ? a.stream - b.stream
    : b.priority - a.priority
}
 
Scheduler.prototype.schedule = function schedule (data) {
  var priority = data.priority
  var stream = data.stream
  var chunks = data.chunks
 
  // Synchronous frames should not be interleaved
  if (priority === false) {
    debug('queue sync', chunks)
    this.sync.push(data)
    this.count += chunks.length
 
    this._read()
    return
  }
 
  debug('queue async priority=%d stream=%d', priority, stream, chunks)
  var item = new SchedulerItem(stream, priority)
  var index = utils.binaryLookup(this.list, item, insertCompare)
 
  // Push new item
  if (index >= this.list.length || insertCompare(this.list[index], item) !== 0) {
    this.list.splice(index, 0, item)
  } else { // Coalesce
    item = this.list[index]
  }
 
  item.push(data)
 
  this.count += chunks.length
 
  this._read()
}
 
Scheduler.prototype._read = function _read () {
  if (this.count === 0) {
    return
  }
 
  if (this.pendingTick) {
    return
  }
  this.pendingTick = true
 
  var self = this
  process.nextTick(function () {
    self.pendingTick = false
    self.tick()
  })
}
 
Scheduler.prototype.tick = function tick () {
  // No luck for async frames
  Iif (!this.tickSync()) { return false }
 
  return this.tickAsync()
}
 
Scheduler.prototype.tickSync = function tickSync () {
  // Empty sync queue first
  var sync = this.sync
  var res = true
  this.sync = []
  for (var i = 0; i < sync.length; i++) {
    var item = sync[i]
    debug('tick sync pending=%d', this.count, item.chunks)
    for (var j = 0; j < item.chunks.length; j++) {
      this.count--
      res = this.push(item.chunks[j])
    }
    debug('after tick sync pending=%d', this.count)
 
    // TODO(indutny): figure out the way to invoke callback on actual write
    if (item.callback) {
      item.callback(null)
    }
  }
  return res
}
 
Scheduler.prototype.tickAsync = function tickAsync () {
  var res = true
  var list = this.list
  if (list.length === 0) {
    return res
  }
 
  var startPriority = list[0].priority
  for (var index = 0; list.length > 0; index++) {
    // Loop index
    index %= list.length
    if (startPriority - list[index].priority > this.window) { index = 0 }
    debug('tick async index=%d start=%d', index, startPriority)
 
    var current = list[index]
    var item = current.shift()
 
    if (current.isEmpty()) {
      list.splice(index, 1)
      if (index === 0 && list.length > 0) {
        startPriority = list[0].priority
      }
      index--
    }
 
    debug('tick async pending=%d', this.count, item.chunks)
    for (var i = 0; i < item.chunks.length; i++) {
      this.count--
      res = this.push(item.chunks[i])
    }
    debug('after tick pending=%d', this.count)
 
    // TODO(indutny): figure out the way to invoke callback on actual write
    if (item.callback) {
      item.callback(null)
    }
    Iif (!res) { break }
  }
 
  return res
}
 
Scheduler.prototype.dump = function dump () {
  this.tickSync()
 
  // Write everything out
  while (!this.tickAsync()) {
    // Intentional no-op
  }
  assert.equal(this.count, 0)
}
 
function SchedulerItem (stream, priority) {
  this.stream = stream
  this.priority = priority
  this.queue = []
}
 
SchedulerItem.prototype.push = function push (chunks) {
  this.queue.push(chunks)
}
 
SchedulerItem.prototype.shift = function shift () {
  return this.queue.shift()
}
 
SchedulerItem.prototype.isEmpty = function isEmpty () {
  return this.queue.length === 0
}