Re-insert into the tree if there was a user in done state that gets a new task

Author: rochoa

batch/scheduler/scheduler.js

@@ -50,6 +50,8 @@ Scheduler.prototype.add = function(user) {
     if (taskEntity) {
         if (taskEntity.status === STATUS.DONE) {
             taskEntity.status = STATUS.PENDING;
+            this.tasksTree.insert(taskEntity);
+            this.emit('add');
         }
 
         return true;
@@ -147,6 +149,10 @@ Scheduler.prototype.acquire = function(callback) {
             return null;
         }
 
+        if (candidate) {
+            self.emit('acquired', candidate.user);
+        }
+
         return callback(null, candidate);
     });
 };

test/integration/batch/scheduler.js

@@ -1,12 +1,15 @@
 'use strict';
 
 require('../../helper');
+var debug = require('../../../batch/util/debug')('scheduler-test');
 var assert = require('../../support/assert');
 var Scheduler = require('../../../batch/scheduler/scheduler');
 var FixedCapacity = require('../../../batch/scheduler/capacity/fixed');
 
 describe('scheduler', function() {
 
+    var USER_FINISHED = true;
+
     var USER_A = 'userA';
     var USER_B = 'userB';
     var USER_C = 'userC';
@@ -24,6 +27,27 @@ describe('scheduler', function() {
         }.bind(this), 50);
     };
 
+    function ManualTaskRunner() {
+        this.userTasks = {};
+    }
+
+    ManualTaskRunner.prototype.run = function(user, callback) {
+        if (!this.userTasks.hasOwnProperty(user)) {
+            this.userTasks[user] = [];
+        }
+        this.userTasks[user].push(callback);
+    };
+
+    ManualTaskRunner.prototype.dispatch = function(user, isDone) {
+        if (this.userTasks.hasOwnProperty(user)) {
+            var cb = this.userTasks[user].shift();
+            if (cb) {
+                return cb(null, isDone);
+            }
+        }
+    };
+
+
     // simulate one by one or infinity capacity
     var capacities = [new FixedCapacity(1), new FixedCapacity(2), new FixedCapacity(Infinity)];
 
@@ -54,42 +78,47 @@ describe('scheduler', function() {
             scheduler.schedule();
         });
 
-        it('regression #2', function (done) {
-            var taskRunner = new TaskRunner({
-                userA: 2,
-                userB: 2,
-                userC: 2,
-                userD: 1
-            });
+        it('regression #2: it should restart task after it was done but got re-scheduled', function (done) {
+            var taskRunner = new ManualTaskRunner();
             var scheduler = new Scheduler(capacity, taskRunner);
+            debug('Adding users A and B');
             scheduler.add(USER_A);
             scheduler.add(USER_B);
 
-            scheduler.on('done', function() {
-                var results = taskRunner.results;
+            var acquiredUsers = [];
 
-                assert.equal(results.length, 7);
+            scheduler.on('done', function() {
+                debug('Users %j', acquiredUsers);
+                assert.equal(acquiredUsers[0], USER_A);
+                assert.equal(acquiredUsers[1], USER_B);
+                assert.equal(acquiredUsers[2], USER_A);
+                assert.equal(acquiredUsers[3], USER_B);
 
-                assert.equal(results[0], USER_A);
-                assert.equal(results[1], USER_B);
-                assert.equal(results[2], USER_C);
-                assert.equal(results[3], 'userD');
-                assert.equal(results[4], USER_A);
-                assert.equal(results[5], USER_B);
-                assert.equal(results[6], USER_C);
+                assert.equal(acquiredUsers.length, 4);
 
                 return done();
             });
 
-            setTimeout(function() {
-                scheduler.add(USER_C);
-            }, 10);
-
-            setTimeout(function() {
-                scheduler.add('userD');
-            }, 20);
+            scheduler.on('acquired', function(user) {
+                debug('Acquired user %s', user);
+                acquiredUsers.push(user);
+            });
 
             scheduler.schedule();
+
+            debug('User A will be mark as DONE');
+            taskRunner.dispatch(USER_A, USER_FINISHED);
+
+            debug('User B should be running');
+            debug('User A submit a new task');
+            scheduler.add(USER_A);
+
+            debug('User B will get another task to run');
+            taskRunner.dispatch(USER_B);
+
+            debug('User A should start working on this new task');
+            taskRunner.dispatch(USER_A, USER_FINISHED);
+            taskRunner.dispatch(USER_B, USER_FINISHED);
         });
 
         it('should run tasks', function (done) {