Add rate limiting to the image processor.

lib/cli.js

@@ -512,6 +512,9 @@ program
   .option('-r, --storage-url <url>'
     , 'URL to storage client'
     , String)
+  .option('-c, --concurrency <num>'
+    , 'maximum number of simultaneous transformations'
+    , Number)
   .option('--cache-dir <dir>'
     , 'where to cache images'
     , String
@@ -525,6 +528,7 @@ program
       port: options.port
     , storageUrl: options.storageUrl
     , cacheDir: options.cacheDir
+    , concurrency: options.concurrency || os.cpus().length
     })
   })
 

lib/roles/storage/plugins/image/index.js

@@ -3,6 +3,7 @@ var http = require('http')
 var express = require('express')
 
 var logger = require('../../../../util/logger')
+var requtil = require('../../../../util/requtil')
 
 var parseCrop = require('./param/crop')
 var parseGravity = require('./param/gravity')
@@ -18,8 +19,10 @@ module.exports = function(options) {
   app.set('case sensitive routing', true)
   app.set('trust proxy', true)
 
-  app.get('/api/v1/s/image/:id/*', function(req, res) {
-    get(req.url, options)
+  app.get(
+    '/api/v1/s/image/:id/*'
+  , requtil.limit(options.concurrency, function(req, res) {
+      return get(req.url, options)
         .then(function(stream) {
           return transform(stream, {
             crop: parseCrop(req.query.crop)
@@ -31,13 +34,18 @@ module.exports = function(options) {
           out.pipe(res)
         })
         .catch(function(err) {
-        log.error('Unable to transform resource "%s"', req.params.id, err.stack)
+          log.error(
+            'Unable to transform resource "%s"'
+          , req.params.id
+          , err.stack
+          )
           res.status(500)
             .json({
               success: false
             })
         })
     })
+  )
 
   server.listen(options.port)
   log.info('Listening on port %d', options.port)

lib/util/requtil.js

@@ -26,3 +26,25 @@ module.exports.validate = function(req, rules) {
     }
   })
 }
+
+module.exports.limit = function(limit, handler) {
+  var queue = []
+  var running = 0
+
+  function done() {
+    running -= 1
+    maybeNext()
+  }
+
+  function maybeNext() {
+    while (running < limit && queue.length) {
+      running += 1
+      handler.apply(null, queue.shift()).finally(done)
+    }
+  }
+
+  return function() {
+    queue.push(arguments)
+    maybeNext()
+  }
+}