Long running operation and stashing guidance

[AlexMeakin]

Hi

We are currently refactoring our application to use Orleans and have a use case we are struggling to implement.

Background
The application in question is an interactive forecasting application that has been around for approximately 15 years. The application is being re-factored to improve performance, scalability and re-platform to web technologies. The initial phase of development migrated the Domain Driven Design application to our own Actor framework. Our actor framework uses a variant of a Limited Concurrency Task Scheduler to ensure only one task is executed at once per Actor. The current implementation uses await inconsistently, in certain circumstance the application expects await to interleave, in other cases it is reliant on deterministic ordering. Specifically calling await within an Actor expects tasks to be scheduled sequentially, but calling await on another Actor expects the caller to interleave/be re-entrant. This is leading to a lot of complexity in the code.

The web application uses a reactive architecture, users issue commands to the server which are validated quickly, once validated the user is informed that processing has started. Users are notified of forecast completion via web sockets. Forecasting is a reasonably time consuming operation 0.5 to 120 seconds, depending on data volumes. Forecasting is completed on the .NET thread pool, not the Orleans or Activation scheduler. The forecast duration is longer than the default message time out in Orleans.

In an attempt to resolve the interleaving problems and support horizontal scalability, the application is being further refactored to Orleans. Generally our grains are not re-entrant to reduce complexity.

Use case
It should be possible for users to update the forecast parameters even whilst forecasting is in progress. When an initial update is made, forecasting should start. If a forecast is in progress when the parameters are updated, the changes should be stashed and persisted until the current forecast completes. Once the original forecast completes, any pending changes are aggregated and a new forecast should start, which includes all of the stashed changes. Our users understand that the application is eventually consistent.

Current Progress
The refactor to Orleans is currently in development and to support the use case above two actors have been created: Actor(State) and Actor(Forecast). The Actor(State) is responsible for manipulating state, persisting state, tracking whether a forecast is in progress and starting a new forecast when appropriate. Actor(Forecast) is responsible for forecasting the State from Actor(State). Once forecasting has completed, the results are stored in Actor(Forecast). All clients require a combination of the state and results when displaying information to users, so all requests for results are proxied via the Actor(State).

To support the use case above, forecasting has been refactored to the Tell pattern. When an update is received, the state is updated and persisted by the Actor(State). Once persisted, the Actor(State) tells (via fire and forget *) the Actor(Forecast) to start processing. Once forecasting finishes the Actor(Forecast) tells the Actor(State) it has completed. The approach frees up the Actor(State)'s mailbox during forecasting. Actor(Forecast) sends aggregated results back to the Actor(State), which the Actor(State) stores in memory, so basic metrics can be returned when forecasting is in progress. In circumstances where the Actor(State) doesn't have the appropriate data available it asks the Actor(Forecast) for the results, so they can be returned to the client.

NOTE: The end goal is to move all forecast interactions to be Tell based, but this is a reasonably large amount of work. In the short term the application needs to be available.

Normally the following happens:

1. User update the forecast criteria
2. Actor(State) update and persists the change
3. Actor(State) tells (Fire and Forget) the Actor(Forecast) to start processing (60 seconds)
4. Actor(State) is able to receive and stash state changes until forecasting completes
5. Actor(Forecast) completes the forecast and tells (Fire and Forget) Actor(State) it has finished
6. Actor(State) notifies the user forecasting has finished
7. User requests the results from Actor(State)
8. Actor(State) asks (await) Actor(Forecast) for the missing data.
9. Data is returned to the client

Required Guidance
Does the current implementation lead to deadlocks in the following case?

1. An update is sent to the Actor(State).
2. Actor(State) updates and persists the current state
3. Actor(State) tells (fire and forget) the Actor(Forecast) to start processing (60 seconds)
4. A user requests the results whilst forecasting is in progress.
5. Actor(State) asks (await) for results from Actor(Forecast)
6. Actor(Forecast) tells (fire and forget) the Actor(State) forecasting has completed
7. Actor(State) returns the results to the client (once step 5 completes)

In this case is Actor(State) blocked because it is awaiting the results from Actor(Forecast) and Actor(Forecast) can't complete because it is trying to send a message to the Actor(State)? Or does the operation complete because the Actor(Forecast) is telling (fire and forget) the Actor(State) it has finished, so it's Task completes.

