From 203aec456f94756d94877ab3bcd920b19272f9a6 Mon Sep 17 00:00:00 2001
From: jkrvivian <jkrvivian@gmail.com>
Date: Mon, 17 Feb 2020 22:08:13 +0800
Subject: [PATCH] Fix: Log dropping packets every 1000 drops (#255)
* Fix: Log dropping packets every 1000 drops
* use an atomic counter for dropped message because Write() could be called concurrently
* removes redundant zero
* Add external check in test
Co-authored-by: Luca Moser <moser.luca@gmail.com>
---
packages/gossip/neighbor.go | 17 ++++++++++++-----
packages/gossip/neighbor_test.go | 8 ++++----
2 files changed, 16 insertions(+), 9 deletions(-)
diff --git a/packages/gossip/neighbor.go b/packages/gossip/neighbor.go
index cbf39e91..a529aab2 100644
--- a/packages/gossip/neighbor.go
+++ b/packages/gossip/neighbor.go
@@ -11,6 +11,7 @@ import (
"github.com/iotaledger/goshimmer/packages/netutil/buffconn"
"github.com/iotaledger/hive.go/autopeering/peer"
"github.com/iotaledger/hive.go/logger"
+ "go.uber.org/atomic"
)
var (
@@ -19,16 +20,18 @@ var (
)
const (
- neighborQueueSize = 5000
- maxNumReadErrors = 10
+ neighborQueueSize = 5000
+ maxNumReadErrors = 10
+ droppedMessagesThreshold = 1000
)
type Neighbor struct {
*peer.Peer
*buffconn.BufferedConnection
- log *logger.Logger
- queue chan []byte
+ log *logger.Logger
+ queue chan []byte
+ messagesDropped atomic.Int32
wg sync.WaitGroup
closing chan struct{}
@@ -150,6 +153,10 @@ func (n *Neighbor) Write(b []byte) (int, error) {
case <-n.closing:
return 0, nil
default:
- return 0, ErrNeighborQueueFull
+ if n.messagesDropped.Inc() >= droppedMessagesThreshold {
+ n.messagesDropped.Store(0)
+ return 0, ErrNeighborQueueFull
+ }
+ return 0, nil
}
}
diff --git a/packages/gossip/neighbor_test.go b/packages/gossip/neighbor_test.go
index 21d02cf9..36c61534 100644
--- a/packages/gossip/neighbor_test.go
+++ b/packages/gossip/neighbor_test.go
@@ -87,8 +87,8 @@ func TestNeighborParallelWrite(t *testing.T) {
go func() {
defer wg.Done()
for i := 0; i < neighborQueueSize; i++ {
- _, err := neighborA.Write(testData)
- if err == ErrNeighborQueueFull {
+ l, err := neighborA.Write(testData)
+ if err == ErrNeighborQueueFull || l == 0 {
continue
}
assert.NoError(t, err)
@@ -99,8 +99,8 @@ func TestNeighborParallelWrite(t *testing.T) {
go func() {
defer wg.Done()
for i := 0; i < neighborQueueSize; i++ {
- _, err := neighborA.Write(testData)
- if err == ErrNeighborQueueFull {
+ l, err := neighborA.Write(testData)
+ if err == ErrNeighborQueueFull || l == 0 {
continue
}
assert.NoError(t, err)
--
GitLab