Expanded tests

Everything working except cancelling reserves

Author: Sushisource

src/Temporalio/Bridge/CustomSlotSupplier.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Runtime.InteropServices;
 using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
 
 namespace Temporalio.Bridge
 {
@@ -10,6 +11,7 @@ namespace Temporalio.Bridge
     /// </summary>
     internal class CustomSlotSupplier : NativeInvokeableClass<Interop.CustomSlotSupplierCallbacks>
     {
+        private readonly ILogger logger;
         private readonly Temporalio.Worker.Tuning.ICustomSlotSupplier userSupplier;
         private readonly Dictionary<uint, GCHandle> permits = new();
         private uint permitId = 1;
@@ -18,8 +20,12 @@ internal class CustomSlotSupplier : NativeInvokeableClass<Interop.CustomSlotSupp
         /// Initializes a new instance of the <see cref="CustomSlotSupplier" /> class.
         /// </summary>
         /// <param name="userSupplier">User's slot supplier implementation'.</param>
-        internal unsafe CustomSlotSupplier(Temporalio.Worker.Tuning.ICustomSlotSupplier userSupplier)
+        /// <param name="loggerFactory">Logger factory.</param>
+        internal unsafe CustomSlotSupplier(
+            Temporalio.Worker.Tuning.ICustomSlotSupplier userSupplier,
+            ILoggerFactory loggerFactory)
         {
+            this.logger = loggerFactory.CreateLogger<CustomSlotSupplier>();
             this.userSupplier = userSupplier;
 
             var interopCallbacks = new Interop.CustomSlotSupplierCallbacks
@@ -42,26 +48,45 @@ private void SafeReserve(Interop.SlotReserveCtx ctx, IntPtr sender)
         {
             var reserveTask = Task.Run(async () =>
             {
-                try
+                while (true)
                 {
-                    var permit = await userSupplier.ReserveSlotAsync(new(ctx)).ConfigureAwait(false);
-                    var usedPermitId = AddPermitToMap(permit);
-                    unsafe
+                    try
                     {
-                        Interop.Methods.complete_async_reserve(sender.ToPointer(), new(usedPermitId));
+                        var permit = await userSupplier.ReserveSlotAsync(new(ctx)).ConfigureAwait(false);
+                        var usedPermitId = AddPermitToMap(permit);
+                        unsafe
+                        {
+                            Interop.Methods.complete_async_reserve(sender.ToPointer(), new(usedPermitId));
+                        }
+                        return;
                     }
-                }
-                catch (Exception e)
-                {
-                    Console.WriteLine("Exception in reserve: " + e.Message);
-                    throw;
+#pragma warning disable CA1031 // We are ok catching all exceptions here
+                    catch (Exception e)
+                    {
+#pragma warning restore CA1031
+                        logger.LogError(e, "Error reserving slot");
+                    }
+                    // Wait for a bit to avoid spamming errors
+                    await Task.Delay(1000).ConfigureAwait(false);
                 }
             });
         }
 
         private unsafe UIntPtr TryReserve(Interop.SlotReserveCtx ctx)
         {
-            var maybePermit = userSupplier.TryReserveSlot(new(ctx));
+            Temporalio.Worker.Tuning.ISlotPermit? maybePermit;
+            try
+            {
+                maybePermit = userSupplier.TryReserveSlot(new(ctx));
+            }
+#pragma warning disable CA1031 // We are ok catching all exceptions here
+            catch (Exception e)
+            {
+#pragma warning restore CA1031
+                logger.LogError(e, "Error trying to reserve slot");
+                return UIntPtr.Zero;
+            }
+
             if (maybePermit == null)
             {
                 return UIntPtr.Zero;
@@ -72,17 +97,35 @@ private unsafe UIntPtr TryReserve(Interop.SlotReserveCtx ctx)
 
         private void MarkUsed(Interop.SlotMarkUsedCtx ctx)
         {
-            userSupplier.MarkSlotUsed(new(ctx, permits[ctx.slot_permit.ToUInt32()]));
+            try
+            {
+                userSupplier.MarkSlotUsed(new(ctx, permits[ctx.slot_permit.ToUInt32()]));
+            }
+#pragma warning disable CA1031 // We are ok catching all exceptions here
+            catch (Exception e)
+            {
+#pragma warning restore CA1031
+                logger.LogError(e, "Error marking slot used");
+            }
         }
 
         private void Release(Interop.SlotReleaseCtx ctx)
         {
             var permitId = ctx.slot_permit.ToUInt32();
-            userSupplier.ReleaseSlot(new(ctx, permits[permitId]));
+            try
+            {
+                userSupplier.ReleaseSlot(new(ctx, permits[permitId]));
+            }
+#pragma warning disable CA1031 // We are ok catching all exceptions here
+            catch (Exception e)
+            {
+#pragma warning restore CA1031
+                logger.LogError(e, "Error releasing slot");
+            }
             permits.Remove(permitId);
         }
 
-        private uint AddPermitToMap(Temporalio.Worker.Tuning.SlotPermit permit)
+        private uint AddPermitToMap(Temporalio.Worker.Tuning.ISlotPermit permit)
         {
             var handle = GCHandle.Alloc(permit);
             lock (permits)

src/Temporalio/Bridge/OptionsExtensions.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
+using Microsoft.Extensions.Logging;
 using Temporalio.Bridge.Interop;
 using Temporalio.Exceptions;
 
@@ -408,11 +409,13 @@ public static Interop.TestServerOptions ToInteropOptions(
         /// <param name="options">Options to convert.</param>
         /// <param name="scope">Scope to use.</param>
         /// <param name="namespace_">Namespace for the worker.</param>
+        /// <param name="loggerFactory">Logger factory.</param>
         /// <returns>Converted options.</returns>
         public static Interop.WorkerOptions ToInteropOptions(
             this Temporalio.Worker.TemporalWorkerOptions options,
             Scope scope,
-            string namespace_)
+            string namespace_,
+            ILoggerFactory loggerFactory)
         {
             if (options.TaskQueue == null)
             {
@@ -458,7 +461,7 @@ public static Interop.WorkerOptions ToInteropOptions(
                 build_id = scope.ByteArray(buildId),
                 identity_override = scope.ByteArray(options.Identity),
                 max_cached_workflows = (uint)options.MaxCachedWorkflows,
-                tuner = tuner.ToInteropTuner(scope),
+                tuner = tuner.ToInteropTuner(scope, loggerFactory),
                 no_remote_activities = (byte)(noRemoteActivities ? 1 : 0),
                 sticky_queue_schedule_to_start_timeout_millis =
                     (ulong)options.StickyQueueScheduleToStartTimeout.TotalMilliseconds,
@@ -504,7 +507,7 @@ public static Interop.WorkerOptions ToInteropOptions(
                 build_id = scope.ByteArray(buildId),
                 identity_override = scope.ByteArray(options.Identity),
                 max_cached_workflows = 2,
-                tuner = Temporalio.Worker.Tuning.WorkerTuner.CreateFixedSize(2, 1, 1).ToInteropTuner(scope),
+                tuner = Temporalio.Worker.Tuning.WorkerTuner.CreateFixedSize(2, 1, 1).ToInteropTuner(scope, options.LoggerFactory),
                 no_remote_activities = 1,
                 sticky_queue_schedule_to_start_timeout_millis = 1000,
                 max_heartbeat_throttle_interval_millis = 1000,
@@ -524,7 +527,8 @@ public static Interop.WorkerOptions ToInteropOptions(
 
         private static Interop.TunerHolder ToInteropTuner(
             this Temporalio.Worker.Tuning.WorkerTuner tuner,
-            Scope scope)
+            Scope scope,
+            ILoggerFactory loggerFactory)
         {
             Temporalio.Worker.Tuning.ResourceBasedTunerOptions? lastTunerOptions = null;
             Temporalio.Worker.Tuning.ISlotSupplier[] suppliers =
@@ -549,17 +553,18 @@ private static Interop.TunerHolder ToInteropTuner(
             return new()
             {
                 workflow_slot_supplier =
-                    tuner.WorkflowTaskSlotSupplier.ToInteropSlotSupplier(true),
+                    tuner.WorkflowTaskSlotSupplier.ToInteropSlotSupplier(true, loggerFactory),
                 activity_slot_supplier =
-                    tuner.ActivityTaskSlotSupplier.ToInteropSlotSupplier(false),
+                    tuner.ActivityTaskSlotSupplier.ToInteropSlotSupplier(false, loggerFactory),
                 local_activity_slot_supplier =
-                    tuner.LocalActivitySlotSupplier.ToInteropSlotSupplier(false),
+                    tuner.LocalActivitySlotSupplier.ToInteropSlotSupplier(false, loggerFactory),
             };
         }
 
         private static Interop.SlotSupplier ToInteropSlotSupplier(
             this Temporalio.Worker.Tuning.ISlotSupplier supplier,
-            bool isWorkflow)
+            bool isWorkflow,
+            ILoggerFactory loggerFactory)
         {
             if (supplier is Temporalio.Worker.Tuning.FixedSizeSlotSupplier fixedSize)
             {
@@ -604,7 +609,7 @@ private static Interop.SlotSupplier ToInteropSlotSupplier(
             }
             else if (supplier is Temporalio.Worker.Tuning.ICustomSlotSupplier custom)
             {
-                var wrapped = new CustomSlotSupplier(custom);
+                var wrapped = new CustomSlotSupplier(custom, loggerFactory);
                 unsafe
                 {
                     return new()

src/Temporalio/Bridge/Worker.cs

@@ -2,6 +2,7 @@
 using System.Runtime.InteropServices;
 using System.Threading.Tasks;
 using Google.Protobuf;
+using Microsoft.Extensions.Logging;
 
 namespace Temporalio.Bridge
 {
@@ -16,11 +17,16 @@ internal class Worker : SafeHandle
         /// <param name="client">Client for the worker.</param>
         /// <param name="namespace_">Namespace for the worker.</param>
         /// <param name="options">Options for the worker.</param>
+        /// <param name="loggerFactory">Logger factory, used instead of the one in options by
+        ///   anything in the bridge that needs it, since it's guaranteed to be set.</param>
         /// <exception cref="Exception">
         /// If any of the options are invalid including improperly defined workflows/activities.
         /// </exception>
         public Worker(
-            Client client, string namespace_, Temporalio.Worker.TemporalWorkerOptions options)
+            Client client,
+            string namespace_,
+            Temporalio.Worker.TemporalWorkerOptions options,
+            ILoggerFactory loggerFactory)
             : base(IntPtr.Zero, true)
         {
             Runtime = client.Runtime;
@@ -29,7 +35,8 @@ public Worker(
                 unsafe
                 {
                     var workerOrFail = Interop.Methods.worker_new(
-                        client.Ptr, scope.Pointer(options.ToInteropOptions(scope, namespace_)));
+                        client.Ptr,
+                        scope.Pointer(options.ToInteropOptions(scope, namespace_, loggerFactory)));
                     if (workerOrFail.fail != null)
                     {
                         string failStr;
@@ -375,4 +382,4 @@ protected override unsafe bool ReleaseHandle()
             return true;
         }
     }
-}
+}

src/Temporalio/Worker/TemporalWorker.cs

@@ -34,6 +34,7 @@ public class TemporalWorker : IDisposable
         public TemporalWorker(IWorkerClient client, TemporalWorkerOptions options)
         {
             this.client = client;
+            var loggerFactory = options.LoggerFactory ?? client.Options.LoggerFactory;
             // Clone the options to discourage mutation (but we aren't completely disabling mutation
             // on the Options field herein).
             Options = (TemporalWorkerOptions)options.Clone();
@@ -42,7 +43,8 @@ public TemporalWorker(IWorkerClient client, TemporalWorkerOptions options)
             BridgeWorker = new(
                 (Bridge.Client)bridgeClient,
                 client.Options.Namespace,
-                options);
+                options,
+                loggerFactory);
             if (options.Activities.Count == 0 && options.Workflows.Count == 0)
             {
                 throw new ArgumentException("Must have at least one workflow and/or activity");
@@ -81,7 +83,7 @@ public TemporalWorker(IWorkerClient client, TemporalWorkerOptions options)
                     Workflows: options.Workflows,
                     DataConverter: client.Options.DataConverter,
                     Interceptors: Interceptors,
-                    LoggerFactory: options.LoggerFactory ?? client.Options.LoggerFactory,
+                    LoggerFactory: loggerFactory,
                     WorkflowInstanceFactory: options.WorkflowInstanceFactory,
                     DebugMode: options.DebugMode,
                     DisableWorkflowTracingEventListener: options.DisableWorkflowTracingEventListener,
@@ -400,4 +402,4 @@ private async Task ExecuteInternalAsync(
             }
         }
     }
-}
+}

src/Temporalio/Worker/Tuning/ICustomSlotSupplier.cs

@@ -18,27 +18,36 @@ public interface ICustomSlotSupplier : ISlotSupplier
         /// invocations of this method may be cancelled. Any other exceptions thrown will be logged
         /// and ignored.
         /// </summary>
+        /// <remarks>
+        /// This method will be called concurrently from multiple threads, so it must be thread-safe.
+        /// </remarks>
         /// <param name="ctx">The context for slot reservation.</param>
         /// <returns>A permit to use the slot which may be populated with your own data.</returns>
         /// <exception cref="OperationCanceledException">Cancellation requested.</exception>
-        public Task<SlotPermit> ReserveSlotAsync(SlotReserveContext ctx);
+        public Task<ISlotPermit> ReserveSlotAsync(SlotReserveContext ctx);
 
         /// <summary>
         /// This function is called when trying to reserve slots for "eager" workflow and activity tasks.
         /// Eager tasks are those which are returned as a result of completing a workflow task, rather than
         /// from polling. Your implementation must not block, and if a slot is available, return a permit
         /// to use that slot.
         /// </summary>
+        /// <remarks>
+        /// This method will be called concurrently from multiple threads, so it must be thread-safe.
+        /// </remarks>
         /// <param name="ctx">The context for slot reservation.</param>
         /// <returns>Maybe a permit to use the slot which may be populated with your own data.</returns>
-        public SlotPermit? TryReserveSlot(SlotReserveContext ctx);
+        public ISlotPermit? TryReserveSlot(SlotReserveContext ctx);
 
         /// <summary>
         /// This function is called once a slot is actually being used to process some task, which may be
         /// some time after the slot was reserved originally. For example, if there is no work for a
         /// worker, a number of slots equal to the number of active pollers may already be reserved, but
         /// none of them are being used yet. This call should be non-blocking.
         /// </summary>
+        /// <remarks>
+        /// This method will be called concurrently from multiple threads, so it must be thread-safe.
+        /// </remarks>
         /// <param name="ctx">The context for marking a slot as used.</param>
         public void MarkSlotUsed(SlotMarkUsedContext ctx);
 
@@ -47,6 +56,9 @@ public interface ICustomSlotSupplier : ISlotSupplier
         /// finished, whether successfully or not, or because the slot was no longer needed (ex: the number
         /// of active pollers decreased). This call should be non-blocking.
         /// </summary>
+        /// <remarks>
+        /// This method will be called concurrently from multiple threads, so it must be thread-safe.
+        /// </remarks>
         /// <param name="ctx">The context for releasing a slot.</param>
         public void ReleaseSlot(SlotReleaseContext ctx);
     }

src/Temporalio/Worker/Tuning/ISlotPermit.cs

@@ -0,0 +1,16 @@
+namespace Temporalio.Worker.Tuning
+{
+    // This interface is intended as a marker type
+#pragma warning disable CA1040
+    /// <summary>
+    /// A permit to use a slot for a workflow/activity/local activity task.
+    /// You can implement this interface to add your own data to the permit.
+    /// </summary>
+    /// <remarks>
+    /// WARNING: Custom slot suppliers are currently experimental.
+    /// </remarks>
+    public interface ISlotPermit
+    {
+    }
+#pragma warning restore CA1040
+}

src/Temporalio/Worker/Tuning/SlotMarkUsedContext.cs

@@ -20,7 +20,7 @@ internal SlotMarkUsedContext(Temporalio.Bridge.Interop.SlotMarkUsedCtx ctx, GCHa
             this.SlotInfo = SlotInfo.FromBridge(ctx.slot_info);
             unsafe
             {
-                this.Permit = (SlotPermit)userData.Target!;
+                this.Permit = (ISlotPermit)userData.Target!;
             }
         }
 
@@ -32,6 +32,6 @@ internal SlotMarkUsedContext(Temporalio.Bridge.Interop.SlotMarkUsedCtx ctx, GCHa
         /// <summary>
         /// Gets the permit that was issued when the slot was reserved.
         /// </summary>
-        public SlotPermit Permit { get; }
+        public ISlotPermit Permit { get; }
     }
 }