[REEF-2054] Elastic group communication: broadcast

[interesaaat]

This PR introduces the new elastic group communication framework. For the moment only the broadcast operator with flat topology is implemented. Further operators and topologies will be added in successive PRs.

This PR uses some of the pieces of PR #1479 so the latter should be merged before merging the former.

JIRA:
REEF-2054 Elastic Broadcast

[motus]

OK, I think I'll start it slow. I'll bundle my comments into groups of 20 or so, and keep reviewing as you push the updates. This way we can work in parallel.

[motus]

Shouldn't we write it to the log instead?

[interesaaat]

Should we? This is the client, I see that often we print directly to console, no?

[motus]

why not just

foreach (var op in _operators)
{
    op.ResetPosition();
}

[interesaaat]

This is tricky. When you have iterators in the pipeline you only reset each time within the context of an iteration. The for loop is used because the position tells where to actually start to reset.

[motus]

oh, I see. I have not noticed that it starts from the _position. Then, are we sure that it never starts from -1?

[interesaaat]

Yes. ResetOperatorPositions is private and only used inside MoveNext where the first operation is _position++.

[motus]

I don't like that cast. Can we enforce that _operators is a list of IElasticIterator elements instead?

[interesaaat]

Not really because _operators contains the sequence of all operators composing the workflow. I could make all operators be an IElasticIterator of size 0, but this is less clean than using a cast I believe. I am open for suggestions tho.

[motus]

I see. I actually would not mind having an empty IElasticIterator implementation, but it is up to you here.

P.S. Note that we don't have to implement the actual iterator: IElasticOperator can just inherit from IEnumerable and return an empty sequence from .GetEnumerator() by default.

[interesaaat]

What about this:

IElasticIterator iteratorOperator;                
switch(_operators[_position])
{
    case IElasticIterator iter:
        iteratorOperator = iter;
        break;
     default:
         throw new IllegalStateException("Operator not Iterator");
         break;
}

[interesaaat]

Ready for the second round!

[motus]