If this does lead to a deadlock, what is the best way to resolve the problem? I'd prefer to avoid making the Grains re-entrant. It may be possible to create a short lived re-entrant grain responsible for mediating between the two Actors, but the performance may be bad at scale.

* Fire and forget calls the target grain, but don't await the response. All fire and forget actions use a method similar to Ignore in https://github.com/dotnet/orleans/blob/master/src/Orleans/Async/TaskExtensions.cs. The main difference is that method logs all exceptions to ELK and supports a continuation. The continuation is used to recover from failures, either using retry or informing the source that the operation failed. This pattern is loosely based on Akka's Tell.

[jason-bragg]

I understand the desire to not make the grains reentrant, in general, but have you considered allowing specific calls to be allowed to interleave using the AlwaysInterleaveAttribute?

[jason-bragg]

Specifically calling await within an Actor expects tasks to be scheduled sequentially

Understood. Yes, this deviates from Orleans task scheduling. It sounds like you're refactoring the logic to conform with Orleans task scheduling, but in areas that need this specific task execution pattern you may want to consider the use of the utility AsyncSerialExecutor. This utility ensures serial execution of actions. It was initially developed to help with the development/maintenance of in reentrant grains or interleaved calls, but may also fit your needs.

[jason-bragg]

5. Actor(State) asks (await) for results from Actor(Forecast)
6. Actor(Forecast) tells (fire and forget) the Actor(State) forecasting has completed
7. Actor(State) returns the results to the client (once step 5 completes)

From my understanding, this should not cause a deadlock, but the result returned in 7, will not include the information delivered in the 6 tell, because Actor(State) will not process the 6 tell until 5 is completed in 7. To get the results from the 6 tell into Actor(State) prior to the results of 5 being returned in 7, the 6 tell call needs be marked AlwaysInterleave.

[AlexMeakin]

@jason-bragg thank you for the detailed responses.

Both the AlwaysInterleaveAttribute and AsyncSerialExecutor look very useful based on where we are today. I need to do a bit more research in to the AlwaysInterleaveAttribute, but if it does what I think it does, this will save a lot of time until the problem can be solved properly.

I suspect the best way to resolve the issue long term is to create a new actor that is responsible for querying and aggregating results, so state change messages have the following flow Actor(State) > Actor(Forecast) > Actor(ResultsAggregator). Any request for results directly calls the Actor(ResultsAggregator).

Is it possible to explain why the operation won't cause a deadlock, but will return incorrect results? There may be some context missing here - the exact interaction in step 7 is that the Actor(State) awaits the response and manipulates before returning it to the Client.

This example may describe the implementation for accurately

    public interface IForecastGrain : IGrainWithIntegerKey
    {
        Task Forecast(int parameter, IStateGrain actorRef);
        Task<int> GetResults(int parameter);
    }

    public interface IStateGrain : IGrainWithIntegerKey
    {
        Task SetDetails(int parameter);
        Task<int> GetResults(int criteria);
        Task ForecastComplete(int result);
    }

    public class StateGrain : Grain, IStateGrain
    {
        private int _parameter = 1;
        private bool _isForecasting;
        private bool _requiresAdditionalForecast;

        public Task SetDetails(int parameter)
        {
            // Modify state
            _parameter += parameter;

            // Only forecast if not busy
            if (!_isForecasting)
            {
                ForecastDetails();
            }
            else
            {
                _requiresAdditionalForecast = true;
            }
            
            return TaskDone.Done;
        }

        public async Task<int> GetResults(int criteria)
        {
            var grain = GrainFactory.GetGrain<IForecastGrain>(1);

            // Get results
            var val = await grain.GetResults(criteria);

            // Post processing
            return val * _parameter;
        }

        public Task ForecastComplete(int result)
        {
            // Update result based on forecast complete
            _parameter -= result;
            _isForecasting = false;

            if (_requiresAdditionalForecast)
            {
                _requiresAdditionalForecast = false;
                ForecastDetails();
            }

            return TaskDone.Done;
        }

        private void ForecastDetails()
        {
            _isForecasting = true;

            var grain = GrainFactory.GetGrain<IForecastGrain>(1);

            // Fire and forget
            grain.Forecast(_parameter, this.AsReference<IStateGrain>()).FireAndForget(r => _isForecasting = false);
        }
    }

    public class ForecastGrain : Grain, IForecastGrain
    {
        private int _parameter = 1;
        public async Task Forecast(int parameter, IStateGrain actorRef)
        {
            // In the real implementation this work is done on the .NET thread pool
            await Task.Delay(60000);
            _parameter *= parameter;
            actorRef.ForecastComplete(_parameter).FireAndForget();
        }

        public Task<int> GetResults(int parameter)
        {
            return Task.FromResult(_parameter * parameter);
        }
    }

