Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Memberlist now forwards only changes, not full original received message #4419

Merged
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Memberlist now forwards only changes, not full original received mess…
…age.

Signed-off-by: Peter Štibraný <[email protected]>
pstibrany committed Aug 13, 2021
commit d934e2a7bd39b872651dd84d2446c47e1e3e4b2b
16 changes: 2 additions & 14 deletions pkg/ring/kv/memberlist/memberlist_client.go
Original file line number Diff line number Diff line change
@@ -980,20 +980,8 @@ func (m *KV) NotifyMsg(msg []byte) {
} else if version > 0 {
m.notifyWatchers(kvPair.Key)

m.addSentMessage(message{
Time: time.Now(),
Size: len(msg),
Pair: kvPair,
Version: version,
Changes: changes,
})

// Forward this message
// Memberlist will modify message once this function returns, so we need to make a copy
msgCopy := append([]byte(nil), msg...)

// forward this message further
m.queueBroadcast(kvPair.Key, mod.MergeContent(), version, msgCopy)
// Don't resend original message, but only changes.
m.broadcastNewValue(kvPair.Key, mod, version, codec)
}
}

94 changes: 94 additions & 0 deletions pkg/ring/kv/memberlist/memberlist_client_test.go
Original file line number Diff line number Diff line change
@@ -1074,3 +1074,97 @@ func TestMessageBuffer(t *testing.T) {
assert.Len(t, buf, 2)
assert.Equal(t, size, 75)
}

func TestNotifyMsgResendsOnlyChanges(t *testing.T) {
codec := dataCodec{}

cfg := KVConfig{}
// We will be checking for number of messages in the broadcast queue, so make sure to use known retransmit factor.
cfg.RetransmitMult = 1
cfg.Codecs = append(cfg.Codecs, codec)

kv := NewKV(cfg, log.NewNopLogger())
require.NoError(t, services.StartAndAwaitRunning(context.Background(), kv))
defer services.StopAndAwaitTerminated(context.Background(), kv)

client, err := NewClient(kv, codec)
require.NoError(t, err)

// No broadcast messages from KV at the beginning.
require.Equal(t, 0, len(kv.GetBroadcasts(0, math.MaxInt32)))

now := time.Now()

require.NoError(t, client.CAS(context.Background(), key, func(in interface{}) (out interface{}, retry bool, err error) {
d := getOrCreateData(in)
d.Members["a"] = member{Timestamp: now.Unix(), State: JOINING}
d.Members["b"] = member{Timestamp: now.Unix(), State: JOINING}
return d, true, nil
}))

// Check that new instance is broadcasted about just once.
assert.Equal(t, 1, len(kv.GetBroadcasts(0, math.MaxInt32)))
require.Equal(t, 0, len(kv.GetBroadcasts(0, math.MaxInt32)))

kv.NotifyMsg(marshalKeyValuePair(t, key, codec, &data{
Members: map[string]member{
"a": {Timestamp: now.Unix() - 5, State: ACTIVE},
"b": {Timestamp: now.Unix() + 5, State: ACTIVE, Tokens: []uint32{1, 2, 3}},
"c": {Timestamp: now.Unix(), State: ACTIVE},
}}))

// Check two things here:
// 1) state of value in KV store
// 2) broadcast message only has changed members

d := getData(t, client, key)
assert.Equal(t, &data{
Members: map[string]member{
"a": {Timestamp: now.Unix(), State: JOINING, Tokens: []uint32{}}, // unchanged, timestamp too old
"b": {Timestamp: now.Unix() + 5, State: ACTIVE, Tokens: []uint32{1, 2, 3}},
"c": {Timestamp: now.Unix(), State: ACTIVE, Tokens: []uint32{}},
}}, d)

bs := kv.GetBroadcasts(0, math.MaxInt32)
require.Equal(t, 1, len(bs))

d = decodeDataFromMarshalledKeyValuePair(t, bs[0], key, codec)
assert.Equal(t, &data{
Members: map[string]member{
// "a" is not here, because it wasn't changed by the message.
"b": {Timestamp: now.Unix() + 5, State: ACTIVE, Tokens: []uint32{1, 2, 3}},
"c": {Timestamp: now.Unix(), State: ACTIVE},
}}, d)
}

func decodeDataFromMarshalledKeyValuePair(t *testing.T, marshalledKVP []byte, key string, codec dataCodec) *data {
kvp := KeyValuePair{}
require.NoError(t, kvp.Unmarshal(marshalledKVP))
require.Equal(t, key, kvp.Key)

val, err := codec.Decode(kvp.Value)
require.NoError(t, err)
d, ok := val.(*data)
require.True(t, ok)
return d
}

func marshalKeyValuePair(t *testing.T, key string, codec codec.Codec, value interface{}) []byte {
data, err := codec.Encode(value)
require.NoError(t, err)

kvp := KeyValuePair{Key: key, Codec: codec.CodecID(), Value: data}
data, err = kvp.Marshal()
require.NoError(t, err)
return data
}

func getOrCreateData(in interface{}) *data {
// Modify value that was passed as a parameter.
// Client takes care of concurrent modifications.
r, ok := in.(*data)
if !ok || r == nil {
return &data{Members: map[string]member{}}
}
return r
}