micro/tunnel/link.go

443 lines
8.4 KiB
Go
Raw Permalink Normal View History

2019-08-29 12:42:27 +01:00
package tunnel
import (
"bytes"
2019-10-23 16:05:21 +01:00
"io"
2019-08-30 20:05:00 +01:00
"sync"
"time"
2019-08-29 12:42:27 +01:00
"github.com/google/uuid"
"github.com/micro/go-micro/transport"
"github.com/micro/go-micro/util/log"
2019-08-29 12:42:27 +01:00
)
2019-08-30 20:05:00 +01:00
type link struct {
transport.Socket
2019-08-30 20:05:00 +01:00
sync.RWMutex
2019-09-11 07:11:40 -07:00
// stops the link
closed chan bool
// link state channel for testing link
state chan *packet
// send queue for sending packets
2019-10-23 16:05:21 +01:00
sendQueue chan *packet
// receive queue for receiving packets
2019-10-23 16:05:21 +01:00
recvQueue chan *packet
2019-08-30 20:05:00 +01: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 19:38:29 +01:00
// the weighted moving average roundtrip
2019-10-23 16:05:21 +01:00
length int64
2019-10-22 19:38:29 +01:00
// weighted moving average of bits flowing
rate float64
// keep an error count on the link
errCount int
2019-08-30 20:05:00 +01:00
}
2019-10-23 16:05:21 +01: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
}
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}
)
2019-08-29 12:42:27 +01:00
func newLink(s transport.Socket) *link {
l := &link{
Socket: s,
id: uuid.New().String(),
lastKeepAlive: time.Now(),
channels: make(map[string]time.Time),
2019-10-23 16:05:21 +01:00
closed: make(chan bool),
state: make(chan *packet, 64),
2019-10-23 16:05:21 +01:00
sendQueue: make(chan *packet, 128),
recvQueue: make(chan *packet, 128),
}
// process inbound/outbound packets
2019-10-23 16:05:21 +01:00
go l.process()
// manage the link state
go l.manage()
return l
}
// 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 18:43:09 +01:00
func (l *link) setRTT(d time.Duration) {
l.Lock()
defer l.Unlock()
2019-10-23 16:05:21 +01:00
if l.length <= 0 {
l.length = d.Nanoseconds()
2019-10-22 18:43:09 +01:00
return
}
// https://fishi.devtail.io/weblog/2015/04/12/measuring-bandwidth-and-round-trip-time-tcp-connection-inside-application-layer/
2019-10-23 16:05:21 +01:00
length := 0.8*float64(l.length) + 0.2*float64(d.Nanoseconds())
// set new length
l.length = int64(length)
}
func (l *link) delChannel(ch string) {
l.Lock()
delete(l.channels, ch)
l.Unlock()
}
2019-10-25 14:24:37 +01:00
func (l *link) getChannel(ch string) time.Time {
l.RLock()
defer l.RUnlock()
return l.channels[ch]
}
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 16:05:21 +01: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 {
l.Lock()
l.errCount++
l.Unlock()
}
// 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:
default:
}
continue
}
// process all messages as is
2019-10-23 16:05:21 +01:00
select {
case l.recvQueue <- pk:
2019-10-23 16:05:21 +01:00
case <-l.closed:
return
}
}
}()
// send messages
for {
select {
case pk := <-l.sendQueue:
// send the message
pk.status <- l.send(pk.message)
case <-l.closed:
return
}
}
2019-10-22 18:43:09 +01: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
t := time.NewTicker(time.Minute)
defer t.Stop()
// used to send link state packets
send := func(b []byte) error {
return l.Send(&transport.Message{
Header: map[string]string{
"Micro-Method": "link",
}, 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 {
case bytes.Compare(p.message.Body, linkRequest) == 0:
log.Tracef("Link %s received link request %v", l.id, p.message.Body)
// send response
if err := send(linkResponse); err != nil {
l.Lock()
l.errCount++
l.Unlock()
}
case bytes.Compare(p.message.Body, linkResponse) == 0:
// set round trip time
d := time.Since(now)
log.Tracef("Link %s received link response in %v", p.message.Body, d)
l.setRTT(d)
}
case <-t.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 11:58:25 +01: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-10-23 16:05:21 +01:00
func (l *link) send(m *transport.Message) error {
if m.Header == nil {
m.Header = make(map[string]string)
2019-10-23 16:05:21 +01:00
}
// send the message
return l.Socket.Send(m)
2019-10-23 16:05:21 +01: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 16:50:00 +01:00
// Delay is the current load on the link
func (l *link) Delay() int64 {
2019-10-23 16:05:21 +01:00
return int64(len(l.sendQueue) + len(l.recvQueue))
2019-10-22 16:50:00 +01:00
}
2019-10-22 16:53:47 +01:00
// Current transfer rate as bits per second (lower is better)
2019-10-22 16:50:00 +01:00
func (l *link) Rate() float64 {
2019-10-22 19:38:29 +01:00
l.RLock()
defer l.RUnlock()
return l.rate
2019-10-22 16:50:00 +01:00
}
2019-10-22 18:43:09 +01:00
// Length returns the roundtrip time as nanoseconds (lower is better).
// Returns 0 where no measurement has been taken.
2019-10-22 16:50:00 +01:00
func (l *link) Length() int64 {
2019-10-22 18:43:09 +01:00
l.RLock()
defer l.RUnlock()
2019-10-23 16:05:21 +01:00
return l.length
2019-10-22 16:50:00 +01:00
}
2019-09-04 15:41:57 +01:00
func (l *link) Id() string {
l.RLock()
defer l.RUnlock()
return l.id
}
func (l *link) Close() error {
2019-10-23 21:24:31 +01:00
l.Lock()
defer l.Unlock()
select {
case <-l.closed:
2019-09-04 15:41:57 +01:00
return nil
default:
2019-10-23 16:39:26 +01:00
l.Socket.Close()
close(l.closed)
2019-08-29 12:42:27 +01:00
}
2019-09-04 15:41:57 +01:00
return nil
2019-08-29 12:42:27 +01:00
}
2019-09-11 07:11:40 -07:00
2019-10-23 16:05:21 +01:00
// Send sencs a message on the link
func (l *link) Send(m *transport.Message) error {
2019-10-23 16:05:21 +01:00
// create a new packet to send over the link
p := &packet{
message: m,
status: make(chan error, 1),
2019-10-22 19:38:29 +01:00
}
// get time now
now := time.Now()
2019-10-23 16:39:26 +01:00
// check if its closed first
2019-10-23 16:05:21 +01:00
select {
case <-l.closed:
return io.EOF
2019-10-23 16:39:26 +01:00
default:
}
// queue the message
select {
2019-10-23 16:05:21 +01:00
case l.sendQueue <- p:
// in the send queue
2019-10-23 16:39:26 +01:00
case <-l.closed:
return io.EOF
2019-10-22 19:38:29 +01:00
}
// error to use
var err error
2019-10-23 16:05:21 +01:00
// wait for response
select {
case <-l.closed:
return io.EOF
case err = <-p.status:
}
l.Lock()
defer l.Unlock()
// there's an error increment the counter and bail
if err != nil {
l.errCount++
2019-10-23 16:05:21 +01:00
return err
}
// reset the counter
l.errCount = 0
// calculate the data sent
dataSent := len(m.Body)
// set header length
for k, v := range m.Header {
dataSent += (len(k) + len(v))
}
// calculate based on data
if dataSent > 0 {
// bit sent
bits := dataSent * 1024
// set the rate
l.setRate(int64(bits), time.Since(now))
}
2019-10-23 16:05:21 +01:00
return nil
}
2019-10-23 16:05:21 +01:00
// Accept accepts a message on the socket
func (l *link) Recv(m *transport.Message) error {
select {
case <-l.closed:
2019-10-23 16:39:26 +01: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 16:05:21 +01: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 07:11:40 -07:00
select {
case <-l.closed:
return "closed"
default:
l.RLock()
defer l.RUnlock()
if l.errCount > 3 {
return "error"
}
2019-09-11 07:11:40 -07:00
return "connected"
}
}