[jason-bragg]

In the code sample, their could be blocking (and possibly timeouts) but not deadlocks. This could occur because while the ForecastGrain is performing Forecast(..) it cannot process any other messages, so all calls to it's GetResults(..) would wait until the forecast was complete, and it's ForecastComplete callback would be queued on the StateGrain only after all other calls queued for that grain.

So given the following calls to StateGrain:

1. SetDetails
2. SetDetails
3. GetResults
4. SetDetails
5. GetResults
6. GetResults
7. SetDetails

The first SetDetails (1) would modify the state and kick off a forecast.
The second SetDetails (2) would modify the state and mark requiresAdditionalForecast to true.
The first GetResults (3) would request the results from the ForecastGrain which would be blocked until the forecast was complete, because the forecast grain is busy forecasting.
All of the remaining calls 4,7 would be queued up for the StateGrain.
When the forecast was complete, it would call ForecastComplete on the StateGrain which would be queued after 7.
So the adjustment to the _parameter from the forecast needed in 3, 5, and 6 would not occur until after 7 was processed, and even then would only be updated (at that point) with the results from the original SetDetails
call (1).

[jason-bragg]

I don't think the behavior you seek is hard, it's just not clear what behavior you want.
If a SetDetails and GetResults call are made, immediately one after another, what is the expected behavior? Should the GetResults return the forecast after it completes, or return the most recently completed forecast?

[sergeybykov]

I'm confused with the code Forecast awaits the long running calculation and notifies the called at the end. Couldn't you do it differently as follows?

1. Instead of Forcast have a BeginForcast method that would start a new computation on the .NET thread pool (unless there is already one running), and return to the caller (StateGrain) confirming that the computation started.
2. Upon completion of the forecasting computation, the code running on the .NET thread pool will enqueue a call to itself (to the ForecastGrain that started it), e.g. a CompleteForcast method as described in http://dotnet.github.io/orleans/Documentation/Advanced-Concepts/External-Tasks-and-Grains.html.
3. CompleteForcast will call to the StateGrain that initiated the computation to notify it about completion. This can be done reliably as a normal grain method calls or via an observer.

In this approach, ForecastGrain is always available for incoming calls. The only call that could potentially cause a temporary deadlock is the notification call back to StateGrain. The possibility for a deadlock can be eliminated by using an observer (probably in conjunction with a periodic poll from StateGrain) or by marking the notification method as AlwaysInterleave. That being said, I see no reason for not marking ForecastGrain as reentrant in this case, as none of its methods will call await, and hence no real interleaving will happen anyway.

Did I miss something?

[AlexMeakin]

@jason-bragg, @sergeybykov

Again thanks for the feedback. The pattern suggested in the Sergey's post is what I am attempting to achieve. I do have a follow on question based on the suggestion.

The documentation suggests creating a Task using Task.Run, awaiting that task and then returning to the caller. My understanding is that awaiting the task dispatched on .NET Thread pool won't complete until the forecast completes. This means the return value won't be available until that Task completes. Is the suggestion to start the Task and store it within the Actor(Forecast)?

