Change over to Message Passing edge communication #1425
Conversation
|
@desa At this stage all nodes are using the legacy wrapper on the new MPI edge type. A few tests are still not passing so something is broken, but I wanted to get your feedback before going too much further. |
edge/consumer.go
Outdated
| func (ec *Consumer) Run() error { | ||
| for msg, ok := ec.edge.Next(); ok; msg, ok = ec.edge.Next() { | ||
| switch typ := msg.Type(); typ { | ||
| case BeginBatch: |
There was a problem hiding this comment.
My only concern, and I'm not sure that it's even valid, is that switch for every message will end up being expensive.
My concern comes from a note in go-lua
This exercises more of the bytecode interpreter’s inner loop. Here we see the performance impact of Go’s switch implementation. Both go-lua and gopher-lua are an order of magnitude slower than the C Lua interpreter.
I'm not exactly sure what to do about it, but my initial though is to do something like you would with a parser, Where we receive a BeginBatchMessage and therefore we know that all further messages should be PointMessages, until we receive a EndBatchMessage. This does assume that that there are some semantics to the order in which messages are received.
There was a problem hiding this comment.
I like that idea. I am too concerned that this will be less performant. For the most part I have ignored those questions and just been focusing on the API design first. Then we can come back through and fix slow spots. We will need to develop good benchmarks to test.
There was a problem hiding this comment.
I played around with this idea, and it currently can only remove two cases from the switch statement and has only one level of nesting. It doesn't seem like much of a gain for the complexity it adds. And I don't foresee many nested messages types emerging.
When we circle back to performance test this we can reconsider.
|
@nathanielc so far I really like it. Just have the one thought about the switch statement on each message. |
nathanielc
force-pushed
the
nc-mpi
branch
2 times, most recently
from
1865678 to
2c26302
Compare
nathanielc
changed the title
|
@desa This is ready for a final review. |
|
|
||
| // ForwardReceiver handles messages as they arrive and can return a message to be forwarded to output edges. | ||
| // If a returned messages is nil, no message is forwarded. | ||
| type ForwardReceiver interface { |
There was a problem hiding this comment.
Why don't all of the nodes use a ForwardReceiver? Reading through the combine.go, it seems like it would have been possible to implement it using a ForwardReceiver instead of a Receiver. Am I missing something?
join.go
Outdated
| } | ||
|
|
||
| func (n *JoinNode) Barrier(src int, b edge.BarrierMessage) error { | ||
| //TODO: implement barrier |
There was a problem hiding this comment.
Doesn't this need to forward the barrier message?
| "github.com/influxdata/kapacitor/expvar" | ||
| "github.com/influxdata/kapacitor/models" | ||
| "github.com/influxdata/kapacitor/pipeline" | ||
| "github.com/influxdata/kapacitor/timer" | ||
| "github.com/pkg/errors" | ||
| ) | ||
|
|
||
| type JoinNode struct { |
There was a problem hiding this comment.
Why JoinNode doesn't need to implement Receiver?
There was a problem hiding this comment.
Ah because there are different types of Consumers :)
Change over to Message Passing edge communication
This PR introduces the new API for using MPI semantics for edge communication between nodes of the task DAG.
A LegacyEdge type is introduced that implements the original behavior on top of the new API, this way the existing code remains the same so it can be incrementally switched over to the new API.
All nodes have been reimplemented using the new API. The LegacyEdge type has been removed.