<commit_before>package dispatcher

import (
	"context"
	"net"
	"sync"
	"time"

	"github.com/scionproto/scion/go/godispatcher/internal/metrics"
	"github.com/scionproto/scion/go/godispatcher/internal/registration"
	"github.com/scionproto/scion/go/godispatcher/internal/respool"
	"github.com/scionproto/scion/go/lib/addr"
	"github.com/scionproto/scion/go/lib/common"
	"github.com/scionproto/scion/go/lib/hpkt"
	"github.com/scionproto/scion/go/lib/log"
	"github.com/scionproto/scion/go/lib/overlay/conn"
	"github.com/scionproto/scion/go/lib/ringbuf"
	"github.com/scionproto/scion/go/lib/scmp"
	"github.com/scionproto/scion/go/lib/serrors"
	"github.com/scionproto/scion/go/lib/spkt"
)

// OverflowLoggingInterval is the minimum amount of time that needs to
// pass before another overflow logging message is printed (if needed).
const OverflowLoggingInterval = 10 * time.Second

// ReceiveBufferSize is the size of receive buffers used by the dispatcher.
const ReceiveBufferSize = 1 << 20

// Server is the main object allowing to create new SCION connections.
type Server struct {
	// routingTable is used to register new connections.
	routingTable *IATable
	ipv4Conn     net.PacketConn
	ipv6Conn     net.PacketConn
}

// NewServer creates new instance of Server. Internally, it opens the dispatcher ports
// for both IPv4 and IPv6. Returns error if the ports can't be opened.
func NewServer(address string) (*Server, error) {
	metaLogger := &throttledMetaLogger{
		Logger:      log.Root(),
		MinInterval: OverflowLoggingInterval,
	}
	ipv4Conn, err := openConn("udp4", address, metaLogger)
	if err != nil {
		return nil, err
	}
	ipv6Conn, err := openConn("udp6", address, metaLogger)
	if err != nil {
		ipv4Conn.Close()
		return nil, err
	}

	return &Server{
		routingTable: NewIATable(1024, 65535),
		ipv4Conn:     ipv4Conn,
		ipv6Conn:     ipv6Conn,
	}, nil
}

// Serve starts reading packets from network and dispatching them to different connections.
// The function blocks and returns if there's an error or when Close has been called.
func (as *Server) Serve() error {
	errChan := make(chan error)
	go func() {
		defer log.LogPanicAndExit()
		netToRingDataplane := &NetToRingDataplane{
			OverlayConn:  as.ipv4Conn,
			RoutingTable: as.routingTable,
		}
		errChan <- netToRingDataplane.Run()
	}()
	go func() {
		defer log.LogPanicAndExit()
		netToRingDataplane := &NetToRingDataplane{
			OverlayConn:  as.ipv6Conn,
			RoutingTable: as.routingTable,
		}
		errChan <- netToRingDataplane.Run()
	}()
	return <-errChan
}

// Register creates a new connection.
func (as *Server) Register(ctx context.Context, ia addr.IA, address *net.UDPAddr,
	svc addr.HostSVC) (net.PacketConn, uint16, error) {

	tableEntry := newTableEntry()
	ref, err := as.routingTable.Register(ia, address, nil, svc, tableEntry)
	if err != nil {
		return nil, 0, err
	}
	conn := &Conn{
		ipv4Conn:     as.ipv4Conn,
		ipv6Conn:     as.ipv6Conn,
		ring:         tableEntry.appIngressRing,
		regReference: ref,
	}
	return conn, uint16(ref.UDPAddr().Port), nil
}

func (as *Server) Close() {
	as.ipv4Conn.Close()
	as.ipv6Conn.Close()
}

// Conn represents a connection bound to a specific SCION port/SVC.
type Conn struct {
	// ipv4Conn is used to send IPv4 packets.
	ipv4Conn net.PacketConn
	// ipv6Conn is used to send IPv6 packets.
	ipv6Conn net.PacketConn
	// ring is used to retrieve incoming packets.
	ring *ringbuf.Ring
	// regReference is the reference to the registration in the routing table.
	regReference registration.RegReference
}

func (ac *Conn) WriteTo(p []byte, addr net.Addr) (int, error) {
	var info spkt.ScnPkt
	hpkt.ParseScnPkt(&info, p)
	if err := registerIfSCMPRequest(ac.regReference, &info); err != nil {
		log.Warn("SCMP Request ID error, packet still sent", "err", err)
	}
	isIPv6 := addr.(*net.UDPAddr).IP.To4() == nil
	if isIPv6 {
		return ac.ipv6Conn.WriteTo(p, addr)
	} else {
		return ac.ipv4Conn.WriteTo(p, addr)
	}
}

// Write is optimized for the use by ConnHandler (avoids reparsing the packet).
func (ac *Conn) Write(pkt *respool.Packet) (int, error) {
	if err := registerIfSCMPRequest(ac.regReference, &pkt.Info); err != nil {
		log.Warn("SCMP Request ID error, packet still sent", "err", err)
	}
	isIPv6 := pkt.OverlayRemote.IP.To4() == nil
	if isIPv6 {
		return pkt.SendOnConn(ac.ipv6Conn, pkt.OverlayRemote)
	} else {
		return pkt.SendOnConn(ac.ipv4Conn, pkt.OverlayRemote)
	}
}

func (ac *Conn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
	pkt := ac.Read()
	if pkt == nil {
		return 0, nil, serrors.New("Connection closed")
	}
	n = pkt.CopyTo(p)
	addr = pkt.OverlayRemote
	pkt.Free()
	return
}

