This needs a response type wrapper.

This seems confusing. The doc says that it returns a stream, but it only seems to take a stream as input. Who is the caller and callee in this scenario?

Why the big change? Was this for testing?

Doesn't this arrive at the same problem? This buffers the whole thing into a single buffer. How does this address the issue?

The problem was being able to stream large snapshots without arbitrarily large gRPC message limits and timeout outs. This addresses that problem. Memory constraint is not the problem we're trying to solve, at least yet.

@anshulpundir Fair enough. :)

This could double (or more!) memory usage.

Here's how we can decide: Do we have to load the entire snapshot into memory to apply it? If yes, then this is fine, but if not, we should buffer this to disk (tmp is fine).

Buffering to disk won't be any better would it ? to apply we'd still need to load the snapshot into memory before passing it to etcd/raft.

Also, I think MemoryStorage::ApplySnapshot() doesn't actually do a memory copy, so we should still just need 1x memory.

There are three components:

1. "The bytes" of the snapshot itself.
2. The structure that gets applied to the store.
3. The store itself (not raft, but the database).

With the current model, I think the minimum is a 2x bump. Ideally, we can stream it in and avoid 1 and 2. Buffering to disk avoids 1, if 2 is necessary.

Looking at the code, there are two components: the entries and the snapshot data, a []byte. From the looks of it, the snapshot is applied always as a []byte, so we can never avoid 2. If a snapshot is applied in this fashion, there may exist a snapshot that has a similar size, which can contribute to bloat.

So, I think I agree with you: this code is probably fine as is, as long as this doesn't get copied (and it doesn't look like it does). Following that, I don't think there is a easy way to avoid the 2x memory bump by applying a snapshot in a stream.

i might be mistaken, but I don't think this code is any worse off than the regular raft message rpc call, except for the allocations and copies required to expand the array from the successive appends.

that said, maybe it's advantageous to take the couple of extra bytes to send the total message length with each stream request, so we can on the first message allocate an appropriately sized byte slice.

I did consider that, but streaming snapshots should not happen that frequently, so this is probably OK for now. We can always make the change to send the size later, if needed.

I would advocate for always using streaming unless the other end doesn't support it.

Can you please elaborate on this ?

I thought a bit more about this. If the message is smaller than GrpcMaxMsgSize, it's less efficient to first try to stream and then fall back to a basic send if unsupported.

Having said that, I do see the value in always streaming, and eventually deprecating the other endpoint, unless there is a performance overhead to streaming a single message.

Can you please also elaborate on this ?

@stevvooe

I also like the idea of always using streaming. It gives the streaming version of the RPC more real-world coverage instead of only using it in unusual corner cases. It would let us eventually deprecate the non-streaming version.

I don't think the mixed-version scenario is important to optimize.

I agree with the point about giving this more real-world coverage. I'll make this change.

Can you put these below with the other non-stdlib imports?

Put this in manager/state/raft/testutils instead of duplicating it in two places.

typo

Using floating point math here doesn't feel right.

holy shit I know right!

you'd prefer just doing if/else ? @aaronlehmann

instead of math.Min why not just do min by hand for integers, so there's no round trip through floats?

chunkSize = payloadSize
if remainingSize < payloadSize {
    chunkSize = remainingSize
}

Use the builtin function copy instead of this byte-by-byte loop.

i don't believe we're mutating the bytes, so there's probably a way to avoid copying and leverage the same underlying storage array for all of the slices.

this would need to be documented, so that someone didn't come later on and break that assumption.

i think i misunderstood this code actually...

Avoiding copy here would be ideal, but I'm not sure if there is a way to do this. I'll make the change if it is.

I think it should be possible to avoid copy @dperny

yeah, we don't need to do any copying at all. instead of this loop, just do

raftMsg.Snapshot.Data = m.Snapshot.Data[snapDataIndex:(snapDataIndex+chunkSize)]

And then you can change the for loop above to be something like

var chunkSize int
for snapDataIndex := 0; snapDataIndex < size; snapDataIndex = snapDataIndex + chunkSize {
    // do the processing in here
}

I also like the idea of always using streaming. It gives the streaming version of the RPC more real-world coverage instead of only using it in unusual corner cases. It would let us eventually deprecate the non-streaming version.

I don't think the mixed-version scenario is important to optimize.

nit: stream.Context(), instead of context.Background(), unless there's some reason I'm not aware of to do so.

same as above, use stream context.

instead of math.Min why not just do min by hand for integers, so there's no round trip through floats?

chunkSize = payloadSize
if remainingSize < payloadSize {
    chunkSize = remainingSize
}

why represent this as 4 << 20? just curious.

Someone suggested it as being cleaner/easier to read than 4 * 1024 * 1024.

lol i disagree strongly with someone but that's bikeshedding.

so this part looks fine to me.

Its a little shorter, but I think its fine either way.

i don't believe we're mutating the bytes, so there's probably a way to avoid copying and leverage the same underlying storage array for all of the slices.

this would need to be documented, so that someone didn't come later on and break that assumption.

dumb question: is GrpcMaxMsgSize actually the optimal message size? Or would a smaller message perform better?

GrpcMaxMsgSize is probably fine for now. I can't think of any reason to treat streamed messages differently than the regular messages.

instead of building an error string, you might want to use golang.google.com/grpc/status.FromError, and compare the code and message separately?

Sounds reasonable

Unsure if I think we should use the stream context here or not.

maybe it's better not to discard the first return value, and instead use it as the return value for stream.SendAndClose below?

Unsure if I think we should use the stream context here or not.

Yea, I'm unclear on this also. I'll think more about this.

i might be mistaken, but I don't think this code is any worse off than the regular raft message rpc call, except for the allocations and copies required to expand the array from the successive appends.

that said, maybe it's advantageous to take the couple of extra bytes to send the total message length with each stream request, so we can on the first message allocate an appropriately sized byte slice.

Trying to use this from testutils causes a circular dependency and moving this test to a transport_test package was just too problematic since unexported fields have been used extensively in the tests. Leaving as is for now, will find a way to address this in the future.

why do you need to use this from testutils?

Keeping it in once place instead of duplicating it. It's also used in raft_test.go

what's this?

The large timeout/grpc message size is no longer needed once we have snapshot streaming. I guess that can be a separate PR.

to be clear, you're derefing m to make a copy of it, yeah?

Yes, since I want to keep the entire message the same except for the fields to be split.

nit, not a blocker. change if you want to, or don't. either way.

check := func(size, expectedMessages int) {
    raftMsg.Snapshot.Data = make([]byte, size)
    msgs := splitSnapshotData(ctx, &raftMsg)
    assert.Equal(t, expectedMesages, len(msgs), "unexpected number of raft messages")
}

numMsgs := int(math.Ceil(float64(snapshotSize) / float64(raftMessagePayloadSize)))
check(raftMessagePayloadSize, numMsgs
check(raftMessagePayloadSize-1, 1)
check(raftMessagePayloadSize*2, 2)
check(0, 0)

These need to have rpc specific types. You can't just reuse the existing request type.