Fix search behaviour on closed Conns, various other fixes

src/tuntap/tun.go

@@ -193,12 +193,11 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
 		delete(tun.conns, remoteNodeID)
 		tun.mutex.Unlock()
 	}()
-	tun.log.Debugln("Start connection reader for", conn.String())
 	b := make([]byte, 65535)
 	for {
 		n, err := conn.Read(b)
 		if err != nil {
-			tun.log.Errorln("TUN/TAP conn read error:", err)
+			tun.log.Errorln(conn.String(), "TUN/TAP conn read error:", err)
 			continue
 		}
 		if n == 0 {
@@ -206,11 +205,11 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
 		}
 		w, err := tun.iface.Write(b[:n])
 		if err != nil {
-			tun.log.Errorln("TUN/TAP iface write error:", err)
+			tun.log.Errorln(conn.String(), "TUN/TAP iface write error:", err)
 			continue
 		}
 		if w != n {
-			tun.log.Errorln("TUN/TAP iface write mismatch:", w, "bytes written vs", n, "bytes given")
+			tun.log.Errorln(conn.String(), "TUN/TAP iface write mismatch:", w, "bytes written vs", n, "bytes given")
 			continue
 		}
 	}
@@ -219,20 +218,24 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error {
 func (tun *TunAdapter) ifaceReader() error {
 	bs := make([]byte, 65535)
 	for {
+		// Wait for a packet to be delivered to us through the TUN/TAP adapter
 		n, err := tun.iface.Read(bs)
 		if err != nil {
 			continue
 		}
-		// Look up if the dstination address is somewhere we already have an
+		// From the IP header, work out what our source and destination addresses
-		// open connection to
+		// and node IDs are. We will need these in order to work out where to send
+		// the packet
 		var srcAddr address.Address
 		var dstAddr address.Address
 		var dstNodeID *crypto.NodeID
 		var dstNodeIDMask *crypto.NodeID
 		var dstSnet address.Subnet
 		var addrlen int
+		// Check the IP protocol - if it doesn't match then we drop the packet and
+		// do nothing with it
 		if bs[0]&0xf0 == 0x60 {
-			// Check if we have a fully-sized header
+			// Check if we have a fully-sized IPv6 header
 			if len(bs) < 40 {
 				continue
 			}
@@ -242,7 +245,7 @@ func (tun *TunAdapter) ifaceReader() error {
 			copy(dstAddr[:addrlen], bs[24:])
 			copy(dstSnet[:addrlen/2], bs[24:])
 		} else if bs[0]&0xf0 == 0x40 {
-			// Check if we have a fully-sized header
+			// Check if we have a fully-sized IPv4 header
 			if len(bs) < 20 {
 				continue
 			}
@@ -251,7 +254,7 @@ func (tun *TunAdapter) ifaceReader() error {
 			copy(srcAddr[:addrlen], bs[12:])
 			copy(dstAddr[:addrlen], bs[16:])
 		} else {
-			// Unknown address length or protocol
+			// Unknown address length or protocol, so drop the packet and ignore it
 			continue
 		}
 		if !dstAddr.IsValid() && !dstSnet.IsValid() {
@@ -261,32 +264,39 @@ func (tun *TunAdapter) ifaceReader() error {
 		dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
 		// Do we have an active connection for this node ID?
 		tun.mutex.RLock()
-		if conn, isIn := tun.conns[*dstNodeID]; isIn {
+		conn, isIn := tun.conns[*dstNodeID]
 		tun.mutex.RUnlock()
+		// If we don't have a connection then we should open one
+		if !isIn {
+			// 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
+				// 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
+			} else {
+				// We weren't able to dial for some reason so there's no point in
+				// continuing this iteration - skip to the next one
+				continue
+			}
+		}
+		// If we have an open connection, either because we already had one or
+		// because we opened one above, try writing the packet to it
+		if isIn && conn != nil {
 			w, err := conn.Write(bs[:n])
 			if err != nil {
-				tun.log.Errorln("TUN/TAP conn write error:", err)
+				tun.log.Errorln(conn.String(), "TUN/TAP conn write error:", err)
 				continue
 			}
 			if w != n {
-				tun.log.Errorln("TUN/TAP conn write mismatch:", w, "bytes written vs", n, "bytes given")
+				tun.log.Errorln(conn.String(), "TUN/TAP conn write mismatch:", w, "bytes written vs", n, "bytes given")
 				continue
 			}
-		} else {
-			tun.mutex.RUnlock()
-			func() {
-				tun.mutex.Lock()
-				defer tun.mutex.Unlock()
-				if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
-					tun.log.Debugln("Opening new session connection")
-					tun.log.Debugln("Node:", dstNodeID)
-					tun.log.Debugln("Mask:", dstNodeIDMask)
-					tun.conns[*dstNodeID] = &conn
-					go tun.connReader(&conn)
-				} else {
-					tun.log.Errorln("TUN/TAP dial error:", err)
-				}
-			}()
 		}
 
 		/*if !r.cryptokey.isValidSource(srcAddr, addrlen) {

src/yggdrasil/conn.go

@@ -21,41 +21,50 @@ type Conn struct {
 	readDeadline  atomic.Value // time.Time // TODO timer
 	writeDeadline atomic.Value // time.Time // TODO timer
 	expired       atomic.Value // bool
+	searching     atomic.Value // bool
 }
 
 func (c *Conn) String() string {
-	return fmt.Sprintf("c=%p", c)
+	return fmt.Sprintf("conn=%p", c)
 }
 
 // This method should only be called from the router goroutine
 func (c *Conn) startSearch() {
 	searchCompleted := func(sinfo *sessionInfo, err error) {
-		if err != nil {
+		c.searching.Store(false)
-			c.core.log.Debugln("DHT search failed:", err)
 		c.mutex.Lock()
+		defer c.mutex.Unlock()
+		if err != nil {
+			c.core.log.Debugln(c.String(), "DHT search failed:", err)
 			c.expired.Store(true)
 			return
 		}
 		if sinfo != nil {
-			c.mutex.Lock()
+			c.core.log.Debugln(c.String(), "DHT search completed")
 			c.session = sinfo
 			c.nodeID, c.nodeMask = sinfo.theirAddr.GetNodeIDandMask()
-			c.mutex.Unlock()
+			c.expired.Store(false)
+		} else {
+			c.core.log.Debugln(c.String(), "DHT search failed: no session returned")
+			c.expired.Store(true)
+			return
 		}
 	}
 	doSearch := func() {
+		c.searching.Store(true)
 		sinfo, isIn := c.core.searches.searches[*c.nodeID]
 		if !isIn {
 			sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
+			c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo)
 		}
 		c.core.searches.continueSearch(sinfo)
 	}
 	c.mutex.RLock()
-	defer c.mutex.RUnlock()
+	sinfo := c.session
+	c.mutex.RUnlock()
 	if c.session == nil {
 		doSearch()
 	} else {
-		sinfo := c.session // In case c.session is somehow changed meanwhile
 		sinfo.worker <- func() {
 			switch {
 			case !sinfo.init:
@@ -74,43 +83,66 @@ 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
 	c.mutex.RUnlock()
+	// If the session is not initialised, do nothing. Currently in this instance
+	// 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 {
+		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
+		// 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)
 		var err error
+		// Hand over to the session worker
 		sinfo.doWorker(func() {
+			// If the nonce is bad then drop the packet and return an error
 			if !sinfo.nonceIsOK(&p.Nonce) {
 				err = errors.New("packet dropped due to invalid nonce")
 				return
 			}
+			// Decrypt the packet
 			bs, isOK := crypto.BoxOpen(&sinfo.sharedSesKey, p.Payload, &p.Nonce)
+			// Check if we were unable to decrypt the packet for some reason and
+			// return an error if we couldn't
 			if !isOK {
 				util.PutBytes(bs)
 				err = errors.New("packet dropped due to decryption failure")
 				return
 			}
+			// Return the newly decrypted buffer back to the slice we were given
 			copy(b, bs)
+			// Trim the slice down to size based on the data we received
 			if len(bs) < len(b) {
 				b = b[:len(bs)]
 			}
+			// Update the session
 			sinfo.updateNonce(&p.Nonce)
 			sinfo.time = time.Now()
 			sinfo.bytesRecvd += uint64(len(b))
 		})
+		// Something went wrong in the session worker so abort
 		if err != nil {
 			return 0, err
 		}
+		// If we've reached this point then everything went to plan, return the
+		// number of bytes we populated back into the given slice
 		return len(b), nil
 		//case <-c.recvTimeout:
 		//case <-c.session.closed:
@@ -120,27 +152,35 @@ func (c *Conn) Read(b []byte) (int, error) {
 }
 
 func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
-	if e, ok := c.expired.Load().(bool); ok && e {
-		return 0, errors.New("session is closed")
-	}
 	c.mutex.RLock()
 	sinfo := c.session
 	c.mutex.RUnlock()
-	if sinfo == nil {
+	// 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
+	// is marked as expired, then see if we can start a new search
+	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
 			c.core.router.doAdmin(func() {
 				c.startSearch()
 			})
-		return 0, errors.New("searching for remote side")
+			return 0, errors.New("starting search")
+		}
+		// 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)
 	var packet []byte
+	// Hand over to the session worker
 	sinfo.doWorker(func() {
-		if !sinfo.init {
+		// Encrypt the packet
-			err = errors.New("waiting for remote side to accept " + c.String())
-			return
-		}
 		payload, nonce := crypto.BoxSeal(&sinfo.sharedSesKey, b, &sinfo.myNonce)
 		defer util.PutBytes(payload)
+		// Construct the wire packet to send to the router
 		p := wire_trafficPacket{
 			Coords:  sinfo.coords,
 			Handle:  sinfo.theirHandle,
@@ -150,13 +190,19 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 		packet = p.encode()
 		sinfo.bytesSent += uint64(len(b))
 	})
+	// Give the packet to the router
 	sinfo.core.router.out(packet)
+	// Finally return the number of bytes we wrote
 	return len(b), nil
 }
 
 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()
+	// This can't fail yet - TODO?
 	return nil
 }
 

src/yggdrasil/dialer.go

@@ -65,8 +65,5 @@ func (d *Dialer) DialByNodeIDandMask(nodeID, nodeMask *crypto.NodeID) (Conn, err
 		nodeMask: nodeMask,
 		recv:     make(chan *wire_trafficPacket, 32),
 	}
-	conn.core.router.admin <- func() {
-		conn.startSearch()
-	}
 	return conn, nil
 }

src/yggdrasil/search.go

@@ -90,11 +90,10 @@ func (s *searches) handleDHTRes(res *dhtRes) {
 	if !isIn || s.checkDHTRes(sinfo, res) {
 		// Either we don't recognize this search, or we just finished it
 		return
-	} else {
+	}
 	// Add to the search and continue
 	s.addToSearch(sinfo, res)
 	s.doSearchStep(sinfo)
-	}
 }
 
 // Adds the information from a dhtRes to an ongoing search.

src/yggdrasil/session.go

@@ -387,6 +387,7 @@ 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.