// Read is optimized for the use by ConnHandler (avoids one copy).
func (ac *Conn) Read() *respool.Packet {<commit_msg>Fix bug in dispatcher (#3637)<commit_after>package dispatcher

import (
	"context"
	"net"
	"sync"
	"time"

	"github.com/scionproto/scion/go/godispatcher/internal/metrics"
	"github.com/scionproto/scion/go/godispatcher/internal/registration"
	"github.com/scionproto/scion/go/godispatcher/internal/respool"
	"github.com/scionproto/scion/go/lib/addr"
	"github.com/scionproto/scion/go/lib/common"
	"github.com/scionproto/scion/go/lib/hpkt"
	"github.com/scionproto/scion/go/lib/log"
	"github.com/scionproto/scion/go/lib/overlay/conn"
	"github.com/scionproto/scion/go/lib/ringbuf"
	"github.com/scionproto/scion/go/lib/scmp"
	"github.com/scionproto/scion/go/lib/serrors"
	"github.com/scionproto/scion/go/lib/spkt"
)

// OverflowLoggingInterval is the minimum amount of time that needs to
// pass before another overflow logging message is printed (if needed).
const OverflowLoggingInterval = 10 * time.Second

// ReceiveBufferSize is the size of receive buffers used by the dispatcher.
const ReceiveBufferSize = 1 << 20

// Server is the main object allowing to create new SCION connections.
type Server struct {
	// routingTable is used to register new connections.
	routingTable *IATable
	ipv4Conn     net.PacketConn
	ipv6Conn     net.PacketConn
}

// NewServer creates new instance of Server. Internally, it opens the dispatcher ports
// for both IPv4 and IPv6. Returns error if the ports can't be opened.
func NewServer(address string) (*Server, error) {
	metaLogger := &throttledMetaLogger{
		Logger:      log.Root(),
		MinInterval: OverflowLoggingInterval,
	}
	ipv4Conn, err := openConn("udp4", address, metaLogger)
	if err != nil {
		return nil, err
	}
	ipv6Conn, err := openConn("udp6", address, metaLogger)
	if err != nil {
		ipv4Conn.Close()
		return nil, err
	}

	return &Server{
		routingTable: NewIATable(1024, 65535),
		ipv4Conn:     ipv4Conn,
		ipv6Conn:     ipv6Conn,
	}, nil
}

// Serve starts reading packets from network and dispatching them to different connections.
// The function blocks and returns if there's an error or when Close has been called.
func (as *Server) Serve() error {
	errChan := make(chan error)
	go func() {
		defer log.LogPanicAndExit()
		netToRingDataplane := &NetToRingDataplane{
			OverlayConn:  as.ipv4Conn,
			RoutingTable: as.routingTable,
		}
		errChan <- netToRingDataplane.Run()
	}()
	go func() {
		defer log.LogPanicAndExit()
		netToRingDataplane := &NetToRingDataplane{
			OverlayConn:  as.ipv6Conn,
			RoutingTable: as.routingTable,
		}
		errChan <- netToRingDataplane.Run()
	}()
	return <-errChan
}

// Register creates a new connection.
func (as *Server) Register(ctx context.Context, ia addr.IA, address *net.UDPAddr,
	svc addr.HostSVC) (net.PacketConn, uint16, error) {

	tableEntry := newTableEntry()
	ref, err := as.routingTable.Register(ia, address, nil, svc, tableEntry)
	if err != nil {
		return nil, 0, err
	}
	var ovConn net.PacketConn
	if address.IP.To4() == nil {
		ovConn = as.ipv6Conn
	} else {
		ovConn = as.ipv4Conn
	}
	conn := &Conn{
		conn:         ovConn,
		ring:         tableEntry.appIngressRing,
		regReference: ref,
	}
	return conn, uint16(ref.UDPAddr().Port), nil
}

func (as *Server) Close() {
	as.ipv4Conn.Close()
	as.ipv6Conn.Close()
}

// Conn represents a connection bound to a specific SCION port/SVC.
type Conn struct {
	// conn is used to send packets.
	conn net.PacketConn
	// ring is used to retrieve incoming packets.
	ring *ringbuf.Ring
	// regReference is the reference to the registration in the routing table.
	regReference registration.RegReference
}

func (ac *Conn) WriteTo(p []byte, addr net.Addr) (int, error) {
	var info spkt.ScnPkt
	hpkt.ParseScnPkt(&info, p)
	if err := registerIfSCMPRequest(ac.regReference, &info); err != nil {
		log.Warn("SCMP Request ID error, packet still sent", "err", err)
	}
	return ac.conn.WriteTo(p, addr)
}

// Write is optimized for the use by ConnHandler (avoids reparsing the packet).
func (ac *Conn) Write(pkt *respool.Packet) (int, error) {
	if err := registerIfSCMPRequest(ac.regReference, &pkt.Info); err != nil {
		log.Warn("SCMP Request ID error, packet still sent", "err", err)
	}
	return pkt.SendOnConn(ac.conn, pkt.OverlayRemote)
}

func (ac *Conn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
	pkt := ac.Read()
	if pkt == nil {
		return 0, nil, serrors.New("Connection closed")
	}
	n = pkt.CopyTo(p)
	addr = pkt.OverlayRemote
	pkt.Free()
	return
}

// Read is optimized for the use by ConnHandler (avoids one copy).
func (ac *Conn) Read() *respool.Packet {
{'fraction_non_alphanumeric': 0.12119571103649951, 'fraction_numerical': 0.010939023069424889, 'mean_word_length': 4.004878048780488, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3299508', 'n_tokens_mistral': 3418, 'n_tokens_neox': 3309, 'n_words': 891}