-
Notifications
You must be signed in to change notification settings - Fork 2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Added Orleans.Streaming.AdoNet project skeleton * Added Streaming Artefacts * Added Streaming Artefacts * Added Orleans.Streaming.AdoNet project skeleton * Added Streaming Artefacts * Added Streaming Artefacts * Added tests * Added tests * Added tests * Added AdoNetQueueDataAdapter * Added adonet streaming artefacts * Added tests * Added tests * Added AdoNetQueueAdapter * Added stream configurators * Refactored * Added AdoNetQueueAdapterFactory * Added AdoNetStreamFailureHandler * Added tests * Added tests * Added tests * Refactored * Added tests * Added tests * Added tests * Added tests * Refactored * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Refactored * Refactored * Refactored * Refactored * Configured defaults * WIP Sweeping * Refactored * Reverted * Refactored * Refactored * Added error handling * Refactored * Wired up options validator * Fixed ArgumentOutOfRangeException thrown from PeriodicTimer * Refactored * Refactored * Refactored * Refactored * Added tests * Cleanup * WIP Scripts * Added MariaDB/MySQL scripts * Added tests * Upgraded Npgsql package with vulnerability warning. Upgraded MySql.Data to fix connection pool deadlock: https://bugs.mysql.com/bug.php?id=114272 Upgraded Microsoft.NET.Test.Sdk to stable release version. * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Added tests * Added safety timeout * Fixed test * Fixed "DROP DATABASE cannot be executed within a pipeline" issue for latest PostgreSQL * Added PostgreSQL scripts * Added tests and refactored * Refactored * Refactored * Fixed argument types * Added check to flaky test * Flaky test * Fixed flaky tests * Clearing pools in MySQL tests * Removed READPAST to address flakiness * Removed SKIP LOCKED to address flakiness * Added benchmark * Added benchmark * Updated benchmark * Updated benchmarks
- Loading branch information
1 parent
71ea69e
commit fd89211
Showing
64 changed files
with
6,499 additions
and
46 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
105 changes: 105 additions & 0 deletions
105
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetBatchContainer.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| namespace Orleans.Streaming.AdoNet; | ||
|
|
||
| /// <summary> | ||
| /// The <see cref="IBatchContainer"/> implementation for the ADONET provider. | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// 1. This class only supports binary serialization as performance and data size is the priority for database storage. | ||
| /// 2. Though the <see cref="SequenceToken"/> is supported here, it is not yet used, as the ADO.NET provider is not rewindable. | ||
| /// </remarks> | ||
| [GenerateSerializer] | ||
| [Alias("Orleans.Streaming.AdoNet.AdoNetBatchContainer")] | ||
| internal class AdoNetBatchContainer : IBatchContainer | ||
| { | ||
| public AdoNetBatchContainer(StreamId streamId, List<object> events, Dictionary<string, object> requestContext) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(events); | ||
|
|
||
| StreamId = streamId; | ||
| Events = events; | ||
| RequestContext = requestContext; | ||
| } | ||
|
|
||
| #region Serialized State | ||
|
|
||
| [Id(0)] | ||
| public StreamId StreamId { get; } | ||
|
|
||
| [Id(1)] | ||
| public List<object> Events { get; } | ||
|
|
||
| [Id(2)] | ||
| public Dictionary<string, object> RequestContext { get; } | ||
|
|
||
| [Id(3)] | ||
| public EventSequenceTokenV2 SequenceToken { get; internal set; } = null!; | ||
|
|
||
| /// <summary> | ||
| /// Holds the receipt for message confirmation. | ||
| /// </summary> | ||
| [Id(4)] | ||
| public int Dequeued { get; internal set; } | ||
|
|
||
| #endregion Serialized State | ||
|
|
||
| #region Interface | ||
|
|
||
| StreamSequenceToken IBatchContainer.SequenceToken => SequenceToken; | ||
|
|
||
| public IEnumerable<Tuple<T, StreamSequenceToken>> GetEvents<T>() | ||
| { | ||
| return SequenceToken is null | ||
| ? throw new InvalidOperationException($"Cannot get events from a half-baked {nameof(AdoNetBatchContainer)}") | ||
| : Events | ||
| .OfType<T>() | ||
| .Select((e, i) => Tuple.Create<T, StreamSequenceToken>(e, SequenceToken.CreateSequenceTokenForEvent(i))); | ||
| } | ||
|
|
||
| public bool ImportRequestContext() | ||
| { | ||
| if (RequestContext is not null) | ||
| { | ||
| RequestContextExtensions.Import(RequestContext); | ||
| return true; | ||
| } | ||
|
|
||
| return false; | ||
| } | ||
|
|
||
| #endregion Interface | ||
|
|
||
| #region Conversion | ||
|
|
||
| /// <summary> | ||
| /// Creates a new <see cref="AdoNetBatchContainer"/> from the specified <see cref="AdoNetStreamMessage"/>. | ||
| /// </summary> | ||
| public static AdoNetBatchContainer FromMessage(Serializer<AdoNetBatchContainer> serializer, AdoNetStreamMessage message) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(serializer); | ||
| ArgumentNullException.ThrowIfNull(message); | ||
|
|
||
| var container = serializer.Deserialize(message.Payload); | ||
| container.SequenceToken = new(message.MessageId); | ||
| container.Dequeued = message.Dequeued; | ||
|
|
||
| return container; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Converts the specified <see cref="AdoNetBatchContainer"/> to a message payload. | ||
| /// </summary> | ||
| public static byte[] ToMessagePayload(Serializer<AdoNetBatchContainer> serializer, StreamId streamId, List<object> events, Dictionary<string, object> requestContext) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(serializer); | ||
| ArgumentNullException.ThrowIfNull(events); | ||
|
|
||
| var container = new AdoNetBatchContainer(streamId, events, requestContext); | ||
| var payload = serializer.SerializeToArray(container); | ||
|
|
||
| return payload; | ||
| } | ||
|
|
||
| #endregion Conversion | ||
|
|
||
| public override string ToString() => $"[{nameof(AdoNetBatchContainer)}:Stream={StreamId},#Items={Events.Count}]"; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| namespace Orleans.Streaming.AdoNet; | ||
|
|
||
| /// <summary> | ||
| /// Stream queue storage adapter for ADO.NET providers. | ||
| /// </summary> | ||
| internal partial class AdoNetQueueAdapter(string name, AdoNetStreamOptions streamOptions, ClusterOptions clusterOptions, SimpleQueueCacheOptions cacheOptions, AdoNetStreamQueueMapper mapper, RelationalOrleansQueries queries, Serializer<AdoNetBatchContainer> serializer, ILogger<AdoNetQueueAdapter> logger, IServiceProvider serviceProvider) : IQueueAdapter | ||
| { | ||
| private readonly ILogger<AdoNetQueueAdapter> _logger = logger; | ||
|
|
||
| /// <summary> | ||
| /// Maps to the ProviderId in the database. | ||
| /// </summary> | ||
| public string Name { get; } = name; | ||
|
|
||
| /// <summary> | ||
| /// The ADO.NET provider is not yet rewindable. | ||
| /// </summary> | ||
| public bool IsRewindable => false; | ||
|
|
||
| /// <summary> | ||
| /// The ADO.NET provider works both ways. | ||
| /// </summary> | ||
| public StreamProviderDirection Direction => StreamProviderDirection.ReadWrite; | ||
|
|
||
| public IQueueAdapterReceiver CreateReceiver(QueueId queueId) | ||
| { | ||
| // map the queue id | ||
| var adoNetQueueId = mapper.GetAdoNetQueueId(queueId); | ||
|
|
||
| // create the receiver | ||
| return ReceiverFactory(serviceProvider, [Name, adoNetQueueId, streamOptions, clusterOptions, cacheOptions, queries]); | ||
| } | ||
|
|
||
| public async Task QueueMessageBatchAsync<T>(StreamId streamId, IEnumerable<T> events, StreamSequenceToken token, Dictionary<string, object> requestContext) | ||
| { | ||
| // the ADO.NET provider is not rewindable so we do not support user supplied tokens | ||
| if (token is not null) | ||
| { | ||
| throw new ArgumentException($"{nameof(AdoNetQueueAdapter)} does not support a user supplied {nameof(StreamSequenceToken)}."); | ||
| } | ||
|
|
||
| // map the Orleans stream id to the corresponding queue id | ||
| var queueId = mapper.GetAdoNetQueueId(streamId); | ||
|
|
||
| // create the payload from the events | ||
| var payload = AdoNetBatchContainer.ToMessagePayload(serializer, streamId, events.Cast<object>().ToList(), requestContext); | ||
|
|
||
| // we can enqueue the message now | ||
| try | ||
| { | ||
| await queries.QueueStreamMessageAsync(clusterOptions.ServiceId, Name, queueId, payload, streamOptions.ExpiryTimeout.TotalSecondsCeiling()); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| LogFailedToQueueStreamMessage(ex, clusterOptions.ServiceId, Name, queueId); | ||
| throw; | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// The receiver factory. | ||
| /// </summary> | ||
| private static readonly ObjectFactory<AdoNetQueueAdapterReceiver> ReceiverFactory = ActivatorUtilities.CreateFactory<AdoNetQueueAdapterReceiver>([typeof(string), typeof(string), typeof(AdoNetStreamOptions), typeof(ClusterOptions), typeof(SimpleQueueCacheOptions), typeof(RelationalOrleansQueries)]); | ||
|
|
||
| #region Logging | ||
|
|
||
| [LoggerMessage(1, LogLevel.Error, "Failed to queue stream message with ({ServiceId}, {ProviderId}, {QueueId})")] | ||
| private partial void LogFailedToQueueStreamMessage(Exception ex, string serviceId, string providerId, string queueId); | ||
|
|
||
| #endregion Logging | ||
| } |
121 changes: 121 additions & 0 deletions
121
src/AdoNet/Orleans.Streaming.AdoNet/AdoNetQueueAdapterFactory.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| using System.Threading; | ||
| using Microsoft.Extensions.Hosting; | ||
|
|
||
| namespace Orleans.Streaming.AdoNet; | ||
|
|
||
| internal class AdoNetQueueAdapterFactory : IQueueAdapterFactory | ||
| { | ||
| public AdoNetQueueAdapterFactory(string name, AdoNetStreamOptions streamOptions, ClusterOptions clusterOptions, SimpleQueueCacheOptions cacheOptions, HashRingStreamQueueMapperOptions hashOptions, ILoggerFactory loggerFactory, IHostApplicationLifetime lifetime, IServiceProvider serviceProvider) | ||
| { | ||
| _name = name; | ||
| _streamOptions = streamOptions; | ||
| _clusterOptions = clusterOptions; | ||
| _cacheOptions = cacheOptions; | ||
| _lifetime = lifetime; | ||
| _serviceProvider = serviceProvider; | ||
|
|
||
| _streamQueueMapper = new HashRingBasedStreamQueueMapper(hashOptions, name); | ||
| _cache = new SimpleQueueAdapterCache(cacheOptions, name, loggerFactory); | ||
| _adoNetQueueMapper = new AdoNetStreamQueueMapper(_streamQueueMapper); | ||
| } | ||
|
|
||
| private readonly string _name; | ||
| private readonly AdoNetStreamOptions _streamOptions; | ||
| private readonly ClusterOptions _clusterOptions; | ||
| private readonly SimpleQueueCacheOptions _cacheOptions; | ||
| private readonly IHostApplicationLifetime _lifetime; | ||
| private readonly IServiceProvider _serviceProvider; | ||
|
|
||
| private readonly HashRingBasedStreamQueueMapper _streamQueueMapper; | ||
| private readonly SimpleQueueAdapterCache _cache; | ||
| private readonly AdoNetStreamQueueMapper _adoNetQueueMapper; | ||
|
|
||
| private RelationalOrleansQueries _queries; | ||
|
|
||
| /// <summary> | ||
| /// Unfortunate implementation detail to account for lack of async lifetime. | ||
| /// Ideally this concern will be moved upstream so this won't be needed. | ||
| /// </summary> | ||
| private readonly SemaphoreSlim _semaphore = new(1); | ||
|
|
||
| /// <summary> | ||
| /// Ensures queries are loaded only once while allowing for recovery if the load fails. | ||
| /// </summary> | ||
| private ValueTask<RelationalOrleansQueries> GetQueriesAsync() | ||
| { | ||
| // attempt fast path | ||
| return _queries is not null ? new(_queries) : new(CoreAsync()); | ||
|
|
||
| // slow path | ||
| async Task<RelationalOrleansQueries> CoreAsync() | ||
| { | ||
| await _semaphore.WaitAsync(_streamOptions.InitializationTimeout, _lifetime.ApplicationStopping); | ||
| try | ||
| { | ||
| // attempt fast path again | ||
| if (_queries is not null) | ||
| { | ||
| return _queries; | ||
| } | ||
|
|
||
| // slow path - the member variable will only be set if the call succeeds | ||
| return _queries = await RelationalOrleansQueries | ||
| .CreateInstance(_streamOptions.Invariant, _streamOptions.ConnectionString) | ||
| .WaitAsync(_streamOptions.InitializationTimeout); | ||
| } | ||
| finally | ||
| { | ||
| _semaphore.Release(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public async Task<IQueueAdapter> CreateAdapter() | ||
| { | ||
| var queries = await GetQueriesAsync(); | ||
|
|
||
| return AdapterFactory(_serviceProvider, [_name, _streamOptions, _clusterOptions, _cacheOptions, _adoNetQueueMapper, queries]); | ||
| } | ||
|
|
||
| public async Task<IStreamFailureHandler> GetDeliveryFailureHandler(QueueId queueId) | ||
| { | ||
| var queries = await GetQueriesAsync(); | ||
|
|
||
| return HandlerFactory(_serviceProvider, [false, _streamOptions, _clusterOptions, _adoNetQueueMapper, queries]); | ||
| } | ||
|
|
||
| public IQueueAdapterCache GetQueueAdapterCache() => _cache; | ||
|
|
||
| public IStreamQueueMapper GetStreamQueueMapper() => _streamQueueMapper; | ||
|
|
||
| /// <summary> | ||
| /// Used by the silo and client configurators as an entry point to set up a stream. | ||
| /// </summary> | ||
| public static IQueueAdapterFactory Create(IServiceProvider serviceProvider, string name) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(serviceProvider); | ||
| ArgumentNullException.ThrowIfNull(name); | ||
|
|
||
| var streamOptions = serviceProvider.GetOptionsByName<AdoNetStreamOptions>(name); | ||
| var clusterOptions = serviceProvider.GetProviderClusterOptions(name).Value; | ||
| var cacheOptions = serviceProvider.GetOptionsByName<SimpleQueueCacheOptions>(name); | ||
| var hashOptions = serviceProvider.GetOptionsByName<HashRingStreamQueueMapperOptions>(name); | ||
|
|
||
| return QueueAdapterFactoryFactory(serviceProvider, [name, streamOptions, clusterOptions, cacheOptions, hashOptions]); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Factory of <see cref="AdoNetQueueAdapterFactory"/> instances. | ||
| /// </summary> | ||
| private static readonly ObjectFactory<AdoNetQueueAdapterFactory> QueueAdapterFactoryFactory = ActivatorUtilities.CreateFactory<AdoNetQueueAdapterFactory>([typeof(string), typeof(AdoNetStreamOptions), typeof(ClusterOptions), typeof(SimpleQueueCacheOptions), typeof(HashRingStreamQueueMapperOptions)]); | ||
|
|
||
| /// <summary> | ||
| /// Factory of <see cref="AdoNetQueueAdapter"/> instances. | ||
| /// </summary> | ||
| private static readonly ObjectFactory<AdoNetQueueAdapter> AdapterFactory = ActivatorUtilities.CreateFactory<AdoNetQueueAdapter>([typeof(string), typeof(AdoNetStreamOptions), typeof(ClusterOptions), typeof(SimpleQueueCacheOptions), typeof(AdoNetStreamQueueMapper), typeof(RelationalOrleansQueries)]); | ||
|
|
||
| /// <summary> | ||
| /// Factory of <see cref="AdoNetStreamFailureHandler"/> instances. | ||
| /// </summary> | ||
| private static readonly ObjectFactory<AdoNetStreamFailureHandler> HandlerFactory = ActivatorUtilities.CreateFactory<AdoNetStreamFailureHandler>([typeof(bool), typeof(AdoNetStreamOptions), typeof(ClusterOptions), typeof(AdoNetStreamQueueMapper), typeof(RelationalOrleansQueries)]); | ||
| } |
Oops, something went wrong.