lib/writeconcurrencylimiter: moved the error generation from incConcurrency() to the caller place

This commit is contained in:
Aliaksandr Valialkin 2023-01-06 23:45:54 -08:00
parent 986a05e18d
commit 45f39e291e
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1

View file

@ -62,7 +62,14 @@ var readerPool sync.Pool
func (r *Reader) Read(p []byte) (int, error) {
n, err := r.r.Read(p)
if !r.increasedConcurrency {
if err := incConcurrency(); err != nil {
if !incConcurrency() {
err = &httpserver.ErrorWithStatusCode{
Err: fmt.Errorf("cannot process insert request for %.3f seconds because %d concurrent insert requests are already executed. "+
"Possible solutions: to reduce workload; to increase compute resources at the server; "+
"to increase -insert.maxQueueDuration; to increase -maxConcurrentInserts",
maxQueueDuration.Seconds(), *maxConcurrentInserts),
StatusCode: http.StatusServiceUnavailable,
}
return 0, err
}
r.increasedConcurrency = true
@ -87,12 +94,12 @@ var (
concurrencyLimitChOnce sync.Once
)
func incConcurrency() error {
func incConcurrency() bool {
concurrencyLimitChOnce.Do(initConcurrencyLimitCh)
select {
case concurrencyLimitCh <- struct{}{}:
return nil
return true
default:
}
@ -101,17 +108,11 @@ func incConcurrency() error {
select {
case concurrencyLimitCh <- struct{}{}:
timerpool.Put(t)
return nil
return true
case <-t.C:
timerpool.Put(t)
concurrencyLimitTimeout.Inc()
return &httpserver.ErrorWithStatusCode{
Err: fmt.Errorf("cannot process insert request for %.3f seconds because %d concurrent insert requests are already executed. "+
"Possible solutions: to reduce workload; to increase compute resources at the server; "+
"to increase -insert.maxQueueDuration; to increase -maxConcurrentInserts",
maxQueueDuration.Seconds(), *maxConcurrentInserts),
StatusCode: http.StatusServiceUnavailable,
}
return false
}
}