feat: Add exponential doubling capacity algorithm. (#21)

Author: kinyoklion

.circleci/config.yml

@@ -0,0 +1,46 @@
+name: ci
+on:
+  push:
+    branches: [main, rlamb/exponential-doubling-capacity-scaling]
+    paths-ignore:
+      - '**.md' #Do not need to run CI for markdown changes.
+  pull_request:
+    branches: [main]
+    paths-ignore:
+      - '**.md'
+
+jobs:
+  build-test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        node: [14, latest]
+      max-parallel: 1
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-node@v3
+        with:
+          node-version: ${{ matrix.node }}
+      - name: NPM Install
+        run: npm install
+      - name: Run Unit Tests
+        run: npm run test
+      - name: Build Contract Tests
+        run: make build-contract-tests
+      - name: Run Contract Tests
+        run: |
+          make start-contract-test-service > test-service.log 2>&1 & disown
+          set -o pipefail
+          make run-contract-tests | tee test-harness.log
+      - name: Upload Test Service Logs
+        uses: actions/upload-artifact@v3
+        if: always()
+        with:
+          name: test-service-${{ matrix.node }}
+          path: test-service.log
+      - name: Upload Test Harness Logs
+        uses: actions/upload-artifact@v3
+        if: always()
+        with:
+          name: test-harness-${{ matrix.node }}
+          path: test-harness.log

.github/workflows/ci.yml

@@ -5,3 +5,4 @@ yarn.lock
 .nyc_output
 /coverage
 package-lock.json
+tmp

.gitignore

@@ -0,0 +1,43 @@
+/**
+ * Determine if a new capacity is needed for a buffer, and if it is, then what
+ * that new capacity should be.
+ *
+ * @param {number} currentCapacity The current capacity of the buffer.
+ * @param {number} requiredCapacity The required capacity from the buffer.
+ * @param {number} maxOverAllocation The maximum extra capacity to allocate.
+ * This is how much the capacity can exceed the required capacity. If the over
+ * allocation exceeds this amount (from doubling), then instead the amount
+ * over allocated will be equal to maxOverAllocation.
+ *
+ * @returns {[boolean, number]} Either [false, 0] if no allocation is needed, or [true, <capacity>] if an
+ * allocation is needed.
+ */
+function CalculateCapacity (currentCapacity, requiredCapacity, maxOverAllocation) {
+  if (requiredCapacity > currentCapacity) {
+    let newCapacity = requiredCapacity
+
+    // Might as well start with a buffer of the maximum size that can be pooled.
+    // It is unlikely the initial buffer would be small enough to encounter
+    // this case.
+    if (newCapacity < Buffer.poolSize) {
+      newCapacity = Buffer.poolSize
+    }
+
+    // Exponential doubling capacity scaling.
+    let doubleCapacity = currentCapacity * 2
+    if (newCapacity < doubleCapacity) {
+      newCapacity = doubleCapacity
+    }
+
+    // Doubling could become problematic over a certain size, so limit the total
+    // amount of extra capacity allocated.
+    const overAllocation = newCapacity - requiredCapacity
+    if (overAllocation > maxOverAllocation) {
+      newCapacity = requiredCapacity + maxOverAllocation
+    }
+    return [true, newCapacity]
+  }
+  return [false, 0]
+}
+
+module.exports = CalculateCapacity

lib/capacity.js

@@ -1,4 +1,5 @@
 var retryDelay = require('./retry-delay')
+const CalculateCapacity = require('./capacity')
 
 var parse = require('url').parse
 var URL = require('url').URL
@@ -18,6 +19,8 @@ var space = 32
 var lineFeed = 10
 var carriageReturn = 13
 
+const MAX_OVER_ALLOCATION = 1024 * 1024 // 1 MiB
+
 function hasBom (buf) {
   return bom.every(function (charCode, index) {
     return buf[index] === charCode
@@ -211,7 +214,7 @@ function EventSource (url, eventSourceInitDict) {
       data = ''
       eventName = ''
       eventId = undefined
-      
+
       readyState = EventSource.OPEN
       res.on('close', function () {
         res.removeAllListeners('close')
@@ -232,16 +235,31 @@ function EventSource (url, eventSourceInitDict) {
       var buf
       var startingPos = 0
       var startingFieldLength = -1
+      let sizeUsed = 0
 
       res.on('data', function (chunk) {
-        buf = buf ? Buffer.concat([buf, chunk]) : chunk
-        if (isFirst && hasBom(buf)) {
-          buf = buf.slice(bom.length)
+        if (!buf) {
+          buf = chunk
+          if (isFirst && hasBom(buf)) {
+            buf = buf.slice(bom.length)
+            sizeUsed -= bom.length
+          }
+        } else {
+          // allocate new buffer
+          const [resize, newCapacity] = CalculateCapacity(buf.length, chunk.length + sizeUsed, MAX_OVER_ALLOCATION)
+          if (resize) {
+            let newBuffer = Buffer.alloc(newCapacity)
+            buf.copy(newBuffer, 0, 0, sizeUsed)
+            buf = newBuffer
+          }
+
+          chunk.copy(buf, sizeUsed)
         }
 
+        sizeUsed += chunk.length
         isFirst = false
-        var pos = 0
-        var length = buf.length
+        let pos = 0
+        const length = sizeUsed
 
         while (pos < length) {
           if (discardTrailingNewline) {
@@ -285,15 +303,17 @@ function EventSource (url, eventSourceInitDict) {
 
         if (pos === length) {
           buf = void 0
+          sizeUsed = 0
         } else if (pos > 0) {
           buf = buf.slice(pos)
+          sizeUsed = buf.length
         }
       })
     }
     var api = isSecure ? https : http
-    req = urlAndOptions.url ?
-      api.request(urlAndOptions.url, urlAndOptions.options, callback) :
-      api.request(urlAndOptions.options, callback)
+    req = urlAndOptions.url
+      ? api.request(urlAndOptions.url, urlAndOptions.options, callback)
+      : api.request(urlAndOptions.options, callback)
 
     if (config.readTimeoutMillis) {
       req.setTimeout(config.readTimeoutMillis)
@@ -376,7 +396,7 @@ function EventSource (url, eventSourceInitDict) {
       } else if (field === 'event') {
         eventName = value
       } else if (field === 'id') {
-        if (!value.includes("\u0000")) {
+        if (!value.includes('\u0000')) {
           eventId = value
         }
       } else if (field === 'retry') {

lib/eventsource.js

@@ -44,7 +44,8 @@
   "scripts": {
     "test": "mocha --reporter spec --exit && standard 'src/**/*.js' 'test/**/*.js'",
     "polyfill": "webpack lib/eventsource-polyfill.js example/eventsource-polyfill.js",
-    "coverage": "nyc --reporter=html --reporter=text _mocha --reporter spec"
+    "coverage": "nyc --reporter=html --reporter=text _mocha --reporter spec",
+    "act": "gh act --artifact-server-path=./tmp --env ACTIONS_RUNTIME_TOKEN=foo --env ACTIONS_RUNTIME_URL=http://localhost:8080/"
   },
   "engines": {
     "node": ">=0.12.0"

package.json

@@ -0,0 +1,39 @@
+const CalculateCapacity = require('../lib/capacity')
+
+const mocha = require('mocha')
+const assert = require('assert')
+const it = mocha.it
+
+it('Uses the minimum capacity', () => {
+  const [resize, newCapacity] = CalculateCapacity(0, 1, 1024 * 1024)
+  assert.equal(resize, true)
+  assert.equal(newCapacity, Buffer.poolSize)
+})
+
+it('Does not increase capacity when there is sufficient capacity', () => {
+  const [resize, newCapacity] = CalculateCapacity(1024, 1023, 1024 * 1024)
+
+  assert.equal(resize, false)
+  assert.equal(newCapacity, 0)
+})
+
+it('Uses exponential doubling capacity scaling', () => {
+  const res1 = CalculateCapacity(8192, 8193, 1024 * 1024)
+  assert.deepEqual(res1, [true, 16384])
+  const res2 = CalculateCapacity(16384, 16385, 1024 * 1024)
+  assert.deepEqual(res2, [true, 32768])
+})
+
+it('Uses required capacity when it exceeds doubling', () => {
+  const [resize, newCapacity] = CalculateCapacity(1024, 16384, 1024 * 1024)
+  assert.equal(resize, true)
+  assert.equal(newCapacity, 16384)
+})
+
+it('Does not exceed max over allocation', () => {
+  const capacity = 1024 * 1024 * 3
+  const maxOverAllocation = 1024 * 1024
+  const [resize, newCapacity] = CalculateCapacity(capacity, capacity + 1, 1024 * 1024)
+  assert.equal(resize, true)
+  assert.equal(newCapacity, capacity + maxOverAllocation + 1)
+})