Skip to content

Commit bc6b4b8

Browse files
nickeskovnickeskov
authored
Proto tx messages deduplicator (#1728)
* Wrap proto messages FSM chan with filtering. * Add comment for 'Handle' function. * Remove debug log record about when a message is filtered. * Fix 'TODO' with proto messages chan size metrics. * Move proto messages chan wrapper to the 'helpers.go'. * Refactored a bit. * Add 'TestDeduplicateProtoTxMessages'. * Renaming for clarity.
1 parent 37207f7 commit bc6b4b8

4 files changed

Lines changed: 276 additions & 8 deletions

File tree

‎pkg/node/helpers.go‎

Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
package node
22

33
import (
4+
"context"
5+
"hash/maphash"
6+
"sync"
7+
8+
"github.com/wavesplatform/gowaves/pkg/p2p/peer"
49
"github.com/wavesplatform/gowaves/pkg/proto"
510
"github.com/wavesplatform/gowaves/pkg/state"
611
"github.com/wavesplatform/gowaves/pkg/types"
@@ -33,3 +38,176 @@ func maybeEnableExtendedApi(state startProvidingExtendedApi, lastBlock *proto.Bl
3338
}
3439
return nil
3540
}
41+
42+
type safeMap[K comparable, V any] struct {
43+
mu sync.Mutex
44+
m map[K]V
45+
}
46+
47+
func newSafeMap[K comparable, V any]() *safeMap[K, V] {
48+
return &safeMap[K, V]{
49+
m: make(map[K]V),
50+
}
51+
}
52+
53+
func (s *safeMap[K, V]) SetIfNew(key K, value V) bool {
54+
s.mu.Lock()
55+
defer s.mu.Unlock()
56+
if _, exists := s.m[key]; exists {
57+
return false
58+
}
59+
s.m[key] = value
60+
return true
61+
}
62+
63+
func (s *safeMap[K, V]) Delete(key K) {
64+
s.mu.Lock()
65+
defer s.mu.Unlock()
66+
delete(s.m, key)
67+
}
68+
69+
type protoMessageWrapper struct {
70+
protoMsg peer.ProtoMessage
71+
messageID uint64
72+
}
73+
74+
// messageIDFilterFunc is a function type that takes a proto.Message and returns a uint64
75+
// representing the deterministic message ID. This is used to filter messages based on
76+
// their message ID: it can be calculated from the message payload, or from the whole message itself.
77+
//
78+
// If the returned message ID is 0, the message bypasses filtering.
79+
type messageIDFilterFunc func(message proto.Message) uint64
80+
81+
type txMessagePayloadIDFilter struct {
82+
seed maphash.Seed
83+
}
84+
85+
func (f txMessagePayloadIDFilter) Filter(message proto.Message) uint64 {
86+
switch msg := message.(type) {
87+
case *proto.TransactionMessage:
88+
return maphash.Bytes(f.seed, msg.Transaction)
89+
case *proto.PBTransactionMessage:
90+
return maphash.Bytes(f.seed, msg.Transaction)
91+
default:
92+
return 0 // non-transaction messages bypass filtering
93+
}
94+
}
95+
96+
func messagesSplitter(
97+
ctx context.Context,
98+
origMessageCh <-chan peer.ProtoMessage,
99+
noFilteredChan, filteredChan chan<- protoMessageWrapper,
100+
sm *safeMap[uint64, struct{}],
101+
messageIDFilter messageIDFilterFunc,
102+
) {
103+
for {
104+
var (
105+
msg peer.ProtoMessage
106+
messageID uint64
107+
)
108+
select {
109+
case <-ctx.Done():
110+
return
111+
case msg = <-origMessageCh:
112+
messageID = messageIDFilter(msg.Message)
113+
}
114+
bypassesFiltering := messageID == 0 // if messageID is 0, it means the message no need to be filtered.
115+
needToWrite := bypassesFiltering || sm.SetIfNew(messageID, struct{}{})
116+
if !needToWrite {
117+
continue // skip a message if message ID already exists in the map
118+
}
119+
outCh := noFilteredChan
120+
if !bypassesFiltering {
121+
outCh = filteredChan
122+
}
123+
select {
124+
case <-ctx.Done():
125+
return
126+
case outCh <- protoMessageWrapper{msg, messageID}:
127+
}
128+
}
129+
}
130+
131+
func runMessagesMerger(
132+
ctx context.Context, wg *sync.WaitGroup,
133+
noFilteredChan, filteredChan <-chan protoMessageWrapper,
134+
sm *safeMap[uint64, struct{}],
135+
) <-chan peer.ProtoMessage {
136+
wg.Add(1)
137+
outMessageCh := make(chan peer.ProtoMessage) // intentionally unbuffered channel because of map usage
138+
go func() {
139+
defer wg.Done()
140+
for {
141+
var wMsg protoMessageWrapper
142+
select {
143+
case <-ctx.Done():
144+
return
145+
case wMsg = <-noFilteredChan:
146+
case wMsg = <-filteredChan:
147+
}
148+
select {
149+
case <-ctx.Done():
150+
return
151+
case outMessageCh <- wMsg.protoMsg:
152+
if wMsg.messageID != 0 {
153+
sm.Delete(wMsg.messageID) // remove message ID from the map after sending the message
154+
}
155+
}
156+
}
157+
}()
158+
return outMessageCh
159+
}
160+
161+
type chanLenProvider[T any] <-chan T
162+
163+
func (c chanLenProvider[T]) Len() int { return len(c) }
164+
165+
type aggregatedLenProvider []lenProvider
166+
167+
func (a aggregatedLenProvider) Len() int {
168+
l := 0
169+
for _, p := range a {
170+
l += p.Len()
171+
}
172+
return l
173+
}
174+
175+
type waiter interface {
176+
Wait()
177+
}
178+
179+
func deduplicateProtoMessages(
180+
ctx context.Context, origMessageCh <-chan peer.ProtoMessage,
181+
messageIDFilter messageIDFilterFunc,
182+
) (<-chan peer.ProtoMessage, lenProvider, waiter) {
183+
const noFilteredChanSize = 1000
184+
185+
wg := &sync.WaitGroup{}
186+
sm := newSafeMap[uint64, struct{}]()
187+
188+
noFilteredChan := make(chan protoMessageWrapper, noFilteredChanSize)
189+
filteredChan := make(chan protoMessageWrapper, max(cap(origMessageCh)-noFilteredChanSize, noFilteredChanSize))
190+
191+
wg.Add(1)
192+
go func() { // run messages splitter with filtering
193+
defer wg.Done()
194+
messagesSplitter(ctx, origMessageCh, noFilteredChan, filteredChan, sm, messageIDFilter)
195+
}()
196+
197+
out := runMessagesMerger(ctx, wg, noFilteredChan, filteredChan, sm)
198+
199+
lp := aggregatedLenProvider{
200+
chanLenProvider[peer.ProtoMessage](origMessageCh),
201+
chanLenProvider[protoMessageWrapper](noFilteredChan),
202+
chanLenProvider[protoMessageWrapper](filteredChan),
203+
chanLenProvider[peer.ProtoMessage](out),
204+
}
205+
return out, lp, wg
206+
}
207+
208+
func deduplicateProtoTxMessages(
209+
ctx context.Context, origMessageCh <-chan peer.ProtoMessage,
210+
) (<-chan peer.ProtoMessage, lenProvider, waiter) {
211+
payloadIDFilter := txMessagePayloadIDFilter{seed: maphash.MakeSeed()}
212+
return deduplicateProtoMessages(ctx, origMessageCh, payloadIDFilter.Filter)
213+
}

‎pkg/node/helpers_internal_test.go‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package node
2+
3+
import (
4+
"context"
5+
"testing"
6+
"time"
7+
8+
"github.com/stretchr/testify/assert"
9+
10+
"github.com/wavesplatform/gowaves/pkg/p2p/peer"
11+
"github.com/wavesplatform/gowaves/pkg/proto"
12+
)
13+
14+
func testContext(t *testing.T) (context.Context, context.CancelFunc) {
15+
t.Helper()
16+
ctx, cancel := context.WithCancel(context.Background())
17+
t.Cleanup(cancel)
18+
return ctx, cancel
19+
}
20+
21+
func filledMsgChan(values ...proto.Message) chan peer.ProtoMessage {
22+
ch := make(chan peer.ProtoMessage, len(values))
23+
for _, v := range values {
24+
ch <- peer.ProtoMessage{
25+
ID: nil, // ID is not used in this test.
26+
Message: v,
27+
}
28+
}
29+
return ch
30+
}
31+
32+
func readProtoMessages(t *testing.T, ch <-chan peer.ProtoMessage, expectedMsgCount int) []proto.Message {
33+
timeout := time.NewTimer(5 * time.Second)
34+
messages := make([]proto.Message, 0, expectedMsgCount)
35+
for range expectedMsgCount {
36+
select {
37+
case <-timeout.C:
38+
t.Fatalf("timed out waiting for messages, check expectedMsgCount=%d", expectedMsgCount)
39+
case msg := <-ch:
40+
messages = append(messages, msg.Message)
41+
}
42+
}
43+
select {
44+
case msg := <-ch:
45+
t.Fatalf("unexpected message=%+v received, expected exactly %d messages", msg, expectedMsgCount)
46+
default:
47+
// No more messages expected, this is fine.
48+
}
49+
return messages
50+
}
51+
52+
func TestDeduplicateProtoTxMessages(t *testing.T) {
53+
messages := filledMsgChan(
54+
&proto.TransactionMessage{Transaction: nil},
55+
&proto.TransactionMessage{Transaction: nil}, // duplicate
56+
&proto.PBTransactionMessage{Transaction: nil}, // considered as duplicate because payloads are equal
57+
&proto.TransactionMessage{Transaction: []byte{1, 2, 3}},
58+
&proto.TransactionMessage{Transaction: []byte{1, 2, 3}}, // duplicate
59+
&proto.PBTransactionMessage{Transaction: []byte{1, 2, 3}}, // duplicate, payloads are equal
60+
&proto.PBTransactionMessage{Transaction: []byte{42}},
61+
&proto.TransactionMessage{Transaction: []byte{42}}, // duplicate, payloads are equal
62+
// -- non-transaction messages
63+
&proto.ScoreMessage{Score: nil},
64+
&proto.ScoreMessage{Score: nil},
65+
&proto.ScoreMessage{Score: []byte{1, 2, 3}},
66+
&proto.ScoreMessage{Score: []byte{1, 2, 3}},
67+
&proto.ScoreMessage{Score: []byte{42}},
68+
&proto.ScoreMessage{Score: []byte{42}},
69+
&proto.ScoreMessage{Score: []byte{21}},
70+
)
71+
72+
expected := []proto.Message{
73+
&proto.TransactionMessage{Transaction: nil},
74+
&proto.TransactionMessage{Transaction: []byte{1, 2, 3}},
75+
&proto.PBTransactionMessage{Transaction: []byte{42}},
76+
// -- non-transaction messages
77+
&proto.ScoreMessage{Score: nil},
78+
&proto.ScoreMessage{Score: nil},
79+
&proto.ScoreMessage{Score: []byte{1, 2, 3}},
80+
&proto.ScoreMessage{Score: []byte{1, 2, 3}},
81+
&proto.ScoreMessage{Score: []byte{42}},
82+
&proto.ScoreMessage{Score: []byte{42}},
83+
&proto.ScoreMessage{Score: []byte{21}},
84+
}
85+
86+
ctx, cancel := testContext(t)
87+
deduplicated, _, wg := deduplicateProtoTxMessages(ctx, messages)
88+
defer wg.Wait()
89+
defer cancel()
90+
91+
actual := readProtoMessages(t, deduplicated, len(expected))
92+
assert.ElementsMatch(t, expected, actual)
93+
}

‎pkg/node/node.go‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -156,10 +156,11 @@ func (a *Node) Run(
156156
ctx context.Context, p peer.Parent, internalMessageCh <-chan messages.InternalMessage,
157157
networkMsgCh <-chan network.InfoMessage, syncPeer *network.SyncPeer,
158158
) {
159-
protoMessagesChan := chanLenProvider[peer.ProtoMessage](p.MessageCh)
159+
messageCh, protoMessagesLenProvider, wg := deduplicateProtoTxMessages(ctx, p.MessageCh)
160+
defer wg.Wait()
160161

161162
go a.runOutgoingConnections(ctx)
162-
go a.runInternalMetrics(ctx, protoMessagesChan)
163+
go a.runInternalMetrics(ctx, protoMessagesLenProvider)
163164
go a.runIncomingConnections(ctx)
164165

165166
tasksCh := make(chan tasks.AsyncTask, 10)
@@ -207,7 +208,7 @@ func (a *Node) Run(
207208
default:
208209
zap.S().Warnf("[%s] Unknown network info message '%T'", m.State.State, msg)
209210
}
210-
case mess := <-p.MessageCh:
211+
case mess := <-messageCh:
211212
zap.S().Named(logging.FSMNamespace).Debugf("[%s] Network message '%T' received from '%s'",
212213
m.State.State, mess.Message, mess.ID.ID())
213214
action, ok := actions[reflect.TypeOf(mess.Message)]
@@ -235,10 +236,6 @@ type lenProvider interface {
235236
Len() int
236237
}
237238

238-
type chanLenProvider[T any] <-chan T
239-
240-
func (c chanLenProvider[T]) Len() int { return len(c) }
241-
242239
func (a *Node) runInternalMetrics(ctx context.Context, protoMessagesChan lenProvider) {
243240
ticker := time.NewTicker(metricInternalChannelSizeUpdateInterval)
244241
defer ticker.Stop()

‎pkg/p2p/peer/handle.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ func Handle(ctx context.Context, peer Peer, parent Parent, remote Remote) error
6464
var errSentToParent bool // if errSentToParent is true then we need to wait ctx cancellation
6565
for {
6666
select {
67-
case <-ctx.Done():
67+
case <-ctx.Done(): // context is unique for each peer, so when passed 'peer' arg is closed, context is canceled
6868
//TODO: On Done() Err() contains only Canceled or DeadlineExceeded.
6969
// Actually, those errors are only logged in different places and not used to alter behavior.
7070
// Consider removing wrapping. For now, if context was canceled no error is passed by.

0 commit comments

Comments
 (0)