feat(backend, udp): add UDP configuration and enhance UDP server functionality

Author: JavierRibaldelRio

backend/cmd/config.toml

@@ -34,6 +34,12 @@ max_retries = 0              # Maximum retries before cycling (0 = infinite retr
 connection_timeout_ms = 1000 # Connection timeout in milliseconds
 keep_alive_ms = 1000         # Keep-alive interval in milliseconds
 
+# UDP Configuration
+# These settings control the UDP server's buffer sizes and performance characteristics
+[udp]
+ring_buffer_size = 64      # Size of the ring buffer for incoming packets (number of packets, not bytes)
+packet_chan_size = 16      # Size of the channel buffer
+
 # TFTP Configuration
 [tftp]
 block_size = 131072    # TFTP block size in bytes (128kB)

backend/cmd/dev-config.toml

@@ -21,7 +21,6 @@ boards = ["HVSCU", "HVSCU-Cabinet", "PCU", "LCU", "BCU", "BMSL"]
 branch = "software" # Leave blank when using ADJ as a submodule (like this: "")
 validate = true # Execute ADJ validator before starting backend
 
-
 # Transport Configuration
 [transport]
 propagate_fault = false
@@ -36,6 +35,12 @@ max_retries = 0                # Maximum retries before cycling (0 = infinite re
 connection_timeout_ms = 999999 # Timeout for the initial connection attempt
 keep_alive_ms = 0              # Keep-alive interval in milliseconds (0 to disable)
 
+# UDP Configuration
+# These settings control the UDP server's buffer sizes and performance characteristics
+[udp]
+ring_buffer_size = 64      # Size of the ring buffer for incoming packets (number of packets, not bytes)
+packet_chan_size = 16      # Size of the channel buffer
+
 # Server Configuration
 [server.ethernet-view]
 address = "127.0.0.1:4040"

backend/cmd/setup_transport.go

@@ -50,7 +50,7 @@ func configureTransport(
 	configureTCPServerTransport(adj, transp)
 
 	// Start handling network packets using UDP server
-	configureUDPServerTransport(adj, transp)
+	configureUDPServerTransport(adj, transp, config)
 
 }
 
