TaskClient Upgrade

Gofer.NET.Tests/GivenATaskClient.cs

@@ -1,7 +1,10 @@
 using System;
 using System.IO;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using FluentAssertions;
+using Gofer.NET.Utils;
 using Xunit;
 
 namespace Gofer.NET.Tests
@@ -29,10 +32,96 @@ public async Task ItContinuesListeningWhenATaskThrowsAnException()
             TaskQueueTestFixture.EnsureSemaphore(semaphoreFile);
         }
 
+        [Fact]
+        public async Task ItDoesNotDelayScheduledTaskPromotionWhenRunningLongTasks()
+        {
+            var waitTime = 4000;
+
+            var taskQueue = TaskQueueTestFixture.UniqueRedisTaskQueue();
+            var taskClient = new TaskClient(taskQueue);
+
+            var semaphoreFile = Path.GetTempFileName();
+            File.Delete(semaphoreFile);
+            File.Exists(semaphoreFile).Should().BeFalse();
+
+            await taskClient.TaskQueue.Enqueue(() => Wait(waitTime));
+
+            await taskClient.TaskScheduler.AddScheduledTask(
+                () => TaskQueueTestFixture.WriteSemaphore(semaphoreFile),
+                TimeSpan.FromMilliseconds(waitTime / 4));
+
+            var task = Task.Run(async () => await taskClient.Listen());
+
+            await Task.Delay(waitTime / 2);
+
+            // Ensure we did not run the scheduled task
+            File.Exists(semaphoreFile).Should().BeFalse();
+
+            var dequeuedScheduledTask = await taskQueue.Dequeue();
+
+            File.Exists(semaphoreFile).Should().BeFalse();
+            dequeuedScheduledTask.Should().NotBeNull();
+            dequeuedScheduledTask.MethodName.Should().Be(nameof(TaskQueueTestFixture.WriteSemaphore));
+
+            taskClient.CancelListen();
+            await task;
+        }
+
+        [Fact]
+        public async Task ItExecutesImmediateAndScheduledTasksInOrder()
+        {
+            const int immediateTasks = 5;
+            const int scheduledTasks = 20;
+
+            const int scheduledTasksStart = -100;
+            const int scheduledTasksIncrement = 50;
+
+            var taskQueue = TaskQueueTestFixture.UniqueRedisTaskQueue();
+            var taskClient = new TaskClient(taskQueue);
+
+            var semaphoreFile = Path.GetTempFileName();
+            File.Delete(semaphoreFile);
+            File.Exists(semaphoreFile).Should().BeFalse();
+
+            for (var i=0; i<immediateTasks; ++i)
+            {
+                await taskClient.TaskQueue.Enqueue(() => 
+                    TaskQueueTestFixture.WriteSemaphoreValue(semaphoreFile, (i+1).ToString()));
+            }
+
+            for (var i=0; i<scheduledTasks; ++i)
+            {
+                await taskClient.TaskScheduler.AddScheduledTask(
+                    () => TaskQueueTestFixture.WriteSemaphoreValue(semaphoreFile, (immediateTasks+i+1).ToString()),
+                    TimeSpan.FromMilliseconds(scheduledTasksStart + (scheduledTasksIncrement*i)));
+            }
+
+            var task = Task.Run(async () => await taskClient.Listen());
+            Thread.Sleep(scheduledTasks * scheduledTasksIncrement + 2000);
+
+            var expectedFileContents = Enumerable
+                .Range(1, immediateTasks + scheduledTasks)
+                .Aggregate("", (acc, v) => acc + v.ToString());
+
+            File.Exists(semaphoreFile).Should().BeTrue();
+            File.ReadAllText(semaphoreFile).Should().Be(expectedFileContents);
+
+            taskClient.CancelListen();
+            await task;
+        }
+
         public static void Throw()
         {
             throw new Exception();
         }
-
+
+        public static async Task Wait(int time)
+        {
+            // REVIEW: The ItDoesNotDelayScheduledTaskPromotionWhenRunningLongTasks 
+            //     test fails when using Thread.Sleep rather than Task.Delay here.
+            // Thread.Sleep(time);
+
+            await Task.Delay(time);
+        }
     }
 }

TaskClient.cs

@@ -17,9 +17,17 @@ public class TaskClient
         private bool IsCanceled { get; set; }
 
         public TaskQueue TaskQueue { get; }