I believe there are two possible solutions, please can you let me know if either summarise the suggestion?

        public Task<bool> BeginForecast(int parameter, IStateGrain actorRef)
        {
            if (_forecastingInProgress)
            {
                return Task.FromResult(false);
            }

            var orleansTs = TaskScheduler.Current;
            _forecastingInProgress = true;

            // Should this be:
            // - _forecastTasks = Task.Run 
            // OR
            // - await Task.Run  
            _forecastingTask = Task.Run(async () =>
            {
                await Task.Delay(60000);
                await Task.Factory.StartNew(async () =>
                {
                    var grain = GrainFactory.GetGrain<IForecastGrain>(this.GetPrimaryKeyLong());
                    var res = _parameter * parameter;
                    await grain.CompleteForecast(res, actorRef);
                }, CancellationToken.None, TaskCreationOptions.None, scheduler: orleansTs);
            });

            return Task.FromResult(true);
        }

        public async Task<bool> BeginForecast2(int parameter, IStateGrain actorRef)
        {
            if (_forecastingInProgress)
            {
                return false;
            }

            var orleansTs = TaskScheduler.Current;
            _forecastingInProgress = true;

            // Should this be:
            // - _forecastTasks = Task.Run 
            // OR
            // - await Task.Run  
            await Task.Factory.StartNew(async () =>
            {
                await Task.Delay(60000);
                await Task.Factory.StartNew(async () =>
                {
                    var grain = GrainFactory.GetGrain<IForecastGrain>(this.GetPrimaryKeyLong());
                    var res = _parameter * parameter;
                    await grain.CompleteForecast(res, actorRef);
                }, CancellationToken.None, TaskCreationOptions.None, scheduler: orleansTs);
            }, CancellationToken.None, TaskCreationOptions.None, scheduler: TaskScheduler.Default);

            return true;
        }

I've extended the example to represent the code more accurately. I've also attempted to incorporate the suggestions, but I'm not quite sure how to achieve the "and return to the caller" quickly.

    public interface IForecastGrain : IGrainWithIntegerKey
    {
        Task<bool> BeginForecast(int parameter, IStateGrain actorRef);
        Task<int> GetResults(int parameter);
        Task CompleteForecast(int result, IStateGrain actorRef);
    }

    public interface IStateGrain : IGrainWithIntegerKey
    {
        Task SetDetails(int parameter);
        Task<ForecastResults> GetResults(int criteria);
        [AlwaysInterleave]
        Task ForecastComplete(int result);
    }

    public class StateGrain : Grain, IStateGrain
    {
        private readonly CachedForecastResults _cachedResults = new CachedForecastResults();
        private bool _isForecasting;
        private bool _requiresAdditionalForecast;
        private int _state;
        private IDisposable _timer;

        public async Task SetDetails(int newState)
        {
            // Modify state
            _state += newState;

            // Only forecast if not busy
            if (!_isForecasting)
            {
                await ForecastDetails();
            }
            else
            {
                _requiresAdditionalForecast = true;
            }
        }

        public async Task<ForecastResults> GetResults(int criteria)
        {
            var grain = GrainFactory.GetGrain<IForecastGrain>(1);

            // Get results
            var val = await _cachedResults.GetOrAdd(criteria, () => grain.GetResults(criteria));

            // Post processing
            return new ForecastResults(_state, val);

        }

        public async Task ForecastComplete(int result)
        {
            _cachedResults.Clear();
            // Update result based on forecast complete
            _isForecasting = false;

            if (_requiresAdditionalForecast)
            {
                _requiresAdditionalForecast = false;
                await ForecastDetails();
            }
            
        }

        private async Task ForecastDetails()
        {
            _isForecasting = true;

            var grain = GrainFactory.GetGrain<IForecastGrain>(1);

            // Fire and forget
            var result = await grain.BeginForecast(_state, this.AsReference<IStateGrain>());
            if (result)
            {
                if (_timer != null) _timer.Dispose();
                _isForecasting = true;
            }
            else
            {
                _requiresAdditionalForecast = true;
                if (_timer == null)
                {
                    _timer = this.RegisterTimer(o => ForecastDetails(), null, TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(10));
                }
            }
        }
    }

    public class ForecastGrain : Grain, IForecastGrain
    {
        private int _parameter = 1;
        private bool _forecastingInProgress;
        private Task _forecastingTask;

        public Task<bool> BeginForecast(int parameter, IStateGrain actorRef)
        {
            if (_forecastingInProgress)
            {
                return Task.FromResult(false);
            }

            var orleansTs = TaskScheduler.Current;
            _forecastingInProgress = true;

            // Should this be:
            // - _forecastTasks = Task.Run 
            // OR
            // - await Task.Run  
            _forecastingTask = Task.Run(async () =>
            {
                await Task.Delay(60000);
                await Task.Factory.StartNew(async () =>
                {
                    var grain = GrainFactory.GetGrain<IForecastGrain>(this.GetPrimaryKeyLong());
                    var res = _parameter * parameter;
                    await grain.CompleteForecast(res, actorRef);
                }, CancellationToken.None, TaskCreationOptions.None, scheduler: orleansTs);
            });

            return Task.FromResult(true);
        }

        public Task<int> GetResults(int parameter)
        {
            return Task.FromResult(_parameter * parameter);
        }

        public async Task CompleteForecast(int result, IStateGrain actorRef)
        {
            _parameter = result;
            _forecastingInProgress = false;
            await actorRef.ForecastComplete(result);
        }
    }

    public class CachedForecastResults
    {
        private Dictionary<int, int> _results = new Dictionary<int, int>();

        public async Task<int> GetOrAdd(int parm, Func<Task<int>> getFunc)
        {
            if (_results.ContainsKey(parm))
            {
                return _results[parm];
            }

            var res = await getFunc();
            _results.Add(parm, res);

            return res;
        }

        public void Clear()
        {
            _results.Clear();
        }
    }

    [Immutable]
    public class ForecastResults
    {
        public ForecastResults(int currentState, int result)
        {
            CurrentState = currentState;
            Result = result;
        }

        public int CurrentState { get; private set; }

        public int Result { get; private set; }
    }

