diff --git a/src/Orleans/Core/Exceptions.cs b/src/Orleans/Core/Exceptions.cs index 08c3538dd36..224f052ef8a 100644 --- a/src/Orleans/Core/Exceptions.cs +++ b/src/Orleans/Core/Exceptions.cs @@ -235,5 +235,26 @@ protected OrleansMessageRejectionException(SerializationInfo info, StreamingCont : base(info, context) { } } + + /// + /// Indicates a lifecycle was canceled, either by request or due to observer error. + /// + [Serializable] + public class OrleansLifecycleCanceledException : OrleansException + { + internal OrleansLifecycleCanceledException(string message) + : base(message) + { + } + + internal OrleansLifecycleCanceledException(string message, + Exception innerException) : base(message, innerException) + { + } + + protected OrleansLifecycleCanceledException(SerializationInfo info, StreamingContext context) + : base(info, context) + { } + } } diff --git a/src/Orleans/Core/Grain.cs b/src/Orleans/Core/Grain.cs index ae3b2521465..722a0a1c886 100644 --- a/src/Orleans/Core/Grain.cs +++ b/src/Orleans/Core/Grain.cs @@ -1,11 +1,13 @@ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; +using System.Threading; using System.Threading.Tasks; using Orleans.Core; using Orleans.Runtime; using Orleans.Storage; using Orleans.Streams; +using System.Diagnostics; namespace Orleans { @@ -268,11 +270,9 @@ private void EnsureRuntime() /// Base class for a Grain with declared persistent state. /// /// The class of the persistent state object - public class Grain : Grain, IStatefulGrain where TGrainState : new() + public class Grain : Grain, ILifecycleParticipant where TGrainState : new() { - private readonly GrainState grainState; - - private IStorage storage; + private IStorage storage; /// /// This constructor should never be invoked. We expose it so that client code (subclasses of this class) do not have to add a constructor. @@ -280,7 +280,6 @@ private void EnsureRuntime() /// protected Grain() { - grainState = new GrainState(); } /// @@ -288,10 +287,9 @@ protected Grain() /// This constructor is particularly useful for unit testing where test code can create a Grain and replace /// the IGrainIdentity, IGrainRuntime and State with test doubles (mocks/stubs). /// - protected Grain(IGrainIdentity identity, IGrainRuntime runtime, TGrainState state, IStorage storage) + protected Grain(IGrainIdentity identity, IGrainRuntime runtime, IStorage storage) : base(identity, runtime) { - grainState = new GrainState(state); this.storage = storage; } @@ -300,18 +298,8 @@ protected Grain(IGrainIdentity identity, IGrainRuntime runtime, TGrainState stat /// protected TGrainState State { - get { return grainState.State; } - set { grainState.State = value; } - } - - void IStatefulGrain.SetStorage(IStorage storage) - { - this.storage = storage; - } - - IGrainState IStatefulGrain.GrainState - { - get { return grainState; } + get { return this.storage.State; } + set { this.storage.State = value; } } /// Clear the current grain state data from backing store. @@ -332,5 +320,32 @@ protected virtual Task ReadStateAsync() { return storage.ReadStateAsync(); } + + public virtual void Participate(ILifecycleObservable lifecycle) + { + lifecycle.Subscribe(GrainLifecycleStage.SetupState, OnSetupState); + } + + private async Task OnSetupState(CancellationToken ct) + { + if (ct.IsCancellationRequested) + return; + IStorageProvider storageProvider = this.GetStorageProvider(this.ServiceProvider); + string grainTypeName = this.GetType().FullName; + this.storage = new StateStorageBridge(grainTypeName, this.GrainReference, storageProvider); + Stopwatch sw = Stopwatch.StartNew(); + try + { + await this.ReadStateAsync(); + sw.Stop(); + StorageStatisticsGroup.OnStorageActivate(grainTypeName, sw.Elapsed); + } + catch (Exception) + { + sw.Stop(); + StorageStatisticsGroup.OnStorageActivateError(grainTypeName); + throw; + } + } } } diff --git a/src/Orleans/Core/GrainAttributes.cs b/src/Orleans/Core/GrainAttributes.cs index 955d047cb76..2bc5cc32683 100644 --- a/src/Orleans/Core/GrainAttributes.cs +++ b/src/Orleans/Core/GrainAttributes.cs @@ -347,7 +347,7 @@ public StorageProviderAttribute() /// The [Orleans.Providers.LogConsistencyProvider] attribute is used to define which consistency provider to use for grains using the log-view state abstraction. /// /// Specifying [Orleans.Providers.LogConsistencyProvider] property is recommended for all grains that derive - /// from ILogConsistentGrain, such as JournaledGrain. + /// from LogConsistentGrain, such as JournaledGrain. /// If no [Orleans.Providers.LogConsistencyProvider] attribute is specified, then the runtime tries to locate /// one as follows. First, it looks for a /// "Default" provider in the configuration file, then it checks if the grain type defines a default. diff --git a/src/Orleans/Core/GrainStateStorageBridge.cs b/src/Orleans/Core/GrainStateStorageBridge.cs deleted file mode 100644 index 476720c96c2..00000000000 --- a/src/Orleans/Core/GrainStateStorageBridge.cs +++ /dev/null @@ -1,184 +0,0 @@ -// #define REREAD_STATE_AFTER_WRITE_FAILED - -using System; -using System.Diagnostics; -using System.Net; -using System.Threading.Tasks; -using Orleans.Runtime; -using Orleans.Storage; - -namespace Orleans.Core -{ - internal class GrainStateStorageBridge : IStorage - { - private readonly IStorageProvider store; - private IStatefulGrain statefulGrain; - private Grain baseGrain; - private readonly string grainTypeName; - - public GrainStateStorageBridge(string grainTypeName, IStorageProvider store) - { - if (grainTypeName == null) - { - throw new ArgumentNullException("grainTypeName", "No grain type name supplied"); - } - if (store == null) - { - throw new ArgumentNullException("store", "No storage provider supplied"); - } - - this.grainTypeName = grainTypeName; - this.store = store; - } - - internal void SetGrain(Grain grain) - { - if(grain == null) - throw new ArgumentNullException(nameof(grain)); - - statefulGrain = grain as IStatefulGrain; - - if(statefulGrain == null) - throw new ArgumentException("Attempt to configure storage bridge for a non-perisstent grain.", nameof(grain)); - - if (statefulGrain.GrainState == null) - throw new ArgumentException("No grain state object supplied", nameof(grain)); - - this.baseGrain = grain; - } - - /// - /// Async method to cause refresh of the current grain state data from backing store. - /// Any previous contents of the grain state data will be overwritten. - /// - public async Task ReadStateAsync() - { - const string what = "ReadState"; - Stopwatch sw = Stopwatch.StartNew(); - GrainReference grainRef = baseGrain.GrainReference; - try - { - await store.ReadStateAsync(grainTypeName, grainRef, statefulGrain.GrainState); - - StorageStatisticsGroup.OnStorageRead(store, grainTypeName, grainRef, sw.Elapsed); - } - catch (Exception exc) - { - StorageStatisticsGroup.OnStorageReadError(store, grainTypeName, grainRef); - - string errMsg = MakeErrorMsg(what, exc); - store.Log.Error((int) ErrorCode.StorageProvider_ReadFailed, errMsg, exc); - throw new OrleansException(errMsg, exc); - } - finally - { - sw.Stop(); - } - } - - /// - /// Async method to cause write of the current grain state data into backing store. - /// - public async Task WriteStateAsync() - { - const string what = "WriteState"; - Stopwatch sw = Stopwatch.StartNew(); - GrainReference grainRef = baseGrain.GrainReference; - Exception errorOccurred; - try - { - await store.WriteStateAsync(grainTypeName, grainRef, statefulGrain.GrainState); - StorageStatisticsGroup.OnStorageWrite(store, grainTypeName, grainRef, sw.Elapsed); - errorOccurred = null; - } - catch (Exception exc) - { - errorOccurred = exc; - } - // Note, we can't do this inside catch block above, because await is not permitted there. - if (errorOccurred != null) - { - StorageStatisticsGroup.OnStorageWriteError(store, grainTypeName, grainRef); - - string errMsgToLog = MakeErrorMsg(what, errorOccurred); - store.Log.Error((int) ErrorCode.StorageProvider_WriteFailed, errMsgToLog, errorOccurred); - // If error is not specialization of OrleansException, wrap it - if (!(errorOccurred is OrleansException)) - { - errorOccurred = new OrleansException(errMsgToLog, errorOccurred); - } - -#if REREAD_STATE_AFTER_WRITE_FAILED - // Force rollback to previously stored state - try - { - sw.Restart(); - store.Log.Warn(ErrorCode.StorageProvider_ForceReRead, "Forcing re-read of last good state for grain Type={0}", grainTypeName); - await store.ReadStateAsync(grainTypeName, grainRef, grain.GrainState); - StorageStatisticsGroup.OnStorageRead(store, grainTypeName, grainRef, sw.Elapsed); - } - catch (Exception exc) - { - StorageStatisticsGroup.OnStorageReadError(store, grainTypeName, grainRef); - - // Should we ignore this secondary error, and just return the original one? - errMsgToLog = MakeErrorMsg("re-read state from store after write error", exc); - errorOccurred = new OrleansException(errMsgToLog, exc); - } -#endif - } - sw.Stop(); - if (errorOccurred != null) - { - throw errorOccurred; - } - } - - /// - /// Async method to cause write of the current grain state data into backing store. - /// - public async Task ClearStateAsync() - { - const string what = "ClearState"; - Stopwatch sw = Stopwatch.StartNew(); - GrainReference grainRef = baseGrain.GrainReference; - try - { - // Clear (most likely Delete) state from external storage - await store.ClearStateAsync(grainTypeName, grainRef, statefulGrain.GrainState); - - // Reset the in-memory copy of the state - statefulGrain.GrainState.State = Activator.CreateInstance(statefulGrain.GrainState.State.GetType()); - - // Update counters - StorageStatisticsGroup.OnStorageDelete(store, grainTypeName, grainRef, sw.Elapsed); - } - catch (Exception exc) - { - StorageStatisticsGroup.OnStorageDeleteError(store, grainTypeName, grainRef); - - string errMsg = MakeErrorMsg(what, exc); - store.Log.Error((int) ErrorCode.StorageProvider_DeleteFailed, errMsg, exc); - throw new OrleansException(errMsg, exc); - } - finally - { - sw.Stop(); - } - } - - private string MakeErrorMsg(string what, Exception exc) - { - var httpStatusCode = HttpStatusCode.Unused; - string errorCode = String.Empty; - - var decoder = store as IRestExceptionDecoder; - if(decoder != null) - decoder.DecodeException(exc, out httpStatusCode, out errorCode, true); - - GrainReference grainReference = baseGrain.GrainReference; - return string.Format("Error from storage provider during {0} for grain Type={1} Pk={2} Id={3} Error={4}" + Environment.NewLine + " {5}", - what, grainTypeName, grainReference.GrainId.ToDetailedString(), grainReference, errorCode, LogFormatter.PrintException(exc)); - } - } -} diff --git a/src/Orleans/Core/IStatefulGrain.cs b/src/Orleans/Core/IStatefulGrain.cs deleted file mode 100644 index dc4aca654d3..00000000000 --- a/src/Orleans/Core/IStatefulGrain.cs +++ /dev/null @@ -1,11 +0,0 @@ -using Orleans.Core; - -namespace Orleans -{ - internal interface IStatefulGrain - { - IGrainState GrainState { get; } - - void SetStorage(IStorage storage); - } -} diff --git a/src/Orleans/Core/IStorage.cs b/src/Orleans/Core/IStorage.cs index 40373d396e9..e5c3b7e8ebd 100644 --- a/src/Orleans/Core/IStorage.cs +++ b/src/Orleans/Core/IStorage.cs @@ -2,8 +2,11 @@ namespace Orleans.Core { - public interface IStorage + public interface IStorage + where TState : new() { + TState State { get; set; } + /// /// Async method to cause the current grain state data to be cleared and reset. /// This will usually mean the state record is deleted from backing store, but the specific behavior is defined by the storage provider instance configured for this grain. diff --git a/src/Orleans/Core/StateStorageBridge.cs b/src/Orleans/Core/StateStorageBridge.cs new file mode 100644 index 00000000000..350dee2d36d --- /dev/null +++ b/src/Orleans/Core/StateStorageBridge.cs @@ -0,0 +1,140 @@ +using System; +using System.Diagnostics; +using System.Net; +using System.Threading.Tasks; +using Orleans.Runtime; +using Orleans.Storage; + +namespace Orleans.Core +{ + internal class StateStorageBridge : IStorage + where TState : new() + { + private readonly string name; + private readonly GrainReference grainRef; + private readonly IStorageProvider store; + private readonly GrainState grainState; + + public TState State + { + get { return grainState.State; } + set { grainState.State = value; } + } + + public StateStorageBridge(string name, GrainReference grainRef, IStorageProvider store) + { + if (name == null) throw new ArgumentNullException(nameof(name)); + if (grainRef == null) throw new ArgumentNullException(nameof(grainRef)); + if (store == null) throw new ArgumentNullException(nameof(store)); + + this.name = name; + this.grainRef = grainRef; + this.store = store; + this.grainState = new GrainState(new TState()); + } + + /// + /// Async method to cause refresh of the current grain state data from backing store. + /// Any previous contents of the grain state data will be overwritten. + /// + public async Task ReadStateAsync() + { + const string what = "ReadState"; + Stopwatch sw = Stopwatch.StartNew(); + try + { + await store.ReadStateAsync(name, grainRef, grainState); + + StorageStatisticsGroup.OnStorageRead(store, name, grainRef, sw.Elapsed); + } + catch (Exception exc) + { + StorageStatisticsGroup.OnStorageReadError(store, name, grainRef); + + string errMsg = MakeErrorMsg(what, exc); + store.Log.Error((int)ErrorCode.StorageProvider_ReadFailed, errMsg, exc); + if (!(exc is OrleansException)) + { + throw new OrleansException(errMsg, exc); + } + throw; + } + finally + { + sw.Stop(); + } + } + + /// + /// Async method to cause write of the current grain state data into backing store. + /// + public async Task WriteStateAsync() + { + const string what = "WriteState"; + try + { + Stopwatch sw = Stopwatch.StartNew(); + await store.WriteStateAsync(name, grainRef, grainState); + sw.Stop(); + StorageStatisticsGroup.OnStorageWrite(store, name, grainRef, sw.Elapsed); + } + catch (Exception exc) + { + StorageStatisticsGroup.OnStorageWriteError(store, name, grainRef); + string errMsgToLog = MakeErrorMsg(what, exc); + store.Log.Error((int)ErrorCode.StorageProvider_WriteFailed, errMsgToLog, exc); + // If error is not specialization of OrleansException, wrap it + if (!(exc is OrleansException)) + { + throw new OrleansException(errMsgToLog, exc); + } + throw; + } + } + + /// + /// Async method to cause write of the current grain state data into backing store. + /// + public async Task ClearStateAsync() + { + const string what = "ClearState"; + try + { + Stopwatch sw = Stopwatch.StartNew(); + // Clear (most likely Delete) state from external storage + await store.ClearStateAsync(name, grainRef, grainState); + sw.Stop(); + + // Reset the in-memory copy of the state + grainState.State = new TState(); + + // Update counters + StorageStatisticsGroup.OnStorageDelete(store, name, grainRef, sw.Elapsed); + } + catch (Exception exc) + { + StorageStatisticsGroup.OnStorageDeleteError(store, name, grainRef); + + string errMsg = MakeErrorMsg(what, exc); + store.Log.Error((int)ErrorCode.StorageProvider_DeleteFailed, errMsg, exc); + if (!(exc is OrleansException)) + { + throw new OrleansException(errMsg, exc); + } + throw; + } + } + + private string MakeErrorMsg(string what, Exception exc) + { + HttpStatusCode httpStatusCode; + string errorCode = string.Empty; + + var decoder = store as IRestExceptionDecoder; + decoder?.DecodeException(exc, out httpStatusCode, out errorCode, true); + + return string.Format("Error from storage provider during {0} for grain Type={1} Pk={2} Id={3} Error={4}" + Environment.NewLine + " {5}", + what, name, grainRef.GrainId.ToDetailedString(), grainRef, errorCode, LogFormatter.PrintException(exc)); + } + } +} diff --git a/src/Orleans/Lifecycle/LifecycleObservable.cs b/src/Orleans/Lifecycle/LifecycleObservable.cs index b5c15fb90c3..ab9558f54ac 100644 --- a/src/Orleans/Lifecycle/LifecycleObservable.cs +++ b/src/Orleans/Lifecycle/LifecycleObservable.cs @@ -29,7 +29,7 @@ public async Task OnStart(CancellationToken ct) { if (ct.IsCancellationRequested) { - throw new OperationCanceledException(); + throw new OrleansLifecycleCanceledException("Lifecycle start canceled by request"); } this.highStage = observerGroup.Key; await Task.WhenAll(observerGroup.Select(orderedObserver => WrapExecution(ct, orderedObserver.Observer.OnStart))); @@ -39,7 +39,7 @@ public async Task OnStart(CancellationToken ct) { string error = $"Lifecycle start canceled due to errors at stage {this.highStage}"; this.logger?.Error(ErrorCode.LifecycleStartFailure, error, ex); - throw new OperationCanceledException(error, ex); + throw new OrleansLifecycleCanceledException(error, ex); } } @@ -56,7 +56,7 @@ public async Task OnStop(CancellationToken ct) { if (ct.IsCancellationRequested) { - throw new OperationCanceledException(); + throw new OrleansLifecycleCanceledException("Lifecycle stop canceled by request"); } await Task.WhenAll(observerGroup.Select(orderedObserver => WrapExecution(ct, orderedObserver.Observer.OnStop))); } diff --git a/src/Orleans/LogConsistency/ILogConsistencyDiagnostics.cs b/src/Orleans/LogConsistency/ILogConsistencyDiagnostics.cs index 080c09d8831..1ae2ae88385 100644 --- a/src/Orleans/LogConsistency/ILogConsistencyDiagnostics.cs +++ b/src/Orleans/LogConsistency/ILogConsistencyDiagnostics.cs @@ -27,7 +27,7 @@ public interface ILogConsistencyDiagnostics } /// - /// A collection of statistics for grains using log-consistency. See + /// A collection of statistics for grains using log-consistency. See /// public class LogConsistencyStatistics { diff --git a/src/Orleans/LogConsistency/ILogConsistentGrain.cs b/src/Orleans/LogConsistency/ILogConsistentGrain.cs deleted file mode 100644 index 09d3e508953..00000000000 --- a/src/Orleans/LogConsistency/ILogConsistentGrain.cs +++ /dev/null @@ -1,46 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Text; -using System.Threading.Tasks; -using Orleans.Concurrency; -using Orleans.Runtime; -using Orleans.Storage; - -namespace Orleans.LogConsistency -{ - /// - /// This interface encapsulates functionality of grains that manage their state - /// based on log consistency, such as JournaledGrain. - /// It is the equivalent of for log-consistent grains. - /// - public interface ILogConsistentGrain - { - /// - /// called right after grain construction to install the log view adaptor - /// - /// The adaptor factory to use - /// The initial state of the view - /// The type name of the grain - /// The storage provider, if needed - /// Protocol services - void InstallAdaptor(ILogViewAdaptorFactory factory, object state, string grainTypeName, IStorageProvider storageProvider, ILogConsistencyProtocolServices services); - - /// - /// Gets the default adaptor factory to use, or null if there is no default - /// (in which case user MUST configure a consistency provider) - /// - ILogViewAdaptorFactory DefaultAdaptorFactory { get; } - } - - - /// - /// Base class for all grains that use log-consistency for managing the state. - /// It is the equivalent of for grains using log-consistency. - /// (SiloAssemblyLoader uses it to extract type) - /// - /// The type of the view - public class LogConsistentGrainBase : Grain - { - } -} diff --git a/src/Orleans/LogConsistency/ILogViewAdaptor.cs b/src/Orleans/LogConsistency/ILogViewAdaptor.cs index eded6b94c62..c1e369d8832 100644 --- a/src/Orleans/LogConsistency/ILogViewAdaptor.cs +++ b/src/Orleans/LogConsistency/ILogViewAdaptor.cs @@ -8,7 +8,7 @@ namespace Orleans.LogConsistency { /// - /// A log view adaptor is the storage interface for , whose state is defined as a log view. + /// A log view adaptor is the storage interface for , whose state is defined as a log view. /// /// There is one adaptor per grain, which is installed by when the grain is activated. /// diff --git a/src/Orleans/LogConsistency/LogConsistentGrain.cs b/src/Orleans/LogConsistency/LogConsistentGrain.cs new file mode 100644 index 00000000000..895e1c76bff --- /dev/null +++ b/src/Orleans/LogConsistency/LogConsistentGrain.cs @@ -0,0 +1,96 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Orleans.Runtime; +using Orleans.Storage; +using Microsoft.Extensions.DependencyInjection; +using Orleans.GrainDirectory; +using System.Reflection; +using Orleans.Providers; + +namespace Orleans.LogConsistency +{ + /// + /// Base class for all grains that use log-consistency for managing the state. + /// It is the equivalent of for grains using log-consistency. + /// (SiloAssemblyLoader uses it to extract type) + /// + /// The type of the view + public abstract class LogConsistentGrain : Grain, ILifecycleParticipant + + { + /// + /// called right after grain construction to install the log view adaptor + /// + /// The adaptor factory to use + /// The initial state of the view + /// The type name of the grain + /// The storage provider, if needed + /// Protocol services + protected abstract void InstallAdaptor(ILogViewAdaptorFactory factory, object state, string grainTypeName, IStorageProvider storageProvider, ILogConsistencyProtocolServices services); + + /// + /// Gets the default adaptor factory to use, or null if there is no default + /// (in which case user MUST configure a consistency provider) + /// + protected abstract ILogViewAdaptorFactory DefaultAdaptorFactory { get; } + + public virtual void Participate(ILifecycleObservable lifecycle) + { + lifecycle.Subscribe(GrainLifecycleStage.SetupState, OnSetupState); + } + + private Task OnSetupState(CancellationToken ct) + { + if (ct.IsCancellationRequested) return Task.CompletedTask; + IGrainActivationContext activationContext = this.ServiceProvider.GetRequiredService(); + Factory protocolServicesFactory = this.ServiceProvider.GetRequiredService>(); + ILogViewAdaptorFactory consistencyProvider = SetupLogConsistencyProvider(activationContext); + IStorageProvider storageProvider = consistencyProvider.UsesStorageProvider ? this.GetStorageProvider(this.ServiceProvider) : null; + InstallLogViewAdaptor(activationContext.RegistrationStrategy, protocolServicesFactory, consistencyProvider, storageProvider); + return Task.CompletedTask; + } + + private void InstallLogViewAdaptor( + IMultiClusterRegistrationStrategy mcRegistrationStrategy, + Factory protocolServicesFactory, + ILogViewAdaptorFactory factory, + IStorageProvider storageProvider) + { + // encapsulate runtime services used by consistency adaptors + ILogConsistencyProtocolServices svc = protocolServicesFactory(this, mcRegistrationStrategy); + + TView state = (TView)Activator.CreateInstance(typeof(TView)); + + this.InstallAdaptor(factory, state, this.GetType().FullName, storageProvider, svc); + } + + + private ILogViewAdaptorFactory SetupLogConsistencyProvider(IGrainActivationContext activationContext) + { + var attr = this.GetType().GetTypeInfo().GetCustomAttributes(true).FirstOrDefault(); + + ILogViewAdaptorFactory defaultFactory = attr != null + ? this.ServiceProvider.GetServiceByName(attr.ProviderName) + : this.ServiceProvider.GetService(); + if (attr != null && defaultFactory == null) + { + var errMsg = attr != null + ? $"Cannot find consistency provider with Name={attr.ProviderName} for grain type {this.GetType().FullName}" + : $"No consistency provider manager found loading grain type {this.GetType().FullName}"; + throw new BadProviderConfigException(errMsg); + } + + // use default if none found + defaultFactory = defaultFactory ?? this.DefaultAdaptorFactory; + if (defaultFactory == null) + { + var errMsg = $"No log consistency provider found loading grain type {this.GetType().FullName}"; + throw new BadProviderConfigException(errMsg); + }; + + return defaultFactory; + } + } +} diff --git a/src/Orleans/Providers/GrainStorageExtensions.cs b/src/Orleans/Providers/GrainStorageExtensions.cs new file mode 100644 index 00000000000..ecaee73d45f --- /dev/null +++ b/src/Orleans/Providers/GrainStorageExtensions.cs @@ -0,0 +1,30 @@ +using Orleans.Providers; +using Orleans.Runtime; +using System; +using System.Reflection; +using Microsoft.Extensions.DependencyInjection; +using System.Linq; + +namespace Orleans.Storage +{ + public static class GrainStorageExtensions + { + /// + /// Aquire the storage provider associated with the grain type. + /// + /// + public static IStorageProvider GetStorageProvider(this Grain grain, IServiceProvider services) + { + StorageProviderAttribute attr = grain.GetType().GetTypeInfo().GetCustomAttributes(true).FirstOrDefault(); + IStorageProvider storageProvider = attr != null + ? services.GetServiceByName(attr.ProviderName) + : services.GetService(); + if (storageProvider == null) + { + var errMsg = string.Format("No storage providers found loading grain type {0}", grain.GetType().FullName); + throw new BadProviderConfigException(errMsg); + } + return storageProvider; + } + } +} diff --git a/src/Orleans/Runtime/IGrainActivationContext.cs b/src/Orleans/Runtime/IGrainActivationContext.cs index 44796a3fd70..89cf42752a1 100644 --- a/src/Orleans/Runtime/IGrainActivationContext.cs +++ b/src/Orleans/Runtime/IGrainActivationContext.cs @@ -1,6 +1,7 @@ using System; using Orleans.Core; using System.Collections.Generic; +using Orleans.GrainDirectory; namespace Orleans.Runtime { @@ -29,5 +30,11 @@ public interface IGrainActivationContext /// Observable Grain life cycle /// IGrainLifecycle ObservableLifecycle { get; } + + /// + /// Multi-cluster registration strategy for this grain activation. + /// Used by protocols that coordinate multiple instances. + /// + IMultiClusterRegistrationStrategy RegistrationStrategy { get; } } } \ No newline at end of file diff --git a/src/Orleans/Runtime/IGrainLifeCycle.cs b/src/Orleans/Runtime/IGrainLifeCycle.cs index 0a26466da28..b2adc21194b 100644 --- a/src/Orleans/Runtime/IGrainLifeCycle.cs +++ b/src/Orleans/Runtime/IGrainLifeCycle.cs @@ -8,7 +8,7 @@ namespace Orleans.Runtime /// stream cleanup should all eventually be triggered by the /// grain lifecycle. /// - public enum GrainLifecyleStage + public enum GrainLifecycleStage { //None, //Register, @@ -20,7 +20,7 @@ public enum GrainLifecyleStage /// /// Grain life cycle /// - public interface IGrainLifecycle : ILifecycleObservable + public interface IGrainLifecycle : ILifecycleObservable { } } diff --git a/src/Orleans/Statistics/StorageStatisticsGroup.cs b/src/Orleans/Statistics/StorageStatisticsGroup.cs index 90aa6f2c650..88752f325b0 100644 --- a/src/Orleans/Statistics/StorageStatisticsGroup.cs +++ b/src/Orleans/Statistics/StorageStatisticsGroup.cs @@ -48,7 +48,7 @@ internal static void OnStorageWrite(IStorageProvider storage, string grainType, StorageWriteLatency.AddSample(latency); } } - internal static void OnStorageActivate(IStorageProvider storage, string grainType, GrainReference grain, TimeSpan latency) + internal static void OnStorageActivate(string grainType, TimeSpan latency) { StorageActivateTotal.Increment(); if (latency > TimeSpan.Zero) @@ -64,7 +64,7 @@ internal static void OnStorageWriteError(IStorageProvider storage, string grainT { StorageWriteErrors.Increment(); } - internal static void OnStorageActivateError(IStorageProvider storage, string grainType, GrainReference grain) + internal static void OnStorageActivateError(string grainType) { StorageActivateErrors.Increment(); } diff --git a/src/OrleansEventSourcing/JournaledGrain.cs b/src/OrleansEventSourcing/JournaledGrain.cs index 8188906d8f7..c952f1e5e25 100644 --- a/src/OrleansEventSourcing/JournaledGrain.cs +++ b/src/OrleansEventSourcing/JournaledGrain.cs @@ -2,7 +2,6 @@ using Orleans.MultiCluster; using Orleans.LogConsistency; using System; -using System.Linq; using System.Collections.Generic; using System.Threading.Tasks; using Orleans.Storage; @@ -26,8 +25,7 @@ public abstract class JournaledGrain : JournaledGrainThe common base class for the events /// public abstract class JournaledGrain : - LogConsistentGrainBase, - ILogConsistentGrain, + LogConsistentGrain, ILogConsistencyProtocolParticipant, ILogViewAdaptorHost where TGrainState : class, new() @@ -274,7 +272,7 @@ protected virtual void TransitionState(TGrainState state, TEventBase @event) /// Called right after grain is constructed, to install the adaptor. /// The log-consistency provider contains a factory method that constructs the adaptor with chosen types for this grain /// - void ILogConsistentGrain.InstallAdaptor(ILogViewAdaptorFactory factory, object initialState, string graintypename, IStorageProvider storageProvider, ILogConsistencyProtocolServices services) + protected override void InstallAdaptor(ILogViewAdaptorFactory factory, object initialState, string graintypename, IStorageProvider storageProvider, ILogConsistencyProtocolServices services) { // call the log consistency provider to construct the adaptor, passing the type argument LogViewAdaptor = factory.MakeLogViewAdaptor(this, (TGrainState)initialState, graintypename, storageProvider, services); @@ -283,7 +281,7 @@ void ILogConsistentGrain.InstallAdaptor(ILogViewAdaptorFactory factory, object i /// /// If there is no log-consistency provider specified, store versioned state using default storage provider /// - ILogViewAdaptorFactory ILogConsistentGrain.DefaultAdaptorFactory + protected override ILogViewAdaptorFactory DefaultAdaptorFactory { get { @@ -370,7 +368,6 @@ void IConnectionIssueListener.OnConnectionIssueResolved(ConnectionIssue connecti OnConnectionIssueResolved(connectionIssue); } - #endregion } diff --git a/src/OrleansRuntime/Catalog/ActivationData.cs b/src/OrleansRuntime/Catalog/ActivationData.cs index 66ee53fddd0..8f09d1e5b6b 100644 --- a/src/OrleansRuntime/Catalog/ActivationData.cs +++ b/src/OrleansRuntime/Catalog/ActivationData.cs @@ -177,7 +177,7 @@ public ActivationData( ActivationAddress addr, string genericArguments, PlacementStrategy placedUsing, - MultiClusterRegistrationStrategy registrationStrategy, + IMultiClusterRegistrationStrategy registrationStrategy, IActivationCollector collector, TimeSpan ageLimit, NodeConfiguration nodeConfiguration, @@ -313,8 +313,6 @@ private static void SetGrainActivationContextInScopedServices(IServiceProvider s contextFactory.Context = context; } - public IStorageProvider StorageProvider { get; set; } - private Streams.StreamDirectory streamDirectory; internal Streams.StreamDirectory GetStreamDirectory() { @@ -454,7 +452,7 @@ public void SetCollectionTicket(DateTime ticket) public PlacementStrategy PlacedUsing { get; private set; } - public MultiClusterRegistrationStrategy RegistrationStrategy { get; private set; } + public IMultiClusterRegistrationStrategy RegistrationStrategy { get; private set; } // Currently, the only supported multi-activation grain is one using the StatelessWorkerPlacement strategy. internal bool IsStatelessWorker { get { return PlacedUsing is StatelessWorkerPlacement; } } diff --git a/src/OrleansRuntime/Catalog/Catalog.cs b/src/OrleansRuntime/Catalog/Catalog.cs index 1dc610224c0..9e0ab344cd4 100644 --- a/src/OrleansRuntime/Catalog/Catalog.cs +++ b/src/OrleansRuntime/Catalog/Catalog.cs @@ -1,5 +1,4 @@ using System; -using System.Collections; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -8,11 +7,8 @@ using System.Threading; using System.Threading.Tasks; using Orleans.CodeGeneration; -using Orleans.Core; using Orleans.GrainDirectory; using Orleans.MultiCluster; -using Orleans.Providers; -using Orleans.LogConsistency; using Orleans.Runtime.Configuration; using Orleans.Runtime.GrainDirectory; using Orleans.Runtime.Messaging; @@ -20,10 +16,9 @@ using Orleans.Runtime.Scheduler; using Orleans.Runtime.Versions; using Orleans.Serialization; -using Orleans.Storage; using Orleans.Streams.Core; using Orleans.Streams; -using Orleans.Versions.Compatibility; +using System.Runtime.ExceptionServices; namespace Orleans.Runtime { @@ -80,8 +75,6 @@ public override void GetObjectData(SerializationInfo info, StreamingContext cont private readonly ILocalGrainDirectory directory; private readonly OrleansTaskScheduler scheduler; private readonly ActivationDirectory activations; - private IStorageProviderManager storageProviderManager; - private ILogConsistencyProviderManager logConsistencyProviderManager; private IStreamProviderRuntime providerRuntime; private IStreamProviderManager providerManager; private IServiceProvider serviceProvider; @@ -101,7 +94,6 @@ public override void GetObjectData(SerializationInfo info, StreamingContext cont private readonly TimeSpan maxRequestProcessingTime; private readonly TimeSpan maxWarningRequestProcessingTime; private readonly SerializationManager serializationManager; - private readonly MultiClusterRegistrationStrategyManager multiClusterRegistrationStrategyManager; private readonly CachedVersionSelectorManager versionSelectorManager; public Catalog( @@ -117,7 +109,6 @@ public Catalog( PlacementDirectorsManager placementDirectorsManager, MessageFactory messageFactory, SerializationManager serializationManager, - MultiClusterRegistrationStrategyManager multiClusterRegistrationStrategyManager, IStreamProviderRuntime providerRuntime, IStreamProviderManager providerManager, IServiceProvider serviceProvider, @@ -135,7 +126,6 @@ public Catalog( this.grainCreator = grainCreator; this.nodeConfig = nodeConfig; this.serializationManager = serializationManager; - this.multiClusterRegistrationStrategyManager = multiClusterRegistrationStrategyManager; this.versionSelectorManager = versionSelectorManager; this.providerRuntime = providerRuntime; this.serviceProvider = serviceProvider; @@ -207,16 +197,6 @@ public IReadOnlyDictionary> GetCompatibleSilo return silos; } - internal void SetStorageManager(IStorageProviderManager storageManager) - { - storageProviderManager = storageManager; - } - - internal void SetLogConsistencyManager(ILogConsistencyProviderManager logConsistencyManager) - { - logConsistencyProviderManager = logConsistencyManager; - } - internal void Start() { if (gcTimer != null) gcTimer.Dispose(); @@ -557,8 +537,6 @@ private async Task InitActivation(ActivationData activation, string grainType, s } initStage = ActivationInitializationStage.SetupState; - await SetupActivationState(activation, - string.IsNullOrEmpty(genericArguments) ? grainType : $"{grainType}[{genericArguments}]"); initStage = ActivationInitializationStage.InvokeActivate; await InvokeActivate(activation, requestContextData); @@ -719,41 +697,12 @@ private void CreateGrainInstance(string grainTypeName, ActivationData data, stri //Get the grain's type Type grainType = grainTypeData.Type; - //Gets the type for the grain's state - Type stateObjectType = grainTypeData.StateObjectType; - lock (data) { data.SetupContext(grainTypeData, this.serviceProvider); - Grain grain; + Grain grain = grainCreator.CreateGrainInstance(data); - if (typeof(IStatefulGrain).IsAssignableFrom(grainType)) - { - //for stateful grains, install storage bridge - SetupStorageProvider(grainType, data); - - var storage = new GrainStateStorageBridge(grainType.FullName, data.StorageProvider); - - grain = grainCreator.CreateGrainInstance(data, stateObjectType, storage); - - storage.SetGrain(grain); - } - else - { - // Create a new instance of the given grain type - grain = grainCreator.CreateGrainInstance(data); - - // for log-view grains, install log-view adaptor - if (grain is ILogConsistentGrain) - { - var consistencyProvider = SetupLogConsistencyProvider(grain, grainType, data); - grainCreator.InstallLogViewAdaptor(grain, grainType, - grainTypeData.StateObjectType, grainTypeData.MultiClusterRegistrationStrategy ?? this.multiClusterRegistrationStrategyManager.DefaultStrategy, - consistencyProvider, data.StorageProvider); - } - } - Dictionary observerProxyMap; //if grain implements IStreamSubscriptionObserver, then can get a set of subscriptionObserver from it if(TryGetStreamSubscriptionObservers(grainType, grain, out observerProxyMap)) @@ -762,8 +711,7 @@ private void CreateGrainInstance(string grainTypeName, ActivationData data, stri grain.Data = data; data.SetGrainInstance(grain); } - - + activations.IncrementGrainCounter(grainClassName); if (logger.IsVerbose) logger.Verbose("CreateGrainInstance {0}{1}", data.Grain, data.ActivationId); @@ -798,155 +746,6 @@ private bool TryGetStreamSubscriptionObservers(Type grainType, IAddressable grai return subObserverProxyMap.Count > 0; } - private void SetupStorageProvider(Type grainType, ActivationData data) - { - var grainTypeName = grainType.FullName; - - // Get the storage provider name, using the default if not specified. - var attr = grainType.GetTypeInfo().GetCustomAttributes(true).FirstOrDefault(); - var storageProviderName = attr != null ? attr.ProviderName : Constants.DEFAULT_STORAGE_PROVIDER_NAME; - - IStorageProvider provider; - if (storageProviderManager == null || storageProviderManager.GetNumLoadedProviders() == 0) - { - var errMsg = string.Format("No storage providers found loading grain type {0}", grainTypeName); - logger.Error(ErrorCode.Provider_CatalogNoStorageProvider_1, errMsg); - throw new BadProviderConfigException(errMsg); - } - if (string.IsNullOrWhiteSpace(storageProviderName)) - { - // Use default storage provider - provider = storageProviderManager.GetDefaultProvider(); - } - else - { - // Look for MemoryStore provider as special case name - bool caseInsensitive = Constants.MEMORY_STORAGE_PROVIDER_NAME.Equals(storageProviderName, StringComparison.OrdinalIgnoreCase); - storageProviderManager.TryGetProvider(storageProviderName, out provider, caseInsensitive); - if (provider == null) - { - var errMsg = string.Format( - "Cannot find storage provider with Name={0} for grain type {1}", storageProviderName, - grainTypeName); - logger.Error(ErrorCode.Provider_CatalogNoStorageProvider_2, errMsg); - throw new BadProviderConfigException(errMsg); - } - } - data.StorageProvider = provider; - - if (logger.IsVerbose2) - { - string msg = string.Format("Assigned storage provider with Name={0} to grain type {1}", - storageProviderName, grainTypeName); - logger.Verbose2(ErrorCode.Provider_CatalogStorageProviderAllocated, msg); - } - } - - private ILogViewAdaptorFactory SetupLogConsistencyProvider(Grain grain, Type grainType, ActivationData data) - { - var attr = grainType.GetTypeInfo().GetCustomAttributes(true).FirstOrDefault(); - var consistencyProviderName = attr?.ProviderName; - - ILogConsistencyProvider consistencyProvider; - - if (logConsistencyProviderManager == null) - { - var errMsg = string.Format("No consistency provider manager found loading grain type {0}", grainType.FullName); - logger.Error(ErrorCode.Provider_CatalogNoLogConsistencyProvider, errMsg); - throw new BadProviderConfigException(errMsg); - } - - if (!string.IsNullOrWhiteSpace(consistencyProviderName)) - { - // find the named consistency provider; throw exception if it is not in the config - if (!logConsistencyProviderManager.TryGetProvider(consistencyProviderName, out consistencyProvider, false)) - { - var errMsg = string.Format( - "Cannot find consistency provider with Name={0} for grain type {1}", attr.ProviderName, - grainType.FullName); - logger.Error(ErrorCode.Provider_CatalogNoLogConsistencyProvider, errMsg); - throw new BadProviderConfigException(errMsg); - } - } - else - { - // See if the config specifies a "Default" consistency provider; if so use that - logConsistencyProviderManager.TryGetProvider(Constants.DEFAULT_LOG_CONSISTENCY_PROVIDER_NAME, out consistencyProvider, true); - } - - if (consistencyProvider != null) - { - // we found a log consistency provider in the configuration file - - // if it depends on a storage provider, find that one too - if (consistencyProvider.UsesStorageProvider) - SetupStorageProvider(grainType, data); - - string msg = string.Format("Assigned log consistency provider with Name={0} to grain type {1}", - consistencyProvider.Name, grainType.FullName); - logger.Verbose2(ErrorCode.Provider_CatalogLogConsistencyProviderAllocated, msg); - - return consistencyProvider; - } - - // Case 2 : no log consistency provider was specified in the configuration file. - // now we check if the grain type specifies a default adaptor factory - - var defaultFactory = ((ILogConsistentGrain)grain).DefaultAdaptorFactory; - - if (defaultFactory == null) - { - var errMsg = string.Format("No log consistency provider found loading grain type {0}", grainType.FullName); - logger.Error(ErrorCode.Provider_CatalogNoLogConsistencyProvider, errMsg); - throw new BadProviderConfigException(errMsg); - }; - - // if it depends on a storage provider, find that one too - if (defaultFactory.UsesStorageProvider) - SetupStorageProvider(grainType, data); - - return defaultFactory; - } - - private async Task SetupActivationState(ActivationData result, string grainType) - { - var statefulGrain = result.GrainInstance as IStatefulGrain; - if (statefulGrain == null) - { - return; - } - - var state = statefulGrain.GrainState; - - if (result.StorageProvider != null && state != null) - { - var sw = Stopwatch.StartNew(); - var innerState = statefulGrain.GrainState.State; - - // Populate state data - try - { - var grainRef = result.GrainReference; - - await scheduler.RunOrQueueTask(() => - result.StorageProvider.ReadStateAsync(grainType, grainRef, state), - result.SchedulingContext); - - sw.Stop(); - StorageStatisticsGroup.OnStorageActivate(result.StorageProvider, grainType, result.GrainReference, sw.Elapsed); - } - catch (Exception ex) - { - StorageStatisticsGroup.OnStorageActivateError(result.StorageProvider, grainType, result.GrainReference); - sw.Stop(); - if (!(ex.GetBaseException() is KeyNotFoundException)) - throw; - - statefulGrain.GrainState.State = innerState; // Just keep original empty state object - } - } - } - /// /// Try to get runtime data for an activation /// @@ -1334,11 +1133,6 @@ private async Task CallGrainActivate(ActivationData activation, Dictionary CallGrainDeactivateAndCleanupStreams(ActivationData activation) diff --git a/src/OrleansRuntime/Catalog/GrainCreator.cs b/src/OrleansRuntime/Catalog/GrainCreator.cs index 5d13630ee5d..c2f6837b946 100644 --- a/src/OrleansRuntime/Catalog/GrainCreator.cs +++ b/src/OrleansRuntime/Catalog/GrainCreator.cs @@ -1,9 +1,4 @@ using System; -using Orleans.Core; -using Orleans.LogConsistency; -using Orleans.Storage; -using Orleans.Runtime.LogConsistency; -using Orleans.GrainDirectory; namespace Orleans.Runtime { @@ -15,9 +10,7 @@ internal class GrainCreator private readonly IGrainActivator grainActivator; private readonly Lazy grainRuntime; - - private readonly Factory protocolServicesFactory; - + /// /// Initializes a new instance of the class. /// @@ -26,11 +19,9 @@ internal class GrainCreator /// public GrainCreator( IGrainActivator grainActivator, - Factory getGrainRuntime, - Factory protocolServicesFactory) + Factory getGrainRuntime) { this.grainActivator = grainActivator; - this.protocolServicesFactory = protocolServicesFactory; this.grainRuntime = new Lazy(() => getGrainRuntime()); } @@ -47,56 +38,13 @@ public Grain CreateGrainInstance(IGrainActivationContext context) grain.Runtime = this.grainRuntime.Value; grain.Identity = context.GrainIdentity; - return grain; - } - - /// - /// Create a new instance of a grain - /// - /// The for the executing action. - /// If the grain is a stateful grain, the type of the state it persists. - /// If the grain is a stateful grain, the storage used to persist the state. - /// - public Grain CreateGrainInstance(IGrainActivationContext context, Type stateType, IStorage storage) - { - //Create a new instance of the grain - var grain = CreateGrainInstance(context); - - var statefulGrain = grain as IStatefulGrain; - - if (statefulGrain == null) - return grain; - - //Inject state and storage data into the grain - statefulGrain.GrainState.State = Activator.CreateInstance(stateType); - statefulGrain.SetStorage(storage); + // wire up to lifecycle + var participant = grain as ILifecycleParticipant; + participant?.Participate(context.ObservableLifecycle); return grain; } - - /// - /// Install the log-view adaptor into a log-consistent grain. - /// - /// The grain. - /// The grain type. - /// The type of the grain state. - /// The multi-cluster registration strategy. - /// The consistency adaptor factory - /// The storage provider, or null if none needed - /// The newly created grain. - public void InstallLogViewAdaptor(Grain grain, Type grainType, - Type stateType, IMultiClusterRegistrationStrategy mcRegistrationStrategy, - ILogViewAdaptorFactory factory, IStorageProvider storageProvider) - { - // encapsulate runtime services used by consistency adaptors - var svc = this.protocolServicesFactory(grain, mcRegistrationStrategy); - - var state = Activator.CreateInstance(stateType); - - ((ILogConsistentGrain)grain).InstallAdaptor(factory, state, grainType.FullName, storageProvider, svc); - } - public void Release(IGrainActivationContext context, object grain) { this.grainActivator.Release(context, grain); diff --git a/src/OrleansRuntime/Catalog/GrainLifecycle.cs b/src/OrleansRuntime/Catalog/GrainLifecycle.cs index 8ec0a844b05..068577289ef 100644 --- a/src/OrleansRuntime/Catalog/GrainLifecycle.cs +++ b/src/OrleansRuntime/Catalog/GrainLifecycle.cs @@ -1,7 +1,7 @@  namespace Orleans.Runtime { - internal class GrainLifecycle : LifecycleObservable, IGrainLifecycle + internal class GrainLifecycle : LifecycleObservable, IGrainLifecycle { public GrainLifecycle(Logger logger) : base(logger) { diff --git a/src/OrleansRuntime/GrainTypeManager/GenericGrainTypeData.cs b/src/OrleansRuntime/GrainTypeManager/GenericGrainTypeData.cs index 665b9b56d27..5dafda1094b 100644 --- a/src/OrleansRuntime/GrainTypeManager/GenericGrainTypeData.cs +++ b/src/OrleansRuntime/GrainTypeManager/GenericGrainTypeData.cs @@ -13,7 +13,7 @@ internal class GenericGrainTypeData : GrainTypeData private readonly Type stateObjectType; public GenericGrainTypeData(Type activationType, Type stateObjectType, MultiClusterRegistrationStrategyManager registrationManager) : - base(activationType, stateObjectType, registrationManager) + base(activationType, registrationManager) { if (!activationType.GetTypeInfo().IsGenericTypeDefinition) throw new ArgumentException("Activation type is not generic: " + activationType.Name); @@ -32,7 +32,7 @@ public GrainTypeData MakeGenericType(Type[] typeArgs) ? GetGrainStateType(concreteActivationType.GetTypeInfo()) : this.stateObjectType; - return new GrainTypeData(concreteActivationType, concreteStateObjectType, this.registrationManager); + return new GrainTypeData(concreteActivationType, this.registrationManager); } private static Type GetGrainStateType(TypeInfo grainType) diff --git a/src/OrleansRuntime/GrainTypeManager/GrainTypeData.cs b/src/OrleansRuntime/GrainTypeManager/GrainTypeData.cs index 3d1e6dc2cfc..57d1bc52147 100644 --- a/src/OrleansRuntime/GrainTypeManager/GrainTypeData.cs +++ b/src/OrleansRuntime/GrainTypeManager/GrainTypeData.cs @@ -22,13 +22,12 @@ internal class GrainTypeData internal Type Type { get; private set; } internal string GrainClass { get; private set; } internal List RemoteInterfaceTypes { get; private set; } - internal Type StateObjectType { get; private set; } internal bool IsReentrant { get; private set; } internal bool IsStatelessWorker { get; private set; } internal Func MayInterleave { get; private set; } internal MultiClusterRegistrationStrategy MultiClusterRegistrationStrategy { get; private set; } - public GrainTypeData(Type type, Type stateObjectType, MultiClusterRegistrationStrategyManager registrationManager) + public GrainTypeData(Type type, MultiClusterRegistrationStrategyManager registrationManager) { var typeInfo = type.GetTypeInfo(); Type = type; @@ -36,8 +35,7 @@ public GrainTypeData(Type type, Type stateObjectType, MultiClusterRegistrationSt // TODO: shouldn't this use GrainInterfaceUtils.IsStatelessWorker? IsStatelessWorker = typeInfo.GetCustomAttributes(typeof(StatelessWorkerAttribute), true).Any(); GrainClass = TypeUtils.GetFullName(typeInfo); - RemoteInterfaceTypes = GetRemoteInterfaces(type); ; - StateObjectType = stateObjectType; + RemoteInterfaceTypes = GetRemoteInterfaces(type); MayInterleave = GetMayInterleavePredicate(typeInfo) ?? (_ => false); MultiClusterRegistrationStrategy = registrationManager?.GetMultiClusterRegistrationStrategy(type); } diff --git a/src/OrleansRuntime/GrainTypeManager/SiloAssemblyLoader.cs b/src/OrleansRuntime/GrainTypeManager/SiloAssemblyLoader.cs index b9119b0f15c..4bbfb5c1f28 100644 --- a/src/OrleansRuntime/GrainTypeManager/SiloAssemblyLoader.cs +++ b/src/OrleansRuntime/GrainTypeManager/SiloAssemblyLoader.cs @@ -96,7 +96,7 @@ public IDictionary GetGrainClassTypes() if (parentTypeInfo.IsGenericType) { var definition = parentTypeInfo.GetGenericTypeDefinition(); - if (definition == typeof(Grain<>) || definition == typeof(LogConsistentGrainBase<>)) + if (definition == typeof(Grain<>) || definition == typeof(LogConsistentGrain<>)) { var stateArg = parentType.GetGenericArguments()[0]; if (stateArg.GetTypeInfo().IsClass || stateArg.GetTypeInfo().IsValueType) @@ -143,7 +143,7 @@ private GrainTypeData GetTypeData(Type grainType, Type stateObjectType) { return grainType.GetTypeInfo().IsGenericTypeDefinition ? new GenericGrainTypeData(grainType, stateObjectType, this.registrationManager) : - new GrainTypeData(grainType, stateObjectType, this.registrationManager); + new GrainTypeData(grainType, this.registrationManager); } private static void LogGrainTypesFound(LoggerImpl logger, Dictionary grainTypeData) diff --git a/src/OrleansRuntime/Hosting/DefaultSiloServices.cs b/src/OrleansRuntime/Hosting/DefaultSiloServices.cs index 7e6127889ff..9d627acef80 100644 --- a/src/OrleansRuntime/Hosting/DefaultSiloServices.cs +++ b/src/OrleansRuntime/Hosting/DefaultSiloServices.cs @@ -1,3 +1,4 @@ +using System; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Orleans.CodeGeneration; @@ -30,7 +31,8 @@ using Orleans.Runtime; using Orleans.Runtime.Storage; using Orleans.Transactions; -using System; +using Orleans.LogConsistency; +using Orleans.Storage; namespace Orleans.Hosting { @@ -63,8 +65,17 @@ internal static void AddDefaultServices(IServiceCollection services) services.TryAddTransient(typeof(IStreamSubscriptionObserver<>), typeof(StreamSubscriptionObserverProxy<>)); services.TryAddSingleton(); + + // storage providers services.TryAddSingleton(); + services.TryAddFromExisting, StorageProviderManager>(); // as named services + services.TryAddSingleton(sp => sp.GetRequiredService().GetDefaultProvider()); // default + + // log concistency providers services.TryAddSingleton(); + services.TryAddFromExisting, LogConsistencyProviderManager>(); // as named services + services.TryAddSingleton(sp => sp.GetRequiredService().GetDefaultProvider()); // default + services.TryAddSingleton(); services.TryAddSingleton(); services.TryAddSingleton(); @@ -123,7 +134,7 @@ internal static void AddDefaultServices(IServiceCollection services) services.TryAddSingleton, GlobalSingleInstanceRegistrar>(); services.TryAddSingleton, ClusterLocalRegistrar>(); services.TryAddSingleton(); - services.TryAddSingleton(FactoryUtility.Create); + services.TryAddSingleton>(FactoryUtility.Create); services.TryAddSingleton(FactoryUtility.Create); // Placement diff --git a/src/OrleansRuntime/LogConsistency/ILogConsistencyProviderManager.cs b/src/OrleansRuntime/LogConsistency/ILogConsistencyProviderManager.cs index 8398b66886a..51714efda2d 100644 --- a/src/OrleansRuntime/LogConsistency/ILogConsistencyProviderManager.cs +++ b/src/OrleansRuntime/LogConsistency/ILogConsistencyProviderManager.cs @@ -12,6 +12,8 @@ internal interface ILogConsistencyProviderManager : IProviderManager int GetLoadedProvidersNum(); + ILogConsistencyProvider GetDefaultProvider(); + bool TryGetProvider(string name, out ILogConsistencyProvider provider, bool caseInsensitive = false); } diff --git a/src/OrleansRuntime/LogConsistency/LogConsistencyProviderManager.cs b/src/OrleansRuntime/LogConsistency/LogConsistencyProviderManager.cs index 6558e272408..d844c2c06bd 100644 --- a/src/OrleansRuntime/LogConsistency/LogConsistencyProviderManager.cs +++ b/src/OrleansRuntime/LogConsistency/LogConsistencyProviderManager.cs @@ -2,17 +2,13 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using Orleans.Core; using Orleans.Providers; using Orleans.Runtime.Configuration; -using Orleans.Runtime.Providers; using Orleans.LogConsistency; -using Orleans.Storage; -using Microsoft.Extensions.DependencyInjection; namespace Orleans.Runtime.LogConsistency { - internal class LogConsistencyProviderManager : ILogConsistencyProviderManager, ILogConsistencyProviderRuntime + internal class LogConsistencyProviderManager : ILogConsistencyProviderManager, ILogConsistencyProviderRuntime, IKeyedServiceCollection { private ProviderLoader providerLoader; private IProviderRuntime runtime; @@ -112,5 +108,22 @@ public IProvider GetProvider(string name) return providerLoader.GetProvider(name, true); } + public ILogConsistencyProvider GetService(IServiceProvider services, string key) + { + ILogConsistencyProvider provider; + return TryGetProvider(key, out provider) ? provider : default(ILogConsistencyProvider); + } + + public ILogConsistencyProvider GetDefaultProvider() + { + try + { + return providerLoader.GetDefaultProvider(Constants.DEFAULT_LOG_CONSISTENCY_PROVIDER_NAME); + } catch(InvalidOperationException) + { + // default ILogConsistencyProvider are optional, will fallback to grain specific if not configured. + return default(ILogConsistencyProvider); + } + } } } diff --git a/src/OrleansRuntime/Silo/Silo.cs b/src/OrleansRuntime/Silo/Silo.cs index 35b9aa57ea5..c772e811ae9 100644 --- a/src/OrleansRuntime/Silo/Silo.cs +++ b/src/OrleansRuntime/Silo/Silo.cs @@ -471,7 +471,6 @@ private void DoStart() () => storageProviderManager.LoadStorageProviders(GlobalConfig.ProviderConfigurations), providerManagerSystemTarget.SchedulingContext) .WaitWithThrow(initTimeout); - catalog.SetStorageManager(storageProviderManager); allSiloProviders.AddRange(storageProviderManager.GetProviders()); ITransactionAgent transactionAgent = this.Services.GetRequiredService(); @@ -489,7 +488,6 @@ private void DoStart() () => logConsistencyProviderManager.LoadLogConsistencyProviders(GlobalConfig.ProviderConfigurations), providerManagerSystemTarget.SchedulingContext) .WaitWithThrow(initTimeout); - catalog.SetLogConsistencyManager(logConsistencyProviderManager); if (logger.IsVerbose) { logger.Verbose("Log consistency provider manager created successfully."); } // Load and init stream providers before silo becomes active diff --git a/src/OrleansRuntime/Storage/StorageProviderManager.cs b/src/OrleansRuntime/Storage/StorageProviderManager.cs index 0f9bb49a888..a2465933fc6 100644 --- a/src/OrleansRuntime/Storage/StorageProviderManager.cs +++ b/src/OrleansRuntime/Storage/StorageProviderManager.cs @@ -5,11 +5,10 @@ using Orleans.Providers; using Orleans.Runtime.Configuration; using Orleans.Storage; -using Microsoft.Extensions.DependencyInjection; namespace Orleans.Runtime.Storage { - internal class StorageProviderManager : IStorageProviderManager, IStorageProviderRuntime + internal class StorageProviderManager : IStorageProviderManager, IStorageProviderRuntime, IKeyedServiceCollection { private readonly IProviderRuntime providerRuntime; private ProviderLoader storageProviderLoader; @@ -120,5 +119,11 @@ internal async Task AddAndInitProvider(string name, IStorageProvider provider, I await provider.Init(name, this, config); storageProviderLoader.AddProvider(name, provider, config); } + + public IStorageProvider GetService(IServiceProvider services, string key) + { + IStorageProvider provider; + return TryGetProvider(key, out provider) ? provider : default(IStorageProvider); + } } } diff --git a/test/TestInternalGrains/PersistenceTestGrains.cs b/test/TestInternalGrains/PersistenceTestGrains.cs index 06b9ddc9906..68826de41da 100644 --- a/test/TestInternalGrains/PersistenceTestGrains.cs +++ b/test/TestInternalGrains/PersistenceTestGrains.cs @@ -14,6 +14,7 @@ using Orleans.Serialization; using UnitTests.GrainInterfaces; using Xunit; +using Orleans.Storage; namespace UnitTests.Grains { @@ -83,7 +84,7 @@ public Task CheckStateInit() public Task CheckProviderType() { - var storageProvider = ((ActivationData) Data).StorageProvider; + IStorageProvider storageProvider = this.GetStorageProvider(this.ServiceProvider); Assert.NotNull(storageProvider); return Task.FromResult(storageProvider.GetType().FullName); } diff --git a/test/Tester/Lifecycle/LifecycleTests.cs b/test/Tester/Lifecycle/LifecycleTests.cs index 61ddfba8f55..5fbe2eb4652 100644 --- a/test/Tester/Lifecycle/LifecycleTests.cs +++ b/test/Tester/Lifecycle/LifecycleTests.cs @@ -6,6 +6,7 @@ using System.Threading.Tasks; using Orleans; using Xunit; +using Orleans.Runtime; namespace Tester { @@ -142,7 +143,7 @@ private async Task>> RunLifecycle(Dictionar // run lifecycle if (failOnStart.HasValue) { - await Assert.ThrowsAsync(() => lifecycle.OnStart()); + await Assert.ThrowsAsync(() => lifecycle.OnStart()); } else { diff --git a/test/Tester/StorageFacet/Feature.Implementations/TableExampleStorage.cs b/test/Tester/StorageFacet/Feature.Implementations/TableExampleStorage.cs index 2cd896ac7ce..93f966f3e1a 100644 --- a/test/Tester/StorageFacet/Feature.Implementations/TableExampleStorage.cs +++ b/test/Tester/StorageFacet/Feature.Implementations/TableExampleStorage.cs @@ -7,7 +7,7 @@ namespace Tester.StorageFacet.Implementations { - public class TableExampleStorage : IExampleStorage, ILifecycleParticipant + public class TableExampleStorage : IExampleStorage, ILifecycleParticipant { private IExampleStorageConfig config; private bool activateCalled; @@ -31,9 +31,9 @@ public Task LoadState(CancellationToken ct) return Task.CompletedTask; } - public void Participate(ILifecycleObservable lifecycle) + public void Participate(ILifecycleObservable lifecycle) { - lifecycle.Subscribe(GrainLifecyleStage.SetupState, LoadState); + lifecycle.Subscribe(GrainLifecycleStage.SetupState, LoadState); } public void Configure(IExampleStorageConfig cfg) diff --git a/test/TesterInternal/ErrorInjectionStorageProvider.cs b/test/TesterInternal/ErrorInjectionStorageProvider.cs index 59179ef0cfc..68ec0633efb 100644 --- a/test/TesterInternal/ErrorInjectionStorageProvider.cs +++ b/test/TesterInternal/ErrorInjectionStorageProvider.cs @@ -27,15 +27,10 @@ public struct ErrorInjectionBehavior } [Serializable] - public class StorageProviderInjectedError : Exception + public class StorageProviderInjectedError : OrleansException { private readonly ErrorInjectionPoint errorInjectionPoint; - public StorageProviderInjectedError(SerializationInfo info, StreamingContext context) - : base(info, context) - { - } - public StorageProviderInjectedError(ErrorInjectionPoint errorPoint) { errorInjectionPoint = errorPoint; @@ -53,6 +48,11 @@ public override string Message return "ErrorInjectionPoint=" + Enum.GetName(typeof(ErrorInjectionPoint), errorInjectionPoint); } } + + protected StorageProviderInjectedError(SerializationInfo info, StreamingContext context) + : base(info, context) + { + } } public class ErrorInjectionStorageProvider : MockStorageProvider, IControllable diff --git a/test/TesterInternal/StorageTests/PersistenceGrainTests.cs b/test/TesterInternal/StorageTests/PersistenceGrainTests.cs index 00220d980b7..528311b23f7 100644 --- a/test/TesterInternal/StorageTests/PersistenceGrainTests.cs +++ b/test/TesterInternal/StorageTests/PersistenceGrainTests.cs @@ -987,8 +987,7 @@ public async Task Persistence_Provider_Loop_Read() public async Task Persistence_Grain_BadProvider() { IBadProviderTestGrain grain = this.HostedCluster.GrainFactory.GetGrain(Guid.NewGuid()); - var oex = await Assert.ThrowsAsync(() => grain.DoSomething()); - Assert.IsAssignableFrom(oex.InnerException); + var oex = await Assert.ThrowsAsync(() => grain.DoSomething()); } [Fact, TestCategory("Functional"), TestCategory("Persistence")] diff --git a/test/TesterInternal/StreamingTests/StreamPubSubReliabilityTests.cs b/test/TesterInternal/StreamingTests/StreamPubSubReliabilityTests.cs index 26369b3b150..9ec0700e524 100644 --- a/test/TesterInternal/StreamingTests/StreamPubSubReliabilityTests.cs +++ b/test/TesterInternal/StreamingTests/StreamPubSubReliabilityTests.cs @@ -83,10 +83,8 @@ public async Task PubSub_Store_WriteError() { SetErrorInjection(PubSubStoreProviderName, ErrorInjectionPoint.BeforeWrite); - var exception = await Assert.ThrowsAsync(() => + var exception = await Assert.ThrowsAsync(() => Test_PubSub_Stream(StreamProviderName, StreamId)); - - Assert.IsAssignableFrom(exception.InnerException); } private async Task Test_PubSub_Stream(string streamProviderName, Guid streamId)