micro/tunnel/link.go

559 lines
10 KiB
Go
Raw Normal View History

2019-08-29 14:42:27 +03:00
package tunnel
import (
"bytes"
"errors"
2019-10-23 18:05:21 +03:00
"io"
2019-08-30 22:05:00 +03:00
"sync"
"time"
2019-08-29 14:42:27 +03:00
"github.com/google/uuid"
"github.com/micro/go-micro/v2/transport"
"github.com/micro/go-micro/v2/util/log"
2019-08-29 14:42:27 +03:00
)
2019-08-30 22:05:00 +03:00
type link struct {
transport.Socket
// transport to use for connections
transport transport.Transport
2019-08-30 22:05:00 +03:00
sync.RWMutex
2019-09-11 17:11:40 +03:00
// stops the link
closed chan bool
2019-12-11 18:23:08 +03:00
// metric used to track metrics
metric chan *metric
// link state channel for testing link
state chan *packet
// send queue for sending packets
2019-10-23 18:05:21 +03:00
sendQueue chan *packet
// receive queue for receiving packets
2019-10-23 18:05:21 +03:00
recvQueue chan *packet
2019-08-30 22:05:00 +03:00
// unique id of this link e.g uuid
// which we define for ourselves
id string
// whether its a loopback connection
// this flag is used by the transport listener
// which accepts inbound quic connections
loopback bool
// whether its actually connected
// dialled side sets it to connected
// after sending the message. the
// listener waits for the connect
connected bool
// the last time we received a keepalive
// on this link from the remote side
lastKeepAlive time.Time
// channels keeps a mapping of channels and last seen
channels map[string]time.Time
2019-10-22 21:38:29 +03:00
// the weighted moving average roundtrip
2019-10-23 18:05:21 +03:00
length int64
2019-10-22 21:38:29 +03:00
// weighted moving average of bits flowing
rate float64
// keep an error count on the link
errCount int
2019-08-30 22:05:00 +03:00
}
2019-10-23 18:05:21 +03:00
// packet send over link
type packet struct {
// message to send or received
message *transport.Message
// status returned when sent
status chan error
// receive related error
err error
}
2019-12-11 18:23:08 +03:00
// metric is used to record link rate
type metric struct {
// amount of data sent
data int
// time taken to send
duration time.Duration
// if an error occurred
status error
}
var (
// the 4 byte 0 packet sent to determine the link state
linkRequest = []byte{0, 0, 0, 0}
// the 4 byte 1 filled packet sent to determine link state
linkResponse = []byte{1, 1, 1, 1}
ErrLinkConnectTimeout = errors.New("link connect timeout")
)
2019-08-29 14:42:27 +03:00
func newLink(s transport.Socket) *link {
l := &link{
Socket: s,
id: uuid.New().String(),
lastKeepAlive: time.Now(),
2019-10-23 18:05:21 +03:00
closed: make(chan bool),
channels: make(map[string]time.Time),
state: make(chan *packet, 64),
2019-10-23 18:05:21 +03:00
sendQueue: make(chan *packet, 128),
recvQueue: make(chan *packet, 128),
2019-12-11 18:23:08 +03:00
metric: make(chan *metric, 128),
}
// process inbound/outbound packets
2019-10-23 18:05:21 +03:00
go l.process()
// manage the link state
go l.manage()
return l
}
func (l *link) connect(addr string) error {
c, err := l.transport.Dial(addr)
if err != nil {
return err
}
l.Lock()
l.Socket = c
l.Unlock()
return nil
}
func (l *link) accept(sock transport.Socket) error {
l.Lock()
l.Socket = sock
l.Unlock()
return nil
}
func (l *link) setLoopback(v bool) {
l.Lock()
l.loopback = v
l.Unlock()
}
// setRate sets the bits per second rate as a float64
func (l *link) setRate(bits int64, delta time.Duration) {
// rate of send in bits per nanosecond
rate := float64(bits) / float64(delta.Nanoseconds())
// default the rate if its zero
if l.rate == 0 {
// rate per second
l.rate = rate * 1e9
} else {
// set new rate per second
l.rate = 0.8*l.rate + 0.2*(rate*1e9)
}
}
// setRTT sets a nanosecond based moving average roundtrip time for the link
2019-10-22 20:43:09 +03:00
func (l *link) setRTT(d time.Duration) {
l.Lock()
2019-10-23 18:05:21 +03:00
if l.length <= 0 {
l.length = d.Nanoseconds()
2019-12-11 17:37:03 +03:00
l.Unlock()
2019-10-22 20:43:09 +03:00
return
}
// https://fishi.devtail.io/weblog/2015/04/12/measuring-bandwidth-and-round-trip-time-tcp-connection-inside-application-layer/
2019-10-23 18:05:21 +03:00
length := 0.8*float64(l.length) + 0.2*float64(d.Nanoseconds())
// set new length
l.length = int64(length)
2019-12-11 17:37:03 +03:00
l.Unlock()
2019-10-23 18:05:21 +03:00
}
func (l *link) delChannel(ch string) {
l.Lock()
delete(l.channels, ch)
l.Unlock()
}
2019-10-25 16:24:37 +03:00
func (l *link) getChannel(ch string) time.Time {
l.RLock()
2019-12-11 17:37:03 +03:00
t := l.channels[ch]
l.RUnlock()
return t
2019-10-25 16:24:37 +03:00
}
func (l *link) setChannel(channels ...string) {
l.Lock()
for _, ch := range channels {
l.channels[ch] = time.Now()
}
l.Unlock()
}
// set the keepalive time
func (l *link) keepalive() {
l.Lock()
l.lastKeepAlive = time.Now()
l.Unlock()
}
2019-10-23 18:05:21 +03:00
// process deals with the send queue
func (l *link) process() {
// receive messages
go func() {
for {
m := new(transport.Message)
err := l.recv(m)
if err != nil {
2019-12-11 18:23:08 +03:00
// record the metric
select {
case l.metric <- &metric{status: err}:
default:
}
2019-10-23 18:05:21 +03:00
}
// process new received message
pk := &packet{message: m, err: err}
// this is our link state packet
if m.Header["Micro-Method"] == "link" {
// process link state message
select {
case l.state <- pk:
2019-12-08 18:02:54 +03:00
case <-l.closed:
return
default:
}
continue
}
// process all messages as is
2019-10-23 18:05:21 +03:00
select {
case l.recvQueue <- pk:
2019-10-23 18:05:21 +03:00
case <-l.closed:
return
}
}
}()
// send messages
for {
select {
case pk := <-l.sendQueue:
// send the message
2019-12-08 18:02:54 +03:00
select {
case pk.status <- l.send(pk.message):
case <-l.closed:
return
}
2019-10-23 18:05:21 +03:00
case <-l.closed:
return
}
}
2019-10-22 20:43:09 +03:00
}
// manage manages the link state including rtt packets and channel mapping expiry
func (l *link) manage() {
// tick over every minute to expire and fire rtt packets
2019-12-11 18:23:08 +03:00
t1 := time.NewTicker(time.Minute)
defer t1.Stop()
// used to batch update link metrics
t2 := time.NewTicker(time.Second * 5)
defer t2.Stop()
// get link id
linkId := l.Id()
// used to send link state packets
send := func(b []byte) error {
return l.Send(&transport.Message{
Header: map[string]string{
"Micro-Method": "link",
"Micro-Link-Id": linkId,
}, Body: b,
})
}
// set time now
now := time.Now()
// send the initial rtt request packet
send(linkRequest)
for {
select {
// exit if closed
case <-l.closed:
return
// process link state rtt packets
case p := <-l.state:
if p.err != nil {
continue
}
// check the type of message
switch {
2019-12-03 10:25:58 +03:00
case bytes.Equal(p.message.Body, linkRequest):
log.Tracef("Link %s received link request", linkId)
// send response
if err := send(linkResponse); err != nil {
l.Lock()
l.errCount++
l.Unlock()
}
2019-12-03 10:25:58 +03:00
case bytes.Equal(p.message.Body, linkResponse):
// set round trip time
d := time.Since(now)
log.Tracef("Link %s received link response in %v", linkId, d)
// set the RTT
l.setRTT(d)
}
2019-12-11 18:23:08 +03:00
case <-t1.C:
// drop any channel mappings older than 2 minutes
var kill []string
killTime := time.Minute * 2
l.RLock()
for ch, t := range l.channels {
if d := time.Since(t); d > killTime {
kill = append(kill, ch)
}
}
l.RUnlock()
2019-09-04 13:58:25 +03:00
// if nothing to kill don't bother with a wasted lock
if len(kill) == 0 {
continue
}
// kill the channels!
l.Lock()
for _, ch := range kill {
delete(l.channels, ch)
}
l.Unlock()
// fire off a link state rtt packet
now = time.Now()
send(linkRequest)
2019-12-11 18:23:08 +03:00
case <-t2.C:
// get a batch of metrics
batch := l.batch()
// skip if there's no metrics
if len(batch) == 0 {
continue
}
// lock once to record a batch
l.Lock()
for _, metric := range batch {
l.record(metric)
}
l.Unlock()
}
}
}
2019-12-11 18:23:08 +03:00
func (l *link) batch() []*metric {
var metrics []*metric
// pull all the metrics
for {
select {
case m := <-l.metric:
metrics = append(metrics, m)
// non blocking return
default:
return metrics
}
}
}
func (l *link) record(m *metric) {
// there's an error increment the counter and bail
if m.status != nil {
l.errCount++
return
}
// reset the counter
l.errCount = 0
// calculate based on data
if m.data > 0 {
// bit sent
bits := m.data * 1024
// set the rate
l.setRate(int64(bits), m.duration)
}
}
2019-10-23 18:05:21 +03:00
func (l *link) send(m *transport.Message) error {
if m.Header == nil {
m.Header = make(map[string]string)
2019-10-23 18:05:21 +03:00
}
// send the message
return l.Socket.Send(m)
2019-10-23 18:05:21 +03:00
}
// recv a message on the link
func (l *link) recv(m *transport.Message) error {
if m.Header == nil {
m.Header = make(map[string]string)
}
// receive the transport message
return l.Socket.Recv(m)
}
2019-10-22 18:50:00 +03:00
// Delay is the current load on the link
func (l *link) Delay() int64 {
2019-10-23 18:05:21 +03:00
return int64(len(l.sendQueue) + len(l.recvQueue))
2019-10-22 18:50:00 +03:00
}
2019-10-22 18:53:47 +03:00
// Current transfer rate as bits per second (lower is better)
2019-10-22 18:50:00 +03:00
func (l *link) Rate() float64 {
2019-10-22 21:38:29 +03:00
l.RLock()
2019-12-11 17:37:03 +03:00
r := l.rate
l.RUnlock()
return r
2019-10-22 18:50:00 +03:00
}
func (l *link) Loopback() bool {
l.RLock()
2019-12-11 17:37:03 +03:00
lo := l.loopback
l.RUnlock()
return lo
}
2019-10-22 20:43:09 +03:00
// Length returns the roundtrip time as nanoseconds (lower is better).
// Returns 0 where no measurement has been taken.
2019-10-22 18:50:00 +03:00
func (l *link) Length() int64 {
2019-10-22 20:43:09 +03:00
l.RLock()
2019-12-11 17:37:03 +03:00
length := l.length
l.RUnlock()
return length
2019-10-22 18:50:00 +03:00
}
2019-09-04 17:41:57 +03:00
func (l *link) Id() string {
l.RLock()
2019-12-11 17:37:03 +03:00
id := l.id
l.RUnlock()
2019-12-11 18:23:08 +03:00
return id
2019-09-04 17:41:57 +03:00
}
func (l *link) Close() error {
2019-10-23 23:24:31 +03:00
l.Lock()
defer l.Unlock()
select {
case <-l.closed:
2019-09-04 17:41:57 +03:00
return nil
default:
2019-10-23 18:39:26 +03:00
l.Socket.Close()
close(l.closed)
2019-08-29 14:42:27 +03:00
}
2019-09-04 17:41:57 +03:00
return nil
2019-08-29 14:42:27 +03:00
}
2019-09-11 17:11:40 +03:00
2019-10-23 18:05:21 +03:00
// Send sencs a message on the link
func (l *link) Send(m *transport.Message) error {
2019-10-23 18:05:21 +03:00
// create a new packet to send over the link
p := &packet{
message: m,
status: make(chan error, 1),
2019-10-22 21:38:29 +03:00
}
2019-12-11 18:23:08 +03:00
// calculate the data sent
dataSent := len(m.Body)
// set header length
for k, v := range m.Header {
dataSent += (len(k) + len(v))
}
// get time now
now := time.Now()
2019-10-23 18:39:26 +03:00
// queue the message
select {
2019-10-23 18:05:21 +03:00
case l.sendQueue <- p:
// in the send queue
2019-10-23 18:39:26 +03:00
case <-l.closed:
return io.EOF
2019-10-22 21:38:29 +03:00
}
// error to use
var err error
2019-10-23 18:05:21 +03:00
// wait for response
select {
case <-l.closed:
return io.EOF
case err = <-p.status:
}
2019-12-11 18:23:08 +03:00
// create a metric with
// time taken, size of package, error status
mt := &metric{
data: dataSent,
duration: time.Since(now),
status: err,
}
2019-12-11 18:23:08 +03:00
// pass back a metric
// do not block
select {
case l.metric <- mt:
default:
}
2019-10-23 18:05:21 +03:00
return nil
}
2019-10-23 18:05:21 +03:00
// Accept accepts a message on the socket
func (l *link) Recv(m *transport.Message) error {
select {
case <-l.closed:
2019-10-23 18:39:26 +03:00
// check if there's any messages left
select {
case pk := <-l.recvQueue:
// check the packet receive error
if pk.err != nil {
return pk.err
}
*m = *pk.message
default:
return io.EOF
}
2019-10-23 18:05:21 +03:00
case pk := <-l.recvQueue:
// check the packet receive error
if pk.err != nil {
return pk.err
}
*m = *pk.message
}
return nil
}
// State can return connected, closed, error
func (l *link) State() string {
2019-09-11 17:11:40 +03:00
select {
case <-l.closed:
return "closed"
default:
l.RLock()
2019-12-11 17:37:03 +03:00
errCount := l.errCount
l.RUnlock()
2019-12-11 18:23:08 +03:00
if errCount > 3 {
return "error"
}
2019-12-11 17:37:03 +03:00
2019-09-11 17:11:40 +03:00
return "connected"
}
}