Pull request: all: upd golibs

Squashed commit of the following:

commit 9b7e21eaf062a569a259d2279f785542e6fadb37
Author: Dimitry Kolyshev <[email protected]>
Date:   Tue Dec 12 10:05:31 2023 +0200

    proxy: imp code

commit d5b40a4a88c8fbd5edf1c207a6f7e3d0fad16d0e
Author: Dimitry Kolyshev <[email protected]>
Date:   Mon Dec 11 12:02:26 2023 +0200

    all: upd golibs

README.md

@@ -90,7 +90,7 @@ Application Options:
       --ipv6-disabled              If specified, all AAAA requests will be replied with NoError RCode and empty answer
       --bogus-nxdomain=            Transform the responses containing at least a single IP that matches specified addresses and CIDRs into NXDOMAIN.  Can be specified multiple times.
       --udp-buf-size=              Set the size of the UDP buffer in bytes. A value <= 0 will use the system default.
-      --max-go-routines=           Set the maximum number of go routines. A value <= 0 will not not set a maximum.
+      --max-go-routines=           Set the maximum number of go routines. A zero value will not not set a maximum.
       --pprof                      If present, exposes pprof information on localhost:6060.
       --version                    Prints the program version
 

go.mod

@@ -3,7 +3,7 @@ module github.com/AdguardTeam/dnsproxy
 go 1.20
 
 require (
-	github.com/AdguardTeam/golibs v0.18.0
+	github.com/AdguardTeam/golibs v0.18.1
 	github.com/ameshkov/dnscrypt/v2 v2.2.7
 	github.com/ameshkov/dnsstamps v1.0.3
 	github.com/beefsack/go-rate v0.0.0-20220214233405-116f4ca011a0

go.sum

@@ -1,5 +1,5 @@
-github.com/AdguardTeam/golibs v0.18.0 h1:ckS2YK7t2Ub6UkXl0fnreVaM15Zb07Hh1gmFqttjpWg=
-github.com/AdguardTeam/golibs v0.18.0/go.mod h1:DKhCIXHcUYtBhU8ibTLKh1paUL96n5zhQBlx763sj+U=
+github.com/AdguardTeam/golibs v0.18.1 h1:6u0fvrIj2qjUsRdbIGJ9AR0g5QRSWdKIo/DYl3tp5aM=
+github.com/AdguardTeam/golibs v0.18.1/go.mod h1:DKhCIXHcUYtBhU8ibTLKh1paUL96n5zhQBlx763sj+U=
 github.com/aead/chacha20 v0.0.0-20180709150244-8b13a72661da h1:KjTM2ks9d14ZYCvmHS9iAKVt9AyzRSqNU1qabPih5BY=
 github.com/aead/chacha20 v0.0.0-20180709150244-8b13a72661da/go.mod h1:eHEWzANqSiWQsof+nXEI9bUVUyV6F53Fp89EuCh2EAA=
 github.com/aead/poly1305 v0.0.0-20180717145839-3fee0db0b635 h1:52m0LGchQBBVqJRyYYufQuIbVqRawmubW3OFGqK1ekw=

main.go

@@ -196,8 +196,8 @@ type Options struct {
 	// UDP buffer size value
 	UDPBufferSize int `yaml:"udp-buf-size" long:"udp-buf-size" description:"Set the size of the UDP buffer in bytes. A value <= 0 will use the system default."`
 
-	// The maximum number of go routines
-	MaxGoRoutines int `yaml:"max-go-routines" long:"max-go-routines" description:"Set the maximum number of go routines. A value <= 0 will not not set a maximum."`
+	// MaxGoRoutines is the maximum number of goroutines.
+	MaxGoRoutines uint `yaml:"max-go-routines" long:"max-go-routines" description:"Set the maximum number of go routines. A zero value will not not set a maximum."`
 
 	// Pprof defines whether the pprof information needs to be exposed via
 	// localhost:6060 or not.

proxy/config.go

@@ -182,7 +182,7 @@ type Config struct {
 	// TODO(a.garipov): Rename this to something like
 	// “MaxDNSRequestGoroutines” in a later major version, as it doesn't
 	// actually limit all goroutines.
-	MaxGoroutines int
+	MaxGoroutines uint
 
 	// The size of the read buffer on the underlying socket. Larger read buffers can handle
 	// larger bursts of requests before packets get dropped.

proxy/proxy.go

@@ -19,6 +19,7 @@ import (
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
 	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/syncutil"
 	"github.com/ameshkov/dnscrypt/v2"
 	"github.com/miekg/dns"
 	gocache "github.com/patrickmn/go-cache"
@@ -166,7 +167,7 @@ type Proxy struct {
 	// RWMutex protects the whole proxy.
 	sync.RWMutex
 
-	// requestGoroutinesSema limits the number of simultaneous requests.
+	// requestsSema limits the number of simultaneous requests.
 	//
 	// TODO(a.garipov): Currently we have to pass this exact semaphore to
 	// the workers, to prevent races on restart.  In the future we will need
@@ -174,7 +175,7 @@ type Proxy struct {
 	// states.
 	//
 	// See also: https://github.com/AdguardTeam/AdGuardHome/issues/2242.
-	requestGoroutinesSema semaphore
+	requestsSema syncutil.Semaphore
 
 	// Config is the proxy configuration.
 	//
@@ -195,12 +196,9 @@ func (p *Proxy) Init() (err error) {
 	if p.MaxGoroutines > 0 {
 		log.Info("dnsproxy: max goroutines is set to %d", p.MaxGoroutines)
 
-		p.requestGoroutinesSema, err = newChanSemaphore(p.MaxGoroutines)
-		if err != nil {
-			return fmt.Errorf("can't init semaphore: %w", err)
-		}
+		p.requestsSema = syncutil.NewChanSemaphore(p.MaxGoroutines)
 	} else {
-		p.requestGoroutinesSema = newNoopSemaphore()
+		p.requestsSema = syncutil.EmptySemaphore{}
 	}
 
 	p.udpOOBSize = proxynetutil.UDPGetOOBSize()

proxy/sema.go

@@ -1,58 +0,0 @@
-package proxy
-
-import (
-	"fmt"
-)
-
-// semaphore is the semaphore interface.  acquire will block until the
-// resource can be acquired.  release never blocks.
-type semaphore interface {
-	acquire()
-	release()
-}
-
-// noopSemaphore is a semaphore that has no limit.
-type noopSemaphore struct{}
-
-// acquire implements the semaphore interface for noopSemaphore.
-func (noopSemaphore) acquire() {}
-
-// release implements the semaphore interface for noopSemaphore.
-func (noopSemaphore) release() {}
-
-// newNoopSemaphore returns a new noopSemaphore.
-func newNoopSemaphore() (s semaphore) { return noopSemaphore{} }
-
-// sig is an alias for struct{} to type less.
-type sig = struct{}
-
-// chanSemaphore is a channel-based semaphore.
-type chanSemaphore struct {
-	c chan sig
-}
-
-// acquire implements the semaphore interface for *chanSemaphore.
-func (c *chanSemaphore) acquire() {
-	c.c <- sig{}
-}
-
-// release implements the semaphore interface for *chanSemaphore.
-func (c *chanSemaphore) release() {
-	select {
-	case <-c.c:
-	default:
-	}
-}
-
-// newChanSemaphore returns a new chanSemaphore with the provided
-// maximum resource number.  maxRes must be greater than zero.
-func newChanSemaphore(maxRes int) (s semaphore, err error) {
-	if maxRes < 1 {
-		return nil, fmt.Errorf("bad maxRes: %d", maxRes)
-	}
-
-	s = &chanSemaphore{
-		c: make(chan sig, maxRes),
-	}
-	return s, nil
-}

proxy/server.go

@@ -44,15 +44,15 @@ func (p *Proxy) startListeners(ctx context.Context) error {
 	}
 
 	for _, l := range p.udpListen {
-		go p.udpPacketLoop(l, p.requestGoroutinesSema)
+		go p.udpPacketLoop(l, p.requestsSema)
 	}
 
 	for _, l := range p.tcpListen {
-		go p.tcpPacketLoop(l, ProtoTCP, p.requestGoroutinesSema)
+		go p.tcpPacketLoop(l, ProtoTCP, p.requestsSema)
 	}
 
 	for _, l := range p.tlsListen {
-		go p.tcpPacketLoop(l, ProtoTLS, p.requestGoroutinesSema)
+		go p.tcpPacketLoop(l, ProtoTLS, p.requestsSema)
 	}
 
 	for _, l := range p.httpsListen {
@@ -64,7 +64,7 @@ func (p *Proxy) startListeners(ctx context.Context) error {
 	}
 
 	for _, l := range p.quicListen {
-		go p.quicPacketLoop(l, p.requestGoroutinesSema)
+		go p.quicPacketLoop(l, p.requestsSema)
 	}
 
 	for _, l := range p.dnsCryptUDPListen {

proxy/server_dnscrypt.go

@@ -1,12 +1,14 @@
 package proxy
 
 import (
+	"context"
 	"fmt"
 	"net"
 
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
 	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/syncutil"
 	"github.com/ameshkov/dnscrypt/v2"
 	"github.com/miekg/dns"
 )
@@ -28,7 +30,7 @@ func (p *Proxy) createDNSCryptListeners() (err error) {
 		Handler: &dnsCryptHandler{
 			proxy: p,
 
-			requestGoroutinesSema: p.requestGoroutinesSema,
+			reqSema: p.requestsSema,
 		},
 	}
 
@@ -61,20 +63,24 @@ func (p *Proxy) createDNSCryptListeners() (err error) {
 type dnsCryptHandler struct {
 	proxy *Proxy
 
-	requestGoroutinesSema semaphore
+	reqSema syncutil.Semaphore
 }
 
 // compile-time type check
 var _ dnscrypt.Handler = &dnsCryptHandler{}
 
 // ServeDNS - processes the DNS query
-func (h *dnsCryptHandler) ServeDNS(rw dnscrypt.ResponseWriter, req *dns.Msg) error {
+func (h *dnsCryptHandler) ServeDNS(rw dnscrypt.ResponseWriter, req *dns.Msg) (err error) {
 	d := h.proxy.newDNSContext(ProtoDNSCrypt, req)
 	d.Addr = netutil.NetAddrToAddrPort(rw.RemoteAddr())
 	d.DNSCryptResponseWriter = rw
 
-	h.requestGoroutinesSema.acquire()
-	defer h.requestGoroutinesSema.release()
+	// TODO(d.kolyshev): Pass and use context from above.
+	err = h.reqSema.Acquire(context.Background())
+	if err != nil {
+		return fmt.Errorf("dnsproxy: dnscrypt: acquiring semaphore: %w", err)
+	}
+	defer h.reqSema.Release()
 
 	return h.proxy.handleDNSRequest(d)
 }

proxy/server_quic.go

@@ -13,6 +13,7 @@ import (
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
 	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/syncutil"
 	"github.com/bluele/gcache"
 	"github.com/miekg/dns"
 	"github.com/quic-go/quic-go"
@@ -82,11 +83,12 @@ func (p *Proxy) createQUICListeners() error {
 
 // quicPacketLoop listens for incoming QUIC packets.
 //
-// See also the comment on Proxy.requestGoroutinesSema.
-func (p *Proxy) quicPacketLoop(l *quic.EarlyListener, requestGoroutinesSema semaphore) {
+// See also the comment on Proxy.requestsSema.
+func (p *Proxy) quicPacketLoop(l *quic.EarlyListener, reqSema syncutil.Semaphore) {
 	log.Info("Entering the DNS-over-QUIC listener loop on %s", l.Addr())
 	for {
-		conn, err := l.Accept(context.Background())
+		ctx := context.Background()
+		conn, err := l.Accept(ctx)
 		if err != nil {
 			if isQUICErrorForDebugLog(err) {
 				log.Debug("accepting quic conn: closed or timed out: %s", err)
@@ -97,10 +99,16 @@ func (p *Proxy) quicPacketLoop(l *quic.EarlyListener, requestGoroutinesSema sema
 			break
 		}
 
-		requestGoroutinesSema.acquire()
+		err = reqSema.Acquire(ctx)
+		if err != nil {
+			log.Error("dnsproxy: quic: acquiring semaphore: %s", err)
+
+			break
+		}
 		go func() {
-			p.handleQUICConnection(conn, requestGoroutinesSema)
-			requestGoroutinesSema.release()
+			defer reqSema.Release()
+
+			p.handleQUICConnection(conn, reqSema)
 		}()
 	}
 }
@@ -108,15 +116,17 @@ func (p *Proxy) quicPacketLoop(l *quic.EarlyListener, requestGoroutinesSema sema
 // handleQUICConnection handles a new QUIC connection.  It waits for new streams
 // and passes them to handleQUICStream.
 //
-// See also the comment on Proxy.requestGoroutinesSema.
-func (p *Proxy) handleQUICConnection(conn quic.Connection, requestGoroutinesSema semaphore) {
+// See also the comment on Proxy.requestsSema.
+func (p *Proxy) handleQUICConnection(conn quic.Connection, reqSema syncutil.Semaphore) {
 	for {
+		ctx := context.Background()
+
 		// The stub to resolver DNS traffic follows a simple pattern in which
 		// the client sends a query, and the server provides a response.  This
 		// design specifies that for each subsequent query on a QUIC connection
 		// the client MUST select the next available client-initiated
 		// bidirectional stream.
-		stream, err := conn.AcceptStream(context.Background())
+		stream, err := conn.AcceptStream(ctx)
 		if err != nil {
 			if isQUICErrorForDebugLog(err) {
 				log.Debug("accepting quic stream: closed or timed out: %s", err)
@@ -130,16 +140,24 @@ func (p *Proxy) handleQUICConnection(conn quic.Connection, requestGoroutinesSema
 			return
 		}
 
-		requestGoroutinesSema.acquire()
+		err = reqSema.Acquire(ctx)
+		if err != nil {
+			log.Error("dnsproxy: quic: acquiring semaphore: %s", err)
+
+			// Close the connection to make sure resources are freed.
+			closeQUICConn(conn, DoQCodeNoError)
+
+			return
+		}
 		go func() {
+			defer reqSema.Release()
+
 			p.handleQUICStream(stream, conn)
 
 			// The server MUST send the response(s) on the same stream and MUST
 			// indicate, after the last response, through the STREAM FIN
 			// mechanism that no further data will be sent on that stream.
 			_ = stream.Close()
-
-			requestGoroutinesSema.release()
 		}()
 	}
 }

proxy/server_tcp.go

@@ -13,6 +13,7 @@ import (
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
 	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/syncutil"
 	"github.com/miekg/dns"
 )
 
@@ -60,8 +61,8 @@ func (p *Proxy) createTLSListeners() (err error) {
 // tcpPacketLoop listens for incoming TCP packets.  proto must be either "tcp"
 // or "tls".
 //
-// See also the comment on Proxy.requestGoroutinesSema.
-func (p *Proxy) tcpPacketLoop(l net.Listener, proto Proto, requestGoroutinesSema semaphore) {
+// See also the comment on Proxy.requestsSema.
+func (p *Proxy) tcpPacketLoop(l net.Listener, proto Proto, reqSema syncutil.Semaphore) {
 	log.Info("dnsproxy: entering %s listener loop on %s", proto, l.Addr())
 
 	for {
@@ -76,10 +77,17 @@ func (p *Proxy) tcpPacketLoop(l net.Listener, proto Proto, requestGoroutinesSema
 			break
 		}
 
-		requestGoroutinesSema.acquire()
+		// TODO(d.kolyshev): Pass and use context from above.
+		err = reqSema.Acquire(context.Background())
+		if err != nil {
+			log.Error("dnsproxy: tcp: acquiring semaphore: %s", err)
+
+			break
+		}
 		go func() {
+			defer reqSema.Release()
+
 			p.handleTCPConnection(clientConn, proto)
-			requestGoroutinesSema.release()
 		}()
 	}
 }

proxy/server_udp.go

@@ -10,6 +10,7 @@ import (
 	"github.com/AdguardTeam/golibs/errors"
 	"github.com/AdguardTeam/golibs/log"
 	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/syncutil"
 	"github.com/miekg/dns"
 )
 
@@ -60,8 +61,8 @@ func (p *Proxy) udpCreate(ctx context.Context, udpAddr *net.UDPAddr) (*net.UDPCo
 
 // udpPacketLoop listens for incoming UDP packets.
 //
-// See also the comment on Proxy.requestGoroutinesSema.
-func (p *Proxy) udpPacketLoop(conn *net.UDPConn, requestGoroutinesSema semaphore) {
+// See also the comment on Proxy.requestsSema.
+func (p *Proxy) udpPacketLoop(conn *net.UDPConn, reqSema syncutil.Semaphore) {
 	log.Info("dnsproxy: entering udp listener loop on %s", conn.LocalAddr())
 
 	b := make([]byte, dns.MaxMsgSize)
@@ -79,10 +80,18 @@ func (p *Proxy) udpPacketLoop(conn *net.UDPConn, requestGoroutinesSema semaphore
 			// we need the contents to survive the call because we're handling them in goroutine
 			packet := make([]byte, n)
 			copy(packet, b)
-			requestGoroutinesSema.acquire()
+
+			// TODO(d.kolyshev): Pass and use context from above.
+			sErr := reqSema.Acquire(context.Background())
+			if sErr != nil {
+				log.Error("dnsproxy: udp: acquiring semaphore: %s", sErr)
+
+				break
+			}
 			go func() {
+				defer reqSema.Release()
+
 				p.udpHandlePacket(packet, localIP, remoteAddr, conn)
-				requestGoroutinesSema.release()
 			}()
 		}
 		if err != nil {