Improve async with custom tick scheduler factory and exception printing

Author: FabianTerhorst

api/AltV.Net.Async/AltVAsync.cs

@@ -10,11 +10,11 @@ internal class AltVAsync
         private readonly TickScheduler scheduler;
         private readonly Thread mainThread;
 
-        public AltVAsync()
+        public AltVAsync(ITickSchedulerFactory tickSchedulerFactory)
         {
             mainThread = Thread.CurrentThread;
             mainThread.Name = "main";
-            scheduler = new TickScheduler(mainThread);
+            scheduler = tickSchedulerFactory.Create(mainThread);
             taskFactory = new TaskFactory(
                 CancellationToken.None, TaskCreationOptions.DenyChildAttach,
                 TaskContinuationOptions.None, scheduler);

api/AltV.Net.Async/AsyncEventHandler.cs

@@ -19,6 +19,15 @@ public async Task CallAsync(Func<TEvent, Task> func)
             await Task.WhenAll(tasks);
         }
 
+        public async void CallAsyncWithoutTask(Func<TEvent, Task> callback)
+        {
+            var events = GetEvents();
+            foreach (var value in events)
+            {
+                await ExecuteEventAsync(value, callback);
+            }
+        }
+
         public static async Task ExecuteEventAsync(TEvent subscription, Func<TEvent, Task> callback)
         {
             try
@@ -27,7 +36,7 @@ public static async Task ExecuteEventAsync(TEvent subscription, Func<TEvent, Tas
             }
             catch (Exception e)
             {
-                Alt.Log($"Execution of {typeof(TEvent)} threw an error: {e}");
+                AltAsync.Log($"Execution of {typeof(TEvent)} threw an error: {e}");
             }
         }
 
@@ -39,7 +48,7 @@ public static async void ExecuteEventAsyncWithoutTask(TEvent subscription, Func<
             }
             catch (Exception e)
             {
-                Alt.Log($"Execution of {typeof(TEvent)} threw an error: {e}");
+                AltAsync.Log($"Execution of {typeof(TEvent)} threw an error: {e}");
             }
         }
     }

api/AltV.Net.Async/AsyncModule.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using AltV.Net.Async.Events;
@@ -48,10 +49,10 @@ internal readonly AsyncEventHandler<PlayerChangeVehicleSeatAsyncDelegate>
 
         internal readonly AsyncEventHandler<ConsoleCommandAsyncDelegate> ConsoleCommandAsyncDelegateHandlers =
             new AsyncEventHandler<ConsoleCommandAsyncDelegate>();
-        
+
         internal readonly AsyncEventHandler<MetaDataChangeAsyncDelegate> MetaDataChangeAsyncDelegateHandlers =
             new AsyncEventHandler<MetaDataChangeAsyncDelegate>();
-        
+
         internal readonly AsyncEventHandler<MetaDataChangeAsyncDelegate> SyncedMetaDataChangeAsyncDelegateHandlers =
             new AsyncEventHandler<MetaDataChangeAsyncDelegate>();
 