[sergeybykov]

BeginForecast2 looks excessive to me - it both awaits the forecasting Task and invokes CompleteForecast at the end. I don't think you need that. I like BeginForecast because is returns to the caller quickly confirming only that the long running operation started successfully. That is more in line with the concept of grains being reactive and responsive, and ensures that the call will not time out even if the forecasting computation takes an unexpectedly long time. I think you missed resetting of forecastingInProgress at the end though.

The documentation suggests creating a Task using Task.Run, awaiting that task and then returning to the caller.

You are right, and I realize it's a bit confusing for this case. It was written primarily with another case in mind - to avoid running blocking operation on silo's scheduler thread. If those operations are relatively fast, awaiting them is the most straightforward thing to do. Your case is different - you are executing long running computations. I hope this clears it a little bit.

[ashkan-saeedi-mazdeh]

@sergeybykov You said

The possibility for a deadlock can be eliminated by using an observer (probably in conjunction with a periodic poll from StateGrain ) or by marking the notification method as AlwaysInterleave .

Why an observer between two grains need additional pulling from State grain?

@AlexMeakin I had an issue which was partially similar. The additional state changes didn't need to be stored for later startup of another long running operation. That doesn't make much difference however.

I've implemented it in a different way compared to what Sergey suggests but BeginOperation is good good. You can even use a simple SMS stream from the Forecast grain to publish the results to subscribers or use observers.

The alternate approach which I choose because the code was more linear in that case was that the LongRunning grain (a.k.a forecast) was marked Reentrant and then when I received BeginOperation calls. I created TaskCompletionSource objects and put them in a list and return that task to my caller, Then when the operation got finished/timedout I would set the result for the TaskCompletionSource and then the await call of the BeginOperaiton from caller grain (State grain in your case) would finish.

My case was a MatchMaker grain. My approach is more complex that using a stream/observer but the calling code would linearly get the result and the API surface might be a bit nicer to some. Back then I was a beginner (even more than now in Orleans) so wanted to avoid using any additional complex features like observers/streams specially in unit and integration tests which might have contributed to my choices as well.
Just thought it is nice to bring another possible way of doing it up.

[sergeybykov]

@ashkan-saeedi-mazdeh

Why an observer between two grains need additional pulling from State grain?

Observer messages are inherently unreliable because the caller gets no indication if a message gets lost due to a connectivity issue or a node failure, and hence doesn't know to resend it. Even though the probability of such an occurrence is fairly low, it's not zero. Periodic polling covers for that by ensuring no operation will be left in an unfinished state.

[ashkan-saeedi-mazdeh]

@sergeybykov fair enough. Thanks. I'll update observer docs to reflect the fact.

[sergeybykov]

Looks like this can be closed now.

[ztl8702]

It sounds like that TaskScheduler.Current in grain code is expected to return the (top-level) OrleansTaskScheduler but not ActivationTaskScheduler?

