gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-12 23:32:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Using ping to iterate state management for dialer and listener. (#44)

Browse Source
This commit is contained in:
Michael Quigley
2020-03-26 10:47:53 -04:00
parent 6f3a11b9b8
commit 682ece0149
4 changed files with 106 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/xlink_transwarp/dialer.go
+1 -1
View File
@@ -37,7 +37,7 @@ func (self *dialer) Dial(addressString string, linkId *identity.TokenId) error {
if err := readMessage(conn, self); err == nil {
select {
case <-waitCh:
xlink := newImpl(linkId, conn)
xlink := newImpl(linkId, conn, address)
if err := self.accepter.Accept(xlink); err != nil {
return fmt.Errorf("error accepting outgoing Xlink (%w)", err)
}
router/xlink_transwarp/listener.go
+41 -8
View File
@@ -22,8 +22,12 @@ import (
"github.com/netfoundry/ziti-foundation/identity/identity"
"github.com/sirupsen/logrus"
"net"
"time"
)
/*
* xlink.Listener
*/
func (self *listener) Listen() error {
listener, err := net.ListenUDP("udp", self.config.bindAddress)
if err != nil {
@@ -31,6 +35,7 @@ func (self *listener) Listen() error {
}
self.listener = listener
go self.listen()
go self.ping()
return nil
}
@@ -38,6 +43,25 @@ func (self *listener) GetAdvertisement() string {
return self.config.advertiseAddress.String()
}
/*
* xlink.Xlink
*/
func (self *listener) HandleHello(linkId *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr) {
xlinkImpl := newImpl(linkId, conn, peer)
if err := self.accepter.Accept(xlinkImpl); err == nil {
self.peers[peer.String()] = xlinkImpl
if err := writeHello(linkId, self.listener, peer); err == nil {
logrus.Infof("sent hello [%s] to peer [%s]", linkId.Token, peer)
} else {
logrus.Errorf("error sending hello [%s] to peer at [%s] (%v)", linkId.Token, peer, err)
}
}
}
/*
* listener
*/
func (self *listener) listen() {
for {
if err := readMessage(self.listener, self); err != nil {
@@ -48,15 +72,21 @@ func (self *listener) listen() {
}
}
func (self *listener) HandleHello(linkId *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr) {
xlink := newImpl(linkId, conn)
if err := self.accepter.Accept(xlink); err == nil {
self.peers[peer.String()] = xlink
func (self *listener) ping() {
for {
time.Sleep(pingCycleDelayMs * time.Millisecond)
if err := writeHello(linkId, self.listener, peer); err == nil {
logrus.Infof("sent hello [%s] to peer [%s]", linkId.Token, peer)
} else {
logrus.Errorf("error sending hello [%s] to peer at [%s] (%v)", linkId.Token, peer, err)
for _, xlinkImpl := range self.peers {
/*
* Send
*/
if time.Since(xlinkImpl.lastPingTx).Milliseconds() <= time.Since(xlinkImpl.lastPingRx).Milliseconds() && time.Since(xlinkImpl.lastPingTx).Milliseconds() >= pingDelayMs {
if err := xlinkImpl.sendPing(); err == nil {
logrus.Infof("sent ping for Xlink [l/%s]", xlinkImpl.id.Token)
} else {
logrus.Errorf("error sending ping for Xlink [l/%s] (%v)", xlinkImpl.id.Token, err)
}
}
}
}
}
@@ -68,3 +98,6 @@ type listener struct {
accepter xlink.Accepter
peers map[string]*impl
}
const pingDelayMs = 2000
const pingCycleDelayMs = 1000
router/xlink_transwarp/message.go
+29 -6
View File
@@ -61,17 +61,18 @@ const (
Acknowledgement
)
const versionLen = 4
const timeoutSeconds = 5
const mss = 1472
const noReplyFor = -1
func createMessage(sequence int32, frame uint8, ofFrames uint8, ct uint8, headers map[int32][]byte, payload []byte) ([]byte, error) {
func createMessage(sequence int32, frame uint8, ofFrames uint8, ct contentType, headers map[int32][]byte, payload []byte) ([]byte, error) {
data := new(bytes.Buffer)
data.Write(magicV1)
if err := binary.Write(data, binary.LittleEndian, sequence); err != nil {
return nil, fmt.Errorf("sequence write (%w)", err)
}
data.Write([]byte{ frame, ofFrames, ct })
data.Write([]byte{ frame, ofFrames, uint8(ct) })
if err := binary.Write(data, binary.LittleEndian, uint16(0)); err != nil { // headers length
return nil, fmt.Errorf("headers length write (%w)", err)
}
@@ -96,13 +97,35 @@ func writeHello(linkId *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr)
payload := new(bytes.Buffer)
payload.Write([]byte(linkId.Token))
data, err := createMessage(-1, 0, 1, uint8(Hello), nil, payload.Bytes())
data, err := createMessage(-1, 0, 1, Hello, nil, payload.Bytes())
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
return fmt.Errorf("unable to set deadline (%w)", err)
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writePing(sequence int32, conn *net.UDPConn, peer *net.UDPAddr, replyFor int32) error {
payload := new(bytes.Buffer)
if err := binary.Write(payload, binary.LittleEndian, replyFor); err != nil {
return fmt.Errorf("reply for write (%w)", err)
}
data, err := createMessage(sequence, 0, 1, Ping, nil, payload.Bytes())
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
router/xlink_transwarp/xlink.go
+35 -6
View File
@@ -21,11 +21,13 @@ import (
"github.com/netfoundry/ziti-fabric/router/xgress"
"github.com/netfoundry/ziti-foundation/identity/identity"
"net"
"sync"
"time"
)
/*
* xlink.Xlink
*/
*/
func (self *impl) Id() *identity.TokenId {
return self.id
}
@@ -41,13 +43,40 @@ func (self *impl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) e
func (self *impl) Close() error {
return self.conn.Close()
}
/* */
func newImpl(id *identity.TokenId, conn *net.UDPConn) *impl {
return &impl{id: id, conn: conn}
/*
* impl
*/
func (self *impl) sendPing() error {
sequence := self.nextSequence()
if err := writePing(sequence, self.conn, self.peer, noReplyFor); err != nil {
return fmt.Errorf("error sending ping (%w)", err)
}
self.lastPingTxSequence = sequence
self.lastPingTx = time.Now()
return nil
}
func (self *impl) nextSequence() int32 {
self.sequenceLock.Lock()
defer self.sequenceLock.Unlock()
sequence := self.sequence
self.sequence++
return sequence
}
func newImpl(id *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr) *impl {
return &impl{id: id, conn: conn, peer: peer}
}
type impl struct {
id *identity.TokenId
conn *net.UDPConn
id *identity.TokenId
conn *net.UDPConn
peer *net.UDPAddr
sequence int32
sequenceLock sync.Mutex
lastPingRx time.Time
lastPingTx time.Time
lastPingTxSequence int32
}