diff --git a/Test/DurableTask.Core.Tests/MessageSorterTests.cs b/Test/DurableTask.Core.Tests/MessageSorterTests.cs new file mode 100644 index 000000000..13bb90170 --- /dev/null +++ b/Test/DurableTask.Core.Tests/MessageSorterTests.cs @@ -0,0 +1,340 @@ +// --------------------------------------------------------------- +// Copyright (c) Microsoft Corporation. All rights reserved. +// --------------------------------------------------------------- + +namespace DurableTask.Core.Tests +{ + using System; + using System.Collections.Generic; + using System.IO; + using System.Linq; + using System.Threading.Tasks; + using DurableTask.Core.Entities; + using DurableTask.Core.Entities.EventFormat; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + [TestClass] + public class MessageSorterTests + { + private static readonly TimeSpan ReorderWindow = TimeSpan.FromMinutes(30); + + [TestMethod] + public void SimpleInOrder() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + + var message1 = Send(senderId, receiverId, "1", senderSorter, DateTime.UtcNow); + var message2 = Send(senderId, receiverId, "2", senderSorter, DateTime.UtcNow); + var message3 = Send(senderId, receiverId, "3", senderSorter, DateTime.UtcNow); + + List batch; + MessageSorter receiverSorter = new MessageSorter(); + + // delivering the sequence in order produces 1 message each time + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("1"); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("2"); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("3"); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + [TestMethod] + public void WackySystemClock() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + + // simulate system clock that goes backwards - mechanism should still guarantee monotonicitty + var message1 = Send(senderId, receiverId, "1", senderSorter, DateTime.UtcNow); + var message2 = Send(senderId, receiverId, "2", senderSorter, DateTime.UtcNow - TimeSpan.FromSeconds(1)); + var message3 = Send(senderId, receiverId, "3", senderSorter, DateTime.UtcNow - TimeSpan.FromSeconds(2)); + + List batch; + MessageSorter receiverSorter = new MessageSorter(); + + // delivering the sequence in order produces 1 message each time + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("1"); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("2"); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("3"); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + [TestMethod] + public void DelayedElement() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + + var message1 = Send(senderId, receiverId, "1", senderSorter, DateTime.UtcNow); + var message2 = Send(senderId, receiverId, "2", senderSorter, DateTime.UtcNow); + var message3 = Send(senderId, receiverId, "3", senderSorter, DateTime.UtcNow); + + List batch; + MessageSorter receiverSorter; + + // delivering first message last delays all messages until getting the first one + receiverSorter = new MessageSorter(); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Collection( + batch, + first => first.Input.Equals("1"), + second => second.Input.Equals("2"), + third => third.Input.Equals("3")); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + [TestMethod] + public void NoFilteringOrSortingPastReorderWindow() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + var now = DateTime.UtcNow; + + // last message is sent after an interval exceeding the reorder window + var message1 = Send(senderId, receiverId, "1", senderSorter, now); + var message2 = Send(senderId, receiverId, "2", senderSorter, now + TimeSpan.FromTicks(1)); + var message3 = Send(senderId, receiverId, "3", senderSorter, now + TimeSpan.FromTicks(2) + ReorderWindow); + + List batch; + MessageSorter receiverSorter = new MessageSorter(); + + // delivering the sequence in order produces 1 message each time + receiverSorter = new MessageSorter(); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("1"); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("2"); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("3"); + + // duplicates are not filtered or sorted, but simply passed through, because we are past the reorder window + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("2"); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("1"); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + [TestMethod] + public void DuplicatedElements() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + + var message1 = Send(senderId, receiverId, "1", senderSorter, DateTime.UtcNow); + var message2 = Send(senderId, receiverId, "2", senderSorter, DateTime.UtcNow); + var message3 = Send(senderId, receiverId, "3", senderSorter, DateTime.UtcNow); + + List batch; + MessageSorter receiverSorter; + + // delivering first message last delays all messages until getting the first one + receiverSorter = new MessageSorter(); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Collection( + batch, + first => first.Input.Equals("1"), + second => second.Input.Equals("2")); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Single(batch).Input.Equals("3"); + batch = receiverSorter.ReceiveInOrder(message3, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message2, ReorderWindow).ToList(); + Assert.That.Empty(batch); + batch = receiverSorter.ReceiveInOrder(message1, ReorderWindow).ToList(); + Assert.That.Empty(batch); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + [TestMethod] + public void RandomShuffleAndDuplication() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + var receiverSorter = new MessageSorter(); + + var messageCount = 100; + var duplicateCount = 100; + + // create a ordered sequence of messages + var messages = new List(); + for (int i = 0; i < messageCount; i++) + { + messages.Add(Send(senderId, receiverId, i.ToString(), senderSorter, DateTime.UtcNow)); + } + + // add some random duplicates + var random = new Random(0); + for (int i = 0; i < duplicateCount; i++) + { + messages.Add(messages[random.Next(messageCount)]); + } + + // shuffle the messages + Shuffle(messages, random); + + // deliver all the messages + var deliveredMessages = new List(); + + foreach (var msg in messages) + { + foreach (var deliveredMessage in receiverSorter.ReceiveInOrder(msg, ReorderWindow)) + { + deliveredMessages.Add(deliveredMessage); + } + } + + // check that the delivered messages are the original sequence + Assert.AreEqual(messageCount, deliveredMessages.Count()); + for (int i = 0; i < messageCount; i++) + { + Assert.AreEqual(i.ToString(), deliveredMessages[i].Input); + } + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + /// + /// Tests that if messages get reordered beyond the supported reorder window, + /// we still deliver them all but they may now be out of order. + /// + [TestMethod] + public void RandomCollection() + { + var senderId = "A"; + var receiverId = "B"; + + var senderSorter = new MessageSorter(); + var receiverSorter = new MessageSorter(); + + var messageCount = 100; + + var random = new Random(0); + var now = DateTime.UtcNow; + + // create a ordered sequence of messages + var messages = new List(); + for (int i = 0; i < messageCount; i++) + { + messages.Add(Send(senderId, receiverId, i.ToString(), senderSorter, now + TimeSpan.FromSeconds(random.Next(5)), TimeSpan.FromSeconds(10))); + } + + // shuffle the messages + Shuffle(messages, random); + + // add a final message + messages.Add(Send(senderId, receiverId, (messageCount + 1).ToString(), senderSorter, now + TimeSpan.FromSeconds(1000), TimeSpan.FromSeconds(10))); + + // deliver all the messages + var deliveredMessages = new List(); + + for (int i = 0; i < messageCount; i++) + { + foreach (var deliveredMessage in receiverSorter.ReceiveInOrder(messages[i], TimeSpan.FromSeconds(10))) + { + deliveredMessages.Add(deliveredMessage); + } + + Assert.AreEqual(i + 1, deliveredMessages.Count + receiverSorter.NumberBufferedRequests); + } + + // receive the final messages + foreach (var deliveredMessage in receiverSorter.ReceiveInOrder(messages[messageCount], TimeSpan.FromSeconds(10))) + { + deliveredMessages.Add(deliveredMessage); + } + + // check that all messages were delivered + Assert.AreEqual(messageCount + 1, deliveredMessages.Count()); + + Assert.AreEqual(0, receiverSorter.NumberBufferedRequests); + } + + private static RequestMessage Send(string senderId, string receiverId, string input, MessageSorter sorter, DateTime now, TimeSpan? reorderWindow = null) + { + var msg = new RequestMessage() + { + Id = Guid.NewGuid(), + ParentInstanceId = senderId, + Input = input, + }; + sorter.LabelOutgoingMessage(msg, receiverId, now, reorderWindow.HasValue ? reorderWindow.Value : ReorderWindow); + return msg; + } + + private static void Shuffle(IList list, Random random) + { + int n = list.Count; + while (n > 1) + { + n--; + int k = random.Next(n + 1); + T value = list[k]; + list[k] = list[n]; + list[n] = value; + } + } + } + + internal static class AssertExtensions + { + public static void Empty(this Assert assert, IEnumerable collection) + { + Assert.AreEqual(0, collection.Count()); + } + + public static T Single(this Assert assert, IEnumerable collection) + { + var e = collection.GetEnumerator(); + Assert.IsTrue(e.MoveNext()); + T element = e.Current; + Assert.IsFalse(e.MoveNext()); + return element; + } + + public static void Collection(this Assert assert, IEnumerable collection, params Action[] elementInspectors) + { + var list = collection.ToList(); + Assert.AreEqual(elementInspectors.Length, list.Count); + for(int i = 0; i < elementInspectors.Length; i++) + { + elementInspectors[i](list[i]); + } + } + } +} \ No newline at end of file diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs index 8967e30cd..e2f7be331 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs @@ -27,6 +27,7 @@ namespace DurableTask.AzureStorage using DurableTask.AzureStorage.Storage; using DurableTask.AzureStorage.Tracking; using DurableTask.Core; + using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; using DurableTask.Core.History; using DurableTask.Core.Query; @@ -41,7 +42,8 @@ public sealed class AzureStorageOrchestrationService : IOrchestrationServiceClient, IDisposable, IOrchestrationServiceQueryClient, - IOrchestrationServicePurgeClient + IOrchestrationServicePurgeClient, + IEntityOrchestrationService { static readonly HistoryEvent[] EmptyHistoryEventList = new HistoryEvent[0]; @@ -277,6 +279,55 @@ public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew /// public int TaskOrchestrationDispatcherCount { get; } = 1; + #region IEntityOrchestrationService + + EntityBackendProperties IEntityOrchestrationService.GetEntityBackendProperties() + => new EntityBackendProperties() + { + EntityMessageReorderWindow = TimeSpan.FromMinutes(this.settings.EntityMessageReorderWindowInMinutes), + MaxEntityOperationBatchSize = this.settings.MaxEntityOperationBatchSize, + MaxConcurrentTaskEntityWorkItems = this.settings.MaxConcurrentTaskEntityWorkItems, + SupportsImplicitEntityDeletion = false, // not supported by this backend + MaximumSignalDelayTime = TimeSpan.FromDays(6), + }; + + bool IEntityOrchestrationService.ProcessEntitiesSeparately() + { + if (this.settings.UseSeparateQueueForEntityWorkItems) + { + this.orchestrationSessionManager.ProcessEntitiesSeparately = true; + return true; + } + else + { + return false; + } + } + + Task IEntityOrchestrationService.LockNextOrchestrationWorkItemAsync( + TimeSpan receiveTimeout, + CancellationToken cancellationToken) + { + if (!orchestrationSessionManager.ProcessEntitiesSeparately) + { + throw new InvalidOperationException("backend was not configured for separate entity processing"); + } + return this.LockNextTaskOrchestrationWorkItemAsync(false, cancellationToken); + } + + Task IEntityOrchestrationService.LockNextEntityWorkItemAsync( + TimeSpan receiveTimeout, + CancellationToken cancellationToken) + { + if (!orchestrationSessionManager.ProcessEntitiesSeparately) + { + throw new InvalidOperationException("backend was not configured for separate entity processing"); + } + return this.LockNextTaskOrchestrationWorkItemAsync(entitiesOnly: true, cancellationToken); + } + + #endregion + #region Management Operations (Create/Delete/Start/Stop) /// /// Deletes and creates the neccesary Azure Storage resources for the orchestration service. @@ -625,9 +676,14 @@ static TaskHubInfo GetTaskHubInfo(string taskHub, int partitionCount) #region Orchestration Work Item Methods /// - public async Task LockNextTaskOrchestrationWorkItemAsync( + public Task LockNextTaskOrchestrationWorkItemAsync( TimeSpan receiveTimeout, CancellationToken cancellationToken) + { + return LockNextTaskOrchestrationWorkItemAsync(entitiesOnly: false, cancellationToken); + } + + async Task LockNextTaskOrchestrationWorkItemAsync(bool entitiesOnly, CancellationToken cancellationToken) { Guid traceActivityId = StartNewLogicalTraceScope(useExisting: true); @@ -641,7 +697,7 @@ public async Task LockNextTaskOrchestrationWorkItemAs try { // This call will block until the next session is ready - session = await this.orchestrationSessionManager.GetNextSessionAsync(linkedCts.Token); + session = await this.orchestrationSessionManager.GetNextSessionAsync(entitiesOnly, linkedCts.Token); if (session == null) { return null; diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs index 8d432b151..819480bde 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs @@ -111,6 +111,12 @@ public class AzureStorageOrchestrationServiceSettings /// public int MaxConcurrentTaskOrchestrationWorkItems { get; set; } = 100; + /// + /// Gets or sets the maximum number of entity operation batches that can be processed concurrently on a single node. + /// The default value is 100. + /// + public int MaxConcurrentTaskEntityWorkItems { get; set; } = 100; + /// /// Gets or sets the maximum number of concurrent storage operations that can be executed in the context /// of a single orchestration instance. @@ -286,5 +292,31 @@ internal LogHelper Logger return this.logHelper; } } + + /// + /// Gets or sets the limit on the number of entity operations that should be processed as a single batch. + /// A null value indicates that no particular limit should be enforced. + /// + /// + /// Limiting the batch size can help to avoid timeouts in execution environments that impose time limitations on work items. + /// If set to 1, batching is disabled, and each operation executes as a separate work item. + /// + /// + /// A positive integer, or null. + /// + public int? MaxEntityOperationBatchSize { get; set; } = null; + + /// + /// Gets or sets the time window within which entity messages get deduplicated and reordered. + /// If set to zero, there is no sorting or deduplication, and all messages are just passed through. + /// + public int EntityMessageReorderWindowInMinutes { get; set; } = 30; + + /// + /// Whether to use separate work item queues for entities and orchestrators. + /// This defaults to false, to avoid issues when using this provider from code that does not support separate dispatch. + /// Consumers that support separate dispatch should explicitly set this to true. + /// + public bool UseSeparateQueueForEntityWorkItems { get; set; } = false; } } diff --git a/src/DurableTask.AzureStorage/Fnv1aHashHelper.cs b/src/DurableTask.AzureStorage/Fnv1aHashHelper.cs index fbff51089..2bea574ca 100644 --- a/src/DurableTask.AzureStorage/Fnv1aHashHelper.cs +++ b/src/DurableTask.AzureStorage/Fnv1aHashHelper.cs @@ -22,32 +22,61 @@ namespace DurableTask.AzureStorage /// See https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function. /// Tested with production data and random guids. The result was good distribution. /// - static class Fnv1aHashHelper + internal static class Fnv1aHashHelper { const uint FnvPrime = unchecked(16777619); const uint FnvOffsetBasis = unchecked(2166136261); + /// + /// Compute a hash for a given string. + /// + /// The string to hash. + /// a four-byte hash public static uint ComputeHash(string value) { return ComputeHash(value, encoding: null); } + /// + /// Compute a hash for a given string and encoding. + /// + /// The string to hash. + /// The encoding. + /// a four-byte hash public static uint ComputeHash(string value, Encoding encoding) { return ComputeHash(value, encoding, hash: FnvOffsetBasis); } + /// + /// Compute a hash for a given string, encoding, and hash modifier. + /// + /// The string to hash. + /// The encoding. + /// The modifier hash. + /// a four-byte hash public static uint ComputeHash(string value, Encoding encoding, uint hash) { byte[] bytes = (encoding ?? Encoding.UTF8).GetBytes(value); return ComputeHash(bytes, hash); } + /// + /// Compute a hash for the given byte array. + /// + /// The byte array to hash. + /// a four-byte hash public static uint ComputeHash(byte[] array) { return ComputeHash(array, hash: FnvOffsetBasis); } + /// + /// Compute a hash for the given byte array. + /// + /// The byte array to hash. + /// The modifier hash. + /// a four-byte hash public static uint ComputeHash(byte[] array, uint hash) { for (var i = 0; i < array.Length; i++) diff --git a/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs b/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs index 6d990c63d..4c0486573 100644 --- a/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs +++ b/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs @@ -32,7 +32,8 @@ class OrchestrationSessionManager : IDisposable readonly Dictionary activeOrchestrationSessions = new Dictionary(StringComparer.OrdinalIgnoreCase); readonly ConcurrentDictionary ownedControlQueues = new ConcurrentDictionary(); readonly LinkedList pendingOrchestrationMessageBatches = new LinkedList(); - readonly AsyncQueue> readyForProcessingQueue = new AsyncQueue>(); + readonly AsyncQueue> orchestrationsReadyForProcessingQueue = new AsyncQueue>(); + readonly AsyncQueue> entitiesReadyForProcessingQueue = new AsyncQueue>(); readonly object messageAndSessionLock = new object(); readonly string storageAccountName; @@ -57,6 +58,14 @@ public OrchestrationSessionManager( internal IEnumerable Queues => this.ownedControlQueues.Values; + /// + /// Recent versions of DurableTask.Core can be configured to use a separate pipeline for processing entity work items, + /// while older versions use a single pipeline for both orchestration and entity work items. To support both scenarios, + /// this property can be modified prior to starting the orchestration service. If set to true, the work items that are ready for + /// processing are stored in and , respectively. + /// + internal bool ProcessEntitiesSeparately { get; set; } + public void AddQueue(string partitionId, ControlQueue controlQueue, CancellationToken cancellationToken) { if (this.ownedControlQueues.TryAdd(partitionId, controlQueue)) @@ -520,7 +529,14 @@ async Task ScheduleOrchestrationStatePrefetch( batch.TrackingStoreContext = history.TrackingStoreContext; } - this.readyForProcessingQueue.Enqueue(node); + if (this.ProcessEntitiesSeparately && DurableTask.Core.Common.Entities.IsEntityInstance(batch.OrchestrationInstanceId)) + { + this.entitiesReadyForProcessingQueue.Enqueue(node); + } + else + { + this.orchestrationsReadyForProcessingQueue.Enqueue(node); + } } catch (OperationCanceledException) { @@ -544,14 +560,16 @@ async Task ScheduleOrchestrationStatePrefetch( } } - public async Task GetNextSessionAsync(CancellationToken cancellationToken) + public async Task GetNextSessionAsync(bool entitiesOnly, CancellationToken cancellationToken) { + var readyForProcessingQueue = entitiesOnly? this.entitiesReadyForProcessingQueue : this.orchestrationsReadyForProcessingQueue; + while (!cancellationToken.IsCancellationRequested) { // This call will block until: // 1) a batch of messages has been received for a particular instance and // 2) the history for that instance has been fetched - LinkedListNode node = await this.readyForProcessingQueue.DequeueAsync(cancellationToken); + LinkedListNode node = await readyForProcessingQueue.DequeueAsync(cancellationToken); lock (this.messageAndSessionLock) { @@ -597,7 +615,7 @@ async Task ScheduleOrchestrationStatePrefetch( // A message arrived for a different generation of an existing orchestration instance. // Put it back into the ready queue so that it can be processed once the current generation // is done executing. - if (this.readyForProcessingQueue.Count == 0) + if (readyForProcessingQueue.Count == 0) { // To avoid a tight dequeue loop, delay for a bit before putting this node back into the queue. // This is only necessary when the queue is empty. The main dequeue thread must not be blocked @@ -607,14 +625,14 @@ async Task ScheduleOrchestrationStatePrefetch( lock (this.messageAndSessionLock) { this.pendingOrchestrationMessageBatches.AddLast(node); - this.readyForProcessingQueue.Enqueue(node); + readyForProcessingQueue.Enqueue(node); } }); } else { this.pendingOrchestrationMessageBatches.AddLast(node); - this.readyForProcessingQueue.Enqueue(node); + readyForProcessingQueue.Enqueue(node); } } } @@ -676,7 +694,8 @@ public void GetStats( public virtual void Dispose() { this.fetchRuntimeStateQueue.Dispose(); - this.readyForProcessingQueue.Dispose(); + this.orchestrationsReadyForProcessingQueue.Dispose(); + this.entitiesReadyForProcessingQueue.Dispose(); } class PendingMessageBatch diff --git a/src/DurableTask.Core/Common/Entities.cs b/src/DurableTask.Core/Common/Entities.cs index 2484153de..dc3ba2434 100644 --- a/src/DurableTask.Core/Common/Entities.cs +++ b/src/DurableTask.Core/Common/Entities.cs @@ -10,14 +10,13 @@ // See the License for the specific language governing permissions and // limitations under the License. // ---------------------------------------------------------------------------------- - -using DurableTask.Core.History; -using System; -using System.Collections.Generic; -using System.Text; - +#nullable enable namespace DurableTask.Core.Common { + using DurableTask.Core.History; + using System; + using System.Collections.Generic; + /// /// Helpers for dealing with special naming conventions around auto-started orchestrations (entities) /// diff --git a/src/DurableTask.Core/Common/Fnv1aHashHelper.cs b/src/DurableTask.Core/Common/Fnv1aHashHelper.cs new file mode 100644 index 000000000..6184d6fdd --- /dev/null +++ b/src/DurableTask.Core/Common/Fnv1aHashHelper.cs @@ -0,0 +1,93 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +namespace DurableTask.Core.Common +{ + using System.Text; + + /// + /// Fast, non-cryptographic hash function helper. + /// + /// + /// See https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function. + /// Tested with production data and random guids. The result was good distribution. + /// + internal static class Fnv1aHashHelper + { + const uint FnvPrime = unchecked(16777619); + const uint FnvOffsetBasis = unchecked(2166136261); + + /// + /// Compute a hash for a given string. + /// + /// The string to hash. + /// a four-byte hash + public static uint ComputeHash(string value) + { + return ComputeHash(value, encoding: null); + } + + /// + /// Compute a hash for a given string and encoding. + /// + /// The string to hash. + /// The encoding. + /// a four-byte hash + public static uint ComputeHash(string value, Encoding encoding) + { + return ComputeHash(value, encoding, hash: FnvOffsetBasis); + } + + /// + /// Compute a hash for a given string, encoding, and hash modifier. + /// + /// The string to hash. + /// The encoding. + /// The modifier hash. + /// a four-byte hash + public static uint ComputeHash(string value, Encoding encoding, uint hash) + { + byte[] bytes = (encoding ?? Encoding.UTF8).GetBytes(value); + return ComputeHash(bytes, hash); + } + + /// + /// Compute a hash for the given byte array. + /// + /// The byte array to hash. + /// a four-byte hash + public static uint ComputeHash(byte[] array) + { + return ComputeHash(array, hash: FnvOffsetBasis); + } + + /// + /// Compute a hash for the given byte array. + /// + /// The byte array to hash. + /// The modifier hash. + /// a four-byte hash + public static uint ComputeHash(byte[] array, uint hash) + { + for (var i = 0; i < array.Length; i++) + { + unchecked + { + hash ^= array[i]; + hash *= FnvPrime; + } + } + + return hash; + } + } +} diff --git a/src/DurableTask.Core/Common/Utils.cs b/src/DurableTask.Core/Common/Utils.cs index 32b5939dd..30495a441 100644 --- a/src/DurableTask.Core/Common/Utils.cs +++ b/src/DurableTask.Core/Common/Utils.cs @@ -152,7 +152,7 @@ public static string SerializeToJson(JsonSerializer serializer, object payload) /// The default value comes from the WEBSITE_SITE_NAME environment variable, which is defined /// in Azure App Service. Other environments can use DTFX_APP_NAME to set this value. /// - public static string AppName { get; set; } = + public static string AppName { get; set; } = Environment.GetEnvironmentVariable("WEBSITE_SITE_NAME") ?? Environment.GetEnvironmentVariable("DTFX_APP_NAME") ?? string.Empty; @@ -624,6 +624,40 @@ public static bool TryGetTaskScheduledId(HistoryEvent historyEvent, out int task } } + /// + /// Creates a determinstic Guid from a string using a hash function. This is a simple hash + /// meant to produce pseudo-random Guids, it is not meant to be cryptographically secure, + /// and does not follow any formatting conventions for UUIDs (such as RFC 4122). + /// + /// The string to hash. + /// A Guid constructed from the hash. + /// + internal static Guid CreateGuidFromHash(string stringToHash) + { + if (string.IsNullOrEmpty(stringToHash)) + { + throw new ArgumentException("string to hash must not be null or empty", nameof(stringToHash)); + } + + var bytes = Encoding.UTF8.GetBytes(stringToHash); + uint hash1 = Fnv1aHashHelper.ComputeHash(bytes, 0xdf0dd395); + uint hash2 = Fnv1aHashHelper.ComputeHash(bytes, 0xa19df4df); + uint hash3 = Fnv1aHashHelper.ComputeHash(bytes, 0xc88599c5); + uint hash4 = Fnv1aHashHelper.ComputeHash(bytes, 0xe24e3e64); + return new Guid( + hash1, + (ushort)(hash2 & 0xFFFF), + (ushort)((hash2 >> 16) & 0xFFFF), + (byte)(hash3 & 0xFF), + (byte)((hash3 >> 8) & 0xFF), + (byte)((hash3 >> 16) & 0xFF), + (byte)((hash3 >> 24) & 0xFF), + (byte)(hash4 & 0xFF), + (byte)((hash4 >> 8) & 0xFF), + (byte)((hash4 >> 16) & 0xFF), + (byte)((hash4 >> 24) & 0xFF)); + } + /// /// Gets the generic return type for a specific . /// diff --git a/src/DurableTask.Core/Entities/ClientEntityHelpers.cs b/src/DurableTask.Core/Entities/ClientEntityHelpers.cs new file mode 100644 index 000000000..18832f44d --- /dev/null +++ b/src/DurableTask.Core/Entities/ClientEntityHelpers.cs @@ -0,0 +1,90 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using DurableTask.Core.Entities; + using DurableTask.Core.Entities.EventFormat; + using Newtonsoft.Json.Linq; + using Newtonsoft.Json; + using System; + + /// + /// Utility functions for clients that interact with entities, either by sending events or by accessing the entity state directly in storage + /// + public static class ClientEntityHelpers + { + /// + /// Create an event to represent an entity signal. + /// + /// The target instance. + /// A unique identifier for the request. + /// The name of the operation. + /// The serialized input for the operation. + /// The time to schedule this signal, or null if not a scheduled signal + /// The event to send. + public static EventToSend EmitOperationSignal(OrchestrationInstance targetInstance, Guid requestId, string operationName, string input, (DateTime Original, DateTime Capped)? scheduledTimeUtc) + { + var request = new RequestMessage() + { + ParentInstanceId = null, // means this was sent by a client + ParentExecutionId = null, + Id = requestId, + IsSignal = true, + Operation = operationName, + ScheduledTime = scheduledTimeUtc?.Original, + Input = input, + }; + + var jrequest = JToken.FromObject(request, Serializer.InternalSerializer); + + var eventName = scheduledTimeUtc.HasValue + ? EntityMessageEventNames.ScheduledRequestMessageEventName(scheduledTimeUtc.Value.Capped) + : EntityMessageEventNames.RequestMessageEventName; + + return new EventToSend(eventName, jrequest, targetInstance); + } + + /// + /// Create an event to represent an entity unlock, which is called by clients to fix orphaned locks. + /// + /// The target instance. + /// The instance id of the entity to be unlocked. + /// The event to send. + public static EventToSend EmitUnlockForOrphanedLock(OrchestrationInstance targetInstance, string lockOwnerInstanceId) + { + var message = new ReleaseMessage() + { + ParentInstanceId = lockOwnerInstanceId, + Id = "fix-orphaned-lock", // we don't know the original id but it does not matter + }; + + var jmessage = JToken.FromObject(message, Serializer.InternalSerializer); + + return new EventToSend(EntityMessageEventNames.ReleaseMessageEventName, jmessage, targetInstance); + } + + /// + /// Extracts the user-defined entity state (as a serialized string) from the scheduler state (also a serialized string). + /// + /// The state of the scheduler, as a serialized string. + /// The entity state + /// True if the entity exists, or false otherwise + public static bool TryGetEntityStateFromSerializedSchedulerState(string serializedSchedulerState, out string? entityState) + { + var schedulerState = JsonConvert.DeserializeObject(serializedSchedulerState, Serializer.InternalSerializerSettings); + entityState = schedulerState!.EntityState; + return schedulerState.EntityExists; + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/EntityBackendProperties.cs b/src/DurableTask.Core/Entities/EntityBackendProperties.cs new file mode 100644 index 000000000..1fe9cd5f3 --- /dev/null +++ b/src/DurableTask.Core/Entities/EntityBackendProperties.cs @@ -0,0 +1,67 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using System; + + /// + /// Entity processing characteristics that are controlled by the backend provider, i.e. the orchestration service. + /// + public class EntityBackendProperties + { + /// + /// The time window within which entity messages should be deduplicated and reordered. + /// This is zero for providers that already guarantee exactly-once and ordered delivery. + /// + public TimeSpan EntityMessageReorderWindow { get; set; } + + /// + /// A limit on the number of entity operations that should be processed as a single batch, or null if there is no limit. + /// + public int? MaxEntityOperationBatchSize { get; set; } + + /// + /// The maximum number of entity operation batches that can be processed concurrently on a single node. + /// + public int MaxConcurrentTaskEntityWorkItems { get; set; } + + /// + /// Whether the backend supports implicit deletion, i.e. setting the entity scheduler state to null implicitly deletes the storage record. + /// + public bool SupportsImplicitEntityDeletion { get; set; } + + /// + /// Value of maximum durable timer delay. Used for delayed signals. + /// + public TimeSpan MaximumSignalDelayTime { get; set; } + + /// + /// Computes a cap on the scheduled time of an entity signal, based on the maximum signal delay time + /// + /// The current time. + /// The scheduled time. + /// + public DateTime GetCappedScheduledTime(DateTime nowUtc, DateTime scheduledUtcTime) + { + if ((scheduledUtcTime - nowUtc) <= this.MaximumSignalDelayTime) + { + return scheduledUtcTime; + } + else + { + return nowUtc + this.MaximumSignalDelayTime; + } + } + } +} diff --git a/src/DurableTask.Core/Entities/EntityExecutionOptions.cs b/src/DurableTask.Core/Entities/EntityExecutionOptions.cs new file mode 100644 index 000000000..f6f04e800 --- /dev/null +++ b/src/DurableTask.Core/Entities/EntityExecutionOptions.cs @@ -0,0 +1,55 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using DurableTask.Core.Serializing; + + /// + /// Options that are used for configuring how a TaskEntity executes entity operations. + /// + public class EntityExecutionOptions + { + /// + /// The data converter used for converting inputs and outputs for operations. + /// + public DataConverter MessageDataConverter { get; set; } = JsonDataConverter.Default; + + /// + /// The data converter used for the entity state. + /// + public DataConverter StateDataConverter { get; set; } = JsonDataConverter.Default; + + /// + /// The data converter used for exceptions. + /// + public DataConverter ErrorDataConverter { get; set; } = JsonDataConverter.Default; + + /// + /// If true, all effects of an entity operation (all state changes and all actions) are rolled back + /// if the entity operation completes with an exception. + /// Implementations may override this setting. + /// + public bool RollbackOnExceptions { get; set; } = true; + + /// + /// Information about backend entity support. + /// + internal EntityBackendProperties? EntityBackendProperties { get; set; } + + /// + /// The mode that is used for propagating errors, as specified in the . + /// + internal ErrorPropagationMode ErrorPropagationMode { get; set; } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/EntityId.cs b/src/DurableTask.Core/Entities/EntityId.cs new file mode 100644 index 000000000..68ed4e943 --- /dev/null +++ b/src/DurableTask.Core/Entities/EntityId.cs @@ -0,0 +1,106 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using System; + using System.Runtime.Serialization; + + /// + /// A unique identifier for an entity, consisting of entity name and entity key. + /// + [DataContract] + public readonly struct EntityId : IEquatable, IComparable + { + /// + /// Create an entity id for an entity. + /// + /// The name of this class of entities. + /// The entity key. + public EntityId(string name, string key) + { + if (string.IsNullOrEmpty(name)) + { + throw new ArgumentNullException(nameof(name), "Invalid entity id: entity name must not be a null or empty string."); + } + + this.Name = name; + this.Key = key ?? throw new ArgumentNullException(nameof(key), "Invalid entity id: entity key must not be null."); + } + + /// + /// The name for this class of entities. + /// + [DataMember(Name = "name", IsRequired = true)] + public readonly string Name { get; } + + /// + /// The entity key. Uniquely identifies an entity among all entities of the same name. + /// + [DataMember(Name = "key", IsRequired = true)] + public readonly string Key { get; } + + /// + public override string ToString() + { + return $"@{this.Name}@{this.Key}"; + } + + /// + /// Returns the entity ID for a given instance ID. + /// + /// The instance ID. + /// the corresponding entity ID. + public static EntityId FromString(string instanceId) + { + if (string.IsNullOrEmpty(instanceId)) + { + throw new ArgumentException(nameof(instanceId)); + } + var pos = instanceId.IndexOf('@', 1); + if (pos <= 0 || instanceId[0] != '@') + { + throw new ArgumentException($"Instance ID '{instanceId}' is not a valid entity ID.", nameof(instanceId)); + } + var entityName = instanceId.Substring(1, pos - 1); + var entityKey = instanceId.Substring(pos + 1); + return new EntityId(entityName, entityKey); + } + + + /// + public override bool Equals(object obj) + { + return (obj is EntityId other) && this.Equals(other); + } + + /// + public bool Equals(EntityId other) + { + return (this.Name, this.Key).Equals((other.Name, other.Key)); + } + + /// + public override int GetHashCode() + { + return (this.Name, this.Key).GetHashCode(); + } + + /// + public int CompareTo(object obj) + { + var other = (EntityId)obj; + return (this.Name, this.Key).CompareTo((other.Name, other.Key)); + } + } +} diff --git a/src/DurableTask.Core/Entities/EventFormat/EntityMessageEventNames.cs b/src/DurableTask.Core/Entities/EventFormat/EntityMessageEventNames.cs new file mode 100644 index 000000000..adecea1dc --- /dev/null +++ b/src/DurableTask.Core/Entities/EventFormat/EntityMessageEventNames.cs @@ -0,0 +1,37 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.EventFormat +{ + using System; + + /// + /// Determines event names to use for messages sent to and from entities. + /// + internal static class EntityMessageEventNames + { + public static string RequestMessageEventName => "op"; + + public static string ReleaseMessageEventName => "release"; + + public static string ContinueMessageEventName => "continue"; + + public static string ScheduledRequestMessageEventName(DateTime scheduledUtc) => $"op@{scheduledUtc:o}"; + + public static string ResponseMessageEventName(Guid requestId) => requestId.ToString(); + + public static bool IsRequestMessage(string eventName) => eventName.StartsWith("op"); + + public static bool IsReleaseMessage(string eventName) => eventName == "release"; + } +} diff --git a/src/DurableTask.Core/Entities/EventFormat/ReleaseMessage.cs b/src/DurableTask.Core/Entities/EventFormat/ReleaseMessage.cs new file mode 100644 index 000000000..433a56e33 --- /dev/null +++ b/src/DurableTask.Core/Entities/EventFormat/ReleaseMessage.cs @@ -0,0 +1,32 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.EventFormat +{ + using System.Runtime.Serialization; + + [DataContract] + internal class ReleaseMessage + { + [DataMember(Name = "parent")] + public string? ParentInstanceId { get; set; } + + [DataMember(Name = "id")] + public string? Id { get; set; } + + public override string ToString() + { + return $"[Release lock {Id} by {ParentInstanceId}]"; + } + } +} diff --git a/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs b/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs new file mode 100644 index 000000000..2fbaf6099 --- /dev/null +++ b/src/DurableTask.Core/Entities/EventFormat/RequestMessage.cs @@ -0,0 +1,114 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.EventFormat +{ + using System; + using System.Runtime.Serialization; + + /// + /// A message sent to an entity, such as operation, signal, lock, or continue messages. + /// + [DataContract] + internal class RequestMessage + { + /// + /// The name of the operation being called (if this is an operation message) or null + /// (if this is a lock request). + /// + [DataMember(Name = "op")] + public string? Operation { get; set; } + + /// + /// Whether or not this is a one-way message. + /// + [DataMember(Name = "signal", EmitDefaultValue = false)] + public bool IsSignal { get; set; } + + /// + /// The operation input. + /// + [DataMember(Name = "input", EmitDefaultValue = false)] + public string? Input { get; set; } + + /// + /// A unique identifier for this operation. + /// + [DataMember(Name = "id", IsRequired = true)] + public Guid Id { get; set; } + + /// + /// The parent instance that called this operation. + /// + [DataMember(Name = "parent", EmitDefaultValue = false)] + public string? ParentInstanceId { get; set; } + + /// + /// The parent instance that called this operation. + /// + [DataMember(Name = "parentExecution", EmitDefaultValue = false)] + public string? ParentExecutionId { get; set; } + + /// + /// Optionally, a scheduled time at which to start the operation. + /// + [DataMember(Name = "due", EmitDefaultValue = false)] + public DateTime? ScheduledTime { get; set; } + + /// + /// A timestamp for this request. + /// Used for duplicate filtering and in-order delivery. + /// + [DataMember] + public DateTime Timestamp { get; set; } + + /// + /// A timestamp for the predecessor request in the stream, or DateTime.MinValue if none. + /// Used for duplicate filtering and in-order delivery. + /// + [DataMember] + public DateTime Predecessor { get; set; } + + /// + /// For lock requests, the set of locks being acquired. Is sorted, + /// contains at least one element, and has no repetitions. + /// + [DataMember(Name = "lockset", EmitDefaultValue = false)] + public EntityId[]? LockSet { get; set; } + + /// + /// For lock requests involving multiple locks, the message number. + /// + [DataMember(Name = "pos", EmitDefaultValue = false)] + public int Position { get; set; } + + /// + /// whether this message is a lock request + /// + [DataMember] + public bool IsLockRequest => LockSet != null; + + /// + public override string ToString() + { + if (IsLockRequest) + { + return $"[Request lock {Id} by {ParentInstanceId} {ParentExecutionId}, position {Position}]"; + } + else + { + return $"[{(IsSignal ? "Signal" : "Call")} '{Operation}' operation {Id} by {ParentInstanceId} {ParentExecutionId}]"; + } + } + } +} diff --git a/src/DurableTask.Core/Entities/EventFormat/ResponseMessage.cs b/src/DurableTask.Core/Entities/EventFormat/ResponseMessage.cs new file mode 100644 index 000000000..eb3a17d47 --- /dev/null +++ b/src/DurableTask.Core/Entities/EventFormat/ResponseMessage.cs @@ -0,0 +1,45 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.EventFormat +{ + using System.Runtime.Serialization; + + [DataContract] + internal class ResponseMessage + { + [DataMember(Name = "result")] + public string? Result { get; set; } + + [DataMember(Name = "exceptionType", EmitDefaultValue = false)] + public string? ErrorMessage { get; set; } + + [DataMember(Name = "failureDetails", EmitDefaultValue = false)] + public FailureDetails? FailureDetails { get; set; } + + [IgnoreDataMember] + public bool IsErrorResult => this.ErrorMessage != null; + + public override string ToString() + { + if (this.IsErrorResult) + { + return $"[ErrorResponse {this.Result}]"; + } + else + { + return $"[Response {this.Result}]"; + } + } + } +} diff --git a/src/DurableTask.Core/Entities/EventToSend.cs b/src/DurableTask.Core/Entities/EventToSend.cs new file mode 100644 index 000000000..2ce2b7b4d --- /dev/null +++ b/src/DurableTask.Core/Entities/EventToSend.cs @@ -0,0 +1,49 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + /// + /// The data associated with sending an event to an orchestration. + /// + public readonly struct EventToSend + { + /// + /// The name of the event. + /// + public readonly string EventName { get; } + + /// + /// The content of the event. + /// + public readonly object EventContent { get; } + + /// + /// The target instance for the event. + /// + public readonly OrchestrationInstance TargetInstance { get; } + + /// + /// Construct an entity message event with the given members. + /// + /// The name of the event. + /// The content of the event. + /// The target of the event. + public EventToSend(string name, object content, OrchestrationInstance target) + { + EventName = name; + EventContent = content; + TargetInstance = target; + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/IEntityOrchestrationService.cs b/src/DurableTask.Core/Entities/IEntityOrchestrationService.cs new file mode 100644 index 000000000..5e8823d72 --- /dev/null +++ b/src/DurableTask.Core/Entities/IEntityOrchestrationService.cs @@ -0,0 +1,51 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using System; + using System.Threading; + using System.Threading.Tasks; + + /// + /// Extends with methods that support processing of entities. + /// + public interface IEntityOrchestrationService : IOrchestrationService + { + /// + /// The entity orchestration service. + /// + /// An object containing properties of the entity backend. + EntityBackendProperties GetEntityBackendProperties(); + + /// + /// Checks whether the backend is configured for separate work-item processing of orchestrations and entities. + /// If this returns true, must use or to + /// pull orchestrations or entities separately. Otherwise, must use . + /// This must be called prior to starting the orchestration service. + /// + bool ProcessEntitiesSeparately(); + + /// + /// Specialized variant of that + /// fetches only work items for true orchestrations, not entities. + /// + Task LockNextOrchestrationWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken); + + /// + /// Specialized variant of that + /// fetches only work items for entities, not plain orchestrations. + /// + Task LockNextEntityWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationAction.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationAction.cs new file mode 100644 index 000000000..467cecda4 --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationAction.cs @@ -0,0 +1,29 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using Newtonsoft.Json; + + /// + /// Defines a set of base properties for an operator action. + /// + [JsonConverter(typeof(OperationActionConverter))] + public abstract class OperationAction + { + /// + /// The type of the orchestrator action. + /// + public abstract OperationActionType OperationActionType { get; } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationActionConverter.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationActionConverter.cs new file mode 100644 index 000000000..96cee6b4f --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationActionConverter.cs @@ -0,0 +1,40 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +namespace DurableTask.Core.Entities.OperationFormat +{ + using System; + using Newtonsoft.Json.Linq; + using DurableTask.Core.Serializing; + + internal class OperationActionConverter : JsonCreationConverter + { + protected override OperationAction CreateObject(Type objectType, JObject jObject) + { + if (jObject.TryGetValue("OperationActionType", StringComparison.OrdinalIgnoreCase, out JToken actionType)) + { + var type = (OperationActionType)int.Parse((string)actionType); + switch (type) + { + case OperationActionType.SendSignal: + return new SendSignalOperationAction(); + case OperationActionType.StartNewOrchestration: + return new StartNewOrchestrationOperationAction(); + default: + throw new NotSupportedException("Unrecognized action type."); + } + } + + throw new NotSupportedException("Action Type not provided."); + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationActionType.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationActionType.cs new file mode 100644 index 000000000..8fb656e1a --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationActionType.cs @@ -0,0 +1,31 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + /// + /// Enumeration of entity operation actions. + /// + public enum OperationActionType + { + /// + /// A signal was sent to an entity + /// + SendSignal, + + /// + /// A new fire-and-forget orchestration was started + /// + StartNewOrchestration, + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationBatchRequest.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationBatchRequest.cs new file mode 100644 index 000000000..0abf0138b --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationBatchRequest.cs @@ -0,0 +1,41 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using System.Collections.Generic; + + /// + /// A request for execution of a batch of operations on an entity. + /// + public class OperationBatchRequest + { + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The instance id for this entity. + /// + public string? InstanceId { get; set; } + + /// + /// The current state of the entity, or null if the entity does not exist. + /// + public string? EntityState { get; set; } + + /// + /// The list of operations to be performed on the entity. + /// + public List? Operations { get; set; } + } +} diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationBatchResult.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationBatchResult.cs new file mode 100644 index 000000000..0d2718ad3 --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationBatchResult.cs @@ -0,0 +1,45 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using System.Collections.Generic; + + /// + /// The results of executing a batch of operations on the entity out of process. + /// + public class OperationBatchResult + { + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The results of executing the operations in the batch. The length of this list must match + /// the size of the batch if all messages were processed; In particular, all execution errors must be reported as a result. + /// However, this list of results can be shorter than the list of operations if + /// some suffix of the operation list was skipped, e.g. due to shutdown, send throttling, or timeouts. + /// + public List? Results { get; set; } + + /// + /// The list of actions (outgoing messages) performed while executing the operations in the batch. Can be empty. + /// + public List? Actions { get; set; } + + /// + /// The state of the entity after executing the batch, + /// or null if the entity has no state (e.g. if it has been deleted). + /// + public string? EntityState { get; set; } + } +} diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationRequest.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationRequest.cs new file mode 100644 index 000000000..ab249f88f --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationRequest.cs @@ -0,0 +1,41 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using System; + + /// + /// A request message sent to an entity when calling or signaling the entity. + /// + public class OperationRequest + { + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The name of the operation. + /// + public string? Operation { get; set; } + + /// + /// The unique GUID of the operation. + /// + public Guid Id { get; set; } + + /// + /// The input for the operation. Can be null if no input was given. + /// + public string? Input { get; set; } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/OperationResult.cs b/src/DurableTask.Core/Entities/OperationFormat/OperationResult.cs new file mode 100644 index 000000000..58f26220c --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/OperationResult.cs @@ -0,0 +1,42 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + /// + /// A response message sent by an entity to a caller after it executes an operation. + /// + public class OperationResult + { + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The serialized result returned by the operation. Can be null, if the operation returned no result. + /// May contain error details, such as a serialized exception, if is not null. + /// + public string? Result { get; set; } + + /// + /// If non-null, this string indicates that this operation did not successfully complete. + /// The actual content and its interpretation varies depending on the SDK used. + /// + public string? ErrorMessage { get; set; } + + /// + /// A structured language-independent representation of the error. Whether this field is present + /// depends on which SDK is used, and on configuration settings. + /// + public FailureDetails? FailureDetails { get; set; } + } +} diff --git a/src/DurableTask.Core/Entities/OperationFormat/SendSignalOperationAction.cs b/src/DurableTask.Core/Entities/OperationFormat/SendSignalOperationAction.cs new file mode 100644 index 000000000..04531ac31 --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/SendSignalOperationAction.cs @@ -0,0 +1,49 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using System; + + /// + /// Operation action for sending a signal. + /// + public class SendSignalOperationAction : OperationAction + { + /// + public override OperationActionType OperationActionType => OperationActionType.SendSignal; + + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The destination entity for the signal. + /// + public string? InstanceId { get; set; } + + /// + /// The name of the operation being signaled. + /// + public string? Name { get; set; } + + /// + /// The input of the operation being signaled. + /// + public string? Input { get; set; } + + /// + /// Optionally, a scheduled delivery time for the signal. + /// + public DateTime? ScheduledTime { get; set; } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OperationFormat/StartNewOrchestrationOperationAction.cs b/src/DurableTask.Core/Entities/OperationFormat/StartNewOrchestrationOperationAction.cs new file mode 100644 index 000000000..925b8b791 --- /dev/null +++ b/src/DurableTask.Core/Entities/OperationFormat/StartNewOrchestrationOperationAction.cs @@ -0,0 +1,54 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities.OperationFormat +{ + using System.Collections.Generic; + + /// + /// Entity operation action for creating sub-orchestrations. + /// + public class StartNewOrchestrationOperationAction : OperationAction + { + /// + public override OperationActionType OperationActionType => OperationActionType.StartNewOrchestration; + + // NOTE: Actions must be serializable by a variety of different serializer types to support out-of-process execution. + // To ensure maximum compatibility, all properties should be public and settable by default. + + /// + /// The name of the sub-orchestrator to start. + /// + public string? Name { get; set; } + + /// + /// The version of the sub-orchestrator to start. + /// + public string? Version { get; set; } + + /// + /// The instance ID of the created sub-orchestration. + /// + public string? InstanceId { get; set; } + + /// + /// The input of the sub-orchestration. + /// + public string? Input { get; set; } + + /// + /// Tags to be applied to the sub-orchestration. + /// + public IDictionary? Tags { get; set; } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs b/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs new file mode 100644 index 000000000..1791eb334 --- /dev/null +++ b/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs @@ -0,0 +1,351 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using DurableTask.Core.Entities; + using DurableTask.Core.Entities.EventFormat; + using DurableTask.Core.Entities.OperationFormat; + using DurableTask.Core.Exceptions; + using Newtonsoft.Json; + using Newtonsoft.Json.Linq; + using System; + using System.Collections.Generic; + using System.Linq; + + /// + /// Tracks the entity-related state of an orchestration. + /// Tracks and validates the synchronization state. + /// + public class OrchestrationEntityContext + { + private readonly string instanceId; + private readonly string executionId; + private readonly OrchestrationContext innerContext; + private readonly MessageSorter messageSorter; + + private bool lockAcquisitionPending; + + // the following are null unless we are inside a critical section + private Guid? criticalSectionId; + private EntityId[]? criticalSectionLocks; + private HashSet? availableLocks; + + /// + /// Constructs an OrchestrationEntityContext. + /// + /// The instance id. + /// The execution id. + /// The inner context. + public OrchestrationEntityContext( + string instanceId, + string executionId, + OrchestrationContext innerContext) + { + this.instanceId = instanceId; + this.executionId = executionId; + this.innerContext = innerContext; + this.messageSorter = new MessageSorter(); + } + + /// + /// Checks whether the configured backend supports entities. + /// + public bool EntitiesAreSupported => this.innerContext.EntityBackendProperties != null; + + /// + /// Whether this orchestration is currently inside a critical section. + /// + public bool IsInsideCriticalSection => this.criticalSectionId != null; + + /// + /// Enumerate all the entities that are available for calling from within a critical section. + /// This set contains all the entities that were locked prior to entering the critical section, + /// and for which there is not currently an operation call pending. + /// + /// An enumeration of all the currently available entities. + public IEnumerable GetAvailableEntities() + { + if (this.IsInsideCriticalSection) + { + foreach (var e in this.availableLocks!) + { + yield return e; + } + } + } + + /// + /// Check that a suborchestration is a valid transition in the current state. + /// + /// The error message, if it is not valid, or null otherwise + /// whether the transition is valid + public bool ValidateSuborchestrationTransition(out string? errorMessage) + { + if (this.IsInsideCriticalSection) + { + errorMessage = "While holding locks, cannot call suborchestrators."; + return false; + } + + errorMessage = null; + return true; + } + + /// + /// Check that acquire is a valid transition in the current state. + /// + /// Whether this is a signal or a call. + /// The target instance id. + /// The error message, if it is not valid, or null otherwise + /// whether the transition is valid + public bool ValidateOperationTransition(string targetInstanceId, bool oneWay, out string? errorMessage) + { + if (this.IsInsideCriticalSection) + { + var lockToUse = EntityId.FromString(targetInstanceId); + if (oneWay) + { + if (this.criticalSectionLocks.Contains(lockToUse)) + { + errorMessage = "Must not signal a locked entity from a critical section."; + return false; + } + } + else + { + if (!this.availableLocks!.Remove(lockToUse)) + { + if (this.lockAcquisitionPending) + { + errorMessage = "Must await the completion of the lock request prior to calling any entity."; + return false; + } + if (this.criticalSectionLocks.Contains(lockToUse)) + { + errorMessage = "Must not call an entity from a critical section while a prior call to the same entity is still pending."; + return false; + } + else + { + errorMessage = "Must not call an entity from a critical section if it is not one of the locked entities."; + return false; + } + } + } + } + + errorMessage = null; + return true; + } + + /// + /// Check that acquire is a valid transition in the current state. + /// + /// The error message, if it is not valid, or null otherwise + /// whether the transition is valid + public bool ValidateAcquireTransition(out string? errorMessage) + { + if (this.IsInsideCriticalSection) + { + errorMessage = "Must not enter another critical section from within a critical section."; + return false; + } + + errorMessage = null; + return true; + } + + /// + /// Called after an operation call within a critical section completes. + /// + /// + public void RecoverLockAfterCall(string targetInstanceId) + { + if (this.IsInsideCriticalSection) + { + var lockToUse = EntityId.FromString(targetInstanceId); + this.availableLocks!.Add(lockToUse); + } + } + + /// + /// Get release messages for all locks in the critical section, and release them + /// + public IEnumerable EmitLockReleaseMessages() + { + if (this.IsInsideCriticalSection) + { + var message = new ReleaseMessage() + { + ParentInstanceId = instanceId, + Id = this.criticalSectionId!.Value.ToString(), + }; + + foreach (var entityId in this.criticalSectionLocks!) + { + var instance = new OrchestrationInstance() { InstanceId = entityId.ToString() }; + var jmessage = JObject.FromObject(message, Serializer.InternalSerializer); + yield return new EventToSend(EntityMessageEventNames.ReleaseMessageEventName, jmessage, instance); + } + + this.criticalSectionLocks = null; + this.availableLocks = null; + this.criticalSectionId = null; + } + } + + /// + /// Creates a request message to be sent to an entity. + /// + /// The target entity. + /// The name of the operation. + /// If true, this is a signal, otherwise it is a call. + /// A unique identifier for this request. + /// A time for which to schedule the delivery, or null if this is not a scheduled message + /// The operation input + /// The event to send. + public EventToSend EmitRequestMessage( + OrchestrationInstance target, + string operationName, + bool oneWay, + Guid operationId, + (DateTime original, DateTime capped)? scheduledTimeUtc, + string input) + { + var request = new RequestMessage() + { + ParentInstanceId = this.instanceId, + ParentExecutionId = this.executionId, + Id = operationId, + IsSignal = oneWay, + Operation = operationName, + ScheduledTime = scheduledTimeUtc?.original, + Input = input, + }; + + this.AdjustOutgoingMessage(target.InstanceId, request, scheduledTimeUtc?.capped, out string eventName); + + // we pre-serialize to JObject so we can avoid exposure to application-specific serialization settings + var jrequest = JObject.FromObject(request, Serializer.InternalSerializer); + + return new EventToSend(eventName, jrequest, target); + } + + /// + /// Creates an acquire message to be sent to an entity. + /// + /// A unique request id. + /// All the entities that are to be acquired. + /// The event to send. + public EventToSend EmitAcquireMessage(Guid lockRequestId, EntityId[] entities) + { + // All the entities in entity[] need to be locked, but to avoid deadlock, the locks have to be acquired + // sequentially, in order. So, we send the lock request to the first entity; when the first lock + // is granted by the first entity, the first entity will forward the lock request to the second entity, + // and so on; after the last entity grants the last lock, a response is sent back here. + + // acquire the locks in a globally fixed order to avoid deadlocks + Array.Sort(entities); + + // remove duplicates if necessary. Probably quite rare, so no need to optimize more. + for (int i = 0; i < entities.Length - 1; i++) + { + if (entities[i].Equals(entities[i + 1])) + { + entities = entities.Distinct().ToArray(); + break; + } + } + + // send lock request to first entity in the lock set + var target = new OrchestrationInstance() { InstanceId = entities[0].ToString() }; + var request = new RequestMessage() + { + Id = lockRequestId, + ParentInstanceId = this.instanceId, + ParentExecutionId = this.executionId, + LockSet = entities, + Position = 0, + }; + + this.criticalSectionId = lockRequestId; + this.criticalSectionLocks = entities; + this.lockAcquisitionPending = true; + + this.AdjustOutgoingMessage(target.InstanceId, request, null, out string eventName); + + // we pre-serialize to JObject so we can avoid exposure to application-specific serialization settings + var jrequest = JObject.FromObject(request, Serializer.InternalSerializer); + + return new EventToSend(eventName, jrequest, target); + } + + /// + /// Called when a response to the acquire message is received from the last entity. + /// + /// The result returned. + /// The guid for the lock operation + public void CompleteAcquire(OperationResult result, Guid criticalSectionId) + { + this.availableLocks = new HashSet(this.criticalSectionLocks); + this.lockAcquisitionPending = false; + } + + internal void AdjustOutgoingMessage(string instanceId, RequestMessage requestMessage, DateTime? cappedTime, out string eventName) + { + if (cappedTime.HasValue) + { + eventName = EntityMessageEventNames.ScheduledRequestMessageEventName(cappedTime.Value); + } + else + { + this.messageSorter.LabelOutgoingMessage( + requestMessage, + instanceId, + this.innerContext.CurrentUtcDateTime, + this.innerContext.EntityBackendProperties.EntityMessageReorderWindow); + + eventName = EntityMessageEventNames.RequestMessageEventName; + } + } + + /// + /// Extracts the operation result from an event that represents an entity response. + /// + /// The serialized event content. + /// + public OperationResult DeserializeEntityResponseEvent(string eventContent) + { + var responseMessage = new ResponseMessage(); + + // for compatibility, we deserialize in a way that is resilient to any typename presence/absence/mismatch + try + { + // restore the scheduler state from the input + JsonConvert.PopulateObject(eventContent, responseMessage, Serializer.InternalSerializerSettings); + } + catch (Exception exception) + { + throw new EntitySchedulerException("Failed to deserialize entity response.", exception); + } + + return new OperationResult() + { + Result = responseMessage.Result, + ErrorMessage = responseMessage.ErrorMessage, + FailureDetails = responseMessage.FailureDetails, + }; + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/Serializer.cs b/src/DurableTask.Core/Entities/Serializer.cs new file mode 100644 index 000000000..0a99b5f71 --- /dev/null +++ b/src/DurableTask.Core/Entities/Serializer.cs @@ -0,0 +1,30 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using Newtonsoft.Json; + + internal static class Serializer + { + /// + /// This serializer is used exclusively for internally defined data structures and cannot be customized by user. + /// This is intentional, to avoid problems caused by our unability to control the exact format. + /// For example, including typenames can cause compatibility problems if the type name is later changed. + /// + public static JsonSerializer InternalSerializer = JsonSerializer.Create(InternalSerializerSettings); + + public static JsonSerializerSettings InternalSerializerSettings + = new JsonSerializerSettings() { TypeNameHandling = TypeNameHandling.None }; + } +} diff --git a/src/DurableTask.Core/Entities/StateFormat/EntityStatus.cs b/src/DurableTask.Core/Entities/StateFormat/EntityStatus.cs new file mode 100644 index 000000000..50c98c7d6 --- /dev/null +++ b/src/DurableTask.Core/Entities/StateFormat/EntityStatus.cs @@ -0,0 +1,43 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using System.Runtime.Serialization; + + /// + /// Information about the current status of an entity. Excludes potentially large data + /// (such as the entity state, or the contents of the queue) so it can always be read with low latency. + /// + [DataContract] + public class EntityStatus + { + /// + /// Whether this entity exists or not. + /// + [DataMember(Name = "entityExists", EmitDefaultValue = false)] + public bool EntityExists { get; set; } + + /// + /// The size of the queue, i.e. the number of operations that are waiting for the current operation to complete. + /// + [DataMember(Name = "queueSize", EmitDefaultValue = false)] + public int QueueSize { get; set; } + + /// + /// The instance id of the orchestration that currently holds the lock of this entity. + /// + [DataMember(Name = "lockedBy", EmitDefaultValue = false)] + public string? LockedBy { get; set; } + } +} diff --git a/src/DurableTask.Core/Entities/StateFormat/MessageSorter.cs b/src/DurableTask.Core/Entities/StateFormat/MessageSorter.cs new file mode 100644 index 000000000..ced0bdb8e --- /dev/null +++ b/src/DurableTask.Core/Entities/StateFormat/MessageSorter.cs @@ -0,0 +1,284 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.Core.Entities +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Runtime.Serialization; + using DurableTask.Core.Entities.EventFormat; + + /// + /// provides message ordering and deduplication of request messages (operations or lock requests) + /// that are sent to entities, from other entities, or from orchestrations. + /// + [DataContract] + internal class MessageSorter + { + // don't update the reorder window too often since the garbage collection incurs some overhead. + private static readonly TimeSpan MinIntervalBetweenCollections = TimeSpan.FromSeconds(10); + + [DataMember(EmitDefaultValue = false)] + public Dictionary LastSentToInstance { get; set; } + + [DataMember(EmitDefaultValue = false)] + public Dictionary ReceivedFromInstance { get; set; } + + [DataMember(EmitDefaultValue = false)] + public DateTime ReceiveHorizon { get; set; } + + [DataMember(EmitDefaultValue = false)] + public DateTime SendHorizon { get; set; } + + /// + /// Used for testing purposes. + /// + [IgnoreDataMember] + internal int NumberBufferedRequests => + ReceivedFromInstance?.Select(kvp => kvp.Value.Buffered?.Count ?? 0).Sum() ?? 0; + + /// + /// Called on the sending side, to fill in timestamp and predecessor fields. + /// + public void LabelOutgoingMessage(RequestMessage message, string destination, DateTime now, TimeSpan reorderWindow) + { + if (reorderWindow.Ticks == 0) + { + return; // we are not doing any message sorting. + } + + DateTime timestamp = now; + + // whenever (SendHorizon + reorderWindow < now) it is possible to advance the send horizon to (now - reorderWindow) + // and we can then clean out all the no-longer-needed entries of LastSentToInstance. + // However, to reduce the overhead of doing this collection, we don't update the send horizon immediately when possible. + // Instead, we make sure at least MinIntervalBetweenCollections passes between collections. + if (SendHorizon + reorderWindow + MinIntervalBetweenCollections < now) + { + SendHorizon = now - reorderWindow; + + // clean out send clocks that are past the reorder window + + if (LastSentToInstance != null) + { + List expired = new List(); + + foreach (var kvp in LastSentToInstance) + { + if (kvp.Value < SendHorizon) + { + expired.Add(kvp.Key); + } + } + + foreach (var t in expired) + { + LastSentToInstance.Remove(t); + } + } + } + + if (LastSentToInstance == null) + { + LastSentToInstance = new Dictionary(StringComparer.OrdinalIgnoreCase); + } + else if (LastSentToInstance.TryGetValue(destination, out var last)) + { + message.Predecessor = last; + + // ensure timestamps are monotonic even if system clock is not + if (timestamp <= last) + { + timestamp = new DateTime(last.Ticks + 1); + } + } + + message.Timestamp = timestamp; + LastSentToInstance[destination] = timestamp; + } + + /// + /// Called on the receiving side, to reorder and deduplicate within the window. + /// + public IEnumerable ReceiveInOrder(RequestMessage message, TimeSpan reorderWindow) + { + // messages sent from clients and forwarded lock messages are not participating in the sorting. + if (reorderWindow.Ticks == 0 || message.ParentInstanceId == null || message.Position > 0) + { + // Just pass the message through. + yield return message; + yield break; + } + + // whenever (ReceiveHorizon + reorderWindow < message.Timestamp), we can advance the receive horizon to (message.Timestamp - reorderWindow) + // and then we can clean out all the no-longer-needed entries of ReceivedFromInstance. + // However, to reduce the overhead of doing this collection, we don't update the receive horizon immediately when possible. + // Instead, we make sure at least MinIntervalBetweenCollections passes between collections. + if (ReceiveHorizon + reorderWindow + MinIntervalBetweenCollections < message.Timestamp) + { + ReceiveHorizon = message.Timestamp - reorderWindow; + + // deliver any messages that were held in the receive buffers + // but are now past the reorder window + + List buffersToRemove = new List(); + + if (ReceivedFromInstance != null) + { + foreach (var kvp in ReceivedFromInstance) + { + if (kvp.Value.Last < ReceiveHorizon) + { + // we reset Last to MinValue; this means all future messages received + // are treated as if they were the first message received. + kvp.Value.Last = DateTime.MinValue; + } + + while (TryDeliverNextMessage(kvp.Value, out var next)) + { + yield return next; + } + + if (kvp.Value.Last == DateTime.MinValue + && (kvp.Value.Buffered == null || kvp.Value.Buffered.Count == 0)) + { + // we no longer need to store this buffer since it contains no relevant information anymore + // (it is back to its initial "empty" state) + buffersToRemove.Add(kvp.Key); + } + } + + foreach (var t in buffersToRemove) + { + ReceivedFromInstance.Remove(t); + } + + if (ReceivedFromInstance.Count == 0) + { + ReceivedFromInstance = null; + } + } + } + + // Messages older than the reorder window are not participating. + if (message.Timestamp < ReceiveHorizon) + { + // Just pass the message through. + yield return message; + yield break; + } + + ReceiveBuffer receiveBuffer; + + if (ReceivedFromInstance == null) + { + ReceivedFromInstance = new Dictionary(StringComparer.OrdinalIgnoreCase); + } + + if (!ReceivedFromInstance.TryGetValue(message.ParentInstanceId, out receiveBuffer)) + { + ReceivedFromInstance[message.ParentInstanceId] = receiveBuffer = new ReceiveBuffer() + { + ExecutionId = message.ParentExecutionId, + }; + } + else if (receiveBuffer.ExecutionId != message.ParentExecutionId) + { + // this message is from a new execution; release all buffered messages and start over + if (receiveBuffer.Buffered != null) + { + foreach (var kvp in receiveBuffer.Buffered) + { + yield return kvp.Value; + } + + receiveBuffer.Buffered.Clear(); + } + + receiveBuffer.Last = DateTime.MinValue; + receiveBuffer.ExecutionId = message.ParentExecutionId; + } + + if (message.Timestamp <= receiveBuffer.Last) + { + // This message was already delivered, it's a duplicate + yield break; + } + + if (message.Predecessor > receiveBuffer.Last + && message.Predecessor >= ReceiveHorizon) + { + // this message is waiting for a non-delivered predecessor in the window, buffer it + if (receiveBuffer.Buffered == null) + { + receiveBuffer.Buffered = new SortedDictionary(); + } + + receiveBuffer.Buffered[message.Timestamp] = message; + } + else + { + yield return message; + + receiveBuffer.Last = message.Timestamp >= ReceiveHorizon ? message.Timestamp : DateTime.MinValue; + + while (TryDeliverNextMessage(receiveBuffer, out var next)) + { + yield return next; + } + } + } + + private bool TryDeliverNextMessage(ReceiveBuffer buffer, out RequestMessage message) + { + if (buffer.Buffered != null) + { + using (var e = buffer.Buffered.GetEnumerator()) + { + if (e.MoveNext()) + { + var pred = e.Current.Value.Predecessor; + + if (pred <= buffer.Last || pred < ReceiveHorizon) + { + message = e.Current.Value; + + buffer.Last = message.Timestamp >= ReceiveHorizon ? message.Timestamp : DateTime.MinValue; + + buffer.Buffered.Remove(message.Timestamp); + + return true; + } + } + } + } + + message = null; + return false; + } + + [DataContract] + public class ReceiveBuffer + { + [DataMember] + public DateTime Last { get; set; }// last message delivered, or DateTime.Min if none + + [DataMember(EmitDefaultValue = false)] + public string ExecutionId { get; set; } // execution id of last message, if any + + [DataMember(EmitDefaultValue = false)] + public SortedDictionary Buffered { get; set; } + } + } +} diff --git a/src/DurableTask.Core/Entities/StateFormat/SchedulerState.cs b/src/DurableTask.Core/Entities/StateFormat/SchedulerState.cs new file mode 100644 index 000000000..8ece4ee73 --- /dev/null +++ b/src/DurableTask.Core/Entities/StateFormat/SchedulerState.cs @@ -0,0 +1,118 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Entities +{ + using System; + using System.Collections.Generic; + using System.Runtime.Serialization; + using DurableTask.Core.Entities.EventFormat; + + /// + /// The persisted state of an entity scheduler, as handed forward between ContinueAsNew instances. + /// + [DataContract] + internal class SchedulerState + { + /// + /// Whether this entity exists or not. + /// + [DataMember(Name = "exists", EmitDefaultValue = false)] + public bool EntityExists { get; set; } + + /// + /// The last serialized entity state. + /// + [DataMember(Name = "state", EmitDefaultValue = false)] + public string? EntityState { get; set; } + + /// + /// The queue of waiting operations, or null if none. + /// + [DataMember(Name = "queue", EmitDefaultValue = false)] + public Queue? Queue { get; private set; } + + /// + /// The instance id of the orchestration that currently holds the lock of this entity. + /// + [DataMember(Name = "lockedBy", EmitDefaultValue = false)] + public string? LockedBy { get; set; } + + /// + /// Whether processing on this entity is currently suspended. + /// + [DataMember(Name = "suspended", EmitDefaultValue = false)] + public bool Suspended { get; set; } + + /// + /// The metadata used for reordering and deduplication of messages sent to entities. + /// + [DataMember(Name = "sorter", EmitDefaultValue = false)] + public MessageSorter MessageSorter { get; set; } = new MessageSorter(); + + [IgnoreDataMember] + public bool IsEmpty => !EntityExists && (Queue == null || Queue.Count == 0) && LockedBy == null; + + internal void Enqueue(RequestMessage operationMessage) + { + if (Queue == null) + { + Queue = new Queue(); + } + + Queue.Enqueue(operationMessage); + } + + internal void PutBack(Queue messages) + { + if (Queue != null) + { + foreach (var message in Queue) + { + messages.Enqueue(message); + } + } + + Queue = messages; + } + + internal bool MayDequeue() + { + return Queue != null + && Queue.Count > 0 + && (LockedBy == null || LockedBy == Queue.Peek().ParentInstanceId); + } + + internal RequestMessage Dequeue() + { + if (this.Queue == null) + { + throw new InvalidOperationException("Queue is empty"); + } + + var result = Queue.Dequeue(); + + if (Queue.Count == 0) + { + Queue = null; + } + + return result; + } + + public override string ToString() + { + return $"exists={EntityExists} queue.count={(Queue != null ? Queue.Count : 0)}"; + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/Entities/TaskEntity.cs b/src/DurableTask.Core/Entities/TaskEntity.cs new file mode 100644 index 000000000..960bf14bc --- /dev/null +++ b/src/DurableTask.Core/Entities/TaskEntity.cs @@ -0,0 +1,30 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +#nullable enable +namespace DurableTask.Core.Entities +{ + using System.Threading.Tasks; + using DurableTask.Core.Entities.OperationFormat; + + /// + /// Abstract base class for entities. + /// + public abstract class TaskEntity + { + /// + /// Execute a batch of operations on an entity. + /// + public abstract Task ExecuteOperationBatchAsync(OperationBatchRequest operations, EntityExecutionOptions options); + } +} diff --git a/src/DurableTask.Core/Exceptions/EntitySchedulerException.cs b/src/DurableTask.Core/Exceptions/EntitySchedulerException.cs new file mode 100644 index 000000000..d79541cf4 --- /dev/null +++ b/src/DurableTask.Core/Exceptions/EntitySchedulerException.cs @@ -0,0 +1,61 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +#nullable enable +namespace DurableTask.Core.Exceptions +{ + using System; + using System.Runtime.Serialization; + + /// + /// Exception used to describe various issues encountered by the entity scheduler. + /// + [Serializable] + public class EntitySchedulerException : Exception + { + /// + /// Initializes a new instance of the class. + /// + public EntitySchedulerException() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message that describes the error. + public EntitySchedulerException(string message) + : base(message) + { + } + + /// + /// Initializes an new instance of the class. + /// + /// The message that describes the error. + /// The exception that was caught. + public EntitySchedulerException(string errorMessage, Exception innerException) + : base(errorMessage, innerException) + { + } + + /// + /// Initializes a new instance of the class with serialized data. + /// + /// The System.Runtime.Serialization.SerializationInfo that holds the serialized object data about the exception being thrown. + /// The System.Runtime.Serialization.StreamingContext that contains contextual information about the source or destination. + protected EntitySchedulerException(SerializationInfo info, StreamingContext context) + : base(info, context) + { + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/FailureDetails.cs b/src/DurableTask.Core/FailureDetails.cs index b99584833..ddf1f1a33 100644 --- a/src/DurableTask.Core/FailureDetails.cs +++ b/src/DurableTask.Core/FailureDetails.cs @@ -19,7 +19,7 @@ namespace DurableTask.Core using Newtonsoft.Json; /// - /// Details of an activity or orchestration failure. + /// Details of an activity, orchestration, or entity operation failure. /// [Serializable] public class FailureDetails : IEquatable diff --git a/src/DurableTask.Core/Logging/EventIds.cs b/src/DurableTask.Core/Logging/EventIds.cs index 963de5f71..2998e5f8d 100644 --- a/src/DurableTask.Core/Logging/EventIds.cs +++ b/src/DurableTask.Core/Logging/EventIds.cs @@ -47,6 +47,10 @@ static class EventIds public const int OrchestrationExecuted = 52; public const int OrchestrationAborted = 53; public const int DiscardingMessage = 54; + public const int EntityBatchExecuting = 55; + public const int EntityBatchExecuted = 56; + public const int EntityLockAcquired = 57; + public const int EntityLockReleased = 58; public const int TaskActivityStarting = 60; public const int TaskActivityCompleted = 61; diff --git a/src/DurableTask.Core/Logging/LogEvents.cs b/src/DurableTask.Core/Logging/LogEvents.cs index a914154a2..786981c0f 100644 --- a/src/DurableTask.Core/Logging/LogEvents.cs +++ b/src/DurableTask.Core/Logging/LogEvents.cs @@ -14,9 +14,11 @@ namespace DurableTask.Core.Logging { using System; + using System.Linq; using System.Text; using DurableTask.Core.Command; using DurableTask.Core.Common; + using DurableTask.Core.Entities.OperationFormat; using DurableTask.Core.History; using Microsoft.Extensions.Logging; @@ -1177,6 +1179,223 @@ void IEventSourceEvent.WriteEventSource() => Utils.PackageVersion); } + /// + /// Log event representing a task hub worker executing a batch of entity operations. + /// + internal class EntityBatchExecuting : StructuredLogEvent, IEventSourceEvent + { + public EntityBatchExecuting(OperationBatchRequest request) + { + this.InstanceId = request.InstanceId; + this.OperationCount = request.Operations.Count; + this.EntityStateLength = request.EntityState?.Length ?? 0; + } + + [StructuredLogField] + public string InstanceId { get; } + + [StructuredLogField] + public int OperationCount { get; } + + [StructuredLogField] + public int EntityStateLength { get; } + + public override EventId EventId => new EventId( + EventIds.EntityBatchExecuting, + nameof(EventIds.EntityBatchExecuting)); + + public override LogLevel Level => LogLevel.Debug; + + protected override string CreateLogMessage() => + $"{this.InstanceId}: executing batch of {this.OperationCount} operations on entity state of length {this.EntityStateLength}."; + + void IEventSourceEvent.WriteEventSource() => + StructuredEventSource.Log.EntityBatchExecuting( + this.InstanceId, + this.OperationCount, + this.EntityStateLength, + Utils.AppName, + Utils.PackageVersion); + } + + /// + /// Log event representing a task hub worker executed a batch of entity operations. + /// + internal class EntityBatchExecuted : StructuredLogEvent, IEventSourceEvent + { + public EntityBatchExecuted(OperationBatchRequest request, OperationBatchResult result) + { + this.InstanceId = request.InstanceId; + this.OperationCount = request.Operations.Count; + this.ResultCount = result.Results.Count; + this.ErrorCount = result.Results.Count(x => x.ErrorMessage != null); + this.ActionCount = result.Actions.Count; + this.EntityStateLength = request.EntityState?.Length ?? 0; + } + + [StructuredLogField] + public string InstanceId { get; } + + [StructuredLogField] + public int OperationCount { get; } + + [StructuredLogField] + public int ResultCount { get; } + + [StructuredLogField] + public int ErrorCount { get; } + + [StructuredLogField] + public int ActionCount { get; } + + [StructuredLogField] + public int EntityStateLength { get; } + + public override EventId EventId => new EventId( + EventIds.EntityBatchExecuting, + nameof(EventIds.EntityBatchExecuting)); + + public override LogLevel Level => LogLevel.Information; + + protected override string CreateLogMessage() => + $"{this.InstanceId}: completed {this.ResultCount} of {this.OperationCount} entity operations, resulting in {this.ErrorCount} errors, {this.ActionCount} actions, and entity state of length {this.EntityStateLength}."; + + void IEventSourceEvent.WriteEventSource() => + StructuredEventSource.Log.EntityBatchExecuted( + this.InstanceId, + this.OperationCount, + this.ResultCount, + this.ErrorCount, + this.ActionCount, + this.EntityStateLength, + Utils.AppName, + Utils.PackageVersion); + } + + /// + /// Logs that an entity processed a lock acquire message. + /// + internal class EntityLockAcquired : StructuredLogEvent, IEventSourceEvent + { + public EntityLockAcquired(string entityId, Core.Entities.EventFormat.RequestMessage message) + { + this.EntityId = entityId; + this.InstanceId = message.ParentInstanceId; + this.ExecutionId = message.ParentExecutionId; + this.CriticalSectionId = message.Id; + this.Position = message.Position; + + if (message.LockSet != null) + { + this.LockSet = string.Join(",", message.LockSet.Select(id => id.ToString())); + } + } + + /// + /// The entity that is being locked. + /// + [StructuredLogField] + public string EntityId { get; } + + /// + /// The instance ID of the orchestration that is executing the critical section. + /// + [StructuredLogField] + public string InstanceId { get; set; } + + /// + /// The execution ID of the orchestration that is executing the critical section. + /// + [StructuredLogField] + public string ExecutionId { get; set; } + + /// + /// The unique ID of the critical section that is acquiring this lock. + /// + [StructuredLogField] + public Guid CriticalSectionId { get; set; } + + /// + /// The ordered set of locks that are being acquired for this critical section. + /// + [StructuredLogField] + public string LockSet { get; set; } + + /// + /// Which of the locks in is being acquired. + /// + [StructuredLogField] + public int Position { get; set; } + + public override EventId EventId => new EventId( + EventIds.EntityLockAcquired, + nameof(EventIds.EntityLockAcquired)); + + public override LogLevel Level => LogLevel.Information; + + protected override string CreateLogMessage() => + $"{this.EntityId}: acquired lock {this.Position+1}/{this.LockSet.Length} for orchestration instanceId={this.InstanceId} executionId={this.ExecutionId} criticalSectionId={this.CriticalSectionId}"; + + void IEventSourceEvent.WriteEventSource() => + StructuredEventSource.Log.EntityLockAcquired( + this.EntityId, + this.InstanceId ?? string.Empty, + this.ExecutionId ?? string.Empty, + this.CriticalSectionId, + this.LockSet ?? string.Empty, + this.Position, + Utils.AppName, + Utils.PackageVersion); + } + + /// + /// Logs that an entity processed a lock release message. + /// + internal class EntityLockReleased : StructuredLogEvent, IEventSourceEvent + { + public EntityLockReleased(string entityId, Core.Entities.EventFormat.ReleaseMessage message) + { + this.EntityId = entityId; + this.InstanceId = message.ParentInstanceId; + this.CriticalSectionId = message.Id; + } + + /// + /// The entity that is being unlocked. + /// + [StructuredLogField] + public string EntityId { get; } + + /// + /// The instance ID of the orchestration that is executing the critical section. + /// + [StructuredLogField] + public string InstanceId { get; set; } + + /// + /// The unique ID of the critical section that is releasing the lock after completing. + /// + [StructuredLogField] + public string CriticalSectionId { get; set; } + + public override EventId EventId => new EventId( + EventIds.EntityLockReleased, + nameof(EventIds.EntityLockReleased)); + + public override LogLevel Level => LogLevel.Information; + + protected override string CreateLogMessage() => + $"{this.EntityId}: released lock for orchestration instanceId={this.InstanceId} criticalSectionId={this.CriticalSectionId}"; + + void IEventSourceEvent.WriteEventSource() => + StructuredEventSource.Log.EntityLockReleased( + this.EntityId, + this.InstanceId ?? string.Empty, + this.CriticalSectionId ?? string.Empty, + Utils.AppName, + Utils.PackageVersion); + } + /// /// Log event indicating that an activity execution is starting. /// diff --git a/src/DurableTask.Core/Logging/LogHelper.cs b/src/DurableTask.Core/Logging/LogHelper.cs index f4d1ffe34..7bea8da2b 100644 --- a/src/DurableTask.Core/Logging/LogHelper.cs +++ b/src/DurableTask.Core/Logging/LogHelper.cs @@ -17,7 +17,7 @@ namespace DurableTask.Core.Logging using System.Collections.Generic; using System.Text; using DurableTask.Core.Command; - using DurableTask.Core.Common; + using DurableTask.Core.Entities.OperationFormat; using DurableTask.Core.History; using Microsoft.Extensions.Logging; @@ -561,6 +561,58 @@ internal void RenewOrchestrationWorkItemFailed(TaskOrchestrationWorkItem workIte } } + + /// + /// Logs that an entity operation batch is about to start executing. + /// + /// The batch request. + internal void EntityBatchExecuting(OperationBatchRequest request) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.EntityBatchExecuting(request)); + } + } + + /// + /// Logs that an entity operation batch completed its execution. + /// + /// The batch request. + /// The batch result. + internal void EntityBatchExecuted(OperationBatchRequest request, OperationBatchResult result) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.EntityBatchExecuted(request, result)); + } + } + + /// + /// Logs that an entity processed a lock acquire message. + /// + /// The entity id. + /// The message. + internal void EntityLockAcquired(string entityId, Core.Entities.EventFormat.RequestMessage message) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.EntityLockAcquired(entityId, message)); + } + } + + /// + /// Logs that an entity processed a lock release message. + /// + /// The entity id. + /// The message. + internal void EntityLockReleased(string entityId, Core.Entities.EventFormat.ReleaseMessage message) + { + if (this.IsStructuredLoggingEnabled) + { + this.WriteStructuredLog(new LogEvents.EntityLockReleased(entityId, message)); + } + } + #endregion #region Activity dispatcher diff --git a/src/DurableTask.Core/Logging/StructuredEventSource.cs b/src/DurableTask.Core/Logging/StructuredEventSource.cs index b9edc0a46..b6ec19a32 100644 --- a/src/DurableTask.Core/Logging/StructuredEventSource.cs +++ b/src/DurableTask.Core/Logging/StructuredEventSource.cs @@ -624,6 +624,102 @@ internal void DiscardingMessage( } } + [Event(EventIds.EntityBatchExecuting, Level = EventLevel.Informational, Version = 1)] + internal void EntityBatchExecuting( + string InstanceId, + int OperationCount, + int EntityStateLength, + string AppName, + string ExtensionVersion) + { + if (this.IsEnabled(EventLevel.Informational)) + { + // TODO: Use WriteEventCore for better performance + this.WriteEvent( + EventIds.EntityBatchExecuting, + InstanceId, + OperationCount, + EntityStateLength, + AppName, + ExtensionVersion); + } + } + + [Event(EventIds.EntityBatchExecuted, Level = EventLevel.Informational, Version = 1)] + internal void EntityBatchExecuted( + string InstanceId, + int OperationCount, + int ResultCount, + int ErrorCount, + int ActionCount, + int EntityStateLength, + string AppName, + string ExtensionVersion) + { + if (this.IsEnabled(EventLevel.Informational)) + { + // TODO: Use WriteEventCore for better performance + this.WriteEvent( + EventIds.EntityBatchExecuted, + InstanceId, + OperationCount, + ResultCount, + ErrorCount, + ActionCount, + EntityStateLength, + AppName, + ExtensionVersion); + } + } + + [Event(EventIds.EntityLockAcquired, Level = EventLevel.Informational, Version = 1)] + internal void EntityLockAcquired( + string EntityId, + string InstanceId, + string ExecutionId, + Guid CriticalSectionId, + string LockSet, + int Position, + string AppName, + string ExtensionVersion) + { + if (this.IsEnabled(EventLevel.Informational)) + { + // TODO: Use WriteEventCore for better performance + this.WriteEvent( + EventIds.EntityLockAcquired, + EntityId, + InstanceId, + ExecutionId, + CriticalSectionId, + LockSet, + Position, + AppName, + ExtensionVersion); + } + } + + [Event(EventIds.EntityLockReleased, Level = EventLevel.Informational, Version = 1)] + internal void EntityLockReleased( + string EntityId, + string InstanceId, + string Id, + string AppName, + string ExtensionVersion) + { + if (this.IsEnabled(EventLevel.Informational)) + { + // TODO: Use WriteEventCore for better performance + this.WriteEvent( + EventIds.EntityLockReleased, + EntityId, + InstanceId, + Id, + AppName, + ExtensionVersion); + } + } + [Event(EventIds.TaskActivityStarting, Level = EventLevel.Informational, Version = 1)] internal void TaskActivityStarting( string InstanceId, diff --git a/src/DurableTask.Core/OrchestrationContext.cs b/src/DurableTask.Core/OrchestrationContext.cs index 52238bbc2..39f907542 100644 --- a/src/DurableTask.Core/OrchestrationContext.cs +++ b/src/DurableTask.Core/OrchestrationContext.cs @@ -18,6 +18,7 @@ namespace DurableTask.Core using System.Threading; using System.Threading.Tasks; using Castle.DynamicProxy; + using DurableTask.Core.Entities; using DurableTask.Core.Serializing; /// @@ -67,6 +68,11 @@ public abstract class OrchestrationContext /// internal ErrorPropagationMode ErrorPropagationMode { get; set; } + /// + /// Information about backend entity support, or null if the configured backend does not support entities. + /// + internal EntityBackendProperties EntityBackendProperties { get; set; } + /// /// Create a proxy client class to schedule remote TaskActivities via a strongly typed interface. /// diff --git a/src/DurableTask.Core/TaskEntityDispatcher.cs b/src/DurableTask.Core/TaskEntityDispatcher.cs new file mode 100644 index 000000000..0e9f0990f --- /dev/null +++ b/src/DurableTask.Core/TaskEntityDispatcher.cs @@ -0,0 +1,864 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- +namespace DurableTask.Core +{ + using System; + using System.Collections.Generic; + using System.Diagnostics; + using System.Threading; + using System.Threading.Tasks; + using DurableTask.Core.Common; + using DurableTask.Core.Entities; + using DurableTask.Core.Entities.EventFormat; + using DurableTask.Core.Entities.OperationFormat; + using DurableTask.Core.Exceptions; + using DurableTask.Core.History; + using DurableTask.Core.Logging; + using DurableTask.Core.Middleware; + using DurableTask.Core.Tracing; + using Newtonsoft.Json; + + /// + /// Dispatcher for orchestrations and entities to handle processing and renewing, completion of orchestration events. + /// + public class TaskEntityDispatcher + { + readonly INameVersionObjectManager objectManager; + readonly IOrchestrationService orchestrationService; + readonly IEntityOrchestrationService entityOrchestrationService; + readonly WorkItemDispatcher dispatcher; + readonly DispatchMiddlewarePipeline dispatchPipeline; + readonly EntityBackendProperties entityBackendProperties; + readonly LogHelper logHelper; + readonly ErrorPropagationMode errorPropagationMode; + readonly TaskOrchestrationDispatcher.NonBlockingCountdownLock concurrentSessionLock; + + internal TaskEntityDispatcher( + IOrchestrationService orchestrationService, + INameVersionObjectManager entityObjectManager, + DispatchMiddlewarePipeline entityDispatchPipeline, + LogHelper logHelper, + ErrorPropagationMode errorPropagationMode) + { + this.objectManager = entityObjectManager ?? throw new ArgumentNullException(nameof(entityObjectManager)); + this.orchestrationService = orchestrationService ?? throw new ArgumentNullException(nameof(orchestrationService)); + this.dispatchPipeline = entityDispatchPipeline ?? throw new ArgumentNullException(nameof(entityDispatchPipeline)); + this.logHelper = logHelper ?? throw new ArgumentNullException(nameof(logHelper)); + this.errorPropagationMode = errorPropagationMode; + this.entityOrchestrationService = (orchestrationService as IEntityOrchestrationService)!; + this.entityBackendProperties = entityOrchestrationService.GetEntityBackendProperties(); + + this.dispatcher = new WorkItemDispatcher( + "TaskEntityDispatcher", + item => item == null ? string.Empty : item.InstanceId, + this.OnFetchWorkItemAsync, + this.OnProcessWorkItemSessionAsync) + { + GetDelayInSecondsAfterOnFetchException = orchestrationService.GetDelayInSecondsAfterOnFetchException, + GetDelayInSecondsAfterOnProcessException = orchestrationService.GetDelayInSecondsAfterOnProcessException, + SafeReleaseWorkItem = orchestrationService.ReleaseTaskOrchestrationWorkItemAsync, + AbortWorkItem = orchestrationService.AbandonTaskOrchestrationWorkItemAsync, + DispatcherCount = orchestrationService.TaskOrchestrationDispatcherCount, + MaxConcurrentWorkItems = this.entityBackendProperties.MaxConcurrentTaskEntityWorkItems, + LogHelper = logHelper, + }; + + // To avoid starvation, we only allow half of all concurrently executing entities to + // leverage extended sessions. + var maxConcurrentSessions = (int)Math.Ceiling(this.dispatcher.MaxConcurrentWorkItems / 2.0); + this.concurrentSessionLock = new TaskOrchestrationDispatcher.NonBlockingCountdownLock(maxConcurrentSessions); + } + + /// + /// The entity options configured, or null if the backend does not support entities. + /// + public EntityBackendProperties EntityBackendProperties => this.entityBackendProperties; + + /// + /// Starts the dispatcher to start getting and processing entity message batches + /// + public async Task StartAsync() + { + await this.dispatcher.StartAsync(); + } + + /// + /// Stops the dispatcher to stop getting and processing entity message batches + /// + /// Flag indicating whether to stop gracefully or immediately + public async Task StopAsync(bool forced) + { + await this.dispatcher.StopAsync(forced); + } + + /// + /// Method to get the next work item to process within supplied timeout + /// + /// The max timeout to wait + /// A cancellation token used to cancel a fetch operation. + /// A new TaskOrchestrationWorkItem + protected Task OnFetchWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) + { + return this.entityOrchestrationService.LockNextEntityWorkItemAsync(receiveTimeout, cancellationToken); + } + + async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) + { + try + { + if (workItem.Session == null) + { + // Legacy behavior + await this.OnProcessWorkItemAsync(workItem); + return; + } + + var isExtendedSession = false; + + var processCount = 0; + try + { + while (true) + { + // While the work item contains messages that need to be processed, execute them. + if (workItem.NewMessages?.Count > 0) + { + bool isCompletedOrInterrupted = await this.OnProcessWorkItemAsync(workItem); + if (isCompletedOrInterrupted) + { + break; + } + + processCount++; + } + + // Fetches beyond the first require getting an extended session lock, used to prevent starvation. + if (processCount > 0 && !isExtendedSession) + { + isExtendedSession = this.concurrentSessionLock.Acquire(); + if (!isExtendedSession) + { + break; + } + } + + Stopwatch timer = Stopwatch.StartNew(); + + // Wait for new messages to arrive for the session. This call is expected to block (asynchronously) + // until either new messages are available or until a provider-specific timeout has expired. + workItem.NewMessages = await workItem.Session.FetchNewOrchestrationMessagesAsync(workItem); + if (workItem.NewMessages == null) + { + break; + } + + workItem.OrchestrationRuntimeState.NewEvents.Clear(); + } + } + finally + { + if (isExtendedSession) + { + this.concurrentSessionLock.Release(); + } + } + } + catch (SessionAbortedException e) + { + // Either the orchestration or the orchestration service explicitly abandoned the session. + OrchestrationInstance instance = workItem.OrchestrationRuntimeState?.OrchestrationInstance ?? new OrchestrationInstance { InstanceId = workItem.InstanceId }; + this.logHelper.OrchestrationAborted(instance, e.Message); + await this.orchestrationService.AbandonTaskOrchestrationWorkItemAsync(workItem); + } + } + + class WorkItemEffects + { + public List ActivityMessages; + public List TimerMessages; + public List InstanceMessages; + public int taskIdCounter; + public string InstanceId; + public OrchestrationRuntimeState RuntimeState; + } + + /// + /// Method to process a new work item + /// + /// The work item to process + protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem workItem) + { + OrchestrationRuntimeState originalOrchestrationRuntimeState = workItem.OrchestrationRuntimeState; + + OrchestrationRuntimeState runtimeState = workItem.OrchestrationRuntimeState; + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + + Task renewTask = null; + using var renewCancellationTokenSource = new CancellationTokenSource(); + if (workItem.LockedUntilUtc < DateTime.MaxValue) + { + // start a task to run RenewUntil + renewTask = Task.Factory.StartNew( + () => TaskOrchestrationDispatcher.RenewUntil(workItem, this.orchestrationService, this.logHelper, nameof(TaskEntityDispatcher), renewCancellationTokenSource.Token), + renewCancellationTokenSource.Token); + } + + WorkItemEffects effects = new WorkItemEffects() + { + ActivityMessages = new List(), + TimerMessages = new List(), + InstanceMessages = new List(), + taskIdCounter = 0, + InstanceId = workItem.InstanceId, + RuntimeState = runtimeState, + }; + + try + { + // Assumes that: if the batch contains a new "ExecutionStarted" event, it is the first message in the batch. + if (!TaskOrchestrationDispatcher.ReconcileMessagesWithState(workItem, nameof(TaskEntityDispatcher), this.logHelper)) + { + // TODO : mark an orchestration as faulted if there is data corruption + this.logHelper.DroppingOrchestrationWorkItem(workItem, "Received work-item for an invalid orchestration"); + } + else + { + + // we start with processing all the requests and figuring out which ones to execute now + // results can depend on whether the entity is locked, what the maximum batch size is, + // and whether the messages arrived out of order + + this.DetermineWork(workItem.OrchestrationRuntimeState, + out SchedulerState schedulerState, + out Work workToDoNow); + + if (workToDoNow.OperationCount > 0) + { + // execute the user-defined operations on this entity, via the middleware + var result = await this.ExecuteViaMiddlewareAsync(workToDoNow, runtimeState.OrchestrationInstance, schedulerState.EntityState); + + // go through all results + // for each operation that is not a signal, send a result message back to the calling orchestrator + for (int i = 0; i < result.Results!.Count; i++) + { + var req = workToDoNow.Operations[i]; + if (!req.IsSignal) + { + this.SendResultMessage(effects, req, result.Results[i]); + } + } + + if (result.Results.Count < workToDoNow.OperationCount) + { + // some operations were not processed + var deferred = workToDoNow.RemoveDeferredWork(result.Results.Count); + schedulerState.PutBack(deferred); + workToDoNow.ToBeContinued(schedulerState); + } + + // update the entity state based on the result + schedulerState.EntityState = result.EntityState; + schedulerState.EntityExists = result.EntityState != null; + + // perform the actions + foreach (var action in result.Actions!) + { + switch (action) + { + case (SendSignalOperationAction sendSignalAction): + this.SendSignalMessage(effects, schedulerState, sendSignalAction); + break; + case (StartNewOrchestrationOperationAction startAction): + this.ProcessSendStartMessage(effects, runtimeState, startAction); + break; + } + } + } + + // process the lock request, if any + if (workToDoNow.LockRequest != null) + { + this.ProcessLockRequest(effects, schedulerState, workToDoNow.LockRequest); + } + + if (workToDoNow.ToBeRescheduled != null) + { + foreach (var request in workToDoNow.ToBeRescheduled) + { + // Reschedule all signals that were received before their time + this.SendScheduledSelfMessage(effects, request); + } + } + + if (workToDoNow.SuspendAndContinue) + { + this.SendContinueSelfMessage(effects); + } + + // this batch is complete. Since this is an entity, we now + // (always) start a new execution, as in continue-as-new + + var serializedSchedulerState = this.SerializeSchedulerStateForNextExecution(schedulerState); + var nextExecutionStartedEvent = new ExecutionStartedEvent(-1, serializedSchedulerState) + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = workItem.InstanceId, + ExecutionId = Guid.NewGuid().ToString("N") + }, + Tags = runtimeState.Tags, + ParentInstance = runtimeState.ParentInstance, + Name = runtimeState.Name, + Version = runtimeState.Version + }; + var entityStatus = new EntityStatus() + { + EntityExists = schedulerState.EntityExists, + QueueSize = schedulerState.Queue?.Count ?? 0, + LockedBy = schedulerState.LockedBy, + }; + var serializedEntityStatus = JsonConvert.SerializeObject(entityStatus, Serializer.InternalSerializerSettings); + + // create the new runtime state for the next execution + runtimeState = new OrchestrationRuntimeState(); + runtimeState.Status = serializedEntityStatus; + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(nextExecutionStartedEvent); + runtimeState.AddEvent(new OrchestratorCompletedEvent(-1)); + } + } + finally + { + if (renewTask != null) + { + try + { + renewCancellationTokenSource.Cancel(); + await renewTask; + } + catch (ObjectDisposedException) + { + // ignore + } + catch (OperationCanceledException) + { + // ignore + } + } + } + + OrchestrationState instanceState = (runtimeState.ExecutionStartedEvent != null) ? + instanceState = Utils.BuildOrchestrationState(runtimeState) : null; + + if (workItem.RestoreOriginalRuntimeStateDuringCompletion) + { + // some backends expect the original runtime state object + workItem.OrchestrationRuntimeState = originalOrchestrationRuntimeState; + } + else + { + workItem.OrchestrationRuntimeState = runtimeState; + } + + await this.orchestrationService.CompleteTaskOrchestrationWorkItemAsync( + workItem, + runtimeState, + effects.ActivityMessages, + effects.InstanceMessages, + effects.TimerMessages, + null, + instanceState); + + if (workItem.RestoreOriginalRuntimeStateDuringCompletion) + { + workItem.OrchestrationRuntimeState = runtimeState; + } + + return true; + } + + void ProcessLockRequest(WorkItemEffects effects, SchedulerState schedulerState, RequestMessage request) + { + this.logHelper.EntityLockAcquired(effects.InstanceId, request); + + // mark the entity state as locked + schedulerState.LockedBy = request.ParentInstanceId; + + request.Position++; + + if (request.Position < request.LockSet.Length) + { + // send lock request to next entity in the lock set + var target = new OrchestrationInstance() { InstanceId = request.LockSet[request.Position].ToString() }; + this.SendLockRequestMessage(effects, schedulerState, target, request); + } + else + { + // send lock acquisition completed response back to originating orchestration instance + var target = new OrchestrationInstance() { InstanceId = request.ParentInstanceId, ExecutionId = request.ParentExecutionId }; + this.SendLockResponseMessage(effects, target, request.Id); + } + } + + string SerializeSchedulerStateForNextExecution(SchedulerState schedulerState) + { + if (this.entityBackendProperties.SupportsImplicitEntityDeletion && schedulerState.IsEmpty && !schedulerState.Suspended) + { + // this entity scheduler is idle and the entity is deleted, so the instance and history can be removed from storage + // we convey this to the durability provider by issuing a continue-as-new with null input + return null; + } + else + { + // we persist the state of the entity scheduler and entity + return JsonConvert.SerializeObject(schedulerState, typeof(SchedulerState), Serializer.InternalSerializerSettings); + } + } + + #region Preprocess to determine work + + void DetermineWork(OrchestrationRuntimeState runtimeState, out SchedulerState schedulerState, out Work batch) + { + string instanceId = runtimeState.OrchestrationInstance.InstanceId; + schedulerState = new SchedulerState(); + batch = new Work(); + + Queue lockHolderMessages = null; + + foreach (HistoryEvent e in runtimeState.Events) + { + switch (e.EventType) + { + case EventType.ExecutionStarted: + + + if (runtimeState.Input != null) + { + try + { + // restore the scheduler state from the input + JsonConvert.PopulateObject(runtimeState.Input, schedulerState, Serializer.InternalSerializerSettings); + } + catch (Exception exception) + { + throw new EntitySchedulerException("Failed to deserialize entity scheduler state - may be corrupted or wrong version.", exception); + } + } + break; + + case EventType.EventRaised: + EventRaisedEvent eventRaisedEvent = (EventRaisedEvent)e; + + if (EntityMessageEventNames.IsRequestMessage(eventRaisedEvent.Name)) + { + // we are receiving an operation request or a lock request + var requestMessage = new RequestMessage(); + + try + { + JsonConvert.PopulateObject(eventRaisedEvent.Input, requestMessage, Serializer.InternalSerializerSettings); + } + catch (Exception exception) + { + throw new EntitySchedulerException("Failed to deserialize incoming request message - may be corrupted or wrong version.", exception); + } + + IEnumerable deliverNow; + + if (requestMessage.ScheduledTime.HasValue) + { + if ((requestMessage.ScheduledTime.Value - DateTime.UtcNow) > TimeSpan.FromMilliseconds(100)) + { + // message was delivered too early. This can happen e.g. if the orchestration service has limits on the delay times for messages. + // We handle this by rescheduling the message instead of processing it. + deliverNow = Array.Empty(); + batch.AddMessageToBeRescheduled(requestMessage); + } + else + { + // the message is scheduled to be delivered immediately. + // There are no FIFO guarantees for scheduled messages, so we skip the message sorter. + deliverNow = new RequestMessage[] { requestMessage }; + } + } + else + { + // run this through the message sorter to help with reordering and duplicate filtering + deliverNow = schedulerState.MessageSorter.ReceiveInOrder(requestMessage, this.entityBackendProperties.EntityMessageReorderWindow); + } + + foreach (var message in deliverNow) + { + if (schedulerState.LockedBy != null && schedulerState.LockedBy == message.ParentInstanceId) + { + if (lockHolderMessages == null) + { + lockHolderMessages = new Queue(); + } + + lockHolderMessages.Enqueue(message); + } + else + { + schedulerState.Enqueue(message); + } + } + } + else if (EntityMessageEventNames.IsReleaseMessage(eventRaisedEvent.Name)) + { + // we are receiving a lock release + var message = new ReleaseMessage(); + try + { + // restore the scheduler state from the input + JsonConvert.PopulateObject(eventRaisedEvent.Input, message, Serializer.InternalSerializerSettings); + } + catch (Exception exception) + { + throw new EntitySchedulerException("Failed to deserialize lock release message - may be corrupted or wrong version.", exception); + } + + if (schedulerState.LockedBy == message.ParentInstanceId) + { + this.logHelper.EntityLockReleased(instanceId, message); + schedulerState.LockedBy = null; + } + } + else + { + // this is a continue message. + // Resumes processing of previously queued operations, if any. + schedulerState.Suspended = false; + } + + break; + } + } + + // lock holder messages go to the front of the queue + if (lockHolderMessages != null) + { + schedulerState.PutBack(lockHolderMessages); + } + + if (!schedulerState.Suspended) + { + // 2. We add as many requests from the queue to the batch as possible, + // stopping at lock requests or when the maximum batch size is reached + while (schedulerState.MayDequeue()) + { + if (batch.OperationCount == this.entityBackendProperties.MaxEntityOperationBatchSize) + { + // we have reached the maximum batch size already + // insert a delay after this batch to ensure write back + batch.ToBeContinued(schedulerState); + break; + } + + var request = schedulerState.Dequeue(); + + if (request.IsLockRequest) + { + batch.AddLockRequest(request); + break; + } + else + { + batch.AddOperation(request); + } + } + } + } + + class Work + { + List operationBatch; // a (possibly empty) sequence of operations to be executed on the entity + RequestMessage lockRequest = null; // zero or one lock request to be executed after all the operations + List toBeRescheduled; // a (possibly empty) list of timed messages that were delivered too early and should be rescheduled + bool suspendAndContinue; // a flag telling as to send ourselves a continue signal + + public int OperationCount => this.operationBatch?.Count ?? 0; + public IReadOnlyList Operations => this.operationBatch; + public IReadOnlyList ToBeRescheduled => this.toBeRescheduled; + public RequestMessage LockRequest => this.lockRequest; + public bool SuspendAndContinue => this.suspendAndContinue; + + public void AddOperation(RequestMessage operationMessage) + { + if (this.operationBatch == null) + { + this.operationBatch = new List(); + } + this.operationBatch.Add(operationMessage); + } + + public void AddLockRequest(RequestMessage lockRequest) + { + Debug.Assert(this.lockRequest == null); + this.lockRequest = lockRequest; + } + + public void AddMessageToBeRescheduled(RequestMessage requestMessage) + { + if (this.toBeRescheduled == null) + { + this.toBeRescheduled = new List(); + } + this.toBeRescheduled.Add(requestMessage); + } + + public void ToBeContinued(SchedulerState schedulerState) + { + if (!schedulerState.Suspended) + { + this.suspendAndContinue = true; + } + } + + public List GetOperationRequests() + { + var operations = new List(this.operationBatch.Count); + for (int i = 0; i < this.operationBatch.Count; i++) + { + var request = this.operationBatch[i]; + operations.Add(new OperationRequest() + { + Operation = request.Operation, + Id = request.Id, + Input = request.Input, + }); + } + return operations; + } + + public Queue RemoveDeferredWork(int index) + { + var deferred = new Queue(); + for (int i = index; i < this.operationBatch.Count; i++) + { + deferred.Enqueue(this.operationBatch[i]); + } + this.operationBatch.RemoveRange(index, this.operationBatch.Count - index); + if (this.lockRequest != null) + { + deferred.Enqueue(this.lockRequest); + this.lockRequest = null; + } + return deferred; + } + } + + #endregion + + #region Send Messages + + void SendResultMessage(WorkItemEffects effects, RequestMessage request, OperationResult result) + { + var destination = new OrchestrationInstance() + { + InstanceId = request.ParentInstanceId, + ExecutionId = request.ParentExecutionId, + }; + var responseMessage = new ResponseMessage() + { + Result = result.Result, + ErrorMessage = result.ErrorMessage, + FailureDetails = result.FailureDetails, + }; + this.ProcessSendEventMessage(effects, destination, EntityMessageEventNames.ResponseMessageEventName(request.Id), responseMessage); + } + + void SendSignalMessage(WorkItemEffects effects, SchedulerState schedulerState, SendSignalOperationAction action) + { + OrchestrationInstance destination = new OrchestrationInstance() + { + InstanceId = action.InstanceId + }; + RequestMessage message = new RequestMessage() + { + ParentInstanceId = effects.InstanceId, + ParentExecutionId = null, // for entities, message sorter persists across executions + Id = Guid.NewGuid(), + IsSignal = true, + Operation = action.Name, + ScheduledTime = action.ScheduledTime, + }; + string eventName; + if (action.ScheduledTime.HasValue) + { + DateTime original = action.ScheduledTime.Value; + DateTime capped = this.entityBackendProperties.GetCappedScheduledTime(DateTime.UtcNow, original); + eventName = EntityMessageEventNames.ScheduledRequestMessageEventName(capped); + } + else + { + eventName = EntityMessageEventNames.RequestMessageEventName; + schedulerState.MessageSorter.LabelOutgoingMessage(message, action.InstanceId, DateTime.UtcNow, this.entityBackendProperties.EntityMessageReorderWindow); + } + this.ProcessSendEventMessage(effects, destination, eventName, message); + } + + void SendLockRequestMessage(WorkItemEffects effects, SchedulerState schedulerState, OrchestrationInstance target, RequestMessage message) + { + schedulerState.MessageSorter.LabelOutgoingMessage(message, target.InstanceId, DateTime.UtcNow, this.entityBackendProperties.EntityMessageReorderWindow); + this.ProcessSendEventMessage(effects, target, EntityMessageEventNames.RequestMessageEventName, message); + } + + void SendLockResponseMessage(WorkItemEffects effects, OrchestrationInstance target, Guid requestId) + { + var message = new ResponseMessage() + { + Result = "Lock Acquisition Completed", // ignored by receiver but shows up in traces + }; + this.ProcessSendEventMessage(effects, target, EntityMessageEventNames.ResponseMessageEventName(requestId), message); + } + + void SendScheduledSelfMessage(WorkItemEffects effects, RequestMessage request) + { + var self = new OrchestrationInstance() + { + InstanceId = effects.InstanceId, + }; + this.ProcessSendEventMessage(effects, self, EntityMessageEventNames.ScheduledRequestMessageEventName(request.ScheduledTime.Value), request); + } + + void SendContinueSelfMessage(WorkItemEffects effects) + { + var self = new OrchestrationInstance() + { + InstanceId = effects.InstanceId, + }; + this.ProcessSendEventMessage(effects, self, EntityMessageEventNames.ContinueMessageEventName, null); + } + + void ProcessSendEventMessage(WorkItemEffects effects, OrchestrationInstance destination, string eventName, object eventContent) + { + string serializedContent = null; + if (eventContent != null) + { + serializedContent = JsonConvert.SerializeObject(eventContent, Serializer.InternalSerializerSettings); + } + + var eventSentEvent = new EventSentEvent(effects.taskIdCounter++) + { + InstanceId = destination.InstanceId, + Name = eventName, + Input = serializedContent, + }; + this.logHelper.RaisingEvent(effects.RuntimeState.OrchestrationInstance, eventSentEvent); + + effects.InstanceMessages.Add(new TaskMessage + { + OrchestrationInstance = destination, + Event = new EventRaisedEvent(-1, serializedContent) + { + Name = eventName, + Input = serializedContent, + }, + }); + } + + void ProcessSendStartMessage(WorkItemEffects effects, OrchestrationRuntimeState runtimeState, StartNewOrchestrationOperationAction action) + { + OrchestrationInstance destination = new OrchestrationInstance() + { + InstanceId = action.InstanceId, + ExecutionId = Guid.NewGuid().ToString("N"), + }; + var executionStartedEvent = new ExecutionStartedEvent(-1, action.Input) + { + Tags = OrchestrationTags.MergeTags(action.Tags, runtimeState.Tags), + OrchestrationInstance = destination, + ParentInstance = new ParentInstance + { + OrchestrationInstance = runtimeState.OrchestrationInstance, + Name = runtimeState.Name, + Version = runtimeState.Version, + TaskScheduleId = effects.taskIdCounter++, + }, + Name = action.Name, + Version = action.Version, + }; + this.logHelper.SchedulingOrchestration(executionStartedEvent); + + effects.InstanceMessages.Add(new TaskMessage + { + OrchestrationInstance = destination, + Event = executionStartedEvent, + }); + } + + #endregion + + async Task ExecuteViaMiddlewareAsync(Work workToDoNow, OrchestrationInstance instance, string serializedEntityState) + { + // the request object that will be passed to the worker + var request = new OperationBatchRequest() + { + InstanceId = instance.InstanceId, + EntityState = serializedEntityState, + Operations = workToDoNow.GetOperationRequests(), + }; + + this.logHelper.EntityBatchExecuting(request); + + var entityId = EntityId.FromString(instance.InstanceId); + string entityName = entityId.Name; + + // Get the TaskEntity implementation. If it's not found, it either means that the developer never + // registered it (which is an error, and we'll throw for this further down) or it could be that some custom + // middleware (e.g. out-of-process execution middleware) is intended to implement the entity logic. + TaskEntity taskEntity = this.objectManager.GetObject(entityName, version: null); + + var dispatchContext = new DispatchMiddlewareContext(); + dispatchContext.SetProperty(request); + + await this.dispatchPipeline.RunAsync(dispatchContext, async _ => + { + // Check to see if the custom middleware intercepted and substituted the orchestration execution + // with its own execution behavior, providing us with the end results. If so, we can terminate + // the dispatch pipeline here. + var resultFromMiddleware = dispatchContext.GetProperty(); + if (resultFromMiddleware != null) + { + return; + } + + if (taskEntity == null) + { + throw TraceHelper.TraceExceptionInstance( + TraceEventType.Error, + "TaskOrchestrationDispatcher-EntityTypeMissing", + instance, + new TypeMissingException($"Entity not found: {entityName}")); + } + + var options = new EntityExecutionOptions() + { + EntityBackendProperties = this.entityBackendProperties, + ErrorPropagationMode = this.errorPropagationMode, + }; + + var result = await taskEntity.ExecuteOperationBatchAsync(request, options); + + dispatchContext.SetProperty(result); + }); + + var result = dispatchContext.GetProperty(); + + this.logHelper.EntityBatchExecuted(request, result); + + return result; + } + } +} \ No newline at end of file diff --git a/src/DurableTask.Core/TaskHubClient.cs b/src/DurableTask.Core/TaskHubClient.cs index fdd27592e..72c5231dd 100644 --- a/src/DurableTask.Core/TaskHubClient.cs +++ b/src/DurableTask.Core/TaskHubClient.cs @@ -32,6 +32,9 @@ public sealed class TaskHubClient readonly DataConverter defaultConverter; readonly LogHelper logHelper; + internal LogHelper LogHelper => this.logHelper; + internal DataConverter DefaultConverter => this.defaultConverter; + /// /// The orchestration service client for this task hub client /// diff --git a/src/DurableTask.Core/TaskHubWorker.cs b/src/DurableTask.Core/TaskHubWorker.cs index 1b93bd62e..e7a6e09b5 100644 --- a/src/DurableTask.Core/TaskHubWorker.cs +++ b/src/DurableTask.Core/TaskHubWorker.cs @@ -21,6 +21,7 @@ namespace DurableTask.Core using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; + using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; using DurableTask.Core.Logging; using DurableTask.Core.Middleware; @@ -34,8 +35,12 @@ public sealed class TaskHubWorker : IDisposable { readonly INameVersionObjectManager activityManager; readonly INameVersionObjectManager orchestrationManager; + readonly INameVersionObjectManager entityManager; + + readonly IEntityOrchestrationService entityOrchestrationService; readonly DispatchMiddlewarePipeline orchestrationDispatchPipeline = new DispatchMiddlewarePipeline(); + readonly DispatchMiddlewarePipeline entityDispatchPipeline = new DispatchMiddlewarePipeline(); readonly DispatchMiddlewarePipeline activityDispatchPipeline = new DispatchMiddlewarePipeline(); readonly SemaphoreSlim slimLock = new SemaphoreSlim(1, 1); @@ -47,10 +52,16 @@ public sealed class TaskHubWorker : IDisposable // ReSharper disable once InconsistentNaming (avoid breaking change) public IOrchestrationService orchestrationService { get; } + /// + /// Indicates whether the configured backend supports entities. + /// + public bool SupportsEntities => this.entityOrchestrationService != null; + volatile bool isStarted; TaskActivityDispatcher activityDispatcher; TaskOrchestrationDispatcher orchestrationDispatcher; + TaskEntityDispatcher entityDispatcher; /// /// Create a new TaskHubWorker with given OrchestrationService @@ -60,7 +71,8 @@ public TaskHubWorker(IOrchestrationService orchestrationService) : this( orchestrationService, new NameVersionObjectManager(), - new NameVersionObjectManager()) + new NameVersionObjectManager(), + new NameVersionObjectManager()) { } @@ -75,6 +87,7 @@ public TaskHubWorker(IOrchestrationService orchestrationService, ILoggerFactory orchestrationService, new NameVersionObjectManager(), new NameVersionObjectManager(), + new NameVersionObjectManager(), loggerFactory) { } @@ -93,11 +106,11 @@ public TaskHubWorker( orchestrationService, orchestrationObjectManager, activityObjectManager, + new NameVersionObjectManager(), loggerFactory: null) { } - /// /// Create a new with given and name version managers /// @@ -110,11 +123,43 @@ public TaskHubWorker( INameVersionObjectManager orchestrationObjectManager, INameVersionObjectManager activityObjectManager, ILoggerFactory loggerFactory = null) + : this( + orchestrationService, + orchestrationObjectManager, + activityObjectManager, + new NameVersionObjectManager(), + loggerFactory: null) + { + } + + /// + /// Create a new TaskHubWorker with given OrchestrationService and name version managers + /// + /// Reference the orchestration service implementation + /// NameVersionObjectManager for Orchestrations + /// NameVersionObjectManager for Activities + /// The NameVersionObjectManager for entities. The version is the entity key. + /// The to use for logging + public TaskHubWorker( + IOrchestrationService orchestrationService, + INameVersionObjectManager orchestrationObjectManager, + INameVersionObjectManager activityObjectManager, + INameVersionObjectManager entityObjectManager, + ILoggerFactory loggerFactory = null) { this.orchestrationManager = orchestrationObjectManager ?? throw new ArgumentException("orchestrationObjectManager"); this.activityManager = activityObjectManager ?? throw new ArgumentException("activityObjectManager"); + this.entityManager = entityObjectManager ?? throw new ArgumentException("entityObjectManager"); this.orchestrationService = orchestrationService ?? throw new ArgumentException("orchestrationService"); this.logHelper = new LogHelper(loggerFactory?.CreateLogger("DurableTask.Core")); + + // If the backend supports a separate work item queue for entities (indicated by it implementing IEntityOrchestrationService), + // we take note of that here, and let the backend know that we are going to pull the work items separately. + if (orchestrationService is IEntityOrchestrationService entityOrchestrationService + && entityOrchestrationService.ProcessEntitiesSeparately()) + { + this.entityOrchestrationService = entityOrchestrationService; + } } /// @@ -153,6 +198,15 @@ public void AddOrchestrationDispatcherMiddleware(Func + /// Adds a middleware delegate to the entity dispatch pipeline. + /// + /// Delegate to invoke whenever a message is dispatched to an entity. + public void AddEntityDispatcherMiddleware(Func, Task> middleware) + { + this.entityDispatchPipeline.Add(middleware ?? throw new ArgumentNullException(nameof(middleware))); + } + /// /// Adds a middleware delegate to the activity dispatch pipeline. /// @@ -184,7 +238,8 @@ public async Task StartAsync() this.orchestrationManager, this.orchestrationDispatchPipeline, this.logHelper, - this.ErrorPropagationMode); + this.ErrorPropagationMode, + this.entityOrchestrationService); this.activityDispatcher = new TaskActivityDispatcher( this.orchestrationService, this.activityManager, @@ -192,10 +247,25 @@ public async Task StartAsync() this.logHelper, this.ErrorPropagationMode); + if (this.SupportsEntities) + { + this.entityDispatcher = new TaskEntityDispatcher( + this.orchestrationService, + this.entityManager, + this.entityDispatchPipeline, + this.logHelper, + this.ErrorPropagationMode); + } + await this.orchestrationService.StartAsync(); await this.orchestrationDispatcher.StartAsync(); await this.activityDispatcher.StartAsync(); + if (this.SupportsEntities) + { + await this.entityDispatcher.StartAsync(); + } + this.logHelper.TaskHubWorkerStarted(sw.Elapsed); this.isStarted = true; } @@ -233,6 +303,7 @@ public async Task StopAsync(bool isForced) { this.orchestrationDispatcher.StopAsync(isForced), this.activityDispatcher.StopAsync(isForced), + this.SupportsEntities ? this.entityDispatcher.StopAsync(isForced) : Task.CompletedTask, }; await Task.WhenAll(dispatcherShutdowns); @@ -282,6 +353,53 @@ public TaskHubWorker AddTaskOrchestrations(params ObjectCreator + /// Loads user defined TaskEntity classes in the TaskHubWorker + /// + /// Types deriving from TaskEntity class + /// + public TaskHubWorker AddTaskEntities(params Type[] taskEntityTypes) + { + if (!this.SupportsEntities) + { + throw new NotSupportedException("The configured backend does not support entities."); + } + + foreach (Type type in taskEntityTypes) + { + ObjectCreator creator = new NameValueObjectCreator( + type.Name, + string.Empty, + type); + + this.entityManager.Add(creator); + } + + return this; + } + + /// + /// Loads user defined TaskEntity classes in the TaskHubWorker + /// + /// + /// User specified ObjectCreators that will + /// create classes deriving TaskEntities with specific names and versions + /// + public TaskHubWorker AddTaskEntities(params ObjectCreator[] taskEntityCreators) + { + if (!this.SupportsEntities) + { + throw new NotSupportedException("The configured backend does not support entities."); + } + + foreach (ObjectCreator creator in taskEntityCreators) + { + this.entityManager.Add(creator); + } + + return this; + } + /// /// Loads user defined TaskActivity objects in the TaskHubWorker /// diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs index a9831ff47..a616d3e4a 100644 --- a/src/DurableTask.Core/TaskOrchestrationContext.cs +++ b/src/DurableTask.Core/TaskOrchestrationContext.cs @@ -21,6 +21,7 @@ namespace DurableTask.Core using System.Threading.Tasks; using DurableTask.Core.Command; using DurableTask.Core.Common; + using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; using DurableTask.Core.History; using DurableTask.Core.Serializing; @@ -47,6 +48,7 @@ public void AddEventToNextIteration(HistoryEvent he) public TaskOrchestrationContext( OrchestrationInstance orchestrationInstance, TaskScheduler taskScheduler, + EntityBackendProperties entityBackendProperties = null, ErrorPropagationMode errorPropagationMode = ErrorPropagationMode.SerializeExceptions) { Utils.UnusedParameter(taskScheduler); @@ -58,6 +60,7 @@ public TaskOrchestrationContext( this.ErrorDataConverter = JsonDataConverter.Default; OrchestrationInstance = orchestrationInstance; IsReplaying = false; + this.EntityBackendProperties = entityBackendProperties; ErrorPropagationMode = errorPropagationMode; this.eventsWhileSuspended = new Queue(); } @@ -416,7 +419,6 @@ public void HandleEventRaisedEvent(EventRaisedEvent eventRaisedEvent, bool skipC } } - public void HandleTaskCompletedEvent(TaskCompletedEvent completedEvent) { int taskId = completedEvent.TaskScheduledId; @@ -497,8 +499,8 @@ public void HandleSubOrchestrationInstanceFailedEvent(SubOrchestrationInstanceFa // When using ErrorPropagationMode.UseFailureDetails we instead use FailureDetails to convey // error information, which doesn't involve any serialization at all. Exception cause = this.ErrorPropagationMode == ErrorPropagationMode.SerializeExceptions ? - Utils.RetrieveCause(failedEvent.Details, this.ErrorDataConverter) : - null; + Utils.RetrieveCause(failedEvent.Details, this.ErrorDataConverter) + : null; var failedException = new SubOrchestrationFailedException(failedEvent.EventId, taskId, info.Name, info.Version, @@ -608,7 +610,7 @@ public void FailOrchestration(Exception failure) details = orchestrationFailureException.Details; } } - else + else { if (this.ErrorPropagationMode == ErrorPropagationMode.UseFailureDetails) { diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index d7e3dcc98..d507fd4af 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -15,13 +15,13 @@ namespace DurableTask.Core { using System; using System.Collections.Generic; - using System.Collections.ObjectModel; using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; using DurableTask.Core.Command; using DurableTask.Core.Common; + using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; using DurableTask.Core.History; using DurableTask.Core.Logging; @@ -43,19 +43,24 @@ public class TaskOrchestrationDispatcher readonly LogHelper logHelper; ErrorPropagationMode errorPropagationMode; readonly NonBlockingCountdownLock concurrentSessionLock; + readonly IEntityOrchestrationService? entityOrchestrationService; + readonly EntityBackendProperties? entityBackendProperties; internal TaskOrchestrationDispatcher( IOrchestrationService orchestrationService, INameVersionObjectManager objectManager, DispatchMiddlewarePipeline dispatchPipeline, LogHelper logHelper, - ErrorPropagationMode errorPropagationMode) + ErrorPropagationMode errorPropagationMode, + IEntityOrchestrationService entityOrchestrationService) { this.objectManager = objectManager ?? throw new ArgumentNullException(nameof(objectManager)); this.orchestrationService = orchestrationService ?? throw new ArgumentNullException(nameof(orchestrationService)); this.dispatchPipeline = dispatchPipeline ?? throw new ArgumentNullException(nameof(dispatchPipeline)); this.logHelper = logHelper ?? throw new ArgumentNullException(nameof(logHelper)); this.errorPropagationMode = errorPropagationMode; + this.entityOrchestrationService = entityOrchestrationService; + this.entityBackendProperties = this.entityOrchestrationService?.GetEntityBackendProperties(); this.dispatcher = new WorkItemDispatcher( "TaskOrchestrationDispatcher", @@ -113,7 +118,18 @@ public async Task StopAsync(bool forced) /// A new TaskOrchestrationWorkItem protected Task OnFetchWorkItemAsync(TimeSpan receiveTimeout, CancellationToken cancellationToken) { - return this.orchestrationService.LockNextTaskOrchestrationWorkItemAsync(receiveTimeout, cancellationToken); + if (this.entityOrchestrationService != null) + { + // only orchestrations should be served by this dispatcher, so we call + // the method which returns work items for orchestrations only. + return this.entityOrchestrationService.LockNextOrchestrationWorkItemAsync(receiveTimeout, cancellationToken); + } + else + { + // both entities and orchestrations are served by this dispatcher, + // so we call the method that may return work items for either. + return this.orchestrationService.LockNextTaskOrchestrationWorkItemAsync(receiveTimeout, cancellationToken); + } } @@ -309,14 +325,14 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work { // start a task to run RenewUntil renewTask = Task.Factory.StartNew( - () => this.RenewUntil(workItem, renewCancellationTokenSource.Token), + () => RenewUntil(workItem, this.orchestrationService, this.logHelper, nameof(TaskOrchestrationDispatcher), renewCancellationTokenSource.Token), renewCancellationTokenSource.Token); } try { // Assumes that: if the batch contains a new "ExecutionStarted" event, it is the first message in the batch. - if (!this.ReconcileMessagesWithState(workItem)) + if (!ReconcileMessagesWithState(workItem, nameof(TaskOrchestrationDispatcher), logHelper)) { // TODO : mark an orchestration as faulted if there is data corruption this.logHelper.DroppingOrchestrationWorkItem(workItem, "Received work-item for an invalid orchestration"); @@ -574,7 +590,6 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work instanceState.Status = runtimeState.Status; } - await this.orchestrationService.CompleteTaskOrchestrationWorkItemAsync( workItem, runtimeState, @@ -597,10 +612,10 @@ static OrchestrationExecutionContext GetOrchestrationExecutionContext(Orchestrat return new OrchestrationExecutionContext { OrchestrationTags = runtimeState.Tags ?? new Dictionary(capacity: 0) }; } - TimeSpan MinRenewalInterval = TimeSpan.FromSeconds(5); // prevents excessive retries if clocks are off - TimeSpan MaxRenewalInterval = TimeSpan.FromSeconds(30); + static TimeSpan MinRenewalInterval = TimeSpan.FromSeconds(5); // prevents excessive retries if clocks are off + static TimeSpan MaxRenewalInterval = TimeSpan.FromSeconds(30); - async Task RenewUntil(TaskOrchestrationWorkItem workItem, CancellationToken cancellationToken) + internal static async Task RenewUntil(TaskOrchestrationWorkItem workItem, IOrchestrationService orchestrationService, LogHelper logHelper, string dispatcher, CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { @@ -623,16 +638,16 @@ async Task RenewUntil(TaskOrchestrationWorkItem workItem, CancellationToken canc try { - this.logHelper.RenewOrchestrationWorkItemStarting(workItem); - TraceHelper.Trace(TraceEventType.Information, "TaskOrchestrationDispatcher-RenewWorkItemStarting", "Renewing work item for instance {0}", workItem.InstanceId); - await this.orchestrationService.RenewTaskOrchestrationWorkItemLockAsync(workItem); - this.logHelper.RenewOrchestrationWorkItemCompleted(workItem); - TraceHelper.Trace(TraceEventType.Information, "TaskOrchestrationDispatcher-RenewWorkItemCompleted", "Successfully renewed work item for instance {0}", workItem.InstanceId); + logHelper.RenewOrchestrationWorkItemStarting(workItem); + TraceHelper.Trace(TraceEventType.Information, $"{dispatcher}-RenewWorkItemStarting", "Renewing work item for instance {0}", workItem.InstanceId); + await orchestrationService.RenewTaskOrchestrationWorkItemLockAsync(workItem); + logHelper.RenewOrchestrationWorkItemCompleted(workItem); + TraceHelper.Trace(TraceEventType.Information, $"{dispatcher}-RenewWorkItemCompleted", "Successfully renewed work item for instance {0}", workItem.InstanceId); } catch (Exception exception) when (!Utils.IsFatal(exception)) { - this.logHelper.RenewOrchestrationWorkItemFailed(workItem, exception); - TraceHelper.TraceException(TraceEventType.Warning, "TaskOrchestrationDispatcher-RenewWorkItemFailed", exception, "Failed to renew work item for instance {0}", workItem.InstanceId); + logHelper.RenewOrchestrationWorkItemFailed(workItem, exception); + TraceHelper.TraceException(TraceEventType.Warning, $"{dispatcher}-RenewWorkItemFailed", exception, "Failed to renew work item for instance {0}", workItem.InstanceId); } } } @@ -677,7 +692,8 @@ await this.dispatchPipeline.RunAsync(dispatchContext, _ => runtimeState, taskOrchestration, this.orchestrationService.EventBehaviourForContinueAsNew, - this.errorPropagationMode); + this.entityBackendProperties, + this.errorPropagationMode); ; OrchestratorExecutionResult resultFromOrchestrator = executor.Execute(); dispatchContext.SetProperty(resultFromOrchestrator); return CompletedTask; @@ -719,8 +735,10 @@ await this.dispatchPipeline.RunAsync(dispatchContext, _ => /// Assumes that: if the batch contains a new "ExecutionStarted" event, it is the first message in the batch. /// /// A batch of work item messages. + /// The name of the dispatcher, used for tracing. + /// The log helper. /// True if workItem should be processed further. False otherwise. - bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem) + internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem, string dispatcher, LogHelper logHelper) { foreach (TaskMessage message in workItem.NewMessages) { @@ -729,7 +747,7 @@ bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem) { throw TraceHelper.TraceException( TraceEventType.Error, - "TaskOrchestrationDispatcher-OrchestrationInstanceMissing", + $"{dispatcher}-OrchestrationInstanceMissing", new InvalidOperationException("Message does not contain any OrchestrationInstance information")); } @@ -747,10 +765,10 @@ bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem) return false; } - this.logHelper.ProcessingOrchestrationMessage(workItem, message); + logHelper.ProcessingOrchestrationMessage(workItem, message); TraceHelper.TraceInstance( TraceEventType.Information, - "TaskOrchestrationDispatcher-ProcessEvent", + $"{dispatcher}-ProcessEvent", orchestrationInstance, "Processing new event with Id {0} and type {1}", message.Event.EventId, @@ -761,10 +779,10 @@ bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem) if (workItem.OrchestrationRuntimeState.ExecutionStartedEvent != null) { // this was caused due to a dupe execution started event, swallow this one - this.logHelper.DroppingOrchestrationMessage(workItem, message, "Duplicate start event"); + logHelper.DroppingOrchestrationMessage(workItem, message, "Duplicate start event"); TraceHelper.TraceInstance( TraceEventType.Warning, - "TaskOrchestrationDispatcher-DuplicateStartEvent", + $"{dispatcher}-DuplicateStartEvent", orchestrationInstance, "Duplicate start event. Ignoring event with Id {0} and type {1} ", message.Event.EventId, @@ -778,13 +796,13 @@ bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workItem) workItem.OrchestrationRuntimeState.OrchestrationInstance?.ExecutionId)) { // eat up any events for previous executions - this.logHelper.DroppingOrchestrationMessage( + logHelper.DroppingOrchestrationMessage( workItem, message, $"ExecutionId of event ({orchestrationInstance.ExecutionId}) does not match current executionId"); TraceHelper.TraceInstance( TraceEventType.Warning, - "TaskOrchestrationDispatcher-ExecutionIdMismatch", + $"{dispatcher}-ExecutionIdMismatch", orchestrationInstance, "ExecutionId of event does not match current executionId. Ignoring event with Id {0} and type {1} ", message.Event.EventId, @@ -1036,7 +1054,7 @@ TaskMessage ProcessSendEventDecision( }; } - class NonBlockingCountdownLock + internal class NonBlockingCountdownLock { int available; diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index b0ca99976..2272fddd4 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -14,7 +14,6 @@ namespace DurableTask.Core { using System; - using System.Collections; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -22,6 +21,7 @@ namespace DurableTask.Core using System.Threading; using System.Threading.Tasks; using DurableTask.Core.Common; + using DurableTask.Core.Entities; using DurableTask.Core.Exceptions; using DurableTask.Core.History; @@ -43,23 +43,44 @@ public class TaskOrchestrationExecutor /// /// /// + /// /// public TaskOrchestrationExecutor( OrchestrationRuntimeState orchestrationRuntimeState, TaskOrchestration taskOrchestration, BehaviorOnContinueAsNew eventBehaviourForContinueAsNew, + EntityBackendProperties? entityBackendProperties, ErrorPropagationMode errorPropagationMode = ErrorPropagationMode.SerializeExceptions) { this.decisionScheduler = new SynchronousTaskScheduler(); this.context = new TaskOrchestrationContext( orchestrationRuntimeState.OrchestrationInstance, this.decisionScheduler, - errorPropagationMode); + entityBackendProperties, + errorPropagationMode + ); this.orchestrationRuntimeState = orchestrationRuntimeState; this.taskOrchestration = taskOrchestration; this.skipCarryOverEvents = eventBehaviourForContinueAsNew == BehaviorOnContinueAsNew.Ignore; } + /// + /// Initializes a new instance of the class. + /// This overload is needed only to avoid breaking changes because this is a public constructor. + /// + /// + /// + /// + /// + public TaskOrchestrationExecutor( + OrchestrationRuntimeState orchestrationRuntimeState, + TaskOrchestration taskOrchestration, + BehaviorOnContinueAsNew eventBehaviourForContinueAsNew, + ErrorPropagationMode errorPropagationMode = ErrorPropagationMode.SerializeExceptions) + : this(orchestrationRuntimeState, taskOrchestration, eventBehaviourForContinueAsNew, null, errorPropagationMode) + { + } + internal bool IsCompleted => this.result != null && (this.result.IsCompleted || this.result.IsFaulted); ///