From f0841ef45a771653d933b43dc162df4137bc386a Mon Sep 17 00:00:00 2001 From: "cilium-renovate[bot]" <134692979+cilium-renovate[bot]@users.noreply.github.com> Date: Mon, 16 Oct 2023 09:10:18 +0000 Subject: [PATCH] fix(deps): update module google.golang.org/grpc to v1.58.3 Signed-off-by: cilium-renovate[bot] <134692979+cilium-renovate[bot]@users.noreply.github.com> --- contrib/rthooks/tetragon-oci-hook/go.mod | 2 +- contrib/rthooks/tetragon-oci-hook/go.sum | 4 +- .../grpc/internal/transport/http2_server.go | 11 +-- .../vendor/google.golang.org/grpc/server.go | 71 +++++++++++++------ .../vendor/google.golang.org/grpc/version.go | 2 +- .../tetragon-oci-hook/vendor/modules.txt | 2 +- 6 files changed, 58 insertions(+), 34 deletions(-) diff --git a/contrib/rthooks/tetragon-oci-hook/go.mod b/contrib/rthooks/tetragon-oci-hook/go.mod index d6baf6ebd75..34d74e1625f 100644 --- a/contrib/rthooks/tetragon-oci-hook/go.mod +++ b/contrib/rthooks/tetragon-oci-hook/go.mod @@ -12,7 +12,7 @@ require ( github.com/sirupsen/logrus v1.9.3 github.com/spf13/pflag v1.0.5 github.com/stretchr/testify v1.8.4 - google.golang.org/grpc v1.58.2 + google.golang.org/grpc v1.58.3 ) require ( diff --git a/contrib/rthooks/tetragon-oci-hook/go.sum b/contrib/rthooks/tetragon-oci-hook/go.sum index 88a5eb4b0db..b74b865ba2b 100644 --- a/contrib/rthooks/tetragon-oci-hook/go.sum +++ b/contrib/rthooks/tetragon-oci-hook/go.sum @@ -71,8 +71,8 @@ google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5 h1: google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5/go.mod h1:5DZzOUPCLYL3mNkQ0ms0F3EuUNZ7py1Bqeq6sxzI7/Q= google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13 h1:N3bU/SQDCDyD6R528GJ/PwW9KjYcJA3dgyH+MovAkIM= google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13/go.mod h1:KSqppvjFjtoCI+KGd4PELB0qLNxdJHRGqRI09mB6pQA= -google.golang.org/grpc v1.58.2 h1:SXUpjxeVF3FKrTYQI4f4KvbGD5u2xccdYdurwowix5I= -google.golang.org/grpc v1.58.2/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0= +google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ= +google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= diff --git a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/internal/transport/http2_server.go b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/internal/transport/http2_server.go index 8d3a353c1d5..c06db679d89 100644 --- a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/internal/transport/http2_server.go +++ b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/internal/transport/http2_server.go @@ -171,15 +171,10 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport, ID: http2.SettingMaxFrameSize, Val: http2MaxFrameLen, }} - // TODO(zhaoq): Have a better way to signal "no limit" because 0 is - // permitted in the HTTP2 spec. - maxStreams := config.MaxStreams - if maxStreams == 0 { - maxStreams = math.MaxUint32 - } else { + if config.MaxStreams != math.MaxUint32 { isettings = append(isettings, http2.Setting{ ID: http2.SettingMaxConcurrentStreams, - Val: maxStreams, + Val: config.MaxStreams, }) } dynamicWindow := true @@ -258,7 +253,7 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport, framer: framer, readerDone: make(chan struct{}), writerDone: make(chan struct{}), - maxStreams: maxStreams, + maxStreams: config.MaxStreams, inTapHandle: config.InTapHandle, fc: &trInFlow{limit: uint32(icwz)}, state: reachable, diff --git a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/server.go b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/server.go index 244123c6c5a..eeae92fbe02 100644 --- a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/server.go +++ b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/server.go @@ -115,12 +115,6 @@ type serviceInfo struct { mdata any } -type serverWorkerData struct { - st transport.ServerTransport - wg *sync.WaitGroup - stream *transport.Stream -} - // Server is a gRPC server to serve RPC requests. type Server struct { opts serverOptions @@ -145,7 +139,7 @@ type Server struct { channelzID *channelz.Identifier czData *channelzData - serverWorkerChannel chan *serverWorkerData + serverWorkerChannel chan func() } type serverOptions struct { @@ -179,6 +173,7 @@ type serverOptions struct { } var defaultServerOptions = serverOptions{ + maxConcurrentStreams: math.MaxUint32, maxReceiveMessageSize: defaultServerMaxReceiveMessageSize, maxSendMessageSize: defaultServerMaxSendMessageSize, connectionTimeout: 120 * time.Second, @@ -404,6 +399,9 @@ func MaxSendMsgSize(m int) ServerOption { // MaxConcurrentStreams returns a ServerOption that will apply a limit on the number // of concurrent streams to each ServerTransport. func MaxConcurrentStreams(n uint32) ServerOption { + if n == 0 { + n = math.MaxUint32 + } return newFuncServerOption(func(o *serverOptions) { o.maxConcurrentStreams = n }) @@ -605,24 +603,19 @@ const serverWorkerResetThreshold = 1 << 16 // [1] https://github.com/golang/go/issues/18138 func (s *Server) serverWorker() { for completed := 0; completed < serverWorkerResetThreshold; completed++ { - data, ok := <-s.serverWorkerChannel + f, ok := <-s.serverWorkerChannel if !ok { return } - s.handleSingleStream(data) + f() } go s.serverWorker() } -func (s *Server) handleSingleStream(data *serverWorkerData) { - defer data.wg.Done() - s.handleStream(data.st, data.stream, s.traceInfo(data.st, data.stream)) -} - // initServerWorkers creates worker goroutines and a channel to process incoming // connections to reduce the time spent overall on runtime.morestack. func (s *Server) initServerWorkers() { - s.serverWorkerChannel = make(chan *serverWorkerData) + s.serverWorkerChannel = make(chan func()) for i := uint32(0); i < s.opts.numServerWorkers; i++ { go s.serverWorker() } @@ -982,21 +975,26 @@ func (s *Server) serveStreams(st transport.ServerTransport) { defer st.Close(errors.New("finished serving streams for the server transport")) var wg sync.WaitGroup + streamQuota := newHandlerQuota(s.opts.maxConcurrentStreams) st.HandleStreams(func(stream *transport.Stream) { wg.Add(1) + + streamQuota.acquire() + f := func() { + defer streamQuota.release() + defer wg.Done() + s.handleStream(st, stream, s.traceInfo(st, stream)) + } + if s.opts.numServerWorkers > 0 { - data := &serverWorkerData{st: st, wg: &wg, stream: stream} select { - case s.serverWorkerChannel <- data: + case s.serverWorkerChannel <- f: return default: // If all stream workers are busy, fallback to the default code path. } } - go func() { - defer wg.Done() - s.handleStream(st, stream, s.traceInfo(st, stream)) - }() + go f() }, func(ctx context.Context, method string) context.Context { if !EnableTracing { return ctx @@ -2091,3 +2089,34 @@ func validateSendCompressor(name, clientCompressors string) error { } return fmt.Errorf("client does not support compressor %q", name) } + +// atomicSemaphore implements a blocking, counting semaphore. acquire should be +// called synchronously; release may be called asynchronously. +type atomicSemaphore struct { + n atomic.Int64 + wait chan struct{} +} + +func (q *atomicSemaphore) acquire() { + if q.n.Add(-1) < 0 { + // We ran out of quota. Block until a release happens. + <-q.wait + } +} + +func (q *atomicSemaphore) release() { + // N.B. the "<= 0" check below should allow for this to work with multiple + // concurrent calls to acquire, but also note that with synchronous calls to + // acquire, as our system does, n will never be less than -1. There are + // fairness issues (queuing) to consider if this was to be generalized. + if q.n.Add(1) <= 0 { + // An acquire was waiting on us. Unblock it. + q.wait <- struct{}{} + } +} + +func newHandlerQuota(n uint32) *atomicSemaphore { + a := &atomicSemaphore{wait: make(chan struct{}, 1)} + a.n.Store(int64(n)) + return a +} diff --git a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/version.go b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/version.go index d3f5bcbfcef..724ad210213 100644 --- a/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/version.go +++ b/contrib/rthooks/tetragon-oci-hook/vendor/google.golang.org/grpc/version.go @@ -19,4 +19,4 @@ package grpc // Version is the current grpc version. -const Version = "1.58.2" +const Version = "1.58.3" diff --git a/contrib/rthooks/tetragon-oci-hook/vendor/modules.txt b/contrib/rthooks/tetragon-oci-hook/vendor/modules.txt index f36a9d7bca1..687f766170b 100644 --- a/contrib/rthooks/tetragon-oci-hook/vendor/modules.txt +++ b/contrib/rthooks/tetragon-oci-hook/vendor/modules.txt @@ -123,7 +123,7 @@ google.golang.org/genproto/googleapis/api/expr/v1alpha1 # google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13 ## explicit; go 1.19 google.golang.org/genproto/googleapis/rpc/status -# google.golang.org/grpc v1.58.2 +# google.golang.org/grpc v1.58.3 ## explicit; go 1.19 google.golang.org/grpc google.golang.org/grpc/attributes