+
         public Action<Exception> OnError { get; }
+
         public TaskScheduler TaskScheduler { get; }
 
+        private Task TaskSchedulerThread { get; set; }
+
+        private Task TaskRunnerThread { get; set; }
+
+        private CancellationTokenSource ListenCancellationTokenSource { get; set; }
+
         public TaskClient(
             TaskQueue taskQueue, 
             Action<Exception> onError=null)
@@ -28,54 +36,76 @@ public class TaskClient
             OnError = onError;
             TaskScheduler = new TaskScheduler(TaskQueue);
             IsCanceled = false;
-
         }
 
         public async Task Listen()
         {
-            while (true)
+            Start();
+
+            await Task.WhenAll(new [] {
+                TaskRunnerThread, 
+                TaskSchedulerThread});
+        }
+
+        public CancellationTokenSource Start()
+        {
+            if (TaskSchedulerThread != null || TaskRunnerThread != null)
             {
-                // REVIEW: Locking for read here may be unnecessary.
-                lock (Locker)
+                throw new Exception("This TaskClient is already listening.");
+            }
+
+            ListenCancellationTokenSource = new CancellationTokenSource();
+            var token = ListenCancellationTokenSource.Token;
+
+            TaskSchedulerThread = Task.Run(async () => {
+                var inThreadTaskScheduler = new TaskScheduler(TaskQueue);
+
+                while (true)
                 {
-                    if (IsCanceled)
+                    if (token.IsCancellationRequested)
                     {
                         return;
                     }
+
+                    await inThreadTaskScheduler.Tick();
                 }
-
-                // Tick the Task Scheduler
-                await TaskScheduler.Tick();
-
-                // Execute Any Queued Tasks
-                var (json, info) = await TaskQueue.SafeDequeue();
-                if (info != null)
-                {
-                    LogTaskStarted(info);
+            }, ListenCancellationTokenSource.Token);
 
-                    try
-                    {
-                        var now = DateTime.Now;
-
-                        await info.ExecuteTask();
-
-                        var completionSeconds = (DateTime.Now - now).TotalSeconds;
-                        LogTaskFinished(info, completionSeconds);
-                    }
-                    catch (Exception e)
-                    {
-                        LogTaskException(info, e);
-                    }
-                    finally
+            TaskRunnerThread = Task.Run(async () => {
+                while (true)
+                {
+                    if (token.IsCancellationRequested)
                     {
-//                        TaskQueue.Backend.RemoveBackup(json);
+                        return;
                     }
+
+                    await ExecuteQueuedTask();
+                }
+            }, ListenCancellationTokenSource.Token);
+
+            return ListenCancellationTokenSource;
+        }
+
+        private async Task ExecuteQueuedTask()
+        {
+            var (json, info) = await TaskQueue.SafeDequeue();
+            if (info != null)
+            {
+                LogTaskStarted(info);
+
+                try
+                {
+                    var now = DateTime.Now;
+
+                    await info.ExecuteTask();
+
+                    var completionSeconds = (DateTime.Now - now).TotalSeconds;
+                    LogTaskFinished(info, completionSeconds);
+                }
+                catch (Exception e)
+                {
+                    LogTaskException(info, e);
                 }
-
-                // Restore any expired backup tasks
-//                TaskQueue.RestoreExpiredBackupTasks();
-
-                Thread.Sleep(PollDelay);
             }
         }
 
@@ -101,10 +131,7 @@ private void LogTaskFinished(TaskInfo info, double completionSeconds)
 
         public void CancelListen()
         {
-            lock (Locker)
-            {
-                IsCanceled = true;
-            }
+            ListenCancellationTokenSource.Cancel();
         }
     }
 }

TaskScheduler.cs

@@ -302,9 +302,11 @@ private string LuaScriptToPromoteScheduledTasks()
                 local serializedTaskInfo
                 local isRecurring
                 local serializedRecurringJob
-
+                local key
                 local values
-                for _,key in ipairs(dueScheduledTasks) do 
+
+                for i = #dueScheduledTasks, 1, -1 do
+                    key = dueScheduledTasks[i]
                     values = redis.call(""HMGET"", KEYS[2], key, ""isRecurring::""..key, ""serializedRecurringTask::""..key)
                     serializedTaskInfo = values[1]
                     isRecurring = values[2]