Change some mutexes to atomics, change conns map to pointers, sort of works but seems to deadlock very easily

src/tuntap/tun.go

@@ -30,8 +30,7 @@ type TunAdapter struct {
 	config      *config.NodeState
 	log         *log.Logger
 	reconfigure chan chan error
-	conns       map[crypto.NodeID]yggdrasil.Conn
-	connsMutex  sync.RWMutex
+	conns       map[crypto.NodeID]*yggdrasil.Conn
 	listener    *yggdrasil.Listener
 	dialer      *yggdrasil.Dialer
 	addr        address.Address
@@ -102,7 +101,7 @@ func (tun *TunAdapter) Init(config *config.NodeState, log *log.Logger, listener
 	tun.log = log
 	tun.listener = listener
 	tun.dialer = dialer
-	tun.conns = make(map[crypto.NodeID]yggdrasil.Conn)
+	tun.conns = make(map[crypto.NodeID]*yggdrasil.Conn)
 }
 
 // Start the setup process for the TUN/TAP adapter. If successful, starts the
@@ -180,6 +179,7 @@ func (tun *TunAdapter) handler() error {
 }
 
 func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
+	tun.conns[conn.RemoteAddr()] = conn
 	b := make([]byte, 65535)
 	for {
 		n, err := conn.Read(b)
@@ -203,7 +203,6 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
 }
 
 func (tun *TunAdapter) ifaceReader() error {
-	tun.log.Println("Start TUN reader")
 	bs := make([]byte, 65535)
 	for {
 		n, err := tun.iface.Read(bs)
@@ -244,6 +243,7 @@ func (tun *TunAdapter) ifaceReader() error {
 		dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
 		// Do we have an active connection for this node ID?
 		if conn, isIn := tun.conns[*dstNodeID]; isIn {
+			tun.log.Println("Got", &conn)
 			w, err := conn.Write(bs)
 			if err != nil {
 				tun.log.Println("Unable to write to remote:", err)
@@ -254,14 +254,12 @@ func (tun *TunAdapter) ifaceReader() error {
 			}
 		} else {
 			tun.log.Println("Opening connection for", *dstNodeID)
-			tun.connsMutex.Lock()
 			if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
-				tun.conns[*dstNodeID] = conn
+				tun.conns[*dstNodeID] = &conn
 				go tun.connReader(&conn)
 			} else {
 				tun.log.Println("Error dialing:", err)
 			}
-			tun.connsMutex.Unlock()
 		}
 
 		/*if !r.cryptokey.isValidSource(srcAddr, addrlen) {

src/yggdrasil/conn.go

@@ -1,8 +1,8 @@
 package yggdrasil
 
 import (
-	"encoding/hex"
 	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -42,27 +42,27 @@ func (c *Conn) startSearch() {
 	doSearch := func() {
 		sinfo, isIn := c.core.searches.searches[*c.nodeID]
 		if !isIn {
-			c.core.log.Debugln("Starting search for", hex.EncodeToString(c.nodeID[:]))
 			sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
 		}
 		c.core.searches.continueSearch(sinfo)
 	}
-	var sinfo *sessionInfo
-	var isIn bool
 	switch {
-	case !isIn || !sinfo.init:
+	case c.session == nil || !c.session.init.Load().(bool):
 		doSearch()
-	case time.Since(sinfo.time) > 6*time.Second:
-		if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second {
+	case time.Since(c.session.time.Load().(time.Time)) > 6*time.Second:
+		sTime := c.session.time.Load().(time.Time)
+		pingTime := c.session.pingTime.Load().(time.Time)
+		if sTime.Before(pingTime) && time.Since(pingTime) > 6*time.Second {
 			doSearch()
 		} else {
+			pingSend := c.session.pingSend.Load().(time.Time)
 			now := time.Now()
-			if !sinfo.time.Before(sinfo.pingTime) {
-				sinfo.pingTime = now
+			if !sTime.Before(pingTime) {
+				c.session.pingTime.Store(now)
 			}
-			if time.Since(sinfo.pingSend) > time.Second {
-				sinfo.pingSend = now
-				c.core.sessions.sendPingPong(sinfo, false)
+			if time.Since(pingSend) > time.Second {
+				c.session.pingSend.Store(now)
+				c.core.sessions.sendPingPong(c.session, false)
 			}
 		}
 	}
@@ -77,12 +77,9 @@ func (c *Conn) Read(b []byte) (int, error) {
 	if c.session == nil {
 		return 0, errors.New("searching for remote side")
 	}
-	c.session.initMutex.RLock()
-	if !c.session.init {
-		c.session.initMutex.RUnlock()
+	if !c.session.init.Load().(bool) {
 		return 0, errors.New("waiting for remote side to accept")
 	}
-	c.session.initMutex.RUnlock()
 	select {
 	case p, ok := <-c.session.recv:
 		if !ok {
@@ -106,9 +103,7 @@ func (c *Conn) Read(b []byte) (int, error) {
 				b = b[:len(bs)]
 			}
 			c.session.updateNonce(&p.Nonce)
-			c.session.timeMutex.Lock()
-			c.session.time = time.Now()
-			c.session.timeMutex.Unlock()
+			c.session.time.Store(time.Now())
 			return nil
 		}()
 		if err != nil {
@@ -129,22 +124,18 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 		return 0, errors.New("session is closed")
 	}
 	if c.session == nil {
+		fmt.Println("No session found, starting search for", &c)
 		c.core.router.doAdmin(func() {
 			c.startSearch()
 		})
 		return 0, errors.New("searching for remote side")
 	}
 	defer util.PutBytes(b)
-	c.session.initMutex.RLock()
-	if !c.session.init {
-		c.session.initMutex.RUnlock()
+	if !c.session.init.Load().(bool) {
 		return 0, errors.New("waiting for remote side to accept")
 	}
-	c.session.initMutex.RUnlock()
 	// code isn't multithreaded so appending to this is safe
-	c.session.coordsMutex.RLock()
 	coords := c.session.coords
-	c.session.coordsMutex.RUnlock()
 	// Prepare the payload
 	c.session.myNonceMutex.Lock()
 	payload, nonce := crypto.BoxSeal(&c.session.sharedSesKey, b, &c.session.myNonce)

src/yggdrasil/dialer.go

@@ -35,7 +35,7 @@ func (d *Dialer) Dial(network, address string) (Conn, error) {
 				return Conn{}, err
 			}
 			copy(nodeID[:], dest)
-			for idx := 0; idx < len; idx++ {
+			for idx := 0; idx <= len; idx++ {
 				nodeMask[idx/8] |= 0x80 >> byte(idx%8)
 			}
 		} else {
@@ -59,15 +59,13 @@ func (d *Dialer) Dial(network, address string) (Conn, error) {
 // NodeID parameters.
 func (d *Dialer) DialByNodeIDandMask(nodeID, nodeMask *crypto.NodeID) (Conn, error) {
 	conn := Conn{
-		mutex: &sync.RWMutex{},
+		core:     d.core,
+		mutex:    &sync.RWMutex{},
+		nodeID:   nodeID,
+		nodeMask: nodeMask,
 	}
-	conn.core = d.core
-	conn.nodeID = nodeID
-	conn.nodeMask = nodeMask
 	conn.core.router.doAdmin(func() {
 		conn.startSearch()
 	})
-	conn.mutex.Lock()
-	defer conn.mutex.Unlock()
 	return conn, nil
 }

src/yggdrasil/router.go

@@ -287,16 +287,15 @@ func (r *router) sendPacket(bs []byte) {
 	if destSnet.IsValid() {
 		sinfo, isIn = r.core.sessions.getByTheirSubnet(&destSnet)
 	}
-	sinfo.timeMutex.Lock()
-	sinfo.initMutex.RLock()
-	defer sinfo.timeMutex.Unlock()
-	defer sinfo.initMutex.RUnlock()
+	sTime := sinfo.time.Load().(time.Time)
+	pingTime := sinfo.pingTime.Load().(time.Time)
+	pingSend := sinfo.pingSend.Load().(time.Time)
 	switch {
-	case !isIn || !sinfo.init:
+	case !isIn || !sinfo.init.Load().(bool):
 		// No or unintiialized session, so we need to search first
 		doSearch(bs)
-	case time.Since(sinfo.time) > 6*time.Second:
-		if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second {
+	case time.Since(sTime) > 6*time.Second:
+		if sTime.Before(pingTime) && time.Since(pingTime) > 6*time.Second {
 			// We haven't heard from the dest in a while
 			// We tried pinging but didn't get a response
 			// They may have changed coords
@@ -307,16 +306,15 @@ func (r *router) sendPacket(bs []byte) {
 			// We haven't heard about the dest in a while
 			now := time.Now()
 
-			if !sinfo.time.Before(sinfo.pingTime) {
+			if !sTime.Before(pingTime) {
 				// Update pingTime to start the clock for searches (above)
-				sinfo.pingTime = now
+				sinfo.pingTime.Store(now)
 			}
-			if time.Since(sinfo.pingSend) > time.Second {
+			if time.Since(pingSend) > time.Second {
 				// Send at most 1 ping per second
-				sinfo.pingSend = now
+				sinfo.pingSend.Store(now)
 				r.core.sessions.sendPingPong(sinfo, false)
 			}
-			sinfo.timeMutex.Unlock()
 		}
 		fallthrough // Also send the packet
 	default:

src/yggdrasil/search.go

@@ -212,9 +212,7 @@ func (s *searches) checkDHTRes(info *searchInfo, res *dhtRes) bool {
 		}
 	}
 	// FIXME (!) replay attacks could mess with coords? Give it a handle (tstamp)?
-	sinfo.coordsMutex.Lock()
 	sinfo.coords = res.Coords
-	sinfo.coordsMutex.Unlock()
 	sinfo.packet = info.packet
 	s.core.sessions.ping(sinfo)
 	info.callback(sinfo, nil)

src/yggdrasil/session.go

@@ -18,41 +18,38 @@ import (
 // All the information we know about an active session.
 // This includes coords, permanent and ephemeral keys, handles and nonces, various sorts of timing information for timeout and maintenance, and some metadata for the admin API.
 type sessionInfo struct {
-	core            *Core
-	reconfigure     chan chan error
-	theirAddr       address.Address
-	theirSubnet     address.Subnet
-	theirPermPub    crypto.BoxPubKey
-	theirSesPub     crypto.BoxPubKey
-	mySesPub        crypto.BoxPubKey
-	mySesPriv       crypto.BoxPrivKey
-	sharedSesKey    crypto.BoxSharedKey // derived from session keys
-	theirHandle     crypto.Handle
-	myHandle        crypto.Handle
-	theirNonce      crypto.BoxNonce
-	theirNonceMask  uint64
-	theirNonceMutex sync.Mutex // protects the above
-	myNonce         crypto.BoxNonce
-	myNonceMutex    sync.Mutex // protects the above
-	theirMTU        uint16
-	myMTU           uint16
-	wasMTUFixed     bool         // Was the MTU fixed by a receive error?
-	time            time.Time    // Time we last received a packet
-	mtuTime         time.Time    // time myMTU was last changed
-	pingTime        time.Time    // time the first ping was sent since the last received packet
-	pingSend        time.Time    // time the last ping was sent
-	timeMutex       sync.RWMutex // protects all time fields above
-	coords          []byte       // coords of destination
-	coordsMutex     sync.RWMutex // protects the above
-	packet          []byte       // a buffered packet, sent immediately on ping/pong
-	init            bool         // Reset if coords change
-	initMutex       sync.RWMutex
-	send            chan []byte
-	recv            chan *wire_trafficPacket
-	closed          chan interface{}
-	tstamp          int64  // ATOMIC - tstamp from their last session ping, replay attack mitigation
-	bytesSent       uint64 // Bytes of real traffic sent in this session
-	bytesRecvd      uint64 // Bytes of real traffic received in this session
+	core            *Core                    //
+	reconfigure     chan chan error          //
+	theirAddr       address.Address          //
+	theirSubnet     address.Subnet           //
+	theirPermPub    crypto.BoxPubKey         //
+	theirSesPub     crypto.BoxPubKey         //
+	mySesPub        crypto.BoxPubKey         //
+	mySesPriv       crypto.BoxPrivKey        //
+	sharedSesKey    crypto.BoxSharedKey      // derived from session keys
+	theirHandle     crypto.Handle            //
+	myHandle        crypto.Handle            //
+	theirNonce      crypto.BoxNonce          //
+	theirNonceMask  uint64                   //
+	theirNonceMutex sync.Mutex               // protects the above
+	myNonce         crypto.BoxNonce          //
+	myNonceMutex    sync.Mutex               // protects the above
+	theirMTU        uint16                   //
+	myMTU           uint16                   //
+	wasMTUFixed     bool                     // Was the MTU fixed by a receive error?
+	time            atomic.Value             // time.Time // Time we last received a packet
+	mtuTime         atomic.Value             // time.Time // time myMTU was last changed
+	pingTime        atomic.Value             // time.Time // time the first ping was sent since the last received packet
+	pingSend        atomic.Value             // time.Time // time the last ping was sent
+	coords          []byte                   // coords of destination
+	packet          []byte                   // a buffered packet, sent immediately on ping/pong
+	init            atomic.Value             // bool      // Reset if coords change
+	send            chan []byte              //
+	recv            chan *wire_trafficPacket //
+	closed          chan interface{}         //
+	tstamp          int64                    // ATOMIC - tstamp from their last session ping, replay attack mitigation
+	bytesSent       uint64                   // Bytes of real traffic sent in this session
+	bytesRecvd      uint64                   // Bytes of real traffic received in this session
 }
 
 // Represents a session ping/pong packet, andincludes information like public keys, a session handle, coords, a timestamp to prevent replays, and the tun/tap MTU.
@@ -60,10 +57,10 @@ type sessionPing struct {
 	SendPermPub crypto.BoxPubKey // Sender's permanent key
 	Handle      crypto.Handle    // Random number to ID session
 	SendSesPub  crypto.BoxPubKey // Session key to use
-	Coords      []byte
-	Tstamp      int64 // unix time, but the only real requirement is that it increases
-	IsPong      bool
-	MTU         uint16
+	Coords      []byte           //
+	Tstamp      int64            // unix time, but the only real requirement is that it increases
+	IsPong      bool             //
+	MTU         uint16           //
 }
 
 // Updates session info in response to a ping, after checking that the ping is OK.
@@ -93,21 +90,15 @@ func (s *sessionInfo) update(p *sessionPing) bool {
 		s.coords = append(make([]byte, 0, len(p.Coords)+11), p.Coords...)
 	}
 	now := time.Now()
-	s.timeMutex.Lock()
-	s.time = now
-	s.timeMutex.Unlock()
+	s.time.Store(now)
 	atomic.StoreInt64(&s.tstamp, p.Tstamp)
-	s.initMutex.Lock()
-	s.init = true
-	s.initMutex.Unlock()
+	s.init.Store(true)
 	return true
 }
 
 // Returns true if the session has been idle for longer than the allowed timeout.
 func (s *sessionInfo) timedout() bool {
-	s.timeMutex.RLock()
-	defer s.timeMutex.RUnlock()
-	return time.Since(s.time) > time.Minute
+	return time.Since(s.time.Load().(time.Time)) > time.Minute
 }
 
 // Struct of all active sessions.
@@ -291,12 +282,10 @@ func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo {
 	sinfo.theirMTU = 1280
 	sinfo.myMTU = 1280
 	now := time.Now()
-	sinfo.timeMutex.Lock()
-	sinfo.time = now
-	sinfo.mtuTime = now
-	sinfo.pingTime = now
-	sinfo.pingSend = now
-	sinfo.timeMutex.Unlock()
+	sinfo.time.Store(now)
+	sinfo.mtuTime.Store(now)
+	sinfo.pingTime.Store(now)
+	sinfo.pingSend.Store(now)
 	higher := false
 	for idx := range ss.core.boxPub {
 		if ss.core.boxPub[idx] > sinfo.theirPermPub[idx] {
@@ -437,7 +426,6 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) {
 	bs := ping.encode()
 	shared := ss.getSharedKey(&ss.core.boxPriv, &sinfo.theirPermPub)
 	payload, nonce := crypto.BoxSeal(shared, bs, nil)
-	sinfo.coordsMutex.RLock()
 	p := wire_protoTrafficPacket{
 		Coords:  sinfo.coords,
 		ToKey:   sinfo.theirPermPub,
@@ -445,13 +433,10 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) {
 		Nonce:   *nonce,
 		Payload: payload,
 	}
-	sinfo.coordsMutex.RUnlock()
 	packet := p.encode()
 	ss.core.router.out(packet)
 	if !isPong {
-		sinfo.timeMutex.Lock()
-		sinfo.pingSend = time.Now()
-		sinfo.timeMutex.Unlock()
+		sinfo.pingSend.Store(time.Now())
 	}
 }
 
@@ -551,8 +536,6 @@ func (sinfo *sessionInfo) updateNonce(theirNonce *crypto.BoxNonce) {
 // Called after coord changes, so attemtps to use a session will trigger a new ping and notify the remote end of the coord change.
 func (ss *sessions) resetInits() {
 	for _, sinfo := range ss.sinfos {
-		sinfo.initMutex.Lock()
-		sinfo.init = false
-		sinfo.initMutex.Unlock()
+		sinfo.init.Store(false)
 	}
 }