@@ -79,14 +80,15 @@ public override void OnCheckPointEvent(ICheckpoint checkpoint, IEntity entity, b
         {
             base.OnCheckPointEvent(checkpoint, entity, state);
             if (!CheckpointAsyncEventHandler.HasEvents()) return;
-            Task.Run(() => CheckpointAsyncEventHandler.CallAsync(@delegate => @delegate(checkpoint, entity, state)));
+            Task.Run(() =>
+                CheckpointAsyncEventHandler.CallAsyncWithoutTask(@delegate => @delegate(checkpoint, entity, state)));
         }
 
         public override void OnPlayerConnectEvent(IPlayer player, string reason)
         {
             base.OnPlayerConnectEvent(player, reason);
             if (!PlayerConnectAsyncEventHandler.HasEvents()) return;
-            Task.Run(() => PlayerConnectAsyncEventHandler.CallAsync(@delegate => @delegate(player, reason)));
+            Task.Run(() => PlayerConnectAsyncEventHandler.CallAsyncWithoutTask(@delegate => @delegate(player, reason)));
         }
 
         public override void OnPlayerDamageEvent(IPlayer player, IEntity attacker, uint weapon, ushort damage)
@@ -96,7 +98,7 @@ public override void OnPlayerDamageEvent(IPlayer player, IEntity attacker, uint
             var oldHealth = player.Health;
             var oldArmor = player.Armor;
             Task.Run(() =>
-                PlayerDamageAsyncEventHandler.CallAsync(@delegate =>
+                PlayerDamageAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(player, attacker, oldHealth, oldArmor, weapon, damage)));
         }
 
@@ -105,7 +107,7 @@ public override void OnPlayerDeathEvent(IPlayer player, IEntity killer, uint wea
             base.OnPlayerDeathEvent(player, killer, weapon);
             if (!PlayerDeadAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                PlayerDeadAsyncEventHandler.CallAsync(@delegate => @delegate(player, killer, weapon)));
+                PlayerDeadAsyncEventHandler.CallAsyncWithoutTask(@delegate => @delegate(player, killer, weapon)));
         }
 
         public override void OnPlayerChangeVehicleSeatEvent(IVehicle vehicle, IPlayer player, byte oldSeat,
@@ -114,7 +116,7 @@ public override void OnPlayerChangeVehicleSeatEvent(IVehicle vehicle, IPlayer pl
             base.OnPlayerChangeVehicleSeatEvent(vehicle, player, oldSeat, newSeat);
             if (!PlayerChangeVehicleSeatAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                PlayerChangeVehicleSeatAsyncEventHandler.CallAsync(@delegate =>
+                PlayerChangeVehicleSeatAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(vehicle, player, oldSeat, newSeat)));
         }
 
@@ -123,7 +125,7 @@ public override void OnPlayerEnterVehicleEvent(IVehicle vehicle, IPlayer player,
             base.OnPlayerEnterVehicleEvent(vehicle, player, seat);
             if (!PlayerEnterVehicleAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                PlayerEnterVehicleAsyncEventHandler.CallAsync(@delegate =>
+                PlayerEnterVehicleAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(vehicle, player, seat)));
         }
 
@@ -132,7 +134,7 @@ public override void OnPlayerLeaveVehicleEvent(IVehicle vehicle, IPlayer player,
             base.OnPlayerLeaveVehicleEvent(vehicle, player, seat);
             if (!PlayerLeaveVehicleAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                PlayerLeaveVehicleAsyncEventHandler.CallAsync(@delegate =>
+                PlayerLeaveVehicleAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(vehicle, player, seat)));
         }
 
@@ -155,7 +157,7 @@ public override void OnPlayerRemoveEvent(IPlayer player)
             base.OnPlayerRemoveEvent(player);
             if (!PlayerRemoveAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                PlayerRemoveAsyncEventHandler.CallAsync(@delegate =>
+                PlayerRemoveAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(player)));
         }
 
@@ -164,7 +166,7 @@ public override void OnVehicleRemoveEvent(IVehicle vehicle)
             base.OnVehicleRemoveEvent(vehicle);
             if (!VehicleRemoveAsyncEventHandler.HasEvents()) return;
             Task.Run(() =>
-                VehicleRemoveAsyncEventHandler.CallAsync(@delegate =>
+                VehicleRemoveAsyncEventHandler.CallAsyncWithoutTask(@delegate =>
                     @delegate(vehicle)));
         }
 
@@ -190,14 +192,25 @@ public override void OnClientEventEvent(IPlayer player, string name, ref MValueA
                     }
                 }
 
