TUN/TAP addr/subnet to Conn mappings, other fixes

src/tuntap/tun.go

@@ -28,19 +28,20 @@ const tun_ETHER_HEADER_LENGTH = 14
 // you should pass this object to the yggdrasil.SetRouterAdapter() function
 // before calling yggdrasil.Start().
 type TunAdapter struct {
-	config      *config.NodeState
+	config       *config.NodeState
-	log         *log.Logger
+	log          *log.Logger
-	reconfigure chan chan error
+	reconfigure  chan chan error
-	listener    *yggdrasil.Listener
+	listener     *yggdrasil.Listener
-	dialer      *yggdrasil.Dialer
+	dialer       *yggdrasil.Dialer
-	addr        address.Address
+	addr         address.Address
-	subnet      address.Subnet
+	subnet       address.Subnet
-	icmpv6      ICMPv6
+	icmpv6       ICMPv6
-	mtu         int
+	mtu          int
-	iface       *water.Interface
+	iface        *water.Interface
-	mutex       sync.RWMutex // Protects the below
+	mutex        sync.RWMutex // Protects the below
-	conns       map[crypto.NodeID]*yggdrasil.Conn
+	addrToConn   map[address.Address]*yggdrasil.Conn
-	isOpen      bool
+	subnetToConn map[address.Subnet]*yggdrasil.Conn
+	isOpen       bool
 }
 
 // Gets the maximum supported MTU for the platform based on the defaults in
@@ -102,7 +103,8 @@ 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.addrToConn = make(map[address.Address]*yggdrasil.Conn)
+	tun.subnetToConn = make(map[address.Subnet]*yggdrasil.Conn)
 }
 
 // Start the setup process for the TUN/TAP adapter. If successful, starts the
