magnetico/cmd/magneticod/dht/mainline/transport.go

package mainline

import (
	"net"

	"github.com/anacrolix/torrent/bencode"
	sockaddr "github.com/libp2p/go-sockaddr/net"
	"go.uber.org/zap"
	"golang.org/x/sys/unix"
)

type Transport struct {
	fd      int
	laddr   *net.UDPAddr
	started bool
	buffer  []byte

	// OnMessage is the function that will be called when Transport receives a packet that is
	// successfully unmarshalled as a syntactically correct Message (but -of course- the checking
	// the semantic correctness of the Message is left to Protocol).
	onMessage func(*Message, *net.UDPAddr)
	// OnCongestion
	onCongestion func()
}

func NewTransport(laddr string, onMessage func(*Message, *net.UDPAddr), onCongestion func()) *Transport {
	t := new(Transport)
	/*   The field size sets a theoretical limit of 65,535 bytes (8 byte header + 65,527 bytes of
	 * data) for a UDP datagram. However the actual limit for the data length, which is imposed by
	 * the underlying IPv4 protocol, is 65,507 bytes (65,535 − 8 byte UDP header − 20 byte IP
	 * header).
	 *
	 *   In IPv6 jumbograms it is possible to have UDP packets of size greater than 65,535 bytes.
	 * RFC 2675 specifies that the length field is set to zero if the length of the UDP header plus
	 * UDP data is greater than 65,535.
	 *
	 * https://en.wikipedia.org/wiki/User_Datagram_Protocol
	 */
	t.buffer       = make([]byte, 65507)
	t.onMessage    = onMessage
	t.onCongestion = onCongestion

	var err error
	t.laddr, err = net.ResolveUDPAddr("udp", laddr)
	if err != nil {
		zap.L().Panic("Could not resolve the UDP address for the trawler!", zap.Error(err))
	}
	if t.laddr.IP.To4() == nil {
		zap.L().Panic("IP address is not IPv4!")
	}

	return t
}

func (t *Transport) Start() {
	// Why check whether the Transport `t` started or not, here and not -for instance- in
	// t.Terminate()?
	// Because in t.Terminate() the programmer (i.e. you & me) would stumble upon an error while
	// trying close an uninitialised net.UDPConn or something like that: it's mostly harmless
	// because its effects are immediate. But if you try to start a Transport `t` for the second
	// (or the third, 4th, ...) time, it will keep spawning goroutines and any small mistake may
	// end up in a debugging horror.
	//                                                                   Here ends my justification.
	if t.started {
		zap.L().Panic("Attempting to Start() a mainline/Transport that has been already started! (Programmer error.)")
	}
	t.started = true

	var err error
	t.fd, err = unix.Socket(unix.SOCK_DGRAM, unix.AF_INET, 0)
	if err != nil {
		zap.L().Fatal("Could NOT create a UDP socket!", zap.Error(err))
	}

	var ip [4]byte
	copy(ip[:], t.laddr.IP.To4())
	err = unix.Bind(t.fd, &unix.SockaddrInet4{Addr: ip, Port: t.laddr.Port})
	if err != nil {
		zap.L().Fatal("Could NOT bind the socket!", zap.Error(err))
	}

	go t.readMessages()
}

func (t *Transport) Terminate() {
	unix.Close(t.fd);
}

// readMessages is a goroutine!
func (t *Transport) readMessages() {
	for {
		n, fromSA, err := unix.Recvfrom(t.fd, t.buffer, 0)
		if err == unix.EPERM || err == unix.ENOBUFS { // todo: are these errors possible for recvfrom?
			zap.L().Warn("READ CONGESTION!", zap.Error(err))
			t.onCongestion()
		} else if err != nil {
			// TODO: isn't there a more reliable way to detect if UDPConn is closed?
			zap.L().Warn("Could NOT read an UDP packet!", zap.Error(err))
		}

		if n == 0 {
			/*   Datagram sockets in various domains  (e.g., the UNIX and Internet domains) permit
			 * zero-length datagrams. When such a datagram is received, the return value (n) is 0.
			 */
			continue
		}

		from := sockaddr.SockaddrToUDPAddr(fromSA)
		if from == nil {
			zap.L().Panic("dht mainline transport SockaddrToUDPAddr: nil")
		}

		var msg Message
		err = bencode.Unmarshal(t.buffer[:n], &msg)
		if err != nil {
			// couldn't unmarshal packet data
			continue
		}

		t.onMessage(&msg, from)
	}
}

func (t *Transport) WriteMessages(msg *Message, addr *net.UDPAddr) {
	data, err := bencode.Marshal(msg)
	if err != nil {
		zap.L().Panic("Could NOT marshal an outgoing message! (Programmer error.)")
	}

	addrSA := sockaddr.NetAddrToSockaddr(addr)
	if addrSA == nil {
		zap.L().Debug("Wrong net address for the remote peer!",
			zap.String("addr", addr.String()))
		return;
	}

	err = unix.Sendto(t.fd, data, 0, addrSA)
	if err == unix.EPERM || err == unix.ENOBUFS {
		/*   EPERM (errno: 1) is kernel's way of saying that "you are far too fast, chill". It is
		 * also likely that we have received a ICMP source quench packet (meaning, that we *really*
		 * need to slow down.
		 *
		 * Read more here: http://www.archivum.info/comp.protocols.tcp-ip/2009-05/00088/UDP-socket-amp-amp-sendto-amp-amp-EPERM.html
		 *
		 * >   Note On BSD systems (OS X, FreeBSD, etc.) flow control is not supported for
		 * > DatagramProtocol, because send failures caused by writing too many packets cannot be
		 * > detected easily. The socket always appears ‘ready’ and excess packets are dropped; an
		 * > OSError with errno set to errno.ENOBUFS may or may not be raised; if it is raised, it
		 * > will be reported to DatagramProtocol.error_received() but otherwise ignored.
		 *
		 * Source: https://docs.python.org/3/library/asyncio-protocol.html#flow-control-callbacks
		 */
		zap.L().Warn("WRITE CONGESTION!", zap.Error(err))
		t.onCongestion()
	} else if err != nil {
		zap.L().Warn("Could NOT write an UDP packet!", zap.Error(err))
	}
}