/
pex_reactor.go
358 lines (313 loc) · 9.8 KB
/
pex_reactor.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
package p2p
import (
"bytes"
"fmt"
"reflect"
"time"
"github.com/tendermint/go-wire"
cmn "github.com/tendermint/tmlibs/common"
)
const (
// PexChannel is a channel for PEX messages
PexChannel = byte(0x00)
// period to ensure peers connected
defaultEnsurePeersPeriod = 30 * time.Second
minNumOutboundPeers = 1
maxPexMessageSize = 1048576 // 1MB
// maximum pex messages one peer can send to us during `msgCountByPeerFlushInterval`
defaultMaxMsgCountByPeer = 1000
msgCountByPeerFlushInterval = 1 * time.Hour
maxPeersConnectNum = 2
)
// PEXReactor handles PEX (peer exchange) and ensures that an
// adequate number of peers are connected to the switch.
//
// It uses `AddrBook` (address book) to store `NetAddress`es of the peers.
//
// ## Preventing abuse
//
// For now, it just limits the number of messages from one peer to
// `defaultMaxMsgCountByPeer` messages per `msgCountByPeerFlushInterval` (1000
// msg/hour).
//
// NOTE [2017-01-17]:
// Limiting is fine for now. Maybe down the road we want to keep track of the
// quality of peer messages so if peerA keeps telling us about peers we can't
// connect to then maybe we should care less about peerA. But I don't think
// that kind of complexity is priority right now.
type PEXReactor struct {
BaseReactor
book *AddrBook
ensurePeersPeriod time.Duration
// tracks message count by peer, so we can prevent abuse
msgCountByPeer *cmn.CMap
maxMsgCountByPeer uint16
}
// NewPEXReactor creates new PEX reactor.
func NewPEXReactor(b *AddrBook) *PEXReactor {
r := &PEXReactor{
book: b,
ensurePeersPeriod: defaultEnsurePeersPeriod,
msgCountByPeer: cmn.NewCMap(),
maxMsgCountByPeer: defaultMaxMsgCountByPeer,
}
r.BaseReactor = *NewBaseReactor("PEXReactor", r)
return r
}
// OnStart implements BaseService
func (r *PEXReactor) OnStart() error {
if err := r.BaseReactor.OnStart(); err != nil {
return err
}
err := r.book.Start()
if err != nil && err != cmn.ErrAlreadyStarted {
return err
}
go r.ensurePeersRoutine()
go r.flushMsgCountByPeer()
return nil
}
// OnStop implements BaseService
func (r *PEXReactor) OnStop() {
r.BaseReactor.OnStop()
r.book.Stop()
}
// GetChannels implements Reactor
func (r *PEXReactor) GetChannels() []*ChannelDescriptor {
return []*ChannelDescriptor{
{
ID: PexChannel,
Priority: 1,
SendQueueCapacity: 10,
},
}
}
// AddPeer implements Reactor by adding peer to the address book (if inbound)
// or by requesting more addresses (if outbound).
func (r *PEXReactor) AddPeer(p Peer) {}
// RemovePeer implements Reactor.
func (r *PEXReactor) RemovePeer(p Peer, reason interface{}) {
// If we aren't keeping track of local temp data for each peer here, then we
// don't have to do anything.
}
// Receive implements Reactor by handling incoming PEX messages.
func (r *PEXReactor) Receive(chID byte, src Peer, msgBytes []byte) {
srcAddrStr := src.NodeInfo().RemoteAddr
srcAddr, err := NewNetAddressString(srcAddrStr)
if err != nil {
// this should never happen. TODO: cancel conn
r.Logger.Error("Error in Receive: invalid peer address", "addr", srcAddrStr, "err", err)
return
}
r.IncrementMsgCountForPeer(srcAddrStr)
if r.ReachedMaxMsgCountForPeer(srcAddrStr) {
r.Logger.Error("Maximum number of messages reached for peer", "peer", srcAddrStr)
// TODO remove src from peers?
return
}
_, msg, err := DecodeMessage(msgBytes)
if err != nil {
r.Logger.Error("Error decoding message", "err", err)
return
}
r.Logger.Debug("Received message", "src", src, "chId", chID, "msg", msg)
switch msg := msg.(type) {
case *pexRequestMessage:
// src requested some peers.
// NOTE: we might send an empty selection
r.SendAddrs(src, r.book.GetSelection())
case *pexAddrsMessage:
// We received some peer addresses from src.
// TODO: (We don't want to get spammed with bad peers)
for _, addr := range msg.Addrs {
if addr != nil {
r.book.AddAddress(addr, srcAddr)
}
}
default:
r.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
}
}
// RequestPEX asks peer for more addresses.
func (r *PEXReactor) RequestPEX(p Peer) {
p.Send(PexChannel, struct{ PexMessage }{&pexRequestMessage{}})
}
// SendAddrs sends addrs to the peer.
func (r *PEXReactor) SendAddrs(p Peer, addrs []*NetAddress) {
p.Send(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
}
// SetEnsurePeersPeriod sets period to ensure peers connected.
func (r *PEXReactor) SetEnsurePeersPeriod(d time.Duration) {
r.ensurePeersPeriod = d
}
// SetMaxMsgCountByPeer sets maximum messages one peer can send to us during 'msgCountByPeerFlushInterval'.
func (r *PEXReactor) SetMaxMsgCountByPeer(v uint16) {
r.maxMsgCountByPeer = v
}
// ReachedMaxMsgCountForPeer returns true if we received too many
// messages from peer with address `addr`.
// NOTE: assumes the value in the CMap is non-nil
func (r *PEXReactor) ReachedMaxMsgCountForPeer(addr string) bool {
return r.msgCountByPeer.Get(addr).(uint16) >= r.maxMsgCountByPeer
}
// Increment or initialize the msg count for the peer in the CMap
func (r *PEXReactor) IncrementMsgCountForPeer(addr string) {
var count uint16
countI := r.msgCountByPeer.Get(addr)
if countI != nil {
count = countI.(uint16)
}
count++
r.msgCountByPeer.Set(addr, count)
}
// Ensures that sufficient peers are connected. (continuous)
func (r *PEXReactor) ensurePeersRoutine() {
// Randomize when routine starts
ensurePeersPeriodMs := r.ensurePeersPeriod.Nanoseconds() / 1e6
time.Sleep(time.Duration(ensurePeersPeriodMs) * time.Millisecond)
// fire once immediately.
r.ensurePeers()
// fire periodically
ticker := time.NewTicker(r.ensurePeersPeriod)
for {
select {
case <-ticker.C:
r.ensurePeers()
case <-r.Quit:
ticker.Stop()
return
}
}
}
// ensurePeers ensures that sufficient peers are connected. (once)
//
// Old bucket / New bucket are arbitrary categories to denote whether an
// address is vetted or not, and this needs to be determined over time via a
// heuristic that we haven't perfected yet, or, perhaps is manually edited by
// the node operator. It should not be used to compute what addresses are
// already connected or not.
//
// TODO Basically, we need to work harder on our good-peer/bad-peer marking.
// What we're currently doing in terms of marking good/bad peers is just a
// placeholder. It should not be the case that an address becomes old/vetted
// upon a single successful connection.
func (r *PEXReactor) ensurePeers() {
numOutPeers, _, numDialing := r.Switch.NumPeers()
numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
r.Logger.Info("Ensure peers", "numOutPeers", numOutPeers, "numDialing", numDialing, "numToDial", numToDial)
if numToDial <= 0 {
return
}
// bias to prefer more vetted peers when we have fewer connections.
// not perfect, but somewhate ensures that we prioritize connecting to more-vetted
// NOTE: range here is [10, 90]. Too high ?
newBias := cmn.MinInt(numOutPeers, 8)*10 + 10
toDial := make(map[string]*NetAddress)
// Try maxAttempts times to pick numToDial addresses to dial
maxAttempts := numToDial * 3
for i := 0; i < maxAttempts && len(toDial) < numToDial; i++ {
try := r.book.PickAddress(newBias)
if try == nil {
continue
}
if r.Switch.HasAddr(try.String()) {
continue
}
if _, selected := toDial[try.IP.String()]; selected {
continue
}
if dialling := r.Switch.IsDialing(try); dialling {
continue
}
// XXX: Should probably use pubkey as peer key ...
if connected := r.Switch.Peers().Has(try.String()); connected {
continue
}
r.Logger.Info("Will dial address", "addr", try)
toDial[try.IP.String()] = try
}
if len(toDial) == 0 {
r.Logger.Info("no address to dial")
r.Logger.Info("Force dial AllSeeds", "addrs", AllSeeds)
netAddrs, _ := NewNetAddressStrings(AllSeeds)
for _, try := range netAddrs {
if r.Switch.HasAddr(try.String()) {
continue
}
if _, selected := toDial[try.IP.String()]; selected {
continue
}
if dialling := r.Switch.IsDialing(try); dialling {
continue
}
// XXX: Should probably use pubkey as peer key ...
if connected := r.Switch.Peers().Has(try.String()); connected {
continue
}
r.Logger.Info("Will dial address", "addr", try)
toDial[try.IP.String()] = try
}
}
// Dial picked addresses
for _, item := range toDial {
if r.Switch.Peers().Size() < maxPeersConnectNum {
_, err := r.Switch.DialPeerWithAddress(item, false)
if err != nil {
r.book.MarkAttempt(item)
continue
}
}
break
}
}
func (r *PEXReactor) flushMsgCountByPeer() {
ticker := time.NewTicker(msgCountByPeerFlushInterval)
for {
select {
case <-ticker.C:
r.msgCountByPeer.Clear()
case <-r.Quit:
ticker.Stop()
return
}
}
}
//-----------------------------------------------------------------------------
// Messages
const (
msgTypeRequest = byte(0x01)
msgTypeAddrs = byte(0x02)
)
// PexMessage is a primary type for PEX messages. Underneath, it could contain
// either pexRequestMessage, or pexAddrsMessage messages.
type PexMessage interface{}
var _ = wire.RegisterInterface(
struct{ PexMessage }{},
wire.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
wire.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
)
// DecodeMessage implements interface registered above.
func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
msgType = bz[0]
n := new(int)
r := bytes.NewReader(bz)
msg = wire.ReadBinary(struct{ PexMessage }{}, r, maxPexMessageSize, n, &err).(struct{ PexMessage }).PexMessage
return
}
/*
A pexRequestMessage requests additional peer addresses.
*/
type pexRequestMessage struct {
}
func (m *pexRequestMessage) String() string {
return "[pexRequest]"
}
/*
A message with announced peer addresses.
*/
type pexAddrsMessage struct {
Addrs []*NetAddress
}
func (m *pexAddrsMessage) String() string {
return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
}