From 9da118c809c08904b7469acc303aafc6caa93983 Mon Sep 17 00:00:00 2001 From: James Allen Date: Mon, 11 Aug 2014 14:16:05 +0100 Subject: [PATCH] Process updates in the background so that workers only act as dispatchers --- services/document-updater/app.coffee | 4 ++-- ...sManager.coffee => DispatchManager.coffee} | 22 +++++++++---------- .../DispatchManagerTests.coffee} | 20 ++++++++--------- 3 files changed, 22 insertions(+), 24 deletions(-) rename services/document-updater/app/coffee/{WorkersManager.coffee => DispatchManager.coffee} (70%) rename services/document-updater/test/unit/coffee/{WorkersManager/WorkersManagerTests.coffee => DispatchManager/DispatchManagerTests.coffee} (72%) diff --git a/services/document-updater/app.coffee b/services/document-updater/app.coffee index 79db012081..e246d366e8 100644 --- a/services/document-updater/app.coffee +++ b/services/document-updater/app.coffee @@ -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() diff --git a/services/document-updater/app/coffee/WorkersManager.coffee b/services/document-updater/app/coffee/DispatchManager.coffee similarity index 70% rename from services/document-updater/app/coffee/WorkersManager.coffee rename to services/document-updater/app/coffee/DispatchManager.coffee index b421422f29..afb1a5d176 100644 --- a/services/document-updater/app/coffee/WorkersManager.coffee +++ b/services/document-updater/app/coffee/DispatchManager.coffee @@ -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() \ No newline at end of file diff --git a/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee b/services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee similarity index 72% rename from services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee rename to services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee index 4a86d7cdd0..3d4be1cfec 100644 --- a/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee +++ b/services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee @@ -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