2019-08-29 12:42:27 +01:00
|
|
|
package tunnel
|
|
|
|
|
|
|
|
import (
|
2019-10-24 17:51:41 +01:00
|
|
|
"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"
|
2019-10-24 17:51:41 +01:00
|
|
|
"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 {
|
2019-09-04 09:48:05 +01:00
|
|
|
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
|
2019-10-25 14:16:22 +01:00
|
|
|
// 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
|
2019-10-25 14:16:22 +01:00
|
|
|
// 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
|
2019-09-04 09:48:05 +01:00
|
|
|
// 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
|
2019-09-25 12:14:09 +01:00
|
|
|
// 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
|
|
|
|
}
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
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 {
|
2019-09-04 09:48:05 +01:00
|
|
|
l := &link{
|
2019-09-12 16:22:43 -07:00
|
|
|
Socket: s,
|
|
|
|
id: uuid.New().String(),
|
|
|
|
lastKeepAlive: time.Now(),
|
2019-10-25 14:16:22 +01:00
|
|
|
channels: make(map[string]time.Time),
|
2019-10-23 16:05:21 +01:00
|
|
|
closed: make(chan bool),
|
2019-10-24 17:51:41 +01:00
|
|
|
state: make(chan *packet, 64),
|
2019-10-23 16:05:21 +01:00
|
|
|
sendQueue: make(chan *packet, 128),
|
|
|
|
recvQueue: make(chan *packet, 128),
|
2019-09-04 09:48:05 +01:00
|
|
|
}
|
2019-10-24 17:51:41 +01:00
|
|
|
|
|
|
|
// process inbound/outbound packets
|
2019-10-23 16:05:21 +01:00
|
|
|
go l.process()
|
2019-10-24 17:51:41 +01:00
|
|
|
// manage the link state
|
|
|
|
go l.manage()
|
|
|
|
|
2019-09-04 09:48:05 +01:00
|
|
|
return l
|
|
|
|
}
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
|
2019-10-25 14:16:22 +01:00
|
|
|
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]
|
|
|
|
}
|
|
|
|
|
2019-10-25 14:16:22 +01: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 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
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
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 {
|
2019-10-24 17:51:41 +01:00
|
|
|
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
|
|
|
}
|
|
|
|
|
2019-10-24 17:51:41 +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
|
2019-09-04 09:48:05 +01:00
|
|
|
t := time.NewTicker(time.Minute)
|
|
|
|
defer t.Stop()
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
// used to send link state packets
|
2019-10-25 14:16:22 +01:00
|
|
|
send := func(b []byte) error {
|
|
|
|
return l.Send(&transport.Message{
|
2019-10-24 17:51:41 +01:00
|
|
|
Header: map[string]string{
|
|
|
|
"Micro-Method": "link",
|
|
|
|
}, Body: b,
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
// set time now
|
|
|
|
now := time.Now()
|
|
|
|
|
|
|
|
// send the initial rtt request packet
|
|
|
|
send(linkRequest)
|
|
|
|
|
2019-09-04 09:48:05 +01:00
|
|
|
for {
|
|
|
|
select {
|
2019-10-24 17:51:41 +01:00
|
|
|
// exit if closed
|
2019-09-04 09:48:05 +01:00
|
|
|
case <-l.closed:
|
|
|
|
return
|
2019-10-24 17:51:41 +01:00
|
|
|
// 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
|
2019-10-25 14:16:22 +01:00
|
|
|
if err := send(linkResponse); err != nil {
|
|
|
|
l.Lock()
|
|
|
|
l.errCount++
|
|
|
|
l.Unlock()
|
|
|
|
}
|
2019-10-24 17:51:41 +01:00
|
|
|
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)
|
|
|
|
}
|
2019-09-04 09:48:05 +01:00
|
|
|
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
|
2019-09-04 09:48:05 +01:00
|
|
|
if len(kill) == 0 {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// kill the channels!
|
|
|
|
l.Lock()
|
|
|
|
for _, ch := range kill {
|
|
|
|
delete(l.channels, ch)
|
|
|
|
}
|
|
|
|
l.Unlock()
|
2019-10-24 17:51:41 +01:00
|
|
|
|
|
|
|
// fire off a link state rtt packet
|
|
|
|
now = time.Now()
|
|
|
|
send(linkRequest)
|
2019-09-04 09:48:05 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-10-23 16:05:21 +01:00
|
|
|
func (l *link) send(m *transport.Message) error {
|
2019-10-23 16:15:39 +01:00
|
|
|
if m.Header == nil {
|
|
|
|
m.Header = make(map[string]string)
|
2019-10-23 16:05:21 +01:00
|
|
|
}
|
|
|
|
// send the message
|
2019-10-23 16:15:39 +01:00
|
|
|
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()
|
|
|
|
|
2019-09-04 09:48:05 +01:00
|
|
|
select {
|
|
|
|
case <-l.closed:
|
2019-09-04 15:41:57 +01:00
|
|
|
return nil
|
2019-09-04 09:48:05 +01:00
|
|
|
default:
|
2019-10-23 16:39:26 +01:00
|
|
|
l.Socket.Close()
|
2019-09-04 09:48:05 +01:00
|
|
|
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
|
2019-09-25 12:14:09 +01:00
|
|
|
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
|
|
|
}
|
|
|
|
|
2019-10-23 16:15:39 +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
|
|
|
}
|
|
|
|
|
2019-10-23 16:15:39 +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
|
2019-10-23 16:15:39 +01:00
|
|
|
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
|
2019-09-25 12:14:09 +01:00
|
|
|
}
|
|
|
|
|
2019-10-23 16:15:39 +01:00
|
|
|
// 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
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
// set the rate
|
|
|
|
l.setRate(int64(bits), time.Since(now))
|
2019-10-23 16:15:39 +01:00
|
|
|
}
|
|
|
|
|
2019-10-23 16:05:21 +01:00
|
|
|
return nil
|
|
|
|
}
|
2019-09-25 12:14:09 +01:00
|
|
|
|
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
|
2019-09-25 12:14:09 +01:00
|
|
|
}
|
|
|
|
|
2019-10-25 14:16:22 +01:00
|
|
|
// 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:
|
2019-09-25 12:14:09 +01:00
|
|
|
l.RLock()
|
|
|
|
defer l.RUnlock()
|
|
|
|
if l.errCount > 3 {
|
|
|
|
return "error"
|
|
|
|
}
|
2019-09-11 07:11:40 -07:00
|
|
|
return "connected"
|
|
|
|
}
|
|
|
|
}
|