forked from FactomProject/factomd
/
peer.go
401 lines (344 loc) · 9.88 KB
/
peer.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
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
package p2p
import (
"encoding/gob"
"encoding/json"
"fmt"
"net"
"sync"
"time"
log "github.com/sirupsen/logrus"
)
var peerLogger = packageLogger.WithField("subpack", "peer")
// Peer is an active connection to an endpoint in the network.
// Represents one lifetime of a connection and should not be restarted
type Peer struct {
net *Network
conn net.Conn
metrics ReadWriteCollector
prot Protocol
// current state, read only "constants" after the handshake
IsIncoming bool
Endpoint Endpoint
NodeID uint32 // a nonce to distinguish multiple nodes behind one endpoint
Hash string // This is more of a connection ID than hash right now.
stopper sync.Once
stop chan bool
lastPeerRequest time.Time
peerShareAsk bool
peerShareDeliver chan *Parcel
lastPeerSend time.Time
// communication channels
send ParcelChannel // parcels from Send() are added here
status chan peerStatus // the controller's notification channel
data chan peerParcel // the controller's data channel
registered bool
// Metrics
metricsMtx sync.RWMutex
connected time.Time
lastReceive time.Time // Keep track of how long ago we talked to the peer.
lastSend time.Time // Keep track of how long ago we talked to the peer.
totalParcelsSent uint64
totalParcelsReceived uint64
totalBytesSent uint64
totalBytesReceived uint64
bpsDown, bpsUp float64
mpsDown, mpsUp float64
dropped uint64
// logging
logger *log.Entry
}
func newPeer(net *Network, status chan peerStatus, data chan peerParcel) *Peer {
p := &Peer{}
p.net = net
p.status = status
p.data = data
p.logger = peerLogger.WithFields(log.Fields{
"node": net.conf.NodeName,
})
p.stop = make(chan bool, 1)
p.peerShareDeliver = nil
p.logger.Debugf("Creating blank peer")
return p
}
func (p *Peer) bootstrapProtocol(hs *Handshake, conn net.Conn, decoder *gob.Decoder, encoder *gob.Encoder) error {
v := hs.Header.Version
if v > p.net.conf.ProtocolVersion {
v = p.net.conf.ProtocolVersion
}
///fmt.Printf("@@@ %d %+v %s\n", v, hs.Header, conn.RemoteAddr())
switch v {
case 9:
v9 := new(ProtocolV9)
v9.init(p, decoder, encoder)
p.prot = v9
// v9 starts with a peer request
hsParcel := new(Parcel)
hsParcel.Address = hs.Header.TargetPeer
hsParcel.Payload = hs.Payload
hsParcel.Type = TypePeerRequest
/* TODO at this point, peer is not fully initialized
if !p.deliver(hsParcel) {
return fmt.Errorf("unable to deliver peer request to controller")
}*/
case 10:
v10 := new(ProtocolV10)
v10.init(p, decoder, encoder)
p.prot = v10
default:
return fmt.Errorf("unknown protocol version %d", v)
}
return nil
}
// StartWithHandshake performs a basic handshake maneouver to establish the validity
// of the connection. Immediately sends a Peer Request upon connection and waits for the
// response, which can be any parcel. The information in the header is verified, especially
// the port.
//
// The handshake ensures that ALL peers have a valid Port field to start with.
// If there is no reply within the specified HandshakeTimeout config setting, the process
// fails
//
// For outgoing connections, it is possible the endpoint will reject due to being full, in which
// case this function returns an error AND a list of alternate endpoints
func (p *Peer) StartWithHandshake(ep Endpoint, con net.Conn, incoming bool) ([]Endpoint, error) {
tmplogger := p.logger.WithField("addr", ep.IP)
timeout := time.Now().Add(p.net.conf.HandshakeTimeout)
nonce := []byte(fmt.Sprintf("%x", p.net.instanceID))
// upgrade connection to a metrics connection
p.metrics = NewMetricsReadWriter(con)
p.conn = con
handshake := newHandshake(p.net.conf, nonce)
decoder := gob.NewDecoder(p.metrics) // pipe gob through the metrics writer
encoder := gob.NewEncoder(p.metrics)
con.SetWriteDeadline(timeout)
con.SetReadDeadline(timeout)
//fmt.Printf("@@@ %+v %s\n", handshake.Header, con.RemoteAddr())
failfunc := func(err error) ([]Endpoint, error) {
tmplogger.WithError(err).Debug("Handshake failed")
p.conn.Close()
return nil, err
}
err := encoder.Encode(handshake)
if err != nil {
return failfunc(fmt.Errorf("Failed to send handshake to incoming connection"))
}
var reply Handshake
err = decoder.Decode(&reply)
if err != nil {
return failfunc(fmt.Errorf("Failed to read handshake from incoming connection"))
}
// check basic structure
if err = reply.Valid(p.net.conf); err != nil {
return failfunc(err)
}
// loopback detection
if string(reply.Payload) == string(nonce) {
return failfunc(fmt.Errorf("loopback"))
}
if err = p.bootstrapProtocol(&reply, con, decoder, encoder); err != nil {
return failfunc(err)
}
if reply.Header.Type == TypeRejectAlternative {
con.Close()
tmplogger.Debug("con rejected with alternatives")
var rawShare []Endpoint
err := json.Unmarshal(reply.Payload, &rawShare)
if err != nil {
return nil, fmt.Errorf("unable to parse alternatives: %s", err.Error())
}
filtered := make([]Endpoint, 0, len(rawShare))
for _, ep := range rawShare {
if ep.Valid() {
filtered = append(filtered, ep)
}
}
return filtered, fmt.Errorf("connection rejected")
}
// initialize channels
ep.Port = reply.Header.PeerPort
p.Endpoint = ep
p.NodeID = uint32(reply.Header.NodeID)
p.Hash = fmt.Sprintf("%s:%s %08x", ep.IP, ep.Port, p.NodeID)
p.send = newParcelChannel(p.net.conf.ChannelCapacity)
p.IsIncoming = incoming
p.connected = time.Now()
p.logger = p.logger.WithFields(log.Fields{
"hash": p.Hash,
"address": p.Endpoint.IP,
"Port": p.Endpoint.Port,
"Version": p.prot.Version(),
})
p.status <- peerStatus{peer: p, online: true}
p.registered = true
go p.sendLoop()
go p.readLoop()
go p.statLoop()
return nil, nil
}
// Stop disconnects the peer from its active connection
func (p *Peer) Stop() {
p.stopper.Do(func() {
p.logger.Debug("Stopping peer")
sc := p.send
p.send = nil
close(p.stop)
if p.conn != nil {
p.conn.Close()
}
if sc != nil {
close(sc)
}
if p.registered {
p.status <- peerStatus{peer: p, online: false}
}
})
}
func (p *Peer) String() string {
return p.Hash
}
func (p *Peer) Send(parcel *Parcel) {
_, dropped := p.send.Send(parcel)
p.dropped += uint64(dropped)
}
func (p *Peer) statLoop() {
ticker := time.NewTicker(time.Second * 5)
for {
select {
case <-ticker.C:
p.metricsMtx.Lock()
mw, mr, bw, br := p.metrics.Collect()
p.bpsDown = float64(br) / 5
p.bpsUp = float64(bw) / 5
p.totalBytesReceived += br
p.totalBytesSent += bw
p.mpsDown = float64(mr) / 5
p.mpsUp = float64(mw) / 5
p.totalParcelsReceived += mr
p.totalParcelsSent += mw
p.metricsMtx.Unlock()
case <-p.stop:
return
}
}
}
func (p *Peer) readLoop() {
if p.net.prom != nil {
p.net.prom.ReceiveRoutines.Inc()
defer p.net.prom.ReceiveRoutines.Dec()
}
defer p.conn.Close() // close connection on fatal error
for {
p.conn.SetReadDeadline(time.Now().Add(p.net.conf.ReadDeadline))
msg, err := p.prot.Receive()
if err != nil {
p.logger.WithError(err).Debug("connection error (readLoop)")
p.Stop()
return
}
if err := msg.Valid(); err != nil {
p.logger.WithError(err).Warnf("received invalid msg, disconnecting peer")
p.Stop()
if p.net.prom != nil {
p.net.prom.Invalid.Inc()
}
return
}
// metrics
p.metricsMtx.Lock()
p.lastReceive = time.Now()
p.metricsMtx.Unlock()
// stats
if p.net.prom != nil {
p.net.prom.ParcelsReceived.Inc()
p.net.prom.ParcelSize.Observe(float64(len(msg.Payload)) / 1024)
if msg.IsApplicationMessage() {
p.net.prom.AppReceived.Inc()
}
}
msg.Address = p.Hash // always set sender = peer
if !p.deliver(msg) {
return
}
}
}
// deliver is a blocking delivery of this peer's messages to the peer manager.
func (p *Peer) deliver(parcel *Parcel) bool {
select {
case p.data <- peerParcel{peer: p, parcel: parcel}:
case <-p.stop:
return false
}
return true
}
// sendLoop listens to the Outgoing channel, pushing all data from there
// to the tcp connection
func (p *Peer) sendLoop() {
if p.net.prom != nil {
p.net.prom.SendRoutines.Inc()
defer p.net.prom.SendRoutines.Dec()
}
defer p.conn.Close() // close connection on fatal error
for {
select {
case <-p.stop:
return
case parcel := <-p.send:
if parcel == nil {
p.logger.Error("Received <nil> pointer from application")
continue
}
p.conn.SetWriteDeadline(time.Now().Add(p.net.conf.WriteDeadline))
err := p.prot.Send(parcel)
if err != nil { // no error is recoverable
p.logger.WithError(err).Debug("connection error (sendLoop)")
p.Stop()
return
}
// metrics
p.metricsMtx.Lock()
p.lastSend = time.Now()
p.metricsMtx.Unlock()
// stats
if p.net.prom != nil {
p.net.prom.ParcelsSent.Inc()
p.net.prom.ParcelSize.Observe(float64(len(parcel.Payload)+32) / 1024) // TODO FIX
if parcel.IsApplicationMessage() {
p.net.prom.AppSent.Inc()
}
}
}
}
}
// GetMetrics returns live metrics for this connection
func (p *Peer) GetMetrics() PeerMetrics {
p.metricsMtx.RLock()
defer p.metricsMtx.RUnlock()
pt := "regular"
if p.net.controller.isSpecial(p.Endpoint) {
pt = "special_config"
}
return PeerMetrics{
Hash: p.Hash,
PeerAddress: p.Endpoint.IP,
MomentConnected: p.connected,
LastReceive: p.lastReceive,
LastSend: p.lastSend,
BytesReceived: p.totalBytesReceived,
BytesSent: p.totalBytesSent,
MessagesReceived: p.totalParcelsReceived,
MessagesSent: p.totalParcelsSent,
Incoming: p.IsIncoming,
PeerType: pt,
MPSDown: p.mpsDown,
MPSUp: p.mpsUp,
BPSDown: p.bpsDown,
BPSUp: p.bpsUp,
ConnectionState: fmt.Sprintf("v%s", p.prot.Version()),
Capacity: p.Capacity(),
Dropped: p.dropped,
}
}
// Capacity is a wrapper for the send channel's Capacity
func (p *Peer) Capacity() float64 {
return p.send.Capacity()
}