@@ -133,9 +133,11 @@ func configureTCPServerTransport(
 func configureUDPServerTransport(
 	adj adj_module.ADJ,
 	transp *transport.Transport,
+	config config.Config,
+
 ) {
 	trace.Info().Msg("Starting UDP server")
-	udpServer := udp.NewServer(adj.Info.Addresses[BACKEND], adj.Info.Ports[UDP], &trace.Logger)
+	udpServer := udp.NewServer(adj.Info.Addresses[BACKEND], adj.Info.Ports[UDP], &trace.Logger, config.UDP.RingBufferSize, config.UDP.PacketChanSize)
 	err := udpServer.Start()
 	if err != nil {
 		trace.Fatal().Err(err).Msg("failed to start UDP server: " + err.Error())

backend/internal/config/config_types.go

@@ -36,6 +36,11 @@ type TCP struct {
 	KeepAlive         int     `toml:"keep_alive_ms"`
 }
 
+type UDP struct {
+	RingBufferSize int `toml:"ring_buffer_size"`
+	PacketChanSize int `toml:"packet_chan_size"`
+}
+
 type Logging struct {
 	TimeUnit    logger.TimeUnit `toml:"time_unit"`
 	LoggingPath string          `toml:"logging_path"`
@@ -49,5 +54,6 @@ type Config struct {
 	Transport Transport
 	TFTP      TFTP
 	TCP       TCP
+	UDP       UDP
 	Logging   Logging
 }

backend/pkg/transport/network/udp/server.go

@@ -3,6 +3,7 @@ package udp
 import (
 	"fmt"
 	"net"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -25,17 +26,32 @@ type Server struct {
 	packetsCh chan Packet
 	errorsCh  chan error
 	stopCh    chan struct{}
+	stopped   bool
+
+	ring      []Packet
+	head      int
+	tail      int
+	count     int
+	ringMutex sync.Mutex
+	notEmpty  *sync.Cond
 }
 
-func NewServer(address string, port uint16, logger *zerolog.Logger) *Server {
-	return &Server{
+func NewServer(address string, port uint16, logger *zerolog.Logger, ringBufferSize int, packetChanSize int) *Server {
+	s := &Server{
 		address:   address,
 		port:      port,
 		logger:    logger,
-		packetsCh: make(chan Packet, 1000),
+		packetsCh: make(chan Packet, packetChanSize),
 		errorsCh:  make(chan error, 100),
 		stopCh:    make(chan struct{}),
 	}
+
+	s.ring = make([]Packet, ringBufferSize)
+	s.head = 0
+	s.tail = 0
+	s.count = 0
+	s.notEmpty = sync.NewCond(&s.ringMutex)
+	return s
 }
 
 func (s *Server) Start() error {
@@ -56,6 +72,7 @@ func (s *Server) Start() error {
 		Msg("UDP server started")
 
 	go s.readLoop()
+	go s.dispatchLoop()
 	return nil
 }
 
@@ -69,7 +86,7 @@ func (s *Server) readLoop() {
 		default:
 			// Set read deadline to allow periodic checking of stop channel
 			s.conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
-			
+
 			n, addr, err := s.conn.ReadFromUDP(buffer)
 			if err != nil {
 				// Check if it's a timeout error (expected)
@@ -104,11 +121,8 @@ func (s *Server) readLoop() {
 				Int("size", len(payload)).
 				Msg("received UDP packet")
 
-			select {
-			case s.packetsCh <- packet:
-			default:
-				s.logger.Warn().Msg("packet channel full, dropping packet")
-			}
+			// Push packet to ring buffer
+			s.push(packet)
 		}
 	}
 }
@@ -122,9 +136,73 @@ func (s *Server) GetErrors() <-chan error {
 }
 
 func (s *Server) Stop() error {
+	s.ringMutex.Lock()
+	s.stopped = true
 	close(s.stopCh)
+	s.notEmpty.Broadcast() // despertar a los que esperan
+	s.ringMutex.Unlock()
+
 	if s.conn != nil {
 		return s.conn.Close()
 	}
 	return nil
-}
+}
+
+func (s *Server) push(p Packet) {
+
+	s.ringMutex.Lock()
+	defer s.ringMutex.Unlock()
+
+	if s.count == len(s.ring) {
+		s.logger.Warn().Msg("Ring buffer full, overwriting oldest UDP packet")
+		s.head = (s.head + 1) % len(s.ring)
+		s.count--
+	}
+
+	s.ring[s.tail] = p
+	s.tail = (s.tail + 1) % len(s.ring)
+	s.count++
+
+	s.notEmpty.Signal()
+}
+
+func (s *Server) pop() (Packet, bool) {
+
+	s.ringMutex.Lock()
+	defer s.ringMutex.Unlock()
+
+	for s.count == 0 && !s.stopped {
+		s.notEmpty.Wait()
+	}
+
+	if s.count == 0 && s.stopped {
+		return Packet{}, false
+	}
+
+	p := s.ring[s.head]
+	s.head = (s.head + 1) % len(s.ring)
+	s.count--
+
+	return p, true
+}
+
+func (s *Server) dispatchLoop() {
+	for {
+		select {
+		case <-s.stopCh:
+			return
+		default:
+		}
+
+		packet, ok := s.pop()
+		if !ok {
+			return
+		}
+
+		select {
+		case s.packetsCh <- packet:
+		case <-s.stopCh:
+			return
+		}
+	}
+}

backend/pkg/transport/transport_test.go

@@ -943,45 +943,6 @@ func TestHandleServer_AcceptsAndDispatches(t *testing.T) {
 	}
 }
 
-func TestHandleUDPServer_Dispatches(t *testing.T) {
-	tr, api := createTestTransport(t)
-	tr.SetpropagateFault(false)
-
-	port := getAvailableUDPPort(t)
-	logger := zerolog.Nop()
-	server := udp.NewServer("127.0.0.1", port, &logger)
-	if err := server.Start(); err != nil {
-		t.Fatalf("failed to start UDP server: %v", err)
-	}
-	defer server.Stop()
-
-	go tr.HandleUDPServer(server)
-
-	packet := data.NewPacket(100)
-	packet.SetTimestamp(time.Unix(0, 0))
-	buf, err := tr.encoder.Encode(packet)
-	if err != nil {
-		t.Fatalf("encode failed: %v", err)
-	}
-	defer tr.encoder.ReleaseBuffer(buf)
-
-	conn, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: int(port)})
-	if err != nil {
-		t.Fatalf("failed to dial UDP server: %v", err)
-	}
-	defer conn.Close()
-
-	if _, err := conn.Write(buf.Bytes()); err != nil {
-		t.Fatalf("failed to send UDP packet: %v", err)
-	}
-
-	if err := waitForCondition(func() bool {
-		return len(api.GetNotifications()) > 0
-	}, 2*time.Second, "Should receive notification from UDP server"); err != nil {
-		t.Fatal(err)
-	}
-}
-
 func TestHandleConversation_DispatchesAndStopsOnError(t *testing.T) {
 	tr, api := createTestTransport(t)