mirror of
https://github.com/yggdrasil-network/yggdrasil-go.git
synced 2024-11-27 12:05:23 +00:00
Clean up
This commit is contained in:
Neil Alexander
parent
9c6cf50684
commit
81545fd9bf
@ -54,7 +54,7 @@ func (l *awdl) create(fromAWDL chan []byte, toAWDL chan []byte, name, local, rem
|
||||
toAWDL: toAWDL,
|
||||
}
|
||||
s := stream{}
|
||||
s.init(rwc, nil)
|
||||
s.init(rwc)
|
||||
link, err := l.core.link.create(&s, name, "awdl", local, remote)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -63,7 +63,6 @@ func (l *awdl) create(fromAWDL chan []byte, toAWDL chan []byte, name, local, rem
|
||||
link: link,
|
||||
rwc: rwc,
|
||||
}
|
||||
intf.stream.init(intf.rwc, nil)
|
||||
l.mutex.Lock()
|
||||
l.interfaces[name] = &intf
|
||||
l.mutex.Unlock()
|
||||
|
||||
@ -208,109 +208,3 @@ func (intf *linkInterface) handler() error {
|
||||
////////////////////////////////////////////////////////////////////////////////
|
||||
return nil
|
||||
}
|
||||
|
||||
/*
|
||||
func (intf *linkInterface) start() {
|
||||
myLinkPub, myLinkPriv := crypto.NewBoxKeys()
|
||||
meta := version_getBaseMetadata()
|
||||
meta.box = intf.link.core.boxPub
|
||||
meta.sig = intf.link.core.sigPub
|
||||
meta.link = *myLinkPub
|
||||
metaBytes := meta.encode()
|
||||
//intf.link.core.log.Println("start: intf.tolink <- metaBytes")
|
||||
intf.tolink <- metaBytes
|
||||
//intf.link.core.log.Println("finish: intf.tolink <- metaBytes")
|
||||
//intf.link.core.log.Println("start: metaBytes = <-intf.fromlink")
|
||||
metaBytes = <-intf.fromlink
|
||||
//intf.link.core.log.Println("finish: metaBytes = <-intf.fromlink")
|
||||
meta = version_metadata{}
|
||||
if !meta.decode(metaBytes) || !meta.check() {
|
||||
intf.link.core.log.Println("Metadata decode failure")
|
||||
return
|
||||
}
|
||||
base := version_getBaseMetadata()
|
||||
if meta.ver > base.ver || meta.ver == base.ver && meta.minorVer > base.minorVer {
|
||||
intf.link.core.log.Println("Failed to connect to node: " + intf.name + " version: " + fmt.Sprintf("%d.%d", meta.ver, meta.minorVer))
|
||||
return
|
||||
}
|
||||
shared := crypto.GetSharedKey(myLinkPriv, &meta.link)
|
||||
intf.peer = intf.link.core.peers.newPeer(&meta.box, &meta.sig, shared, intf.name)
|
||||
if intf.peer == nil {
|
||||
intf.link.mutex.Lock()
|
||||
delete(intf.link.interfaces, intf.name)
|
||||
intf.link.mutex.Unlock()
|
||||
return
|
||||
}
|
||||
intf.peer.linkOut = make(chan []byte, 1) // protocol traffic
|
||||
intf.peer.out = func(msg []byte) {
|
||||
defer func() { recover() }()
|
||||
intf.tolink <- msg
|
||||
} // called by peer.sendPacket()
|
||||
intf.link.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle
|
||||
intf.peer.close = func() {
|
||||
close(intf.fromlink)
|
||||
close(intf.tolink)
|
||||
}
|
||||
go intf.handler()
|
||||
go intf.peer.linkLoop()
|
||||
}
|
||||
|
||||
func (l *link) getInterface(identity string) *linkInterface {
|
||||
l.mutex.RLock()
|
||||
defer l.mutex.RUnlock()
|
||||
if intf, ok := l.interfaces[identity]; ok {
|
||||
return intf
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *link) shutdown(identity string) error {
|
||||
if intf, ok := l.interfaces[identity]; ok {
|
||||
intf.shutdown <- true
|
||||
l.core.peers.removePeer(intf.peer.port)
|
||||
l.mutex.Lock()
|
||||
delete(l.interfaces, identity)
|
||||
l.mutex.Unlock()
|
||||
return nil
|
||||
} else {
|
||||
return fmt.Errorf("interface '%s' doesn't exist or already shutdown", identity)
|
||||
}
|
||||
}
|
||||
|
||||
func (ai *linkInterface) handler() {
|
||||
send := func(msg []byte) {
|
||||
ai.tolink <- msg
|
||||
atomic.AddUint64(&ai.peer.bytesSent, uint64(len(msg)))
|
||||
util.PutBytes(msg)
|
||||
}
|
||||
for {
|
||||
timerInterval := tcp_ping_interval
|
||||
timer := time.NewTimer(timerInterval)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case p := <-ai.peer.linkOut:
|
||||
send(p)
|
||||
continue
|
||||
default:
|
||||
}
|
||||
timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
default:
|
||||
}
|
||||
timer.Reset(timerInterval)
|
||||
select {
|
||||
case _ = <-timer.C:
|
||||
send([]byte{})
|
||||
case p := <-ai.peer.linkOut:
|
||||
send(p)
|
||||
continue
|
||||
case r := <-ai.fromlink:
|
||||
ai.peer.handlePacket(r)
|
||||
ai.link.core.switchTable.idleIn <- ai.peer.port
|
||||
case <-ai.shutdown:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
@ -14,8 +14,6 @@ var _ = linkInterfaceMsgIO(&stream{})
|
||||
type stream struct {
|
||||
rwc io.ReadWriteCloser
|
||||
inputBuffer []byte // Incoming packet stream
|
||||
// TODO remove the rest, it shouldn't matter in the long run
|
||||
handlePacket func([]byte)
|
||||
}
|
||||
|
||||
func (s *stream) close() error {
|
||||
@ -26,11 +24,9 @@ const streamMsgSize = 2048 + 65535
|
||||
|
||||
var streamMsg = [...]byte{0xde, 0xad, 0xb1, 0x75} // "dead bits"
|
||||
|
||||
func (s *stream) init(rwc io.ReadWriteCloser, in func([]byte)) {
|
||||
func (s *stream) init(rwc io.ReadWriteCloser) {
|
||||
// TODO have this also do the metadata handshake and create the peer struct
|
||||
s.rwc = rwc
|
||||
s.handlePacket = in
|
||||
|
||||
// TODO call something to do the metadata exchange
|
||||
}
|
||||
|
||||
@ -112,31 +108,6 @@ func (s *stream) _recvMetaBytes() ([]byte, error) {
|
||||
return metaBytes, nil
|
||||
}
|
||||
|
||||
// This reads from the channel into a []byte buffer for incoming messages. It
|
||||
// copies completed messages out of the cache into a new slice, and passes them
|
||||
// to the peer struct via the provided `in func([]byte)` argument. Then it
|
||||
// shifts the incomplete fragments of data forward so future reads won't
|
||||
// overwrite it.
|
||||
func (s *stream) handleInput(bs []byte) error {
|
||||
if len(bs) > 0 {
|
||||
s.inputBuffer = append(s.inputBuffer, bs...)
|
||||
buf := s.inputBuffer
|
||||
msg, ok, err2 := stream_chopMsg(&buf)
|
||||
if err2 != nil {
|
||||
return fmt.Errorf("message error: %v", err2)
|
||||
}
|
||||
if !ok {
|
||||
// We didn't get the whole message yet
|
||||
return nil
|
||||
}
|
||||
newMsg := append(util.GetBytes(), msg...)
|
||||
s.inputBuffer = append(s.inputBuffer[:0], buf...)
|
||||
s.handlePacket(newMsg)
|
||||
util.Yield() // Make sure we give up control to the scheduler
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// This takes a pointer to a slice as an argument. It checks if there's a
|
||||
// complete message and, if so, slices out those parts and returns the message,
|
||||
// true, and nil. If there's no error, but also no complete message, it returns
|
||||
|
||||
@ -20,14 +20,11 @@ import (
|
||||
"math/rand"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/proxy"
|
||||
|
||||
"github.com/yggdrasil-network/yggdrasil-go/src/address"
|
||||
"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
|
||||
"github.com/yggdrasil-network/yggdrasil-go/src/util"
|
||||
)
|
||||
|
||||
const default_timeout = 6 * time.Second
|
||||
@ -284,7 +281,7 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
|
||||
defer sock.Close()
|
||||
iface.setExtraOptions(sock)
|
||||
stream := stream{}
|
||||
stream.init(sock, nil)
|
||||
stream.init(sock)
|
||||
local, _, _ := net.SplitHostPort(sock.LocalAddr().String())
|
||||
remote, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
|
||||
name := "tcp://" + sock.RemoteAddr().String()
|
||||
@ -297,185 +294,3 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
|
||||
err = link.handler()
|
||||
iface.core.log.Println("DEBUG: stopped handler for", name, err)
|
||||
}
|
||||
|
||||
// This exchanges/checks connection metadata, sets up the peer struct, sets up the writer goroutine, and then runs the reader within the current goroutine.
|
||||
// It defers a bunch of cleanup stuff to tear down all of these things when the reader exists (e.g. due to a closed connection or a timeout).
|
||||
func (iface *tcpInterface) handler_old(sock net.Conn, incoming bool) {
|
||||
defer sock.Close()
|
||||
iface.setExtraOptions(sock)
|
||||
// Get our keys
|
||||
myLinkPub, myLinkPriv := crypto.NewBoxKeys() // ephemeral link keys
|
||||
meta := version_getBaseMetadata()
|
||||
meta.box = iface.core.boxPub
|
||||
meta.sig = iface.core.sigPub
|
||||
meta.link = *myLinkPub
|
||||
metaBytes := meta.encode()
|
||||
_, err := sock.Write(metaBytes)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if iface.timeout > 0 {
|
||||
sock.SetReadDeadline(time.Now().Add(iface.timeout))
|
||||
}
|
||||
_, err = sock.Read(metaBytes)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
meta = version_metadata{} // Reset to zero value
|
||||
if !meta.decode(metaBytes) || !meta.check() {
|
||||
// Failed to decode and check the metadata
|
||||
// If it's a version mismatch issue, then print an error message
|
||||
base := version_getBaseMetadata()
|
||||
if meta.meta == base.meta {
|
||||
if meta.ver > base.ver {
|
||||
iface.core.log.Println("Failed to connect to node:", sock.RemoteAddr().String(), "version:", meta.ver)
|
||||
} else if meta.ver == base.ver && meta.minorVer > base.minorVer {
|
||||
iface.core.log.Println("Failed to connect to node:", sock.RemoteAddr().String(), "version:", fmt.Sprintf("%d.%d", meta.ver, meta.minorVer))
|
||||
}
|
||||
}
|
||||
// TODO? Block forever to prevent future connection attempts? suppress future messages about the same node?
|
||||
return
|
||||
}
|
||||
remoteAddr, _, e1 := net.SplitHostPort(sock.RemoteAddr().String())
|
||||
localAddr, _, e2 := net.SplitHostPort(sock.LocalAddr().String())
|
||||
if e1 != nil || e2 != nil {
|
||||
return
|
||||
}
|
||||
info := tcpInfo{ // used as a map key, so don't include ephemeral link key
|
||||
box: meta.box,
|
||||
sig: meta.sig,
|
||||
localAddr: localAddr,
|
||||
remoteAddr: remoteAddr,
|
||||
}
|
||||
if iface.isAlreadyConnected(info) {
|
||||
return
|
||||
}
|
||||
// Quit the parent call if this is a connection to ourself
|
||||
equiv := func(k1, k2 []byte) bool {
|
||||
for idx := range k1 {
|
||||
if k1[idx] != k2[idx] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
if equiv(meta.box[:], iface.core.boxPub[:]) {
|
||||
return
|
||||
}
|
||||
if equiv(meta.sig[:], iface.core.sigPub[:]) {
|
||||
return
|
||||
}
|
||||
// Check if we're authorized to connect to this key / IP
|
||||
if incoming && !iface.core.peers.isAllowedEncryptionPublicKey(&meta.box) {
|
||||
// Allow unauthorized peers if they're link-local
|
||||
raddrStr, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
|
||||
raddr := net.ParseIP(raddrStr)
|
||||
if !raddr.IsLinkLocalUnicast() {
|
||||
return
|
||||
}
|
||||
}
|
||||
// Check if we already have a connection to this node, close and block if yes
|
||||
iface.mutex.Lock()
|
||||
/*if blockChan, isIn := iface.conns[info]; isIn {
|
||||
iface.mutex.Unlock()
|
||||
sock.Close()
|
||||
<-blockChan
|
||||
return
|
||||
}*/
|
||||
blockChan := make(chan struct{})
|
||||
iface.conns[info] = blockChan
|
||||
iface.mutex.Unlock()
|
||||
defer func() {
|
||||
iface.mutex.Lock()
|
||||
delete(iface.conns, info)
|
||||
iface.mutex.Unlock()
|
||||
close(blockChan)
|
||||
}()
|
||||
// Note that multiple connections to the same node are allowed
|
||||
// E.g. over different interfaces
|
||||
p := iface.core.peers.newPeer(&meta.box, &meta.sig, crypto.GetSharedKey(myLinkPriv, &meta.link), sock.RemoteAddr().String())
|
||||
p.linkOut = make(chan []byte, 1)
|
||||
out := make(chan []byte, 1)
|
||||
defer close(out)
|
||||
go func() {
|
||||
// This goroutine waits for outgoing packets, link protocol traffic, or sends idle keep-alive traffic
|
||||
send := func(msg []byte) {
|
||||
msgLen := wire_encode_uint64(uint64(len(msg)))
|
||||
buf := net.Buffers{streamMsg[:], msgLen, msg}
|
||||
buf.WriteTo(sock)
|
||||
atomic.AddUint64(&p.bytesSent, uint64(len(streamMsg)+len(msgLen)+len(msg)))
|
||||
util.PutBytes(msg)
|
||||
}
|
||||
timerInterval := tcp_ping_interval
|
||||
timer := time.NewTimer(timerInterval)
|
||||
defer timer.Stop()
|
||||
for {
|
||||
select {
|
||||
case msg := <-p.linkOut:
|
||||
// Always send outgoing link traffic first, if needed
|
||||
send(msg)
|
||||
continue
|
||||
default:
|
||||
}
|
||||
// Otherwise wait reset the timer and wait for something to do
|
||||
timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
default:
|
||||
}
|
||||
timer.Reset(timerInterval)
|
||||
select {
|
||||
case _ = <-timer.C:
|
||||
send(nil) // TCP keep-alive traffic
|
||||
case msg := <-p.linkOut:
|
||||
send(msg)
|
||||
case msg, ok := <-out:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
send(msg) // Block until the socket write has finished
|
||||
// Now inform the switch that we're ready for more traffic
|
||||
p.core.switchTable.idleIn <- p.port
|
||||
}
|
||||
}
|
||||
}()
|
||||
p.core.switchTable.idleIn <- p.port // Start in the idle state
|
||||
p.out = func(msg []byte) {
|
||||
defer func() { recover() }()
|
||||
out <- msg
|
||||
}
|
||||
p.close = func() { sock.Close() }
|
||||
go p.linkLoop()
|
||||
defer func() {
|
||||
// Put all of our cleanup here...
|
||||
p.core.peers.removePeer(p.port)
|
||||
}()
|
||||
us, _, _ := net.SplitHostPort(sock.LocalAddr().String())
|
||||
them, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
|
||||
themNodeID := crypto.GetNodeID(&meta.box)
|
||||
themAddr := address.AddrForNodeID(themNodeID)
|
||||
themAddrString := net.IP(themAddr[:]).String()
|
||||
themString := fmt.Sprintf("%s@%s", themAddrString, them)
|
||||
iface.core.log.Printf("Connected: %s, source: %s", themString, us)
|
||||
//iface.stream.init(sock, p.handlePacket)
|
||||
bs := make([]byte, 2*streamMsgSize)
|
||||
var n int
|
||||
for {
|
||||
if iface.timeout > 0 {
|
||||
sock.SetReadDeadline(time.Now().Add(iface.timeout))
|
||||
}
|
||||
n, err = sock.Read(bs)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
if n > 0 {
|
||||
//iface.stream.handleInput(bs[:n])
|
||||
}
|
||||
}
|
||||
if err == nil {
|
||||
iface.core.log.Printf("Disconnected: %s, source: %s", themString, us)
|
||||
} else {
|
||||
iface.core.log.Printf("Disconnected: %s, source: %s, error: %s", themString, us, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user