start migrating the TunAdapter to the actor model

src/tuntap/conn.go

@@ -113,8 +113,7 @@ func (s *tunConn) _read(bs []byte) (err error) {
 		util.PutBytes(bs)
 		return
 	}
-	// FIXME this send can block if the tuntap isn't running, which isn't really safe...
+	s.tun.writer.writeFrom(s, bs)
-	s.tun.send <- bs
 	s.stillAlive()
 	return
 }
@@ -208,7 +207,7 @@ func (s *tunConn) _write(bs []byte) (err error) {
 					Data: bs[:900],
 				}
 				if packet, err := CreateICMPv6(bs[8:24], bs[24:40], ipv6.ICMPTypePacketTooBig, 0, ptb); err == nil {
-					s.tun.send <- packet
+					s.tun.writer.writeFrom(s, packet)
 				}
 			} else {
 				if e.Closed() {

src/tuntap/iface.go

@@ -9,264 +9,279 @@ import (
 	"github.com/yggdrasil-network/yggdrasil-go/src/address"
 	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 	"github.com/yggdrasil-network/yggdrasil-go/src/util"
+
+	"github.com/Arceliar/phony"
 )
 
-func (tun *TunAdapter) writer() error {
+type tunWriter struct {
-	var w int
+	phony.Inbox
+	tun *TunAdapter
+}
+
+func (w *tunWriter) writeFrom(from phony.Actor, b []byte) {
+	w.RecvFrom(from, func() {
+		w._write(b)
+	})
+}
+
+// write is pretty loose with the memory safety rules, e.g. it assumes it can read w.tun.iface.IsTap() safely
+func (w *tunWriter) _write(b []byte) {
+	var written int
 	var err error
-	for {
+	n := len(b)
-		b := <-tun.send
+	if n == 0 {
-		n := len(b)
+		return
-		if n == 0 {
-			continue
-		}
-		if tun.iface.IsTAP() {
-			sendndp := func(dstAddr address.Address) {
-				neigh, known := tun.icmpv6.getNeighbor(dstAddr)
-				known = known && (time.Since(neigh.lastsolicitation).Seconds() < 30)
-				if !known {
-					tun.icmpv6.Solicit(dstAddr)
-				}
-			}
-			peermac := net.HardwareAddr{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
-			var dstAddr address.Address
-			var peerknown bool
-			if b[0]&0xf0 == 0x40 {
-				dstAddr = tun.addr
-			} else if b[0]&0xf0 == 0x60 {
-				if !bytes.Equal(tun.addr[:16], dstAddr[:16]) && !bytes.Equal(tun.subnet[:8], dstAddr[:8]) {
-					dstAddr = tun.addr
-				}
-			}
-			if neighbor, ok := tun.icmpv6.getNeighbor(dstAddr); ok && neighbor.learned {
-				// If we've learned the MAC of a 300::/7 address, for example, or a CKR
-				// address, use the MAC address of that
-				peermac = neighbor.mac
-				peerknown = true
-			} else if neighbor, ok := tun.icmpv6.getNeighbor(tun.addr); ok && neighbor.learned {
-				// Otherwise send directly to the MAC address of the host if that's
-				// known instead
-				peermac = neighbor.mac
-				peerknown = true
-			} else {
-				// Nothing has been discovered, try to discover the destination
-				sendndp(tun.addr)
-			}
-			if peerknown {
-				var proto ethernet.Ethertype
-				switch {
-				case b[0]&0xf0 == 0x60:
-					proto = ethernet.IPv6
-				case b[0]&0xf0 == 0x40:
-					proto = ethernet.IPv4
-				}
-				var frame ethernet.Frame
-				frame.Prepare(
-					peermac[:6],          // Destination MAC address
-					tun.icmpv6.mymac[:6], // Source MAC address
-					ethernet.NotTagged,   // VLAN tagging
-					proto,                // Ethertype
-					len(b))               // Payload length
-				copy(frame[tun_ETHER_HEADER_LENGTH:], b[:n])
-				n += tun_ETHER_HEADER_LENGTH
-				w, err = tun.iface.Write(frame[:n])
-			} else {
-				tun.log.Errorln("TUN/TAP iface write error: no peer MAC known for", net.IP(dstAddr[:]).String(), "- dropping packet")
-			}
-		} else {
-			w, err = tun.iface.Write(b[:n])
-			util.PutBytes(b)
-		}
-		if err != nil {
-			if !tun.isOpen {
-				return err
-			}
-			tun.log.Errorln("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")
-			continue
-		}
 	}
-}
+	if w.tun.iface.IsTAP() {
-
+		sendndp := func(dstAddr address.Address) {
-// Run in a separate goroutine by the reader
+			neigh, known := w.tun.icmpv6.getNeighbor(dstAddr)
-// Does all of the per-packet ICMP checks, passes packets to the right Conn worker
+			known = known && (time.Since(neigh.lastsolicitation).Seconds() < 30)
-func (tun *TunAdapter) readerPacketHandler(ch chan []byte) {
+			if !known {
-	for recvd := range ch {
+				w.tun.icmpv6.Solicit(dstAddr)
-		// If it's a TAP adapter, update the buffer slice so that we no longer
-		// include the ethernet headers
-		offset := 0
-		if tun.iface.IsTAP() {
-			// Set our offset to beyond the ethernet headers
-			offset = tun_ETHER_HEADER_LENGTH
-			// Check first of all that we can go beyond the ethernet headers
-			if len(recvd) <= offset {
-				continue
 			}
 		}
-		// Offset the buffer from now on so that we can ignore ethernet frames if
+		peermac := net.HardwareAddr{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
-		// they are present
-		bs := recvd[offset:]
-		// If we detect an ICMP packet then hand it to the ICMPv6 module
-		if bs[6] == 58 {
-			// Found an ICMPv6 packet - we need to make sure to give ICMPv6 the full
-			// Ethernet frame rather than just the IPv6 packet as this is needed for
-			// NDP to work correctly
-			if err := tun.icmpv6.ParsePacket(recvd); err == nil {
-				// We acted on the packet in the ICMPv6 module so don't forward or do
-				// anything else with it
-				continue
-			}
-		}
-		if offset != 0 {
-			// Shift forward to avoid leaking bytes off the front of the slice when we eventually store it
-			bs = append(recvd[:0], bs...)
-		}
-		// From the IP header, work out what our source and destination addresses
-		// and node IDs are. We will need these in order to work out where to send
-		// the packet
 		var dstAddr address.Address
-		var dstSnet address.Subnet
+		var peerknown bool
-		var addrlen int
+		if b[0]&0xf0 == 0x40 {
-		n := len(bs)
+			dstAddr = w.tun.addr
-		// Check the IP protocol - if it doesn't match then we drop the packet and
+		} else if b[0]&0xf0 == 0x60 {
-		// do nothing with it
+			if !bytes.Equal(w.tun.addr[:16], dstAddr[:16]) && !bytes.Equal(w.tun.subnet[:8], dstAddr[:8]) {
-		if bs[0]&0xf0 == 0x60 {
+				dstAddr = w.tun.addr
-			// Check if we have a fully-sized IPv6 header
-			if len(bs) < 40 {
-				continue
 			}
-			// Check the packet size
+		}
-			if n-tun_IPv6_HEADER_LENGTH != 256*int(bs[4])+int(bs[5]) {
+		if neighbor, ok := w.tun.icmpv6.getNeighbor(dstAddr); ok && neighbor.learned {
-				continue
+			// If we've learned the MAC of a 300::/7 address, for example, or a CKR
-			}
+			// address, use the MAC address of that
-			// IPv6 address
+			peermac = neighbor.mac
-			addrlen = 16
+			peerknown = true
-			copy(dstAddr[:addrlen], bs[24:])
+		} else if neighbor, ok := w.tun.icmpv6.getNeighbor(w.tun.addr); ok && neighbor.learned {
-			copy(dstSnet[:addrlen/2], bs[24:])
+			// Otherwise send directly to the MAC address of the host if that's
-		} else if bs[0]&0xf0 == 0x40 {
+			// known instead
-			// Check if we have a fully-sized IPv4 header
+			peermac = neighbor.mac
-			if len(bs) < 20 {
+			peerknown = true
-				continue
-			}
-			// Check the packet size
-			if n != 256*int(bs[2])+int(bs[3]) {
-				continue
-			}
-			// IPv4 address
-			addrlen = 4
-			copy(dstAddr[:addrlen], bs[16:])
 		} else {
-			// Unknown address length or protocol, so drop the packet and ignore it
+			// Nothing has been discovered, try to discover the destination
-			tun.log.Traceln("Unknown packet type, dropping")
+			sendndp(w.tun.addr)
-			continue
 		}
-		if tun.ckr.isEnabled() {
+		if peerknown {
-			if addrlen != 16 || (!dstAddr.IsValid() && !dstSnet.IsValid()) {
+			var proto ethernet.Ethertype
-				if key, err := tun.ckr.getPublicKeyForAddress(dstAddr, addrlen); err == nil {
+			switch {
-					// A public key was found, get the node ID for the search
+			case b[0]&0xf0 == 0x60:
-					dstNodeID := crypto.GetNodeID(&key)
+				proto = ethernet.IPv6
-					dstAddr = *address.AddrForNodeID(dstNodeID)
+			case b[0]&0xf0 == 0x40:
-					dstSnet = *address.SubnetForNodeID(dstNodeID)
+				proto = ethernet.IPv4
-					addrlen = 16
-				}
 			}
+			var frame ethernet.Frame
+			frame.Prepare(
+				peermac[:6],            // Destination MAC address
+				w.tun.icmpv6.mymac[:6], // Source MAC address
+				ethernet.NotTagged,     // VLAN tagging
+				proto,                  // Ethertype
+				len(b))                 // Payload length
+			copy(frame[tun_ETHER_HEADER_LENGTH:], b[:n])
+			n += tun_ETHER_HEADER_LENGTH
+			written, err = w.tun.iface.Write(frame[:n])
+		} else {
+			w.tun.log.Errorln("TUN/TAP iface write error: no peer MAC known for", net.IP(dstAddr[:]).String(), "- dropping packet")
 		}
-		if addrlen != 16 || (!dstAddr.IsValid() && !dstSnet.IsValid()) {
+	} else {
-			// Couldn't find this node's ygg IP
+		written, err = w.tun.iface.Write(b[:n])
-			continue
+		util.PutBytes(b)
-		}
+	}
-		// Do we have an active connection for this node address?
+	if err != nil {
-		var dstNodeID, dstNodeIDMask *crypto.NodeID
+		w.tun.mutex.Lock()
-		tun.mutex.RLock()
+		open := w.tun.isOpen
-		session, isIn := tun.addrToConn[dstAddr]
+		w.tun.mutex.Unlock()
-		if !isIn || session == nil {
+		if !open {
-			session, isIn = tun.subnetToConn[dstSnet]
+			return
-			if !isIn || session == nil {
-				// Neither an address nor a subnet mapping matched, therefore populate
-				// the node ID and mask to commence a search
-				if dstAddr.IsValid() {
-					dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
-				} else {
-					dstNodeID, dstNodeIDMask = dstSnet.GetNodeIDandMask()
-				}
-			}
-		}
-		tun.mutex.RUnlock()
-		// If we don't have a connection then we should open one
-		if !isIn || session == 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
-			go func() {
-				// FIXME just spitting out a goroutine to do this is kind of ugly and means we drop packets until the dial finishes
-				tun.mutex.Lock()
-				_, known := tun.dials[*dstNodeID]
-				tun.dials[*dstNodeID] = append(tun.dials[*dstNodeID], bs)
-				for len(tun.dials[*dstNodeID]) > 32 {
-					util.PutBytes(tun.dials[*dstNodeID][0])
-					tun.dials[*dstNodeID] = tun.dials[*dstNodeID][1:]
-				}
-				tun.mutex.Unlock()
-				if known {
-					return
-				}
-				var tc *tunConn
-				if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
-					// We've been given a connection so prepare the session wrapper
-					if tc, err = tun.wrap(conn); err != nil {
-						// Something went wrong when storing the connection, typically that
-						// something already exists for this address or subnet
-						tun.log.Debugln("TUN/TAP iface wrap:", err)
-					}
-				}
-				tun.mutex.Lock()
-				packets := tun.dials[*dstNodeID]
-				delete(tun.dials, *dstNodeID)
-				tun.mutex.Unlock()
-				if tc != nil {
-					for _, packet := range packets {
-						p := packet // Possibly required because of how range
-						<-tc.SyncExec(func() { tc._write(p) })
-					}
-				}
-			}()
-			// While the dial is going on we can't do much else
-			// continuing this iteration - skip to the next one
-			continue
-		}
-		// If we have a connection now, try writing to it
-		if isIn && session != nil {
-			<-session.SyncExec(func() { session._write(bs) })
 		}
+		w.tun.log.Errorln("TUN/TAP iface write error:", err)
+	}
+	if written != n {
+		w.tun.log.Errorln("TUN/TAP iface write mismatch:", written, "bytes written vs", n, "bytes given")
 	}
 }
 
-func (tun *TunAdapter) reader() error {
+type tunReader struct {
-	toWorker := make(chan []byte, 32)
+	phony.Inbox
-	defer close(toWorker)
+	tun *TunAdapter
-	go tun.readerPacketHandler(toWorker)
+}
-	for {
+
-		// Get a slice to store the packet in
+func (r *tunReader) _read() {
-		recvd := util.ResizeBytes(util.GetBytes(), 65535+tun_ETHER_HEADER_LENGTH)
+	// Get a slice to store the packet in
-		// Wait for a packet to be delivered to us through the TUN/TAP adapter
+	recvd := util.ResizeBytes(util.GetBytes(), 65535+tun_ETHER_HEADER_LENGTH)
-		n, err := tun.iface.Read(recvd)
+	// Wait for a packet to be delivered to us through the TUN/TAP adapter
-		if err != nil {
+	n, err := r.tun.iface.Read(recvd)
-			if !tun.isOpen {
+	if n == 0 {
-				return err
+		util.PutBytes(recvd)
-			}
+	} else {
-			panic(err)
+		r.tun.handlePacketFrom(r, recvd[:n], err)
-		}
+	}
-		if n == 0 {
+	if err == nil {
-			util.PutBytes(recvd)
+		// Now read again
-			continue
+		r.RecvFrom(nil, r._read)
-		}
+	}
-		// Send the packet to the worker
+}
-		toWorker <- recvd[:n]
+
+func (tun *TunAdapter) handlePacketFrom(from phony.Actor, packet []byte, err error) {
+	tun.RecvFrom(from, func() {
+		tun._handlePacket(packet, err)
+	})
+}
+
+// does the work of reading a packet and sending it to the correct tunConn
+func (tun *TunAdapter) _handlePacket(recvd []byte, err error) {
+	if err != nil {
+		tun.log.Errorln("TUN/TAP iface read error:", err)
+		return
+	}
+	// If it's a TAP adapter, update the buffer slice so that we no longer
+	// include the ethernet headers
+	offset := 0
+	if tun.iface.IsTAP() {
+		// Set our offset to beyond the ethernet headers
+		offset = tun_ETHER_HEADER_LENGTH
+		// Check first of all that we can go beyond the ethernet headers
+		if len(recvd) <= offset {
+			return
+		}
+	}
+	// Offset the buffer from now on so that we can ignore ethernet frames if
+	// they are present
+	bs := recvd[offset:]
+	// If we detect an ICMP packet then hand it to the ICMPv6 module
+	if bs[6] == 58 {
+		// Found an ICMPv6 packet - we need to make sure to give ICMPv6 the full
+		// Ethernet frame rather than just the IPv6 packet as this is needed for
+		// NDP to work correctly
+		if err := tun.icmpv6.ParsePacket(recvd); err == nil {
+			// We acted on the packet in the ICMPv6 module so don't forward or do
+			// anything else with it
+			return
+		}
+	}
+	if offset != 0 {
+		// Shift forward to avoid leaking bytes off the front of the slice when we eventually store it
+		bs = append(recvd[:0], bs...)
+	}
+	// From the IP header, work out what our source and destination addresses
+	// and node IDs are. We will need these in order to work out where to send
+	// the packet
+	var dstAddr address.Address
+	var dstSnet address.Subnet
+	var addrlen int
+	n := len(bs)
+	// 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 IPv6 header
+		if len(bs) < 40 {
+			return
+		}
+		// Check the packet size
+		if n-tun_IPv6_HEADER_LENGTH != 256*int(bs[4])+int(bs[5]) {
+			return
+		}
+		// IPv6 address
+		addrlen = 16
+		copy(dstAddr[:addrlen], bs[24:])
+		copy(dstSnet[:addrlen/2], bs[24:])
+	} else if bs[0]&0xf0 == 0x40 {
+		// Check if we have a fully-sized IPv4 header
+		if len(bs) < 20 {
+			return
+		}
+		// Check the packet size
+		if n != 256*int(bs[2])+int(bs[3]) {
+			return
+		}
+		// IPv4 address
+		addrlen = 4
+		copy(dstAddr[:addrlen], bs[16:])
+	} else {
+		// Unknown address length or protocol, so drop the packet and ignore it
+		tun.log.Traceln("Unknown packet type, dropping")
+		return
+	}
+	if tun.ckr.isEnabled() {
+		if addrlen != 16 || (!dstAddr.IsValid() && !dstSnet.IsValid()) {
+			if key, err := tun.ckr.getPublicKeyForAddress(dstAddr, addrlen); err == nil {
+				// A public key was found, get the node ID for the search
+				dstNodeID := crypto.GetNodeID(&key)
+				dstAddr = *address.AddrForNodeID(dstNodeID)
+				dstSnet = *address.SubnetForNodeID(dstNodeID)
+				addrlen = 16
+			}
+		}
+	}
+	if addrlen != 16 || (!dstAddr.IsValid() && !dstSnet.IsValid()) {
+		// Couldn't find this node's ygg IP
+		return
+	}
+	// Do we have an active connection for this node address?
+	var dstNodeID, dstNodeIDMask *crypto.NodeID
+	tun.mutex.RLock()
+	session, isIn := tun.addrToConn[dstAddr]
+	if !isIn || session == nil {
+		session, isIn = tun.subnetToConn[dstSnet]
+		if !isIn || session == nil {
+			// Neither an address nor a subnet mapping matched, therefore populate
+			// the node ID and mask to commence a search
+			if dstAddr.IsValid() {
+				dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask()
+			} else {
+				dstNodeID, dstNodeIDMask = dstSnet.GetNodeIDandMask()
+			}
+		}
+	}
+	tun.mutex.RUnlock()
+	// If we don't have a connection then we should open one
+	if !isIn || session == 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
+		go func() {
+			// FIXME just spitting out a goroutine to do this is kind of ugly and means we drop packets until the dial finishes
+			tun.mutex.Lock()
+			_, known := tun.dials[*dstNodeID]
+			tun.dials[*dstNodeID] = append(tun.dials[*dstNodeID], bs)
+			for len(tun.dials[*dstNodeID]) > 32 {
+				util.PutBytes(tun.dials[*dstNodeID][0])
+				tun.dials[*dstNodeID] = tun.dials[*dstNodeID][1:]
+			}
+			tun.mutex.Unlock()
+			if known {
+				return
+			}
+			var tc *tunConn
+			if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
+				// We've been given a connection so prepare the session wrapper
+				if tc, err = tun.wrap(conn); err != nil {
+					// Something went wrong when storing the connection, typically that
+					// something already exists for this address or subnet
+					tun.log.Debugln("TUN/TAP iface wrap:", err)
+				}
+			}
+			tun.mutex.Lock()
+			packets := tun.dials[*dstNodeID]
+			delete(tun.dials, *dstNodeID)
+			tun.mutex.Unlock()
+			if tc != nil {
+				for _, packet := range packets {
+					p := packet // Possibly required because of how range
+					<-tc.SyncExec(func() { tc._write(p) })
+				}
+			}
+		}()
+		// While the dial is going on we can't do much else
+		return
+	}
+	// If we have a connection now, try writing to it
+	if isIn && session != nil {
+		session.RecvFrom(tun, func() { session._write(bs) })
 	}
 }

src/tuntap/tun.go

@@ -15,6 +15,7 @@ import (
 	"net"
 	"sync"
 
+	"github.com/Arceliar/phony"
 	"github.com/gologme/log"
 	"github.com/yggdrasil-network/water"
 
@@ -33,6 +34,8 @@ const tun_ETHER_HEADER_LENGTH = 14
 // you should pass this object to the yggdrasil.SetRouterAdapter() function
 // before calling yggdrasil.Start().
 type TunAdapter struct {
+	writer       tunWriter
+	reader       tunReader
 	config       *config.NodeState
 	log          *log.Logger
 	reconfigure  chan chan error
@@ -44,7 +47,7 @@ type TunAdapter struct {
 	icmpv6       ICMPv6
 	mtu          int
 	iface        *water.Interface
-	send         chan []byte
+	phony.Inbox               // Currently only used for _handlePacket from the reader, TODO: all the stuff that currently needs a mutex below
 	mutex        sync.RWMutex // Protects the below
 	addrToConn   map[address.Address]*tunConn
 	subnetToConn map[address.Subnet]*tunConn
@@ -114,6 +117,8 @@ func (tun *TunAdapter) Init(config *config.NodeState, log *log.Logger, listener
 	tun.addrToConn = make(map[address.Address]*tunConn)
 	tun.subnetToConn = make(map[address.Subnet]*tunConn)
 	tun.dials = make(map[crypto.NodeID][][]byte)
+	tun.writer.tun = tun
+	tun.reader.tun = tun
 }
 
 // Start the setup process for the TUN/TAP adapter. If successful, starts the
@@ -147,7 +152,6 @@ func (tun *TunAdapter) Start() error {
 	}
 	tun.mutex.Lock()
 	tun.isOpen = true
-	tun.send = make(chan []byte, 32) // TODO: is this a sensible value?
 	tun.reconfigure = make(chan chan error)
 	tun.mutex.Unlock()
 	go func() {
@@ -157,8 +161,7 @@ func (tun *TunAdapter) Start() error {
 		}
 	}()
 	go tun.handler()
-	go tun.reader()
+	tun.reader.RecvFrom(nil, tun.reader._read) // Start the reader
-	go tun.writer()
 	tun.icmpv6.Init(tun)
 	if iftapmode {
 		go tun.icmpv6.Solicit(tun.addr)