Add multi-threaded UDP server example

Add a demo to show how to use multiple sockets with the SO_REUSEPORT
socket option to scale TURN/UDP servers to multiple CPUs.

examples/README.md

@@ -31,6 +31,9 @@ You could also intercept these reads/writes if you want to filter traffic going
 #### simple
 This example is the most minimal invocation of a Pion TURN instance possible. It has no custom behavior, and could be a good starting place for running your own TURN server.
 
+#### simple-multithreaded
+A multithreaded version of the `simple` Pion TURN server, demonstrating how to scale a Pion UDP TURN server to multiple CPU cores. By default, Pion TURN servers use a single UDP socket that is shared across all clients, which limits Pion UDP/TURN servers to a single CPU thread. This example passes a configurable number of UDP sockets to the TURN server, which share the same local `address:port` pair using the `SO_REUSEPORT` socket option. This then lets the server to create a separate readloop to drain each socket. The OS kernel will distribute packets received on the `address:port` pair across the sockets by the IP 5-tuple, which makes sure that all packets of a TURN allocation will be correctly processed in a single readloop.
+
 #### tcp
 This example demonstrates listening on TCP. You could combine this example with `simple` and you will have a Pion TURN instance that is available via TCP and UDP.
 

examples/turn-server/simple-multithreaded/main.go

@@ -0,0 +1,111 @@
+//go:build !wasm
+
+// Package main implements a multi-threaded TURN server
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"net"
+	"os"
+	"os/signal"
+	"regexp"
+	"strconv"
+	"syscall"
+
+	"github.com/pion/turn/v2"
+	"golang.org/x/sys/unix"
+)
+
+func main() {
+	publicIP := flag.String("public-ip", "", "IP Address that TURN can be contacted by.")
+	port := flag.Int("port", 3478, "Listening port.")
+	users := flag.String("users", "", "List of username and password (e.g. \"user=pass,user=pass\")")
+	realm := flag.String("realm", "pion.ly", "Realm (defaults to \"pion.ly\")")
+	threadNum := flag.Int("thread-num", 1, "Number of server threads (defaults to 1)")
+	flag.Parse()
+
+	if len(*publicIP) == 0 {
+		log.Fatalf("'public-ip' is required")
+	} else if len(*users) == 0 {
+		log.Fatalf("'users' is required")
+	}
+
+	addr, err := net.ResolveUDPAddr("udp", "0.0.0.0:"+strconv.Itoa(*port))
+	if err != nil {
+		log.Fatalf("Failed to parse server address: %s", err)
+	}
+
+	// Cache -users flag for easy lookup later
+	// If passwords are stored they should be saved to your DB hashed using turn.GenerateAuthKey
+	usersMap := map[string][]byte{}
+	for _, kv := range regexp.MustCompile(`(\w+)=(\w+)`).FindAllStringSubmatch(*users, -1) {
+		usersMap[kv[1]] = turn.GenerateAuthKey(kv[1], *realm, kv[2])
+	}
+
+	// Create `numThreads` UDP listeners to pass into pion/turn
+	// pion/turn itself doesn't allocate any UDP sockets, but lets the user pass them in
+	// this allows us to add logging, storage or modify inbound/outbound traffic
+	// UDP listeners share the same local address:port with setting SO_REUSEPORT and the kernel
+	// will load-balance received packets per the IP 5-tuple
+	listenerConfig := &net.ListenConfig{
+		Control: func(network, address string, conn syscall.RawConn) error {
+			var operr error
+			if err = conn.Control(func(fd uintptr) {
+				operr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+			}); err != nil {
+				return err
+			}
+
+			return operr
+		},
+	}
+
+	relayAddressGenerator := &turn.RelayAddressGeneratorStatic{
+		RelayAddress: net.ParseIP(*publicIP), // Claim that we are listening on IP passed by user
+		Address:      "0.0.0.0",              // But actually be listening on every interface
+	}
+
+	packetConnConfigs := make([]turn.PacketConnConfig, *threadNum)
+	for i := 0; i < *threadNum; i++ {
+		conn, listErr := listenerConfig.ListenPacket(context.Background(), addr.Network(), addr.String())
+		if listErr != nil {
+			log.Fatalf("Failed to allocate UDP listener at %s:%s", addr.Network(), addr.String())
+		}
+
+		packetConnConfigs[i] = turn.PacketConnConfig{
+			PacketConn:            conn,
+			RelayAddressGenerator: relayAddressGenerator,
+		}
+
+		log.Printf("Server %d listening on %s\n", i, conn.LocalAddr().String())
+	}
+
+	s, err := turn.NewServer(turn.ServerConfig{
+		Realm: *realm,
+		// Set AuthHandler callback
+		// This is called every time a user tries to authenticate with the TURN server
+		// Return the key for that user, or false when no user is found
+		AuthHandler: func(username string, realm string, srcAddr net.Addr) ([]byte, bool) {
+			if key, ok := usersMap[username]; ok {
+				return key, true
+			}
+			return nil, false
+		},
+		// PacketConnConfigs is a list of UDP Listeners and the configuration around them
+		PacketConnConfigs: packetConnConfigs,
+	})
+	if err != nil {
+		log.Panicf("Failed to create TURN server: %s", err)
+	}
+
+	// Block until user sends SIGINT or SIGTERM
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	<-sigs
+
+	if err = s.Close(); err != nil {
+		log.Panicf("Failed to close TURN server: %s", err)
+	}
+}

go.mod

@@ -8,4 +8,5 @@ require (
 	github.com/pion/stun v0.4.0
 	github.com/pion/transport/v2 v2.0.2
 	github.com/stretchr/testify v1.8.1
+	golang.org/x/sys v0.5.0
 )