@@ -181,23 +183,40 @@ func (tun *TunAdapter) handler() error {
 
 func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
 	remoteNodeID := conn.RemoteAddr()
-	tun.mutex.Lock()
+	remoteAddr := address.AddrForNodeID(&remoteNodeID)
-	if _, isIn := tun.conns[remoteNodeID]; isIn {
+	remoteSubnet := address.SubnetForNodeID(&remoteNodeID)
-		tun.mutex.Unlock()
+	err := func() error {
-		return errors.New("duplicate connection")
+		tun.mutex.RLock()
+		defer tun.mutex.RUnlock()
+		if _, isIn := tun.addrToConn[*remoteAddr]; isIn {
+			return errors.New("duplicate connection for address " + net.IP(remoteAddr[:]).String())
+		}
+		if _, isIn := tun.subnetToConn[*remoteSubnet]; isIn {
+			return errors.New("duplicate connection for subnet " + net.IP(remoteSubnet[:]).String())
+		}
+		return nil
+	}()
+	if err != nil {
+		//return err
+		panic(err)
 	}
-	tun.conns[remoteNodeID] = conn
+	// Store the connection mapped to address and subnet
+	tun.mutex.Lock()
+	tun.addrToConn[*remoteAddr] = conn
+	tun.subnetToConn[*remoteSubnet] = conn
 	tun.mutex.Unlock()
+	// Make sure to clean those up later when the connection is closed
 	defer func() {
 		tun.mutex.Lock()
-		delete(tun.conns, remoteNodeID)
+		delete(tun.addrToConn, *remoteAddr)
+		delete(tun.subnetToConn, *remoteSubnet)
 		tun.mutex.Unlock()
 	}()
 	b := make([]byte, 65535)
 	for {
 		n, err := conn.Read(b)
 		if err != nil {
-			tun.log.Errorln(conn.String(), "TUN/TAP conn read error:", err)
+			//tun.log.Errorln(conn.String(), "TUN/TAP conn read error:", err)
 			continue
 		}
 		if n == 0 {
@@ -261,21 +280,28 @@ func (tun *TunAdapter) ifaceReader() error {
 			// For now don't deal with any non-Yggdrasil ranges
 			continue
 		}
-		dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
+		// Do we have an active connection for this node address?
-		// Do we have an active connection for this node ID?
 		tun.mutex.RLock()
-		conn, isIn := tun.conns[*dstNodeID]
+		conn, isIn := tun.addrToConn[dstAddr]
+		if !isIn || conn == nil {
+			conn, isIn = tun.subnetToConn[dstSnet]
+			if !isIn || conn == nil {
+				// Neither an address nor a subnet mapping matched, therefore populate
+				// the node ID and mask to commence a search
+				dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
+			}
+		}
 		tun.mutex.RUnlock()
 		// If we don't have a connection then we should open one
-		if !isIn {
+		if !isIn || conn == nil {
+			// Check we haven't been given empty node ID, really this shouldn't ever
+			// happen but just to be sure...
+			if dstNodeID == nil || dstNodeIDMask == nil {
+				panic("Given empty dstNodeID and dstNodeIDMask - this shouldn't happen")
+			}
 			// Dial to the remote node
 			if c, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
-				// We've been given a connection, so save it in our connections so we
+				// We've been given a connection so start the connection reader goroutine
-				// can refer to it the next time we send a packet to this destination
-				tun.mutex.Lock()
-				tun.conns[*dstNodeID] = &c
-				tun.mutex.Unlock()
-				// Start the connection reader goroutine
 				go tun.connReader(&c)
 				// Then update our reference to the connection
 				conn, isIn = &c, true
@@ -285,9 +311,10 @@ func (tun *TunAdapter) ifaceReader() error {
 				continue
 			}
 		}
-		// If we have an open connection, either because we already had one or
+		// If we have a connection now, try writing to it
-		// because we opened one above, try writing the packet to it
+		if conn != nil {
-		if isIn && conn != nil {
+			// If we have an open connection, either because we already had one or
+			// because we opened one above, try writing the packet to it
 			w, err := conn.Write(bs[:n])
 			if err != nil {
 				tun.log.Errorln(conn.String(), "TUN/TAP conn write error:", err)

src/yggdrasil/conn.go

@@ -20,7 +20,6 @@ type Conn struct {
 	session       *sessionInfo
 	readDeadline  atomic.Value // time.Time // TODO timer
 	writeDeadline atomic.Value // time.Time // TODO timer
-	expired       atomic.Value // bool
 	searching     atomic.Value // bool
 }
 
@@ -30,39 +29,58 @@ func (c *Conn) String() string {
 
 // This method should only be called from the router goroutine
 func (c *Conn) startSearch() {
+	// The searchCompleted callback is given to the search
 	searchCompleted := func(sinfo *sessionInfo, err error) {
+		// Update the connection with the fact that the search completed, which
+		// allows another search to be triggered if necessary
 		c.searching.Store(false)
-		c.mutex.Lock()
+		// If the search failed for some reason, e.g. it hit a dead end or timed
-		defer c.mutex.Unlock()
+		// out, then do nothing
 		if err != nil {
 			c.core.log.Debugln(c.String(), "DHT search failed:", err)
-			c.expired.Store(true)
 			return
 		}
+		// Take the connection mutex
+		c.mutex.Lock()
+		defer c.mutex.Unlock()
+		// Were we successfully given a sessionInfo pointeR?
 		if sinfo != nil {
+			// Store it, and update the nodeID and nodeMask (which may have been
+			// wildcarded before now) with their complete counterparts
 			c.core.log.Debugln(c.String(), "DHT search completed")
 			c.session = sinfo
-			c.nodeID, c.nodeMask = sinfo.theirAddr.GetNodeIDandMask()
+			c.nodeID = crypto.GetNodeID(&sinfo.theirPermPub)
-			c.expired.Store(false)
+			for i := range c.nodeMask {
+				c.nodeMask[i] = 0xFF
+			}
 		} else {
-			c.core.log.Debugln(c.String(), "DHT search failed: no session returned")
+			// No session was returned - this shouldn't really happen because we
-			c.expired.Store(true)
+			// should always return an error reason if we don't return a session
-			return
+			panic("DHT search didn't return an error or a sessionInfo")
 		}
 	}
+	// doSearch will be called below in response to one or more conditions
 	doSearch := func() {
+		// Store the fact that we're searching, so that we don't start additional
+		// searches until this one has completed
 		c.searching.Store(true)
+		// Check to see if there is a search already matching the destination
 		sinfo, isIn := c.core.searches.searches[*c.nodeID]
 		if !isIn {
+			// Nothing was found, so create a new search
 			sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
 			c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo)
 		}
+		// Continue the search
 		c.core.searches.continueSearch(sinfo)
 	}
+	// Take a copy of the session object, in case it changes later
 	c.mutex.RLock()
 	sinfo := c.session
 	c.mutex.RUnlock()
 	if c.session == nil {
+		// No session object is present so previous searches, if we ran any, have
+		// not yielded a useful result (dead end, remote host not found)
 		doSearch()
 	} else {
 		sinfo.worker <- func() {
@@ -83,10 +101,6 @@ func (c *Conn) startSearch() {
 }
 
 func (c *Conn) Read(b []byte) (int, error) {
-	// If the session is marked as expired then do nothing at this point
-	if e, ok := c.expired.Load().(bool); ok && e {
-		return 0, errors.New("session is closed")
-	}
 	// Take a copy of the session object
 	c.mutex.RLock()
 	sinfo := c.session
@@ -95,17 +109,15 @@ func (c *Conn) Read(b []byte) (int, error) {
 	// in a write, we would trigger a new session, but it doesn't make sense for
 	// us to block forever here if the session will not reopen.
 	// TODO: should this return an error or just a zero-length buffer?
-	if !sinfo.init {
+	if sinfo == nil || !sinfo.init {
 		return 0, errors.New("session is closed")
 	}
 	// Wait for some traffic to come through from the session
 	select {
 	// TODO...
 	case p, ok := <-c.recv:
-		// If the channel was closed then mark the connection as expired, this will
+		// If the session is closed then do nothing
-		// mean that the next write will start a new search and reopen the session
 		if !ok {
-			c.expired.Store(true)
 			return 0, errors.New("session is closed")
 		}
 		defer util.PutBytes(p.Payload)
@@ -155,13 +167,9 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 	c.mutex.RLock()
 	sinfo := c.session
 	c.mutex.RUnlock()
-	// Check whether the connection is expired, if it is we can start a new
-	// search to revive it
-	expired, eok := c.expired.Load().(bool)
 	// If the session doesn't exist, or isn't initialised (which probably means
-	// that the session was never set up or it closed by timeout), or the conn
+	// that the search didn't complete successfully) then try to search again
-	// is marked as expired, then see if we can start a new search
+	if sinfo == nil || !sinfo.init {
-	if sinfo == nil || !sinfo.init || (eok && expired) {
 		// Is a search already taking place?
 		if searching, sok := c.searching.Load().(bool); !sok || (sok && !searching) {
 			// No search was already taking place so start a new one
@@ -173,7 +181,7 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 		// A search is already taking place so wait for it to finish
 		return 0, errors.New("waiting for search to complete")
 	}
-	//defer util.PutBytes(b)
+	// defer util.PutBytes(b)
 	var packet []byte
 	// Hand over to the session worker
 	sinfo.doWorker(func() {
@@ -197,11 +205,9 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 }
 
 func (c *Conn) Close() error {
-	// Mark the connection as expired, so that a future read attempt will fail
-	// and a future write attempt will start a new search
-	c.expired.Store(true)
 	// Close the session, if it hasn't been closed already
 	c.session.close()
+	c.session = nil
 	// This can't fail yet - TODO?
 	return nil
 }

src/yggdrasil/session.go

@@ -107,15 +107,6 @@ func (s *sessionInfo) update(p *sessionPing) bool {
 	return true
 }
 
-// Returns true if the session has been idle for longer than the allowed timeout.
-func (s *sessionInfo) timedout() bool {
-	var timedout bool
-	s.doWorker(func() {
-		timedout = time.Since(s.time) > time.Minute
-	})
-	return timedout
-}
-
 // Struct of all active sessions.
 // Sessions are indexed by handle.
 // Additionally, stores maps of address/subnet onto keys, and keys onto handles.
@@ -233,10 +224,6 @@ func (ss *sessions) isSessionAllowed(pubkey *crypto.BoxPubKey, initiator bool) b
 // Gets the session corresponding to a given handle.
 func (ss *sessions) getSessionForHandle(handle *crypto.Handle) (*sessionInfo, bool) {
 	sinfo, isIn := ss.sinfos[*handle]
-	if isIn && sinfo.timedout() {
-		// We have a session, but it has timed out
-		return nil, false
-	}
 	return sinfo, isIn
 }
 
@@ -280,8 +267,9 @@ func (ss *sessions) getByTheirSubnet(snet *address.Subnet) (*sessionInfo, bool)
 	return sinfo, isIn
 }
 
-// Creates a new session and lazily cleans up old/timedout existing sessions.
+// Creates a new session and lazily cleans up old existing sessions. This
-// This includse initializing session info to sane defaults (e.g. lowest supported MTU).
+// includse initializing session info to sane defaults (e.g. lowest supported
+// MTU).
 func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo {
 	if !ss.isSessionAllowed(theirPermKey, true) {
 		return nil
@@ -341,11 +329,6 @@ func (ss *sessions) cleanup() {
 	if time.Since(ss.lastCleanup) < time.Minute {
 		return
 	}
-	for _, s := range ss.sinfos {
-		if s.timedout() {
-			s.close()
-		}
-	}
 	permShared := make(map[crypto.BoxPubKey]*crypto.BoxSharedKey, len(ss.permShared))
 	for k, v := range ss.permShared {
 		permShared[k] = v
@@ -387,7 +370,6 @@ func (sinfo *sessionInfo) close() {
 	delete(sinfo.core.sessions.addrToPerm, sinfo.theirAddr)
 	delete(sinfo.core.sessions.subnetToPerm, sinfo.theirSubnet)
 	close(sinfo.worker)
-	sinfo.init = false
 }
 
 // Returns a session ping appropriate for the given session info.
@@ -465,17 +447,16 @@ func (ss *sessions) handlePing(ping *sessionPing) {
 			return
 		}
 	}
-	if !isIn || sinfo.timedout() {
+	if !isIn {
-		if isIn {
-			sinfo.close()
-		}
 		ss.createSession(&ping.SendPermPub)
 		sinfo, isIn = ss.getByTheirPerm(&ping.SendPermPub)
 		if !isIn {
 			panic("This should not happen")
 		}
 		ss.listenerMutex.Lock()
-		if ss.listener != nil {
+		// Check and see if there's a Listener waiting to accept connections
+		// TODO: this should not block if nothing is accepting
+		if !ping.IsPong && ss.listener != nil {
 			conn := &Conn{
 				core:     ss.core,
 				session:  sinfo,
@@ -488,8 +469,6 @@ func (ss *sessions) handlePing(ping *sessionPing) {
 				conn.nodeMask[i] = 0xFF
 			}
 			ss.listener.conn <- conn
-		} else {
-			ss.core.log.Debugln("Received new session but there is no listener, ignoring")
 		}
 		ss.listenerMutex.Unlock()
 	}