Rethink channels, more error throwing

This commit is contained in:
Neil Alexander 2019-01-05 12:06:45 +00:00
parent 90366dd853
commit 6bbd8c1b30
No known key found for this signature in database
GPG Key ID: A02A2019A2BB0944
3 changed files with 53 additions and 37 deletions

View File

@ -1,9 +1,14 @@
package yggdrasil package yggdrasil
import ( import (
"errors"
"fmt"
"sync" "sync"
"sync/atomic"
"time"
"github.com/yggdrasil-network/yggdrasil-go/src/crypto" "github.com/yggdrasil-network/yggdrasil-go/src/crypto"
"github.com/yggdrasil-network/yggdrasil-go/src/util"
) )
type awdl struct { type awdl struct {
@ -14,8 +19,8 @@ type awdl struct {
type awdlInterface struct { type awdlInterface struct {
awdl *awdl awdl *awdl
recv <-chan []byte // traffic received from the network fromAWDL chan []byte
send chan<- []byte // traffic to send to the network toAWDL chan []byte
shutdown chan bool shutdown chan bool
peer *peer peer *peer
} }
@ -29,11 +34,11 @@ func (l *awdl) init(c *Core) error {
return nil return nil
} }
func (l *awdl) create(boxPubKey *crypto.BoxPubKey, sigPubKey *crypto.SigPubKey, name string) *awdlInterface { func (l *awdl) create(boxPubKey *crypto.BoxPubKey, sigPubKey *crypto.SigPubKey, name string) (*awdlInterface, error) {
shared := crypto.GetSharedKey(&l.core.boxPriv, boxPubKey) shared := crypto.GetSharedKey(&l.core.boxPriv, boxPubKey)
intf := awdlInterface{ intf := awdlInterface{
recv: make(<-chan []byte), fromAWDL: make(chan []byte, 32),
send: make(chan<- []byte), toAWDL: make(chan []byte, 32),
shutdown: make(chan bool), shutdown: make(chan bool),
peer: l.core.peers.newPeer(boxPubKey, sigPubKey, shared, name), peer: l.core.peers.newPeer(boxPubKey, sigPubKey, shared, name),
} }
@ -41,21 +46,21 @@ func (l *awdl) create(boxPubKey *crypto.BoxPubKey, sigPubKey *crypto.SigPubKey,
l.mutex.Lock() l.mutex.Lock()
l.interfaces[name] = &intf l.interfaces[name] = &intf
l.mutex.Unlock() l.mutex.Unlock()
intf.peer.linkOut = make(chan []byte, 1) intf.peer.linkOut = make(chan []byte, 1) // protocol traffic
intf.peer.out = func(msg []byte) { intf.peer.out = func(msg []byte) {
defer func() { recover() }() defer func() { recover() }()
intf.send <- msg intf.toAWDL <- msg
l.core.switchTable.idleIn <- intf.peer.port } // called by peer.sendPacket()
} l.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle
intf.peer.close = func() { intf.peer.close = func() {
close(intf.send) close(intf.fromAWDL)
close(intf.toAWDL)
} }
go intf.peer.linkLoop() go intf.handler() // start listening for packets from switch
go intf.handler() go intf.peer.linkLoop() // start link loop
l.core.switchTable.idleIn <- intf.peer.port return &intf, nil
return &intf
} }
return nil return nil, errors.New("l.core.peers.newPeer failed")
} }
func (l *awdl) getInterface(identity string) *awdlInterface { func (l *awdl) getInterface(identity string) *awdlInterface {
@ -67,45 +72,52 @@ func (l *awdl) getInterface(identity string) *awdlInterface {
return nil return nil
} }
func (l *awdl) shutdown(identity string) { func (l *awdl) shutdown(identity string) error {
if intf, ok := l.interfaces[identity]; ok { if intf, ok := l.interfaces[identity]; ok {
intf.shutdown <- true intf.shutdown <- true
l.core.peers.removePeer(intf.peer.port) l.core.peers.removePeer(intf.peer.port)
l.mutex.Lock() l.mutex.Lock()
delete(l.interfaces, identity) delete(l.interfaces, identity)
l.mutex.Unlock() l.mutex.Unlock()
return nil
} else {
return errors.New(fmt.Sprintf("Interface '%s' doesn't exist or already shutdown", identity))
} }
} }
func (ai *awdlInterface) handler() { func (ai *awdlInterface) handler() {
send := func(msg []byte) {
ai.toAWDL <- msg
atomic.AddUint64(&ai.peer.bytesSent, uint64(len(msg)))
util.PutBytes(msg)
}
for { for {
/*timerInterval := tcp_ping_interval timerInterval := tcp_ping_interval
timer := time.NewTimer(timerInterval) timer := time.NewTimer(timerInterval)
defer timer.Stop()*/ defer timer.Stop()
select { select {
case p := <-ai.peer.linkOut: case p := <-ai.peer.linkOut:
ai.send <- p send(p)
ai.awdl.core.switchTable.idleIn <- ai.peer.port
continue continue
default: default:
} }
/*timer.Stop() timer.Stop()
select { select {
case <-timer.C: case <-timer.C:
default: default:
} }
timer.Reset(timerInterval)*/ timer.Reset(timerInterval)
select { select {
//case _ = <-timer.C: case _ = <-timer.C:
// ai.send <- nil send([]byte{'H', 'E', 'L', 'L', 'O'})
case r := <-ai.recv: // traffic received from AWDL case p := <-ai.peer.linkOut:
send(p)
continue
case r := <-ai.fromAWDL:
ai.peer.handlePacket(r) ai.peer.handlePacket(r)
ai.awdl.core.switchTable.idleIn <- ai.peer.port
case <-ai.shutdown: case <-ai.shutdown:
return return
case p := <-ai.peer.linkOut:
ai.send <- p
ai.awdl.core.switchTable.idleIn <- ai.peer.port
continue
} }
} }
} }

View File

@ -108,20 +108,24 @@ func (c *Core) AWDLCreateInterface(boxPubKey string, sigPubKey string, name stri
} }
copy(boxPub[:], boxPubHex) copy(boxPub[:], boxPubHex)
copy(sigPub[:], sigPubHex) copy(sigPub[:], sigPubHex)
if intf := c.awdl.create(&boxPub, &sigPub, name); intf != nil { if intf, err := c.awdl.create(&boxPub, &sigPub, name); err == nil {
return nil if intf != nil {
return err
} else { } else {
return errors.New("No interface was created") return errors.New("c.awdl.create didn't return an interface")
}
} else {
return err
} }
} }
func (c *Core) AWDLShutdownInterface(name string) { func (c *Core) AWDLShutdownInterface(name string) error {
c.awdl.shutdown(name) return c.awdl.shutdown(name)
} }
func (c *Core) AWDLRecvPacket(identity string) ([]byte, error) { func (c *Core) AWDLRecvPacket(identity string) ([]byte, error) {
if intf := c.awdl.getInterface(identity); intf != nil { if intf := c.awdl.getInterface(identity); intf != nil {
return <-intf.recv, nil return <-intf.toAWDL, nil
} }
return nil, errors.New("identity not known: " + identity) return nil, errors.New("identity not known: " + identity)
} }
@ -129,7 +133,7 @@ func (c *Core) AWDLRecvPacket(identity string) ([]byte, error) {
func (c *Core) AWDLSendPacket(identity string, buf []byte) error { func (c *Core) AWDLSendPacket(identity string, buf []byte) error {
packet := append(util.GetBytes(), buf[:]...) packet := append(util.GetBytes(), buf[:]...)
if intf := c.awdl.getInterface(identity); intf != nil { if intf := c.awdl.getInterface(identity); intf != nil {
intf.send <- packet intf.fromAWDL <- packet
return nil return nil
} }
return errors.New("identity not known: " + identity) return errors.New("identity not known: " + identity)