feat(peerlog): add a bit of backoff logic
Steven Allen committed
Apr 27, 2020 at 19:37 UTC
a10a14fc0629229f2d8b42438ba79c05209d3a57
1 file changed
+66
-31
plugin/plugins/peerlog/peerlog.go
+66
-31
@@ -3,6 +3,7 @@ package peerlog
3
import (
4
"fmt"
5
"sync/atomic"
6
+ "time"
7
8
core "github.com/ipfs/go-ipfs/core"
9
plugin "github.com/ipfs/go-ipfs/plugin"
@@ -18,6 +19,13 @@ var log = logging.Logger("plugin/peerlog")
19
20
type eventType int
21
22
+var (
23
+ // size of the event queue buffer
24
+ eventQueueSize = 64 * 1024
25
+ // number of events to drop when busy.
26
+ busyDropAmount = eventQueueSize / 8
27
+)
28
+
29
const (
30
eventConnect eventType = iota
31
eventIdentify
@@ -60,53 +68,80 @@ func (*peerLogPlugin) Version() string {
68
69
// Init initializes plugin
70
func (pl *peerLogPlugin) Init(*plugin.Environment) error {
63
- pl.events = make(chan plEvent, 64*1024)
71
+ pl.events = make(chan plEvent, eventQueueSize)
72
return nil
73
}
74
75
func (pl *peerLogPlugin) collectEvents(node *core.IpfsNode) {
68
- go func() {
69
- ctx := node.Context()
76
+ ctx := node.Context()
77
71
- dlog := log.Desugar()
72
- for {
73
- dropped := atomic.SwapUint64(&pl.droppedCount, 0)
74
- if dropped > 0 {
75
- dlog.Error("dropped events", zap.Uint64("count", dropped))
76
- }
78
+ busyCounter := 0
79
+ dlog := log.Desugar()
80
+ for {
81
+ // Deal with dropped events.
82
+ dropped := atomic.SwapUint64(&pl.droppedCount, 0)
83
+ if dropped > 0 {
84
+ busyCounter++
85
78
- var e plEvent
86
+ // sleep a bit to give the system a chance to catch up with logging.
87
select {
88
+ case <-time.After(time.Duration(busyCounter) * time.Second):
89
case <-ctx.Done():
90
return
82
- case e = <-pl.events:
91
}
92
85
- peerID := zap.String("peer", e.peer.Pretty())
86
-
87
- switch e.kind {
88
- case eventConnect:
89
- dlog.Info("connected", peerID)
90
- case eventIdentify:
91
- agent, err := node.Peerstore.Get(e.peer, "AgentVersion")
92
- switch err {
93
- case nil:
94
- case peerstore.ErrNotFound:
95
- continue
93
+ // drain 1/8th of the backlog backlog so we
94
+ // don't immediately run into this situation
95
+ // again.
96
+ loop:
97
+ for i := 0; i < busyDropAmount; i++ {
98
+ select {
99
+ case <-pl.events:
100
+ dropped++
101
default:
97
- dlog.Error("failed to get agent version", zap.Error(err))
98
- continue
102
+ break loop
103
}
100
-
101
- agentS, ok := agent.(string)
102
- if !ok {
103
- continue
104
- }
105
- dlog.Info("identified", peerID, zap.String("agent", agentS))
104
}
105
+
106
+ // Add in any events we've dropped in the mean-time.
107
+ dropped += atomic.SwapUint64(&pl.droppedCount, 0)
108
+
109
+ // Report that we've dropped events.
110
+ dlog.Error("dropped events", zap.Uint64("count", dropped))
111
+ } else {
112
+ busyCounter = 0
113
}
108
- }()
114
115
+ var e plEvent
116
+ select {
117
+ case <-ctx.Done():
118
+ return
119
+ case e = <-pl.events:
120
+ }
121
+
122
+ peerID := zap.String("peer", e.peer.Pretty())
123
+
124
+ switch e.kind {
125
+ case eventConnect:
126
+ dlog.Info("connected", peerID)
127
+ case eventIdentify:
128
+ agent, err := node.Peerstore.Get(e.peer, "AgentVersion")
129
+ switch err {
130
+ case nil:
131
+ case peerstore.ErrNotFound:
132
+ continue
133
+ default:
134
+ dlog.Error("failed to get agent version", zap.Error(err))
135
+ continue
136
+ }
137
+
138
+ agentS, ok := agent.(string)
139
+ if !ok {
140
+ continue
141
+ }
142
+ dlog.Info("identified", peerID, zap.String("agent", agentS))
143
+ }
144
+ }
145
}
146
147
func (pl *peerLogPlugin) emit(evt eventType, p peer.ID) {