-                Task.Run(() =>
+                Task.Run(async () =>
                 {
                     foreach (var eventHandler in eventHandlers)
                     {
                         var invokeValues = eventHandler.CalculateInvokeValues(objects, player);
                         if (invokeValues != null)
                         {
-                            eventHandler.InvokeNoResult(invokeValues);
+                            try
+                            {
+                                var task = eventHandler.InvokeTaskOrNull(invokeValues);
+                                if (task != null)
+                                {
+                                    await task;
+                                }
+                            }
+                            catch (Exception e)
+                            {
+                                AltAsync.Log($"Execution of ${name} threw an error: {e}");
+                            }
                         }
                         else
                         {
@@ -284,14 +297,25 @@ public override void OnServerEventEvent(string name, ref MValueArray args, MValu
                     }
                 }
 
-                Task.Run(() =>
+                Task.Run(async () =>
                 {
                     foreach (var eventHandler in eventHandlers)
                     {
                         var invokeValues = eventHandler.CalculateInvokeValues(objects);
                         if (invokeValues != null)
                         {
-                            eventHandler.InvokeNoResult(invokeValues);
+                            try
+                            {
+                                var task = eventHandler.InvokeTaskOrNull(invokeValues);
+                                if (task != null)
+                                {
+                                    await task;
+                                }
+                            }
+                            catch (Exception e)
+                            {
+                                AltAsync.Log($"Execution of ${name} threw an error: {e}");
+                            }
                         }
                         else
                         {
@@ -335,7 +359,7 @@ public override void OnConsoleCommandEvent(string name, string[] args)
             base.OnConsoleCommandEvent(name, args);
             if (!ConsoleCommandAsyncDelegateHandlers.HasEvents()) return;
             Task.Run(() =>
-                ConsoleCommandAsyncDelegateHandlers.CallAsync(@delegate =>
+                ConsoleCommandAsyncDelegateHandlers.CallAsyncWithoutTask(@delegate =>
                     @delegate(name, args)));
         }
 
@@ -344,7 +368,7 @@ public override void OnMetaDataChangeEvent(IEntity entity, string key, object va
             base.OnMetaDataChangeEvent(entity, key, value);
             if (!MetaDataChangeAsyncDelegateHandlers.HasEvents()) return;
             Task.Run(() =>
-                MetaDataChangeAsyncDelegateHandlers.CallAsync(@delegate =>
+                MetaDataChangeAsyncDelegateHandlers.CallAsyncWithoutTask(@delegate =>
                     @delegate(entity, key, value)));
         }
 
@@ -353,7 +377,7 @@ public override void OnSyncedMetaDataChangeEvent(IEntity entity, string key, obj
             base.OnSyncedMetaDataChangeEvent(entity, key, value);
             if (!SyncedMetaDataChangeAsyncDelegateHandlers.HasEvents()) return;
             Task.Run(() =>
-                SyncedMetaDataChangeAsyncDelegateHandlers.CallAsync(@delegate =>
+                SyncedMetaDataChangeAsyncDelegateHandlers.CallAsyncWithoutTask(@delegate =>
                     @delegate(entity, key, value)));
         }
 

api/AltV.Net.Async/AsyncResource.cs

@@ -5,7 +5,16 @@ namespace AltV.Net.Async
 {
     public abstract class AsyncResource : Resource
     {
-        private readonly AltVAsync altVAsync = new AltVAsync();
+        private readonly AltVAsync altVAsync;
+
+        public AsyncResource() : this(new DefaultTickSchedulerFactory())
+        {
+        }
+
+        public AsyncResource(ITickSchedulerFactory tickSchedulerFactory)
+        {
+            altVAsync = new AltVAsync(tickSchedulerFactory);
+        }
 
         public override void OnTick()
         {

api/AltV.Net.Async/ChannelTickScheduler.cs

@@ -0,0 +1,45 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace AltV.Net.Async
+{
+    internal class ChannelTickScheduler : TickScheduler
+    {
+        private readonly Thread mainThread;
+
+        public override int MaximumConcurrencyLevel { get; } = 1;
+
+        private readonly Channel<Task> tasks = Channel.CreateUnbounded<Task>(new UnboundedChannelOptions
+            {SingleReader = true});
+
+        private Task currentTask;
+
+        private readonly ChannelReader<Task> reader;
+
+        private readonly ChannelWriter<Task> writer;
+
+        public ChannelTickScheduler(Thread mainThread)
+        {
+            this.mainThread = mainThread;
+            reader = tasks.Reader;
+            writer = tasks.Writer;
+        }
+
+        protected override IEnumerable<Task> GetScheduledTasks() => null;
+
+        protected override void QueueTask(Task task) => writer.WriteAsync(task);
+
+        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) =>
+            Thread.CurrentThread == mainThread && TryExecuteTask(task);
+
+        public override void Tick()
+        {
+            if (reader.TryRead(out currentTask))
+            {
+                TryExecuteTask(currentTask);
+            }
+        }
+    }
+}

api/AltV.Net.Async/DefaultTickSchedulerFactory.cs

@@ -0,0 +1,12 @@
+using System.Threading;
+
+namespace AltV.Net.Async
+{
+    public class DefaultTickSchedulerFactory : ITickSchedulerFactory
+    {
+        public TickScheduler Create(Thread mainThread)
+        {
+            return new ChannelTickScheduler(mainThread);
+        }
+    }
+}

api/AltV.Net.Async/ITickSchedulerFactory.cs

@@ -0,0 +1,9 @@
+using System.Threading;
+
+namespace AltV.Net.Async
+{
+    public interface ITickSchedulerFactory
+    {
+        TickScheduler Create(Thread mainThread);
+    }
+}

api/AltV.Net.Async/QueueTickScheduler.cs

@@ -0,0 +1,42 @@
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace AltV.Net.Async
+{
+    internal class QueueTickScheduler : TickScheduler
+    {
+        private readonly Thread mainThread;
+
+        public override int MaximumConcurrencyLevel { get; } = 1;
+
+        private readonly ConcurrentQueue<Task> tasks = new ConcurrentQueue<Task>();
+
+        private Task currentTask;
+
+        private int runs;
+
+        public QueueTickScheduler(Thread mainThread)
+        {
+            this.mainThread = mainThread;
+        }
+
+        protected override IEnumerable<Task> GetScheduledTasks() => null;
+
+        protected override void QueueTask(Task task) => tasks.Enqueue(task);
+
+        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) =>
+            Thread.CurrentThread == mainThread && TryExecuteTask(task);
+
+        public override void Tick()
+        {
+            runs = tasks.Count;
+
+            while (runs-- > 0 && tasks.TryDequeue(out currentTask))
+            {
+                TryExecuteTask(currentTask);
+            }
+        }
+    }
+}