What I found out when trying to replicate the above example is that var orleansTs = TaskScheduler.Current; gives the specific ActivationTaskScheduler of that Activation when BeginForcast is called. But if the Grain gets deactivated while the long-running job is executing, then this:

        await Task.Factory.StartNew(async () =>
                {
                    var grain = GrainFactory.GetGrain<IForecastGrain>(this.GetPrimaryKeyLong());
                    var res = _parameter * parameter;
                    await grain.CompleteForecast(res, actorRef);
                }, CancellationToken.None, TaskCreationOptions.None, scheduler: orleansTs);

is scheduling a Task on the orleansTs which was the ActivationTaskScheduler of the (deactivated) activation. And I would get the following log message when stepping over this Task.Factory.StartNew(.... scheduler: orleansTs) line:

warn: Orleans.Runtime.Scheduler.WorkItemGroup[101217]
      Enqueuing task System.Threading.Tasks.Task`1[System.Threading.Tasks.Task] to a stopped work item group. Going to ignore and not execute it. The likely reason is that the task is not being 'awaited' properly. WorkItemGroup:Name=[Activation: S127.0.0.1:22900:283671474*grn/27B66166/047c11b9@439fb987 #GrainType=Zezo.Core.Grains.StepGrain Placement=RandomPlacement],WorkGroupStatus=Shutdown. Currently QueuedWorkItems=0; Total EnQueued=24; Total processed=24; Quantum expirations=0; TaskRunner=ActivationTaskScheduler-34:Queued=0; Detailed SchedulingContext=<[Activation: S127.0.0.1:22900:283671474*grn/27B66166/4d0204423a966e1833f92889047c11b90300000027b66166-0xC74003E7@439fb987 #GrainType=Zezo.Core.Grains.StepGrain Placement=RandomPlacement State=Invalid NonReentrancyQueueSize=0 EnqueuedOnDispatcher=0 InFlightCount=0 NumRunning=0 IdlenessTimeSpan=00:00:18.4740700 CollectionAgeLimit=02:00:00]>

What did I miss here? (I am running Orleans 2.2.0)

[ReubenBond]

@ztl8702 in the case that the task is very long running, you could use Task.Run for the background work as long as you also enable the direct client during configuration time by calling siloHostBuilder.EnableDirectClient().

[ztl8702]

@ReubenBond thanks!

My follow-up question is: how do I properly interact with the Orleans runtime from within the Task.Run lambda (non-Orleans thread) then?

It appears that:

• this.GrainFactory within the non-Orleans thread throws an access violation error (because the property is calling the Orleans runtime on the spot?)
• storing a grain reference in the Grain context (var selfReference = this.GrainFactory.GetGrain<IWhateverGrain>(this.GetPrimaryKey()); ) and using it in the non-Orleans thread works
• storing a reference to the GrainFactory in the Grain context and using it in the non-Orleans thread also works
• i can just await a grain call on the non-Orleans thread (trying to schedule a Task on orleansTs is irrelevant here because of the de-activation issue.)

(The latter three requires siloHostBuilder.EnableDirectClient(), right?)

So it also appears that:

• both the grain reference object and GrainFactory are reusable across different threads (including non-Orleans threads).
• the instance of the Grain's implementation object is still accessible from inside the non-Orleans thread, even after the de-activation of that particular activation. (Because the long-running lambda is still referencing the Grain implementation object, so the .NET GC won't collect it even after the activation has been GC'ed by Orleans?) But now this instance of Grain implementation is no longer used by Orleans, as it has been considered to be de-activated. Another instance of the Grain implementation will be created if the grain is re-activated.
  • the implication is to be careful, and not interact with the internals of the (now-dead) Grain object from a non-Orleans long running thread?

Just want to check that my understanding is correct. :)

[ReubenBond]

Your solutions (point 2 & 3) are both fine approaches.

both the grain reference object and GrainFactory are reusable across different threads (including non-Orleans threads).

Yes, that is correct.

the implication is to be careful, and not interact with the internals of the (now-dead) Grain object from a non-Orleans long running thread?

Yes. I would go further and suggest that you never access the internals of a grain object from the Task.Run delegate - always do it through a grain call via the selfReference.

[ztl8702]

Yes. I would go further and suggest that you never access the internals of a grain object from the Task.Run delegate - always do it through a grain call via the selfReference.

Got it: essentially the code within Task.Run delegate is like client code.

Thanks for helping out!