server: prohibit more than MaxConcurrentStreams handlers from running at once (#6703) #6706
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -115,12 +115,6 @@ type serviceInfo struct { | |
| mdata interface{} | ||
| } | ||
|
|
||
| // 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 func() | ||
| } | ||
|
|
||
| type serverOptions struct { | ||
|
|
@@ -178,6 +172,7 @@ type serverOptions struct { | |
| } | ||
|
|
||
| var defaultServerOptions = serverOptions{ | ||
| maxConcurrentStreams: math.MaxUint32, | ||
| maxReceiveMessageSize: defaultServerMaxReceiveMessageSize, | ||
| maxSendMessageSize: defaultServerMaxSendMessageSize, | ||
| connectionTimeout: 120 * time.Second, | ||
|
|
@@ -389,6 +384,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 | ||
| }) | ||
|
|
@@ -590,24 +588,19 @@ const serverWorkerResetThreshold = 1 << 16 | |
| // [1] https://github.com/golang/go/issues/18138 | ||
| func (s *Server) serverWorker() { | ||
| for completed := 0; completed < serverWorkerResetThreshold; completed++ { | ||
| f, ok := <-s.serverWorkerChannel | ||
| if !ok { | ||
| return | ||
| } | ||
| f() | ||
| } | ||
| go s.serverWorker() | ||
| } | ||
|
|
||
| // 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 func()) | ||
| for i := uint32(0); i < s.opts.numServerWorkers; i++ { | ||
| go s.serverWorker() | ||
| } | ||
|
|
@@ -966,21 +959,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 { | ||
| select { | ||
| case s.serverWorkerChannel <- f: | ||
| return | ||
| default: | ||
| // If all stream workers are busy, fallback to the default code path. | ||
| } | ||
| } | ||
| go f() | ||
| }, func(ctx context.Context, method string) context.Context { | ||
| if !EnableTracing { | ||
| return ctx | ||
|
|
@@ -2075,3 +2073,32 @@ 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 int64 // accessed atomically | ||
| wait chan struct{} | ||
| } | ||
|
|
||
| func (q *atomicSemaphore) acquire() { | ||
| if atomic.AddInt64(&q.n, -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 atomic.AddInt64(&q.n, -1) <= 0 { | ||
|
There was a problem hiding this comment. Suspect this |
||
| // An acquire was waiting on us. Unblock it. | ||
| q.wait <- struct{}{} | ||
| } | ||
| } | ||
|
|
||
| func newHandlerQuota(n uint32) *atomicSemaphore { | ||
| return &atomicSemaphore{n: int64(n), wait: make(chan struct{}, 1)} | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does it mean that it will NOT take effect any more if applications set MaxConcurrentStreams as
math.MaxUint32?Note the default value set by etcd is
math.MaxUint32, reference:There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It seems OK, since the default maxConcurrentStreams is math.MaxUint32