-
Notifications
You must be signed in to change notification settings - Fork 4.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #15569 from dmage/maxconnections
Automatic merge from submit-queue Add limit for number of concurrent connections to registry The registry might have excessive resource usage under heavy load. To avoid this, we limit the number of concurrent requests. Requests over the MaxRunning limit are enqueued. Requests are rejected if there are MaxInQueue requests in the queue. Request may stay in the queue no more than MaxWaitInQueue. See also #15448.
- Loading branch information
Showing
10 changed files
with
573 additions
and
3 deletions.
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
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 |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| package maxconnections | ||
|
|
||
| import ( | ||
| "reflect" | ||
| "sync" | ||
| "testing" | ||
| ) | ||
|
|
||
| type countM map[interface{}]int | ||
|
|
||
| type counter struct { | ||
| mu sync.Mutex | ||
| m countM | ||
| } | ||
|
|
||
| func newCounter() *counter { | ||
| return &counter{ | ||
| m: make(countM), | ||
| } | ||
| } | ||
|
|
||
| func (c *counter) Add(key interface{}, delta int) { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| c.m[key] += delta | ||
| } | ||
|
|
||
| func (c *counter) Values() countM { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| m := make(map[interface{}]int) | ||
| for k, v := range c.m { | ||
| m[k] = v | ||
| } | ||
| return m | ||
| } | ||
|
|
||
| func (c *counter) Equal(m countM) bool { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| for k, v := range m { | ||
| if c.m[k] != v { | ||
| return false | ||
| } | ||
| } | ||
| for k, v := range c.m { | ||
| if _, ok := m[k]; !ok && v != 0 { | ||
| return false | ||
| } | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| func TestCounter(t *testing.T) { | ||
| c := newCounter() | ||
| c.Add(100, 1) | ||
| c.Add(200, 2) | ||
| c.Add(300, 3) | ||
| if expected := (countM{100: 1, 200: 2, 300: 3}); !reflect.DeepEqual(c.m, expected) { | ||
| t.Fatalf("c.m = %v, want %v", c.m, expected) | ||
| } | ||
| if expected := (countM{100: 1, 200: 2, 300: 3}); !c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is false, want true", c.m, expected) | ||
| } | ||
|
|
||
| c.Add(200, -2) | ||
| if expected := (countM{100: 1, 200: 0, 300: 3}); !c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is false, want true", c.m, expected) | ||
| } | ||
| if expected := (countM{100: 1, 300: 3}); !c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is false, want true", c.m, expected) | ||
| } | ||
| if expected := (countM{100: 1, 300: 3, 400: 0}); !c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is false, want true", c.m, expected) | ||
| } | ||
|
|
||
| if expected := (countM{100: 1}); c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is true, want false", c.m, expected) | ||
| } | ||
| if expected := (countM{100: 1, 300: 3, 400: 4}); c.Equal(expected) { | ||
| t.Fatalf("counter(%v).Equal(%v) is true, want false", c.m, expected) | ||
| } | ||
| } |
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 |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package maxconnections | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
| ) | ||
|
|
||
| // A Limiter controls starting of jobs. | ||
| type Limiter interface { | ||
| // Start decides whether a new job can be started. The decision may be | ||
| // returned after a delay if the limiter wants to throttle jobs. | ||
| Start(context.Context) bool | ||
|
|
||
| // Done must be called when a job is finished. | ||
| Done() | ||
| } | ||
|
|
||
| // limiter ensures that there are no more than maxRunning jobs at the same | ||
| // time. It can enqueue up to maxInQueue jobs awaiting to be run, for other | ||
| // jobs Start will return false immediately. | ||
| type limiter struct { | ||
| // running is a buffered channel. Before starting a job, an empty struct is | ||
| // sent to the channel. When the job is finished, one element is received | ||
| // back from the channel. If the channel's buffer is full, the job is | ||
| // enqueued. | ||
| running chan struct{} | ||
|
|
||
| // queue is a buffered channel. An empty struct is placed into the channel | ||
| // while a job is waiting for a spot in the running channel's buffer. | ||
| // If the queue channel's buffer is full, the job is declined. | ||
| queue chan struct{} | ||
|
|
||
| // maxWaitInQueue is a maximum wait time in the queue, zero means forever. | ||
| maxWaitInQueue time.Duration | ||
|
|
||
| // newTimer allows to override the function time.NewTimer for tests. | ||
| newTimer func(d time.Duration) *time.Timer | ||
| } | ||
|
|
||
| // NewLimiter return a limiter that allows no more than maxRunning jobs at the | ||
| // same time. It can enqueue up to maxInQueue jobs awaiting to be run, and a | ||
| // job may wait in the queue no more than maxWaitInQueue. | ||
| func NewLimiter(maxRunning, maxInQueue int, maxWaitInQueue time.Duration) Limiter { | ||
| return &limiter{ | ||
| running: make(chan struct{}, maxRunning), | ||
| queue: make(chan struct{}, maxInQueue), | ||
| maxWaitInQueue: maxWaitInQueue, | ||
| newTimer: time.NewTimer, | ||
| } | ||
| } | ||
|
|
||
| func (l *limiter) Start(ctx context.Context) bool { | ||
| select { | ||
| case l.running <- struct{}{}: | ||
| return true | ||
| default: | ||
| } | ||
|
|
||
| // Slow-path. | ||
| select { | ||
| case l.queue <- struct{}{}: | ||
| defer func() { | ||
| <-l.queue | ||
| }() | ||
| default: | ||
| return false | ||
| } | ||
|
|
||
| var timeout <-chan time.Time | ||
| // if l.maxWaitInQueue is 0, timeout will stay nil which practically means wait forever. | ||
| if l.maxWaitInQueue > 0 { | ||
| timer := l.newTimer(l.maxWaitInQueue) | ||
| defer timer.Stop() | ||
| timeout = timer.C | ||
| } | ||
|
|
||
| select { | ||
| case l.running <- struct{}{}: | ||
| return true | ||
| case <-timeout: | ||
| case <-ctx.Done(): | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func (l *limiter) Done() { | ||
| <-l.running | ||
| } |
Oops, something went wrong.