Process updates in the background so that workers only act as dispatchers

This commit is contained in:
James Allen
2014-08-11 14:16:05 +01:00
parent 773e17df66
commit 9da118c809
3 changed files with 22 additions and 24 deletions
+2 -2
View File
@@ -5,7 +5,7 @@ logger = require('logger-sharelatex')
logger.initialize("documentupdater")
RedisManager = require('./app/js/RedisManager')
UpdateManager = require('./app/js/UpdateManager')
WorkersManager = require('./app/js/WorkersManager')
DispatchManager = require('./app/js/DispatchManager')
Keys = require('./app/js/RedisKeyBuilder')
redis = require('redis')
Errors = require "./app/js/Errors"
@@ -35,7 +35,7 @@ rclient.on "message", (channel, doc_key) ->
else
logger.log project_id: project_id, doc_id: doc_id, "ignoring incoming update"
WorkersManager.createAndStartWorkers(Settings.workerCount || 10)
DispatchManager.createAndStartDispatchers(Settings.dispatcherCount || 10)
UpdateManager.resumeProcessing()
@@ -5,15 +5,15 @@ redis = require('redis')
UpdateManager = require('./UpdateManager')
Metrics = require('./Metrics')
module.exports = WorkersManager =
createWorker: () ->
module.exports = DispatchManager =
createDispatcher: () ->
redisConf = Settings.redis.web
client = redis.createClient(redisConf.port, redisConf.host)
client.auth(redisConf.password)
worker = {
client: client
waitForAndProcessUpdate: (callback = (error) ->) ->
_waitForUpdateThenDispatchWorker: (callback = (error) ->) ->
timer = new Metrics.Timer "worker.waiting"
worker.client.blpop "pending-updates-list", 0, (error, result) ->
timer.done()
@@ -21,26 +21,24 @@ module.exports = WorkersManager =
return callback() if !result?
[list_name, doc_key] = result
[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)
# Dispatch this in the background
UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) ->
logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error?
return callback(error) if error?
return callback()
callback()
run: () ->
return if Settings.shuttingDown
worker.waitForAndProcessUpdate (error) =>
worker._waitForUpdateThenDispatchWorker (error) =>
if error?
logger.error err: error, "Error in worker process, waiting 1 second before continuing"
setTimeout () ->
worker.run()
, 1000
logger.error err: error, "Error in worker process"
throw error
else
worker.run()
}
return worker
createAndStartWorkers: (number) ->
createAndStartDispatchers: (number) ->
for i in [1..number]
worker = WorkersManager.createWorker()
worker = DispatchManager.createDispatcher()
worker.run()
@@ -1,12 +1,12 @@
sinon = require('sinon')
chai = require('chai')
should = chai.should()
modulePath = "../../../../app/js/WorkersManager.js"
modulePath = "../../../../app/js/DispatchManager.js"
SandboxedModule = require('sandboxed-module')
describe "WorkersManager", ->
describe "DispatchManager", ->
beforeEach ->
@WorkersManager = SandboxedModule.require modulePath, requires:
@DispatchManager = SandboxedModule.require modulePath, requires:
"./UpdateManager" : @UpdateManager = {}
"logger-sharelatex": @logger = { log: sinon.stub() }
"settings-sharelatex": @settings =
@@ -21,12 +21,12 @@ describe "WorkersManager", ->
auth: sinon.stub()
@redis.createClient = sinon.stub().returns @client
@worker = @WorkersManager.createWorker()
@worker = @DispatchManager.createDispatcher()
it "should create a new redis client", ->
@redis.createClient.called.should.equal true
describe "waitForAndProcessUpdate", ->
describe "_waitForUpdateThenDispatchWorker", ->
beforeEach ->
@project_id = "project-id-123"
@doc_id = "doc-id-123"
@@ -34,7 +34,7 @@ describe "WorkersManager", ->
@client.blpop = sinon.stub().callsArgWith(2, null, ["pending-updates-list", @doc_key])
@UpdateManager.processOutstandingUpdatesWithLock = sinon.stub().callsArg(2)
@worker.waitForAndProcessUpdate @callback
@worker._waitForUpdateThenDispatchWorker @callback
it "should call redis with BLPOP", ->
@client.blpop
@@ -50,22 +50,22 @@ describe "WorkersManager", ->
@callback.called.should.equal true
describe "run", ->
it "should call waitForAndProcessUpdate until shutting down", (done) ->
it "should call _waitForUpdateThenDispatchWorker until shutting down", (done) ->
callCount = 0
@worker.waitForAndProcessUpdate = (callback = (error) ->) =>
@worker._waitForUpdateThenDispatchWorker = (callback = (error) ->) =>
callCount++
if callCount == 3
@settings.shuttingDown = true
setTimeout () ->
callback()
, 10
sinon.spy @worker, "waitForAndProcessUpdate"
sinon.spy @worker, "_waitForUpdateThenDispatchWorker"
@worker.run()
setTimeout () =>
@worker.waitForAndProcessUpdate.callCount.should.equal 3
@worker._waitForUpdateThenDispatchWorker.callCount.should.equal 3
done()
, 100