From 682ece01491b375ac7724ebb451eff4810a4f63e Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Thu, 26 Mar 2020 10:47:53 -0400 Subject: [PATCH] Using ping to iterate state management for dialer and listener. (#44) --- router/xlink_transwarp/dialer.go | 2 +- router/xlink_transwarp/listener.go | 49 +++++++++++++++++++++++++----- router/xlink_transwarp/message.go | 35 +++++++++++++++++---- router/xlink_transwarp/xlink.go | 41 +++++++++++++++++++++---- 4 files changed, 106 insertions(+), 21 deletions(-) diff --git a/router/xlink_transwarp/dialer.go b/router/xlink_transwarp/dialer.go index 5a8b0801c..87ec728ba 100644 --- a/router/xlink_transwarp/dialer.go +++ b/router/xlink_transwarp/dialer.go @@ -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) } diff --git a/router/xlink_transwarp/listener.go b/router/xlink_transwarp/listener.go index d8d0ed7da..b785ddbef 100644 --- a/router/xlink_transwarp/listener.go +++ b/router/xlink_transwarp/listener.go @@ -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 diff --git a/router/xlink_transwarp/message.go b/router/xlink_transwarp/message.go index 881f0d791..5de36d54e 100644 --- a/router/xlink_transwarp/message.go +++ b/router/xlink_transwarp/message.go @@ -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 { diff --git a/router/xlink_transwarp/xlink.go b/router/xlink_transwarp/xlink.go index 2ea32d9db..e92013908 100644 --- a/router/xlink_transwarp/xlink.go +++ b/router/xlink_transwarp/xlink.go @@ -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 }