Reactive Queries

Hi all,

I'm interning at MSR this summer and will be working on what we (currently) call reactive queries. Since there seems to be quite some interest and previous ideas already, I decided to start a thread to get the community's feedback.

The main idea is to allow programmers to write declarative "queries" that get updated whenever the result of that query changes (where a query is actually just a method calling some other grain methods, for now). This declarative (pull-based) query is then transformed into a reactive (push-based) version by the runtime. Thus, executing such query will set up the required subscriptions (dependency tracking) for you in the runtime in order to get updated whenever the result of the query changes.

Concretely, at the client or consumer end one would write something like this:

var Grain = GrainFactory.GetGrain<IMyGrain>(0);
Query Query = Grain.MyQuery(args).KeepAlive(2000, CancellationToken);
while (Interested) {
    Result result = await Query.NextUpdateAsync();
    // Use the entire result of the query (so not a delta change), e.g. to display on screen
}

And your query in your grain would look like this, for example:

public class MyGrain : IGrainReactive {
    List<IOtherGrain> OtherGrains;

    public Query<Result> MyQuery(args){
       return Task.WhenAll(this.OtherGrains.Select((g) => await g.OtherQuery());
    }

    public Task AddGrain(IOtherGrain grain) {
        this.OtherGrains.Add(grain);
        return TaskDone.Done;
    }
}

This has the following semantics:

• Whenever the result of the Query changes, this will be propagated to the client and Query.NextUpdateAsync() will return with the new result.
• Even though multiple updates might have been propagated in between 2 calls, .NextUpdateAsync() will just return the latest result (thus possibly skipping some versions of the state), i.e. we are only interested in the latest state.
• These updates will stop whenever the invoked grain hasn't received a keep-alive notification for the configured time or when it is explicitly cancelled using the CancellationToken (this is recursively propagated to other grains on which this query depends).

In order to achieve this, the runtime will perform the following actions:

• When a query is initiated for the first time, it will be executed just as if it is a regular async method call, together with the following additions:
  • Every grain that calls a query will memoize the result of this query.
  • Every grain on which a query was called will set up a subscription to subscribe the caller to this query. The result of this query will also be memoized on this end.
• Whenever a non-query is called on a IReactiveGrain, all the queries that other grains depend on from this grain will be re-executed. If the result of the query is different from the memoized result, the new value will be pushed to the dependents. They will recursively do the same thing, all the way up to the client that executed the query, where Query.NextUpdateAsync() will then return with the new result of the query. Finally, whenever a query is re-executed and it contains calls to other queries, the re-execution will use the memoized results instead of re-invoking those queries.

Thus, In the example code AddGrain(IOtherGrain grain) will trigger a re-execution of MyQuery(args) for this client, but will re-use the memoized results from g.OtherQuery(). Re-executing the queries on every non-query method invocation is very conservative, but it is a good start to keep the implementation simple.

(Note: Since the result of a query is memoized on both ends, update propagation can later be optimised to let the grains only communicate via diff operations.)

This proposal relates to https://github.com/dotnet/orleans/issues/940, except that we plan to go one step further by not requiring the programmer to manually construct an IObservable. Instead we keep track of when data that is dependent on might affect a query and then automatically propagate this.

We have a pretty good idea about the core of this concept, but the API in which this is exposed to the programmer is still in transit. So any feedback is very welcome.

Thanks for your opinion! :)