Function udp_acceptor

src/main.rs:418–507

(
    globals: Arc<Globals>,
    net_udp_socket: std::net::UdpSocket,
)

#[allow(unreachable_code)]
async fn udp_acceptor(
    globals: Arc<Globals>,
    net_udp_socket: std::net::UdpSocket,
) -> Result<(), Error> {
    let runtime_handle = globals.runtime_handle.clone();
    let tokio_udp_socket = UdpSocket::try_from(net_udp_socket.try_clone()?)?;
    let timeout = globals.udp_timeout;
    let concurrent_connections = globals.udp_concurrent_connections.clone();
    let active_connections = globals.udp_active_connections.clone();
    loop {
        let mut packet = vec![0u8; DNSCRYPT_UDP_QUERY_MAX_SIZE];
        let (packet_len, client_addr) = tokio_udp_socket.recv_from(&mut packet).await?;
        if packet_len < DNS_HEADER_SIZE {
            continue;
        }

        if let Some(ref rate_limiter) = globals.rate_limiter {
            if !rate_limiter.is_allowed(client_addr.ip()) {
                debug!("Rate limit exceeded for {}", client_addr.ip());
                #[cfg(feature = "metrics")]
                globals.varz.client_queries_rate_limited.inc();
                continue;
            }
        }

        packet.truncate(packet_len);

        let active_count = concurrent_connections.load(Ordering::Relaxed);
        if active_count >= globals.udp_max_active_connections {
            debug!("UDP connection limit reached, dropping packet");
            continue;
        }

        // Clone the socket for this request
        let net_udp_socket = match net_udp_socket.try_clone() {
            Ok(socket) => socket,
            Err(e) => {
                error!("Failed to clone UDP socket: {}", e);
                // Add a small delay to avoid spinning on socket errors
                tokio::time::sleep(Duration::from_millis(100)).await;
                continue;
            }
        };

        let client_ctx = ClientCtx::Udp(UdpClientCtx {
            net_udp_socket,
            client_addr,
        });
        let (tx, rx) = oneshot::channel::<()>();
        let tx_channel_index = {
            let mut active_connections = active_connections.lock();
            if active_connections.is_full() {
                let tx_oldest = active_connections.pop_back().unwrap();
                let _ = tx_oldest.send(());
            }
            active_connections.push_front(tx)?
        };
        let _count = concurrent_connections.fetch_add(1, Ordering::Relaxed);