Few very minor comments for today. Check for naming conventions and make sure you use {0}-style string interpolation in logging instead of string concatenation or $"". (Note that for exceptions it's the other way around)

[motus]

A few more substantial comments this time. I think we can shorten the code quite a bit

[motus]

Few very minor comments for today. Check for naming conventions and make > sure you use {0}-style string interpolation in logging instead of string > concatenation or $"". (Note that for exceptions it's the other way > around) Can you please explain to me the intuition behind this?

Sure! In case of explicit string concatenation or $"..." - style interpolation, the entire string is evaluated *before* the method invocation. This is exactly what we need for exceptions and most of other methods and constructors as well. For logging, the situation is different, as we don't know if the string will be used at all. If the log level is higher than the priority of the current log message, the logging call is basically a no-op, and the string parameter will be ignored. That is, if we build a string *before* the logging call, it can be a big waste, especially if the log message is long and expensive to construct. This is why for logging we want to perform string interpolation *inside* the logging method, so the logger can do it only when it knows for sure that we need to log that string. Sometimes we even wrap the logging call into Logger.IsLoggable() check, if interpolation at the logger level is not possible. Internal logging string interpolation uses {0}-style placeholders, hence the rule: use {0} style for logging, $"..." for everything else. Hope that helps! :) Cheers, Sergiy.

[motus]

Added some more comments. will continue tomorrow

[motus]

left a few minor comments. Will add more soon!

[motus]

we can define constructor

public DefaultFailureState(DefaultFailureStates state)

to avoid the explicit cast here

[interesaaat]

Mmm not sure. Because I will have to do a cast later on because FailureState is an int (cannot be a DefaultFailureState because it's defined in the base class).

[motus]

Yeah we have to discuss that. I still cannot think of something elegant.

[motus]

Another round of comments - will add more tomorrow!

[motus]

oh, I see. I have not noticed that it starts from the _position. Then, are we sure that it never starts from -1?

[motus]

added a few more comments. To be continued!

[motus]

few more things

[motus]

added a few more comments. will continue tomorrow

[motus]

Note to self: why ref? isn't it against the idea of asynchronous event handlers? Also, maybe we should separate input and output messages? i.e. do one of:

// mesages in, messages out, in the most generic container
IEnumerable<IElasticDriverMessage> OnTaskMessage(
    ITaskMessage message, IEnumerable<IElasticDriverMessage> driverMessages);

// messages in, async handler for messages out
void OnTaskMessage(ITaskMessage message,
    IEnumerable<IElasticDriverMessage> driverMessages,
    EventHandler<IEnumerable<IElasticDriverMessage>> handler = null); // or a delegate

need to track the usage of this interface and give it more thought later...

[motus]

By the way, why are we serializing everything like that? is it for Java interoperability?

[interesaaat]

Not really. Where you suggesting to use codecs? I found that the previous group communication uses a similar strategy and I simply piggyback on that. I actually like to have the serialization logic within the message class.

[motus]

Here's another round of comments. Also, please remove the trailing spaces that popped up in the last commit 55aedfa

[motus]

man, this +/- 1 thing is really confusing. let's come back to it later and discuss some more elegant solution

[interesaaat]

I know! the fact is that task infos are internally stored 0-indexed while externally they are used as 1-indexed. The solution here could be to just 0-index everywhere.

[interesaaat]

Here's another round of comments. Also, please remove the trailing spaces that popped up in the last commit 55aedfa

Sorry @motus which trailing spaces are you referring to?

[motus]

here's some more

[motus]

just

System.Threading.Tasks.Task.Factory.StartNew(Clock.Run, TaskCreationOptions.LongRunning);

[motus]

Also, I feel a bit uneasy about starting a new thread from the injectable constructor. it's hard to track when exactly it will run and when we'll stop it. I would rather have a separate .Start() method (maybe not in this class), and the corresponding Stop()/Dispose()

[interesaaat]

I didn't 100% get it. You are suggesting something like starting the clock AFTER (outside) the constructor? So to flow will be (1) inject the constructor, (2) invoke Start?

[motus]

maybe

_completed |= _stages.Values.All(stage => stage.IsCompleted);

and there is no need for if

[interesaaat]

Why there is no need for if? I want to print only once when we move from not complete to complete.

[motus]

few more comments

[motus]

I really don't like that we have three different representations for the failure state - int, enum, and class/interface. That leads to awkward casts and news. I am still thinking about the most elegant way to resolve this; one approach would be to remove the DefaultFailureStates enum, and go old Java style, like this:

public sealed class DefaultFailureState : IFailureState
{
    public static readonly DefaultFailureState Continue = new DefaultFailureState(0);
    public static readonly DefaultFailureState ContinueAndReconfigure = new DefaultFailureState(1);
    // ...

then here we can just write

_failureStatus = _failureStatus.Merge(DefaultFailureState.ContinueAndReconfigure);

but I am not 100% sure... maybe we can keep the enum, ditch the IFailureState interface, and use implicit operators to convert between the class and enum..

[interesaaat]

I agree that is weird. I personally like the IFailureState interface because it allows to implement different failure states and related ways of merging them (this is the behavior I want for failure states). If I have to chose, I would prefer to maintain the interface and implement the default behavior in a different way (e.g., as you were suggesting, without the enum).

[motus]

also, if we are OK with mutable the FailureState property (and it is currently fully read and write accessible from the outside - maybe we should tighten it up a bit?), we can write e.g.

_failureStatus.Update(DefaultFailureState.ContinueAndReschedule);

[interesaaat]

Are you proposing to implement a default behavior for Merge when mutating? I like it!

[interesaaat]

Let me take track of the major changes I will have to do (once you are done with the "syntactic" pass). I will update this list as we agree on the changes.

• remove the id - 1, +1 thing. Move everything to 0-indexed;
• configurations are immutable: fix the methods passing ref configurations
• change custom retry logic with Microsoft.Practices.EnterpriseLibrary.TransientFaultHandling
• discuss on incompatibility between DefaultFailureStates and IFailureState. Should we remove enum or remove IFailureState?

[motus]

added a few minor things. Will write more later today

[motus]

Note that we don't lock the _schedule here - is that Ok?

[interesaaat]

I copied this from wake. I think it is probably ok: additions are synchronized so in the worst case you will remove and event that was just added and not the one you were waiting for.

[motus]

Do we really need _isDisposed flag, when we already have separate ones for task and the context? I would simply write the .Dispose() method as

public void Dispose()
{
    DisposeTask();
    DisposeActiveContext();
}

-- after all, we don't call .Dispose() that often (and we don't set _isDisposed in .DropRuntime() anyway 😄)

[interesaaat]

We should discuss on this. Disposing and dropping runtimes are different. You drop runtimes when the tasks fail, while you dispose them when the task manager finishes / fails. I think that more than removing _isDisposed I should add it in other places so that we cannot add a runtime to a disposed task for instance.

[motus]

Done with the Failures folder. Left a few comments, all minor. Still, the thing that bothers me most is type incompatibility between DefaultFailureStates and IFailureState. We should think of something more elegant.

[motus]

added a few nit picks

[motus]

Yeah we have to discuss that. I still cannot think of something elegant.

[motus]

Should we make T contravariant? i.e.

public interface ISender<in T>

[interesaaat]

I made it contravariant, but I am not sure about the use case. In the sense that users will specify anyway the type on the driver side. Did you have any specific use case in mind?

[motus]

Done with the logical operators. Switching to the physical ones now..

[motus]

halfway through the Task package; will add more comments tomorrow.

[motus]

we should probably think about using protobuf instead of our own serialization. Let's discuss it later.

[motus]

added a few more comments

[motus]

extra space before !Send(); also, this code looks like another good candidate to be replaced with retry policy mechanism

[motus]

By the way, I read up on the retry policies available for .NET, and it looks like the Microsoft.Practices.TransientFaultHandling I've mentioned earlier (and used elsewhere in REEF) is now deprecated. The best replacement seems to be the Polly project. So we can use MS Practices now as we already have the dependency, and later switch to Polly (in a separate PR, of course)

[motus]

few more nit picks

[motus]

a few minor suggestions. will add more soon

[motus]

left a few comments on ITopology and its implementations

[motus]

_taskStage is now null by default. maybe,

_taskStage = _taskStage ?? Utils.GetTaskStages(taskId);

[interesaaat]

But this will do a check and reassignment every time instead of just a check, no?

[motus]

shorter version:

return _rootId + "\n" + string.Join(" ", _root.Children.Select(
    node => node.FailState == DataNodeState.Reachable ? "" + node.TaskId : "X"));

(by the way, shouldn't we print IDs of unreachable tasks? say, $"UNREACHABLE:{node.TaskId}"?)

[interesaaat]

I dunno. The failing node should be already present in the log because the task manager logs it once it receives a failure. This is more sort of like "hey we got a failure, this is the new state of world".

[motus]

Few more comments, and that concludes my second full pass over the entire code

[motus]

trim trailing spaces, add blank line after each multiline declaration

[interesaaat]

Wait, no indentation on multiline declarations?

[motus]

is there a chance we'll need to override these handlers in derived classes? if yes, we should probably we make them virtual

[interesaaat]

I don't think OnError etc. will be ever used because I believe errors will surface directly as crashes.

[motus]

in fact, we can pass children into the OperatorTopologyWithDefaultCommunication constructor and initialize the dictionary once:

protected OperatorTopologyWithDefaultCommunication(
    // ...
    int disposeTimeout,
    IEnumerable<KeyValuePair<int, string>> children = null)
{
    _children = children == null
                ? new ConcurrentDictionary<int, string>()
                : new ConcurrentDictionary<int, string>(children);

and here do something like

: base(
    // ...
    disposeTimeout,
    children.Select(child =>
        new KeyValuePair<int, string>(child, Utils.BuildTaskId(stageName, child))))

[interesaaat]

I am not sure about this. I want to maintain the population of _children in the subclass because it is operator specific.

[interesaaat]

@motus I am checking the protobuf thing and is not that easy to do. I mean, I can probably change the serialization/deserialization of messages to use protobuff, but to completely use protobuff end to end I will have to go and change the network service (which is not my code) and this may require some indefinite amount of time. Do you think it is ok if I only try protobuff for my messages?

[motus]

Hi Matteo, Glad that you are back to REEF! Thanks for looking at it. I meant only the messages that you create and serialize on your own, not all the Wake stuff. I am on vacation toll Wednesday - let's discuss it next week when I am back! Thank you, Sergiy.

…

On Tue, Jul 16, 2019 at 5:22 PM Matteo Interlandi ***@***.***> wrote: @motus <https://github.com/motus> I am checking the protobuf thing and is not that easy to do. I mean, I can probably change the serialization/deserialization of messages to use protobuff, but to completely use protobuff end to end I will have to go and change the network service (which is not my code) and this may require some indefinite amount of time. Do you think it is ok if I only try protobuff for my messages? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1487?email_source=notifications&email_token=AAAHVFCQMSEEVFNR24PNZBTP7ZQ43A5CNFSM4GNZUE7KYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD2CUH2Q#issuecomment-512050154>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAAHVFHM7ER6NYM4KTXP2A3P7ZQ43ANCNFSM4GNZUE7A> .