fair queues

Author: coronabytes

Core.TaskProcessor.Tests/ProcessorTests.cs

@@ -15,7 +15,7 @@ public ProcessorTests(ITestOutputHelper output)
         _processor = new TaskProcessor(new TaskProcessorOptions
         {
             Prefix = "{dev}",
-            MaxWorkers = 1,
+            MaxWorkers = 2,
             Queues = new[] { "q1", "fair_q2", "q3" },
             Redis = "localhost:6379,abortConnect=false",
             Retries = 3,
@@ -45,7 +45,7 @@ public ProcessorTests(ITestOutputHelper output)
     [Fact]
     public async Task EnqueueFairness()
     {
-        for (int i = 0; i < 10; i++)
+        for (int i = 0; i < 100; i++)
         {
             await _processor.EnqueueTaskAsync("fair_q2", "1001", new TaskData
             {
@@ -55,7 +55,7 @@ public async Task EnqueueFairness()
 
         await Task.Delay(500);
 
-        for (int i = 0; i < 10; i++)
+        for (int i = 0; i < 100; i++)
         {
             await _processor.EnqueueTaskAsync("fair_q2", "1002", new TaskData
             {

Core.TaskProcessor/TaskProcessor.cs

@@ -596,7 +596,15 @@ public async Task<long> RetryDeadTasksAsync(string queue, int? retries = null, l
   redis.call('hset', '{Prefix("task:")}'..taskId, 'retries', ARGV[2]);
   redis.call('hincrby', '{Prefix("batch:")}'..batchId, 'failed', -1)
   redis.call('hset', '{Prefix("batch:")}'..batchId, 'state', 'go')
-  redis.call('lpush', KEYS[1], taskId);
+
+  if string.find(KEYS[1], 'fair_') then
+    local tenant = redis.call('hget', ""{Prefix("task:")}""..taskId, 'tenant');
+    redis.call('hincrby', KEYS[1]..':fairness', tenant);
+    redis.call('lpush', KEYS[1]..':'..tenant, taskId);
+  else  
+    redis.call('lpush', KEYS[1], taskId);
+  end
+  
   redis.call('zrem', KEYS[2], taskId);
   redis.call('publish', KEYS[1]..':event', 'fetch');
 end;