Repository: VictoriaMetrics/VictoriaMetrics
Pull request #3609 (2023-01-06 15:06:39)

lib/promscrape/discoveryutils: use correct timeout for blocking requests
Fixes timeout usage to use Read timeout since context sets limit for the whole request lifecycle.

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
Base: **57.58**% // Head: **57.56**% // Decreases project coverage by **`-0.01%`** :warning:
> Coverage data is based on head [(`98731e8`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) compared to base [(`463b957`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/463b957e54c446106d7ee4ad411deda5b036a1fd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Patch coverage: 0.00% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3609      +/-   ##
==========================================
- Coverage   57.58%   57.56%   -0.02%     
==========================================
  Files         310      310              
  Lines       51868    51868              
==========================================
- Hits        29870    29860      -10     
- Misses      20119    20127       +8     
- Partials     1879     1881       +2     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [lib/promscrape/discoveryutils/client.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3Byb21zY3JhcGUvZGlzY292ZXJ5dXRpbHMvY2xpZW50Lmdv) | `0.00% <0.00%> (ø)` | |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `85.28% <0.00%> (-2.41%)` | :arrow_down: |
| [app/vmselect/promql/transform.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC90cmFuc2Zvcm0uZ28=) | `88.03% <0.00%> (-0.16%)` | :arrow_down: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `64.38% <0.00%> (+0.68%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3609?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/README.md b/README.md
index 9403e4d2a..248a539b0 100644
--- a/README.md
+++ b/README.md
@@ -2235,7 +2235,7 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index 291ff15ce..da5c85755 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -1247,7 +1247,7 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
diff --git a/app/vmagent/csvimport/request_handler.go b/app/vmagent/csvimport/request_handler.go
index 72dadc888..e17207c0d 100644
--- a/app/vmagent/csvimport/request_handler.go
+++ b/app/vmagent/csvimport/request_handler.go
@@ -10,7 +10,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -26,10 +25,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req, func(rows []parser.Row) error {
-			return insertRows(at, rows, extraLabels)
-		})
+	return parser.ParseStream(req, func(rows []parser.Row) error {
+		return insertRows(at, rows, extraLabels)
 	})
 }
 
diff --git a/app/vmagent/datadog/request_handler.go b/app/vmagent/datadog/request_handler.go
index 2b7aa5a27..adac86be3 100644
--- a/app/vmagent/datadog/request_handler.go
+++ b/app/vmagent/datadog/request_handler.go
@@ -10,7 +10,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -28,11 +27,9 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		ce := req.Header.Get("Content-Encoding")
-		return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
-			return insertRows(at, series, extraLabels)
-		})
+	ce := req.Header.Get("Content-Encoding")
+	return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
+		return insertRows(at, series, extraLabels)
 	})
 }
 
diff --git a/app/vmagent/graphite/request_handler.go b/app/vmagent/graphite/request_handler.go
index c3ef22d8d..d9ff2a1b2 100644
--- a/app/vmagent/graphite/request_handler.go
+++ b/app/vmagent/graphite/request_handler.go
@@ -7,7 +7,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/graphite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -20,9 +19,7 @@ var (
 //
 // See https://graphite.readthedocs.io/en/latest/feeding-carbon.html#the-plaintext-protocol
 func InsertHandler(r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, insertRows)
-	})
+	return parser.ParseStream(r, insertRows)
 }
 
 func insertRows(rows []parser.Row) error {
diff --git a/app/vmagent/influx/request_handler.go b/app/vmagent/influx/request_handler.go
index 19177203d..d7211e385 100644
--- a/app/vmagent/influx/request_handler.go
+++ b/app/vmagent/influx/request_handler.go
@@ -16,7 +16,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -37,10 +36,8 @@ var (
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
 func InsertHandlerForReader(r io.Reader, isGzipped bool) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, isGzipped, "", "", func(db string, rows []parser.Row) error {
-			return insertRows(nil, db, rows, nil)
-		})
+	return parser.ParseStream(r, isGzipped, "", "", func(db string, rows []parser.Row) error {
+		return insertRows(nil, db, rows, nil)
 	})
 }
 
@@ -52,15 +49,13 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		q := req.URL.Query()
-		precision := q.Get("precision")
-		// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
-		db := q.Get("db")
-		return parser.ParseStream(req.Body, isGzipped, precision, db, func(db string, rows []parser.Row) error {
-			return insertRows(at, db, rows, extraLabels)
-		})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	q := req.URL.Query()
+	precision := q.Get("precision")
+	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
+	db := q.Get("db")
+	return parser.ParseStream(req.Body, isGzipped, precision, db, func(db string, rows []parser.Row) error {
+		return insertRows(at, db, rows, extraLabels)
 	})
 }
 
diff --git a/app/vmagent/main.go b/app/vmagent/main.go
index 412580f8b..ce2ab12d1 100644
--- a/app/vmagent/main.go
+++ b/app/vmagent/main.go
@@ -38,7 +38,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -104,7 +103,6 @@ func main() {
 	startTime := time.Now()
 	remotewrite.Init()
 	common.StartUnmarshalWorkers()
-	writeconcurrencylimiter.Init()
 	if len(*influxListenAddr) > 0 {
 		influxServer = influxserver.MustStart(*influxListenAddr, func(r io.Reader) error {
 			return influx.InsertHandlerForReader(r, false)
diff --git a/app/vmagent/native/request_handler.go b/app/vmagent/native/request_handler.go
index f0869e042..898394ae8 100644
--- a/app/vmagent/native/request_handler.go
+++ b/app/vmagent/native/request_handler.go
@@ -12,7 +12,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -31,10 +30,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	isGzip := req.Header.Get("Content-Encoding") == "gzip"
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req.Body, isGzip, func(block *parser.Block) error {
-			return insertRows(at, block, extraLabels)
-		})
+	return parser.ParseStream(req.Body, isGzip, func(block *parser.Block) error {
+		return insertRows(at, block, extraLabels)
 	})
 }
 
diff --git a/app/vmagent/opentsdb/request_handler.go b/app/vmagent/opentsdb/request_handler.go
index 2721912d6..e1e42ba4b 100644
--- a/app/vmagent/opentsdb/request_handler.go
+++ b/app/vmagent/opentsdb/request_handler.go
@@ -7,7 +7,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdb"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -20,9 +19,7 @@ var (
 //
 // See http://opentsdb.net/docs/build/html/api_telnet/put.html
 func InsertHandler(r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, insertRows)
-	})
+	return parser.ParseStream(r, insertRows)
 }
 
 func insertRows(rows []parser.Row) error {
diff --git a/app/vmagent/opentsdbhttp/request_handler.go b/app/vmagent/opentsdbhttp/request_handler.go
index 98b02e5e3..968ea936b 100644
--- a/app/vmagent/opentsdbhttp/request_handler.go
+++ b/app/vmagent/opentsdbhttp/request_handler.go
@@ -9,7 +9,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -25,10 +24,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req, func(rows []parser.Row) error {
-			return insertRows(at, rows, extraLabels)
-		})
+	return parser.ParseStream(req, func(rows []parser.Row) error {
+		return insertRows(at, rows, extraLabels)
 	})
 }
 
diff --git a/app/vmagent/prometheusimport/request_handler.go b/app/vmagent/prometheusimport/request_handler.go
index e5b86a799..571860e25 100644
--- a/app/vmagent/prometheusimport/request_handler.go
+++ b/app/vmagent/prometheusimport/request_handler.go
@@ -11,7 +11,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -31,21 +30,17 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
-			return insertRows(at, rows, extraLabels)
-		}, nil)
-	})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
+		return insertRows(at, rows, extraLabels)
+	}, nil)
 }
 
 // InsertHandlerForReader processes metrics from given reader with optional gzip format
 func InsertHandlerForReader(r io.Reader, isGzipped bool) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, 0, isGzipped, func(rows []parser.Row) error {
-			return insertRows(nil, rows, nil)
-		}, nil)
-	})
+	return parser.ParseStream(r, 0, isGzipped, func(rows []parser.Row) error {
+		return insertRows(nil, rows, nil)
+	}, nil)
 }
 
 func insertRows(at *auth.Token, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
diff --git a/app/vmagent/promremotewrite/request_handler.go b/app/vmagent/promremotewrite/request_handler.go
index f2ed1b6f6..4fd0ce0e6 100644
--- a/app/vmagent/promremotewrite/request_handler.go
+++ b/app/vmagent/promremotewrite/request_handler.go
@@ -13,7 +13,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -29,19 +28,15 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error {
-			return insertRows(at, tss, extraLabels)
-		})
+	return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error {
+		return insertRows(at, tss, extraLabels)
 	})
 }
 
 // InsertHandlerForReader processes metrics from given reader
 func InsertHandlerForReader(at *auth.Token, r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, func(tss []prompb.TimeSeries) error {
-			return insertRows(at, tss, nil)
-		})
+	return parser.ParseStream(r, func(tss []prompb.TimeSeries) error {
+		return insertRows(at, tss, nil)
 	})
 }
 
diff --git a/app/vmagent/vmimport/request_handler.go b/app/vmagent/vmimport/request_handler.go
index 0afe7389b..781f75753 100644
--- a/app/vmagent/vmimport/request_handler.go
+++ b/app/vmagent/vmimport/request_handler.go
@@ -13,7 +13,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -31,20 +30,16 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
-			return insertRows(at, rows, extraLabels)
-		})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
+		return insertRows(at, rows, extraLabels)
 	})
 }
 
 // InsertHandlerForReader processes metrics from given reader
 func InsertHandlerForReader(r io.Reader, isGzipped bool) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, isGzipped, func(rows []parser.Row) error {
-			return insertRows(nil, rows, nil)
-		})
+	return parser.ParseStream(r, isGzipped, func(rows []parser.Row) error {
+		return insertRows(nil, rows, nil)
 	})
 }
 
diff --git a/app/vminsert/common/insert_ctx.go b/app/vminsert/common/insert_ctx.go
index d2c63feb0..153e003bd 100644
--- a/app/vminsert/common/insert_ctx.go
+++ b/app/vminsert/common/insert_ctx.go
@@ -144,6 +144,9 @@ func (ctx *InsertCtx) FlushBufs() error {
 			return nil
 		}
 	}
+	// There is no need in limiting the number of concurrent calls to vmstorage.AddRows() here,
+	// since the number of concurrent FlushBufs() calls should be already limited via writeconcurrencylimiter
+	// used at every ParseStream() call under lib/protoparser/*/streamparser.go
 	err := vmstorage.AddRows(ctx.mrs)
 	ctx.Reset(0)
 	if err == nil {
diff --git a/app/vminsert/common/streamaggr.go b/app/vminsert/common/streamaggr.go
index 9f4cbc233..ed028e089 100644
--- a/app/vminsert/common/streamaggr.go
+++ b/app/vminsert/common/streamaggr.go
@@ -111,6 +111,8 @@ func pushAggregateSeries(tss []prompbmarshal.TimeSeries) {
 			return
 		}
 	}
+	// There is no need in limiting the number of concurrent calls to vmstorage.AddRows() here,
+	// since the number of concurrent pushAggregateSeries() calls should be already limited by lib/streamaggr.
 	if err := vmstorage.AddRows(ctx.mrs); err != nil {
 		logger.Errorf("cannot flush aggregate series: %s", err)
 	}
diff --git a/app/vminsert/csvimport/request_handler.go b/app/vminsert/csvimport/request_handler.go
index 6997dd4cb..971f3abbd 100644
--- a/app/vminsert/csvimport/request_handler.go
+++ b/app/vminsert/csvimport/request_handler.go
@@ -8,7 +8,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -23,10 +22,8 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req, func(rows []parser.Row) error {
-			return insertRows(rows, extraLabels)
-		})
+	return parser.ParseStream(req, func(rows []parser.Row) error {
+		return insertRows(rows, extraLabels)
 	})
 }
 
diff --git a/app/vminsert/datadog/request_handler.go b/app/vminsert/datadog/request_handler.go
index 6dbfbbbec..48dd6f83f 100644
--- a/app/vminsert/datadog/request_handler.go
+++ b/app/vminsert/datadog/request_handler.go
@@ -1,7 +1,6 @@
 package datadog
 
 import (
-	"fmt"
 	"net/http"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
@@ -9,7 +8,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -26,15 +24,9 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		ce := req.Header.Get("Content-Encoding")
-		err := parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
-			return insertRows(series, extraLabels)
-		})
-		if err != nil {
-			return fmt.Errorf("headers: %q; err: %w", req.Header, err)
-		}
-		return nil
+	ce := req.Header.Get("Content-Encoding")
+	return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
+		return insertRows(series, extraLabels)
 	})
 }
 
diff --git a/app/vminsert/graphite/request_handler.go b/app/vminsert/graphite/request_handler.go
index 5aa24f929..5d2d0a504 100644
--- a/app/vminsert/graphite/request_handler.go
+++ b/app/vminsert/graphite/request_handler.go
@@ -6,7 +6,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/graphite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -19,9 +18,7 @@ var (
 //
 // See https://graphite.readthedocs.io/en/latest/feeding-carbon.html#the-plaintext-protocol
 func InsertHandler(r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, insertRows)
-	})
+	return parser.ParseStream(r, insertRows)
 }
 
 func insertRows(rows []parser.Row) error {
diff --git a/app/vminsert/influx/request_handler.go b/app/vminsert/influx/request_handler.go
index 40ede91c8..daa7334d7 100644
--- a/app/vminsert/influx/request_handler.go
+++ b/app/vminsert/influx/request_handler.go
@@ -15,7 +15,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -35,10 +34,8 @@ var (
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
 func InsertHandlerForReader(r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, false, "", "", func(db string, rows []parser.Row) error {
-			return insertRows(db, rows, nil)
-		})
+	return parser.ParseStream(r, false, "", "", func(db string, rows []parser.Row) error {
+		return insertRows(db, rows, nil)
 	})
 }
 
@@ -50,15 +47,13 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		q := req.URL.Query()
-		precision := q.Get("precision")
-		// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
-		db := q.Get("db")
-		return parser.ParseStream(req.Body, isGzipped, precision, db, func(db string, rows []parser.Row) error {
-			return insertRows(db, rows, extraLabels)
-		})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	q := req.URL.Query()
+	precision := q.Get("precision")
+	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
+	db := q.Get("db")
+	return parser.ParseStream(req.Body, isGzipped, precision, db, func(db string, rows []parser.Row) error {
+		return insertRows(db, rows, extraLabels)
 	})
 }
 
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index 8ad6b2023..118726236 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -35,7 +35,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -71,7 +70,6 @@ func Init() {
 	storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries)
 	storage.SetMaxLabelValueLen(*maxLabelValueLen)
 	common.StartUnmarshalWorkers()
-	writeconcurrencylimiter.Init()
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, graphite.InsertHandler)
 	}
diff --git a/app/vminsert/native/request_handler.go b/app/vminsert/native/request_handler.go
index f821b0889..666848e4a 100644
--- a/app/vminsert/native/request_handler.go
+++ b/app/vminsert/native/request_handler.go
@@ -12,7 +12,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -28,10 +27,8 @@ func InsertHandler(req *http.Request) error {
 		return err
 	}
 	isGzip := req.Header.Get("Content-Encoding") == "gzip"
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req.Body, isGzip, func(block *parser.Block) error {
-			return insertRows(block, extraLabels)
-		})
+	return parser.ParseStream(req.Body, isGzip, func(block *parser.Block) error {
+		return insertRows(block, extraLabels)
 	})
 }
 
diff --git a/app/vminsert/opentsdb/request_handler.go b/app/vminsert/opentsdb/request_handler.go
index 49a6157a1..44bf3eb98 100644
--- a/app/vminsert/opentsdb/request_handler.go
+++ b/app/vminsert/opentsdb/request_handler.go
@@ -6,7 +6,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdb"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -19,9 +18,7 @@ var (
 //
 // See http://opentsdb.net/docs/build/html/api_telnet/put.html
 func InsertHandler(r io.Reader) error {
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(r, insertRows)
-	})
+	return parser.ParseStream(r, insertRows)
 }
 
 func insertRows(rows []parser.Row) error {
diff --git a/app/vminsert/opentsdbhttp/request_handler.go b/app/vminsert/opentsdbhttp/request_handler.go
index b5927ecc7..3dafe29f9 100644
--- a/app/vminsert/opentsdbhttp/request_handler.go
+++ b/app/vminsert/opentsdbhttp/request_handler.go
@@ -9,7 +9,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -28,10 +27,8 @@ func InsertHandler(req *http.Request) error {
 		if err != nil {
 			return err
 		}
-		return writeconcurrencylimiter.Do(func() error {
-			return parser.ParseStream(req, func(rows []parser.Row) error {
-				return insertRows(rows, extraLabels)
-			})
+		return parser.ParseStream(req, func(rows []parser.Row) error {
+			return insertRows(rows, extraLabels)
 		})
 	default:
 		return fmt.Errorf("unexpected path requested on HTTP OpenTSDB server: %q", path)
diff --git a/app/vminsert/prometheusimport/request_handler.go b/app/vminsert/prometheusimport/request_handler.go
index aa65b7da5..0afccf45b 100644
--- a/app/vminsert/prometheusimport/request_handler.go
+++ b/app/vminsert/prometheusimport/request_handler.go
@@ -8,7 +8,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -27,12 +26,10 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
-			return insertRows(rows, extraLabels)
-		}, nil)
-	})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
+		return insertRows(rows, extraLabels)
+	}, nil)
 }
 
 func insertRows(rows []parser.Row, extraLabels []prompbmarshal.Label) error {
diff --git a/app/vminsert/promremotewrite/request_handler.go b/app/vminsert/promremotewrite/request_handler.go
index eef326361..8d94e3235 100644
--- a/app/vminsert/promremotewrite/request_handler.go
+++ b/app/vminsert/promremotewrite/request_handler.go
@@ -9,7 +9,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -24,10 +23,8 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error {
-			return insertRows(tss, extraLabels)
-		})
+	return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error {
+		return insertRows(tss, extraLabels)
 	})
 }
 
diff --git a/app/vminsert/vmimport/request_handler.go b/app/vminsert/vmimport/request_handler.go
index f0a531a54..2a3a45039 100644
--- a/app/vminsert/vmimport/request_handler.go
+++ b/app/vminsert/vmimport/request_handler.go
@@ -12,7 +12,6 @@ import (
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -29,11 +28,9 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	return writeconcurrencylimiter.Do(func() error {
-		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
-			return insertRows(rows, extraLabels)
-		})
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
+		return insertRows(rows, extraLabels)
 	})
 }
 
diff --git a/app/vmselect/main.go b/app/vmselect/main.go
index f021a46c0..f17f2e438 100644
--- a/app/vmselect/main.go
+++ b/app/vmselect/main.go
@@ -62,7 +62,7 @@ func Init() {
 	netstorage.InitTmpBlocksDir(tmpDirPath)
 	promql.InitRollupResultCache(*vmstorage.DataPath + "/cache/rollupResult")
 
-	concurrencyCh = make(chan struct{}, *maxConcurrentRequests)
+	concurrencyLimitCh = make(chan struct{}, *maxConcurrentRequests)
 	initVMAlertProxy()
 }
 
@@ -71,17 +71,17 @@ func Stop() {
 	promql.StopRollupResultCache()
 }
 
-var concurrencyCh chan struct{}
+var concurrencyLimitCh chan struct{}
 
 var (
 	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_select_limit_reached_total`)
 	concurrencyLimitTimeout = metrics.NewCounter(`vm_concurrent_select_limit_timeout_total`)
 
 	_ = metrics.NewGauge(`vm_concurrent_select_capacity`, func() float64 {
-		return float64(cap(concurrencyCh))
+		return float64(cap(concurrencyLimitCh))
 	})
 	_ = metrics.NewGauge(`vm_concurrent_select_current`, func() float64 {
-		return float64(len(concurrencyCh))
+		return float64(len(concurrencyLimitCh))
 	})
 )
 
@@ -99,8 +99,8 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 
 	// Limit the number of concurrent queries.
 	select {
-	case concurrencyCh <- struct{}{}:
-		defer func() { <-concurrencyCh }()
+	case concurrencyLimitCh <- struct{}{}:
+		defer func() { <-concurrencyLimitCh }()
 	default:
 		// Sleep for a while until giving up. This should resolve short bursts in requests.
 		concurrencyLimitReached.Inc()
@@ -110,18 +110,18 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		}
 		t := timerpool.Get(d)
 		select {
-		case concurrencyCh <- struct{}{}:
-			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", *maxConcurrentRequests)
+		case concurrencyLimitCh <- struct{}{}:
 			timerpool.Put(t)
-			defer func() { <-concurrencyCh }()
+			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", *maxConcurrentRequests)
+			defer func() { <-concurrencyLimitCh }()
 		case <-t.C:
 			timerpool.Put(t)
 			concurrencyLimitTimeout.Inc()
 			err := &httpserver.ErrorWithStatusCode{
-				Err: fmt.Errorf("cannot handle more than %d concurrent search requests during %s; possible solutions: "+
-					"increase `-search.maxQueueDuration`; increase `-search.maxQueryDuration`; increase `-search.maxConcurrentRequests`; "+
-					"increase server capacity",
-					*maxConcurrentRequests, d),
+				Err: fmt.Errorf("couldn't start executing the request in %.3f seconds, since -search.maxConcurrentRequests=%d concurrent requests "+
+					"are executed. Possible solutions: to reduce query load; to add more compute resources to the server; "+
+					"to increase -search.maxQueueDuration; to increase -search.maxQueryDuration; to increase -search.maxConcurrentRequests",
+					d.Seconds(), *maxConcurrentRequests),
 				StatusCode: http.StatusServiceUnavailable,
 			}
 			httpserver.Errorf(w, r, "%s", err)
diff --git a/app/vmstorage/main.go b/app/vmstorage/main.go
index f9eaf7423..d9efd46d9 100644
--- a/app/vmstorage/main.go
+++ b/app/vmstorage/main.go
@@ -147,6 +147,8 @@ var WG syncwg.WaitGroup
 var resetResponseCacheIfNeeded func(mrs []storage.MetricRow)
 
 // AddRows adds mrs to the storage.
+//
+// The caller should limit the number of concurrent calls to AddRows() in order to limit memory usage.
 func AddRows(mrs []storage.MetricRow) error {
 	if Storage.IsReadOnly() {
 		return errReadOnly
@@ -650,22 +652,6 @@ func registerStorageMetrics(strg *storage.Storage) {
 		return float64(m().TooSmallTimestampRows)
 	})
 
-	metrics.NewGauge(`vm_concurrent_addrows_limit_reached_total`, func() float64 {
-		return float64(m().AddRowsConcurrencyLimitReached)
-	})
-	metrics.NewGauge(`vm_concurrent_addrows_limit_timeout_total`, func() float64 {
-		return float64(m().AddRowsConcurrencyLimitTimeout)
-	})
-	metrics.NewGauge(`vm_concurrent_addrows_dropped_rows_total`, func() float64 {
-		return float64(m().AddRowsConcurrencyDroppedRows)
-	})
-	metrics.NewGauge(`vm_concurrent_addrows_capacity`, func() float64 {
-		return float64(m().AddRowsConcurrencyCapacity)
-	})
-	metrics.NewGauge(`vm_concurrent_addrows_current`, func() float64 {
-		return float64(m().AddRowsConcurrencyCurrent)
-	})
-
 	metrics.NewGauge(`vm_search_delays_total`, func() float64 {
 		return float64(m().SearchDelays)
 	})
diff --git a/dashboards/victoriametrics-cluster.json b/dashboards/victoriametrics-cluster.json
index 51fd6350a..e43f206f2 100644
--- a/dashboards/victoriametrics-cluster.json
+++ b/dashboards/victoriametrics-cluster.json
@@ -5021,7 +5021,7 @@
               },
               "editorMode": "code",
               "exemplar": true,
-              "expr": "max(\n    max_over_time(vm_concurrent_addrows_current{job=~\"$job_storage\", \n    instance=~\"$instance\"}[$__rate_interval])\n)",
+              "expr": "max(\n    max_over_time(vm_concurrent_insert_current{job=~\"$job_storage\", \n    instance=~\"$instance\"}[$__rate_interval])\n)",
               "interval": "",
               "legendFormat": "current",
               "range": true,
@@ -5034,7 +5034,7 @@
               },
               "editorMode": "code",
               "exemplar": true,
-              "expr": "min(vm_concurrent_addrows_capacity{job=~\"$job_storage\", instance=~\"$instance\"})",
+              "expr": "min(vm_concurrent_insert_capacity{job=~\"$job_storage\", instance=~\"$instance\"})",
               "hide": false,
               "interval": "",
               "legendFormat": "max",
diff --git a/dashboards/victoriametrics.json b/dashboards/victoriametrics.json
index 51cc849c5..5afd6087e 100644
--- a/dashboards/victoriametrics.json
+++ b/dashboards/victoriametrics.json
@@ -4609,7 +4609,7 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "max_over_time(vm_concurrent_addrows_capacity{job=~\"$job\", instance=~\"$instance\"}[$__rate_interval])",
+              "expr": "max_over_time(vm_concurrent_insert_capacity{job=~\"$job\", instance=~\"$instance\"}[$__rate_interval])",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
@@ -4622,7 +4622,7 @@
                 "type": "prometheus",
                 "uid": "$ds"
               },
-              "expr": "sum(vm_concurrent_addrows_current{job=~\"$job\", instance=~\"$instance\"})",
+              "expr": "sum(vm_concurrent_insert_current{job=~\"$job\", instance=~\"$instance\"})",
               "format": "time_series",
               "intervalFactor": 1,
               "legendFormat": "current",
@@ -5319,4 +5319,4 @@
   "uid": "wNf0q_kZk",
   "version": 1,
   "weekStart": ""
-}
\ No newline at end of file
+}
diff --git a/deployment/docker/alerts-cluster.yml b/deployment/docker/alerts-cluster.yml
index da36260de..f1161a48b 100644
--- a/deployment/docker/alerts-cluster.yml
+++ b/deployment/docker/alerts-cluster.yml
@@ -81,7 +81,7 @@ groups:
             Possible reasons for errors are misconfiguration, overload, network blips or unreachable components."
 
       - alert: ConcurrentFlushesHitTheLimit
-        expr: avg_over_time(vm_concurrent_addrows_current[1m]) >= vm_concurrent_addrows_capacity
+        expr: avg_over_time(vm_concurrent_insert_current[1m]) >= vm_concurrent_insert_capacity
         for: 15m
         labels:
           severity: warning
diff --git a/deployment/docker/alerts.yml b/deployment/docker/alerts.yml
index 9a773d95d..954f037b8 100644
--- a/deployment/docker/alerts.yml
+++ b/deployment/docker/alerts.yml
@@ -61,7 +61,7 @@ groups:
             Please verify if clients are sending correct requests."
 
       - alert: ConcurrentFlushesHitTheLimit
-        expr: avg_over_time(vm_concurrent_addrows_current[1m]) >= vm_concurrent_addrows_capacity
+        expr: avg_over_time(vm_concurrent_insert_current[1m]) >= vm_concurrent_insert_capacity
         for: 15m
         labels:
           severity: warning
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index f123dd2bb..f5f27a701 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -15,7 +15,12 @@ The following tip changes can be tested by building VictoriaMetrics components f
 
 ## tip
 
+**Update note 1:** This release changes the logic behind `-maxConcurrentInserts` command-line flag. Previously this flag was limiting the number of concurrent connections established from clients, which send data to VictoriaMetrics. Some of these connections could be temporarily idle. Such connections do not take significant CPU and memory resources, so there is no need in limiting their count. The new logic takes into account only those connections, which **actively** ingest new data to VictoriaMetrics and to [vmagent](https://docs.victoriametrics.com/vmagent.html). This means that the default `-maxConcurrentInserts` value should handle cases, which could require increasing the value in the previous releases. So it is recommended trying to remove the explicitly set `-maxConcurrentInserts` command-line flag after upgrading to this release and verifying whether this reduces CPU and memory usage.
+
+**Update note 2:** The `vm_concurrent_addrows_current` and `vm_concurrent_addrows_capacity` metrics [exported](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#monitoring) by `vmstorage` are replaced with `vm_concurrent_insert_current` and `vm_concurrent_insert_capacity` metrics in order to be consistent with the corresponding metrics exported by `vminsert`. Please update queries in dahsboards and alerting rules with new metric names if old metric names are used there.
+
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add support for aggregation of incoming [samples](https://docs.victoriametrics.com/keyConcepts.html#raw-samples) by time and by labels. See [these docs](https://docs.victoriametrics.com/stream-aggregation.html) and [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3460).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): reduce memory usage when scraping big number of targets without the need to enable [stream parsing mode](https://docs.victoriametrics.com/vmagent.html#stream-parsing-mode).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add support for Prometheus-compatible target discovery for [HashiCorp Nomad](https://www.nomadproject.io/) services via [nomad_sd_configs](https://docs.victoriametrics.com/sd_configs.html#nomad_sd_configs). See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3367). Thanks to @mr-karan for [the implementation](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/3549).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): automatically pre-fetch `metric_relabel_configs` and the target labels when clicking on the `debug metrics relabeling` link at the `http://vmagent:8429/targets` page at the particular target. See [these docs](https://docs.victoriametrics.com/vmagent.html#relabel-debug).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): add ability to explore metrics exported by a particular `job` / `instance`. See [these docs](https://docs.victoriametrics.com/#metrics-explorer) and [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3386).
@@ -24,6 +29,13 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * FEATURE: [relabeling](https://docs.victoriametrics.com/vmagent.html#relabeling): add support for `keepequal` and `dropequal` relabeling actions, which are supported by Prometheus starting from [v2.41.0](https://github.com/prometheus/prometheus/releases/tag/v2.41.0). These relabeling actions are almost identical to `keep_if_equal` and `drop_if_equal` relabeling actions supported by VictoriaMetrics since `v1.38.0` - see [these docs](https://docs.victoriametrics.com/vmagent.html#relabeling-enhancements) - so it is recommended sticking to `keep_if_equal` and `drop_if_equal` actions instead of switching to `keepequal` and `dropequal`.
 * FEATURE: [csvimport](https://docs.victoriametrics.com/#how-to-import-csv-data): support empty values for imported metrics. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3540).
 * FEATURE: [vmalert](httpпоs://docs.victoriametrics.com/vmalert.html): allow configuring the default number of stored rule's update states in memory via global `-rule.updateEntriesLimit` command-line flag or per-rule via rule's `update_entries_limit` configuration param. See [these docs](https://docs.victoriametrics.com/vmalert.html#rules) and [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/3556).
+* FEATURE: improve the logic benhind `-maxConcurrentInserts` command-line flag. Previously this flag was limiting the number of concurrent connections from clients, which write data to VictoriaMetrics or [vmagent](https://docs.victoriametrics.com/vmagent.html). Some of these connections could be idle for some time. These connections do not need significant amounts of CPU and memory, so there is no sense in limiting their count. The updated logic behind `-maxConcurrentInserts` limits the number of **active** insert requests, not counting idle connections.
+* FEATURE: [VictoriaMetrics cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html): add `-maxConcurrentInserts` and `-insert.maxQueueDuration` command-line flags to `vmstorage`, so they could be tuned if needed in the same way as at `vminsert` nodes.
+* FEATURE: [VictoriaMetrics cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html): limit the number of concurrently executed requests at `vmstorage` proportionally to the number of available CPU cores, since every request can saturate a single CPU core at `vmstorage`. Previously a single `vmstorage` could accept and start processing arbitrary number of concurrent requests received from big number of `vmselect` nodes. This could result in increased RAM, CPU and disk IO usage or event to out of memory crash at `vmstorage` side under high load. The limit can be fine-tuned if needed via `-search.maxConcurrentRequests` command-line flag at `vmstorage` according to [these docs](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#resource-usage-limits). `vmstorage` now [exposes](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#monitoring) the following additional metrics at `http://vmstorage:8482/metrics` page:
+  - `vm_vmselect_concurrent_requests_capacity` - the maximum number of requests allowed to execute concurrently
+  - `vm_vmselect_concurrent_requests_current` - the current number of concurrently executed requests
+  - `vm_vmselect_concurrent_requests_limit_reached_total` - the total number of requests, which were put in the wait queue when `-search.maxConcurrentRequests` concurrent requests are being executed
+  - `vm_vmselect_concurrent_requests_limit_timeout_total` - the total number of canceled requests because they were sitting in the wait queue for more than `-search.maxQueueDuration`
 
 * BUGFIX: [vmui](https://docs.victoriametrics.com/#vmui): properly update the `step` value in url after the `step` input field has been manually changed. This allows preserving the proper `step` when copy-n-pasting the url to another instance of web browser. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3513).
 * BUGFIX: [vmui](https://docs.victoriametrics.com/#vmui): properly update tooltip when quickly hovering multiple lines on the graph. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3530).
@@ -32,6 +44,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): fix a panic during target discovery when `vmagent` runs with `-promscrape.dropOriginalLabels` command-line flag. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3580). The bug has been introduced in [v1.85.0](https://docs.victoriametrics.com/CHANGELOG.html#v1850).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): [dockerswarm_sd_configs](https://docs.victoriametrics.com/sd_configs.html#dockerswarm_sd_configs): properly encode `filters` field. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3579).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): fix possible resource leak after hot reload of the updated [consul_sd_configs](https://docs.victoriametrics.com/sd_configs.html#consul_sd_configs). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3468).
+* BUGFIX: properly return label names starting from uppercase such as `CamelCaseLabel` from [/api/v1/labels](https://docs.victoriametrics.com/url-examples.html#apiv1labels). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3566).
 
 
 ## [v1.85.3](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.85.3)
diff --git a/docs/Cluster-VictoriaMetrics.md b/docs/Cluster-VictoriaMetrics.md
index a22f59132..1976119fc 100644
--- a/docs/Cluster-VictoriaMetrics.md
+++ b/docs/Cluster-VictoriaMetrics.md
@@ -523,22 +523,61 @@ See also [resource usage limits docs](#resource-usage-limits).
 
 ## Resource usage limits
 
-By default cluster components of VictoriaMetrics are tuned for an optimal resource usage under typical workloads. Some workloads may need fine-grained resource usage limits. In these cases the following command-line flags may be useful:
-
-- `-memory.allowedPercent` and `-memory.allowedBytes` limit the amounts of memory, which may be used for various internal caches at all the cluster components of VictoriaMetrics - `vminsert`, `vmselect` and `vmstorage`. Note that VictoriaMetrics components may use more memory, since these flags don't limit additional memory, which may be needed on a per-query basis.
-- `-search.maxMemoryPerQuery` limits the amounts of memory, which can be used for processing a single query at `vmselect` node. Queries, which need more memory, are rejected. Heavy queries, which select big number of time series, may exceed the per-query memory limit by a small percent. The total memory limit for concurrently executed queries can be estimated as `-search.maxMemoryPerQuery` multiplied by `-search.maxConcurrentRequests`.
-- `-search.maxUniqueTimeseries` at `vmselect` component limits the number of unique time series a single query can find and process. `vmselect` passes the limit to `vmstorage` component, which keeps in memory some metainformation about the time series located by each query and spends some CPU time for processing the found time series. This means that the maximum memory usage and CPU usage a single query can use at `vmstorage` is proportional to `-search.maxUniqueTimeseries`.
-- `-search.maxQueryDuration` at `vmselect` limits the duration of a single query. If the query takes longer than the given duration, then it is canceled. This allows saving CPU and RAM at `vmselect` and `vmstorage` when executing unexpected heavy queries.
-- `-search.maxConcurrentRequests` at `vmselect` limits the number of concurrent requests a single `vmselect` node can process. Bigger number of concurrent requests usually means bigger memory usage at both `vmselect` and `vmstorage`. For example, if a single query needs 100 MiB of additional memory during its execution, then 100 concurrent queries may need `100 * 100 MiB = 10 GiB` of additional memory. So it is better to limit the number of concurrent queries, while suspending additional incoming queries if the concurrency limit is reached. `vmselect` provides `-search.maxQueueDuration` command-line flag for limiting the max wait time for suspended queries. See also `-search.maxMemoryPerQuery` command-line flag.
-- `-search.maxSamplesPerSeries` at `vmselect` limits the number of raw samples the query can process per each time series. `vmselect` sequentially processes raw samples per each found time series during the query. It unpacks raw samples on the selected time range per each time series into memory and then applies the given [rollup function](https://docs.victoriametrics.com/MetricsQL.html#rollup-functions). The `-search.maxSamplesPerSeries` command-line flag allows limiting memory usage at `vmselect` in the case when the query is executed on a time range, which contains hundreds of millions of raw samples per each located time series.
+By default cluster components of VictoriaMetrics are tuned for an optimal resource usage under typical workloads.
+Some workloads may need fine-grained resource usage limits. In these cases the following command-line flags may be useful:
+
+- `-memory.allowedPercent` and `-memory.allowedBytes` limit the amounts of memory, which may be used for various internal caches
+  at all the cluster components of VictoriaMetrics - `vminsert`, `vmselect` and `vmstorage`. Note that VictoriaMetrics components may use more memory,
+  since these flags don't limit additional memory, which may be needed on a per-query basis.
+- `-search.maxMemoryPerQuery` limits the amounts of memory, which can be used for processing a single query at `vmselect` node.
+  Queries, which need more memory, are rejected. Heavy queries, which select big number of time series,
+  may exceed the per-query memory limit by a small percent. The total memory limit for concurrently executed queries can be estimated
+  as `-search.maxMemoryPerQuery` multiplied by `-search.maxConcurrentRequests`.
+- `-search.maxUniqueTimeseries` at `vmselect` component limits the number of unique time series a single query can find and process.
+  `vmselect` passes the limit to `vmstorage` component, which keeps in memory some metainformation about the time series located
+  by each query and spends some CPU time for processing the found time series. This means that the maximum memory usage and CPU usage
+  a single query can use at `vmstorage` is proportional to `-search.maxUniqueTimeseries`.
+- `-search.maxQueryDuration` at `vmselect` limits the duration of a single query. If the query takes longer than the given duration, then it is canceled.
+  This allows saving CPU and RAM at `vmselect` and `vmstorage` when executing unexpectedly heavy queries.
+- `-search.maxConcurrentRequests` at `vmselect` and `vmstorage` limits the number of concurrent requests a single `vmselect` / `vmstorage` node can process.
+  Bigger number of concurrent requests usually require bigger amounts of memory at both `vmselect` and `vmstorage`.
+  For example, if a single query needs 100 MiB of additional memory during its execution, then 100 concurrent queries
+  may need `100 * 100 MiB = 10 GiB` of additional memory. So it is better to limit the number of concurrent queries,
+  while suspending additional incoming queries if the concurrency limit is reached.
+  `vmselect` and `vmstorage` provides `-search.maxQueueDuration` command-line flag for limiting the maximum wait time for suspended queries.
+  See also `-search.maxMemoryPerQuery` command-line flag at `vmselect`.
+- `-search.maxQueueDuration` at `vmselect` and `vmstorage` limits the maximum duration queries may wait for execution when `-search.maxConcurrentRequests`
+  concurrent queries are executed.
+- `-search.maxSamplesPerSeries` at `vmselect` limits the number of raw samples the query can process per each time series.
+  `vmselect` processes raw samples sequentially per each found time series during the query. It unpacks raw samples on the selected time range
+  per each time series into memory and then applies the given [rollup function](https://docs.victoriametrics.com/MetricsQL.html#rollup-functions).
+  The `-search.maxSamplesPerSeries` command-line flag allows limiting memory usage at `vmselect` in the case when the query is executed on a time range,
+  which contains hundreds of millions of raw samples per each located time series.
 - `-search.maxSamplesPerQuery` at `vmselect` limits the number of raw samples a single query can process. This allows limiting CPU usage at `vmselect` for heavy queries.
-- `-search.maxPointsPerTimeseries` limits the number of calculated points, which can be returned per each matching time series from [range query](https://docs.victoriametrics.com/keyConcepts.html#range-query).
-- `-search.maxPointsSubqueryPerTimeseries` limits the number of calculated points, which can be generated per each matching time series during [subquery](https://docs.victoriametrics.com/MetricsQL.html#subqueries) evaluation.
-- `-search.maxSeries` at `vmselect` limits the number of time series, which may be returned from [/api/v1/series](https://prometheus.io/docs/prometheus/latest/querying/api/#finding-series-by-label-matchers). This endpoint is used mostly by Grafana for auto-completion of metric names, label names and label values. Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect` when the database contains big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate). In this case it might be useful to set the `-search.maxSeries` to quite low value in order limit CPU and memory usage.
-- `-search.maxTagKeys` at `vmstorage` limits the number of items, which may be returned from [/api/v1/labels](https://prometheus.io/docs/prometheus/latest/querying/api/#getting-label-names). This endpoint is used mostly by Grafana for auto-completion of label names. Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect` when the database contains big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate). In this case it might be useful to set the `-search.maxTagKeys` to quite low value in order to limit CPU and memory usage.
-- `-search.maxTagValues` at `vmstorage` limits the number of items, which may be returned from [/api/v1/label/.../values](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values). This endpoint is used mostly by Grafana for auto-completion of label values. Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect` when the database contains big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate). In this case it might be useful to set the `-search.maxTagValues` to quite low value in order to limit CPU and memory usage.
-- `-storage.maxDailySeries` at `vmstorage` can be used for limiting the number of time series seen per day aka [time series churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate). See [cardinality limiter docs](#cardinality-limiter).
-- `-storage.maxHourlySeries` at `vmstorage` can be used for limiting the number of [active time series](https://docs.victoriametrics.com/FAQ.html#what-is-an-active-time-series). See [cardinality limiter docs](#cardinality-limiter).
+- `-search.maxPointsPerTimeseries` limits the number of calculated points, which can be returned per each
+  matching time series from [range query](https://docs.victoriametrics.com/keyConcepts.html#range-query).
+- `-search.maxPointsSubqueryPerTimeseries` limits the number of calculated points, which can be generated
+  per each matching time series during [subquery](https://docs.victoriametrics.com/MetricsQL.html#subqueries) evaluation.
+- `-search.maxSeries` at `vmselect` limits the number of time series, which may be returned from
+  [/api/v1/series](https://prometheus.io/docs/prometheus/latest/querying/api/#finding-series-by-label-matchers).
+  This endpoint is used mostly by Grafana for auto-completion of metric names, label names and label values.
+  Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect` when the database contains
+  big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate).
+  In this case it might be useful to set the `-search.maxSeries` to quite low value in order limit CPU and memory usage.
+- `-search.maxTagKeys` at `vmstorage` limits the number of items, which may be returned from
+  [/api/v1/labels](https://prometheus.io/docs/prometheus/latest/querying/api/#getting-label-names). This endpoint is used mostly by Grafana
+  for auto-completion of label names. Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect`
+  when the database contains big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate).
+  In this case it might be useful to set the `-search.maxTagKeys` to quite low value in order to limit CPU and memory usage.
+- `-search.maxTagValues` at `vmstorage` limits the number of items, which may be returned from
+  [/api/v1/label/.../values](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values). This endpoint is used mostly by Grafana
+  for auto-completion of label values. Queries to this endpoint may take big amounts of CPU time and memory at `vmstorage` and `vmselect`
+  when the database contains big number of unique time series because of [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate).
+  In this case it might be useful to set the `-search.maxTagValues` to quite low value in order to limit CPU and memory usage.
+- `-storage.maxDailySeries` at `vmstorage` can be used for limiting the number of time series seen per day aka
+  [time series churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate). See [cardinality limiter docs](#cardinality-limiter).
+- `-storage.maxHourlySeries` at `vmstorage` can be used for limiting the number of [active time series](https://docs.victoriametrics.com/FAQ.html#what-is-an-active-time-series).
+  See [cardinality limiter docs](#cardinality-limiter).
 
 See also [capacity planning docs](#capacity-planning) and [cardinality limiter in vmagent](https://docs.victoriametrics.com/vmagent.html#cardinality-limiter).
 
@@ -821,7 +860,7 @@ Below is the output for `/path/to/vminsert -help`:
   -influxTrimTimestamp duration
      Trim timestamps for InfluxDB line protocol data to this duration. Minimum practical duration is 1ms. Higher duration (i.e. 1s) may be used for reducing disk space usage for timestamp data (default 1ms)
   -insert.maxQueueDuration duration
-     The maximum duration for waiting in the queue for insert requests due to -maxConcurrentInserts (default 1m0s)
+     The maximum duration to wait in the queue when -maxConcurrentInserts concurrent insert requests are executed (default 1m0s)
   -loggerDisableTimestamps
      Whether to disable writing timestamps in logs
   -loggerErrorsPerSecondLimit int
@@ -839,7 +878,7 @@ Below is the output for `/path/to/vminsert -help`:
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
@@ -932,6 +971,10 @@ Below is the output for `/path/to/vmselect -help`:
      Path to client-side TLS key file to use when connecting to -storageNode if -cluster.tls flag is set. See https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#mtls-protection . This flag is available only in enterprise version of VictoriaMetrics
   -clusternative.disableCompression
      Whether to disable compression of the data sent to vmselect via -clusternativeListenAddr. This reduces CPU usage at the cost of higher network bandwidth usage
+  -clusternative.maxConcurrentRequests int
+     The maximum number of concurrent vmselect requests the server can process at -clusternativeListenAddr. It shouldn't be high, since a single request usually saturates a CPU core at the underlying vmstorage nodes, and many concurrently executed requests may require high amounts of memory. See also -clusternative.maxQueueDuration (default 8)
+  -clusternative.maxQueueDuration duration
+     The maximum time the incoming query to -clusternativeListenAddr waits for execution when -clusternative.maxConcurrentRequests limit is reached (default 10s)
   -clusternative.maxTagKeys int
      The maximum number of tag keys returned per search at -clusternativeListenAddr (default 100000)
   -clusternative.maxTagValueSuffixesPerSearch int
@@ -1197,6 +1240,8 @@ Below is the output for `/path/to/vmstorage -help`:
      Username for HTTP Basic Auth. The authentication is disabled if empty. See also -httpAuth.password
   -httpListenAddr string
      Address to listen for http connections (default ":8482")
+  -insert.maxQueueDuration duration
+     The maximum duration to wait in the queue when -maxConcurrentInserts concurrent insert requests are executed (default 1m0s)
   -logNewSeries
      Whether to log new series. This option is for debug purposes only. It can lead to performance issues when big number of new series are ingested into VictoriaMetrics
   -loggerDisableTimestamps
@@ -1215,6 +1260,8 @@ Below is the output for `/path/to/vmstorage -help`:
      Timezone to use for timestamps in logs. Timezone must be a valid IANA Time Zone. For example: America/New_York, Europe/Berlin, Etc/GMT+3 or Local (default "UTC")
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
+  -maxConcurrentInserts int
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -memory.allowedBytes size
      Allowed size of system memory VictoriaMetrics caches may occupy. This option overrides -memory.allowedPercent if set to a non-zero value. Too low a value may increase the cache miss rate usually resulting in higher CPU and disk IO usage. Too high a value may evict too much data from OS page cache resulting in higher disk IO usage
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 0)
@@ -1246,6 +1293,10 @@ Below is the output for `/path/to/vmstorage -help`:
      The offset for performing indexdb rotation. If set to 0, then the indexdb rotation is performed at 4am UTC time per each -retentionPeriod. If set to 2h, then the indexdb rotation is performed at 4am EET time (the timezone with +2h offset)
   -rpc.disableCompression
      Whether to disable compression of the data sent from vmstorage to vmselect. This reduces CPU usage at the cost of higher network bandwidth usage
+  -search.maxConcurrentRequests int
+     The maximum number of concurrent vmselect requests the vmstorage can process at -vmselectAddr. It shouldn't be high, since a single request usually saturates a CPU core, and many concurrently executed requests may require high amounts of memory. See also -search.maxQueueDuration (default 8)
+  -search.maxQueueDuration duration
+     The maximum time the incoming vmselect request waits for execution when -search.maxConcurrentRequests limit is reached (default 10s)
   -search.maxTagKeys int
      The maximum number of tag keys returned per search (default 100000)
   -search.maxTagValueSuffixesPerSearch int
diff --git a/docs/README.md b/docs/README.md
index 24151b7d6..b86483327 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -2236,7 +2236,7 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 5006c924e..b8f6d47d6 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -2239,7 +2239,7 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 34b1f64fd..fbd61c471 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -1251,7 +1251,7 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
   -loggerWarnsPerSecondLimit int
      Per-second limit on the number of WARN messages. If more than the given number of warns are emitted per second, then the remaining warns are suppressed. Zero values disable the rate limit
   -maxConcurrentInserts int
-     The maximum number of concurrent inserts. Default value should work for most cases, since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration (default 16)
+     The maximum number of concurrent insert requests. Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. See also -insert.maxQueueDuration (default 8)
   -maxInsertRequestSize size
      The maximum size in bytes of a single Prometheus remote_write API request
      Supports the following optional suffixes for size values: KB, MB, GB, TB, KiB, MiB, GiB, TiB (default 33554432)
diff --git a/lib/promscrape/discoveryutils/client.go b/lib/promscrape/discoveryutils/client.go
index 74bab4f3c..a8c69c85c 100644
--- a/lib/promscrape/discoveryutils/client.go
+++ b/lib/promscrape/discoveryutils/client.go
@@ -207,7 +207,7 @@ func (c *Client) getAPIResponseWithParamsAndClient(client *HTTPClient, path stri
 		return nil, fmt.Errorf("cannot parse %q: %w", requestURL, err)
 	}
 
-	deadline := time.Now().Add(client.WriteTimeout)
+	deadline := time.Now().Add(client.ReadTimeout)
 	ctx, cancel := context.WithDeadline(c.clientCtx, deadline)
 	defer cancel()
 	req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil)
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index e18855e8b..f88cb48c6 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -3,7 +3,6 @@ package promscrape
 import (
 	"flag"
 	"fmt"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"io"
 	"math"
 	"math/bits"
@@ -11,8 +10,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
@@ -416,6 +417,24 @@ func (sw *scrapeWork) scrapeInternal(scrapeTimestamp, realTimestamp int64) error
 	body := leveledbytebufferpool.Get(sw.prevBodyLen)
 	var err error
 	body.B, err = sw.ReadData(body.B[:0])
+	releaseBody, err := sw.processScrapedData(scrapeTimestamp, realTimestamp, body, err)
+	if releaseBody {
+		leveledbytebufferpool.Put(body)
+	}
+	return err
+}
+
+var concurrencyLimitCh = make(chan struct{}, cgroup.AvailableCPUs())
+
+func (sw *scrapeWork) processScrapedData(scrapeTimestamp, realTimestamp int64, body *bytesutil.ByteBuffer, err error) (bool, error) {
+	// This function is CPU-bound, while it may allocate big amounts of memory.
+	// That's why it is a good idea to limit the number of concurrent calls to this function
+	// in order to limit memory usage under high load without sacrificing the performance.
+	concurrencyLimitCh <- struct{}{}
+	defer func() {
+		<-concurrencyLimitCh
+	}()
+
 	endTimestamp := time.Now().UnixNano() / 1e6
 	duration := float64(endTimestamp-realTimestamp) / 1e3
 	scrapeDuration.Update(duration)
@@ -489,13 +508,8 @@ func (sw *scrapeWork) scrapeInternal(scrapeTimestamp, realTimestamp int64) error
 		sw.storeLastScrape(body.B)
 	}
 	sw.finalizeLastScrape()
-	if !mustSwitchToStreamParse {
-		// Return body to the pool only if its size is smaller than -promscrape.minResponseSizeForStreamParse
-		// This should reduce memory usage when scraping targets which return big responses.
-		leveledbytebufferpool.Put(body)
-	}
 	tsmGlobal.Update(sw, up == 1, realTimestamp, int64(duration*1000), samplesScraped, err)
-	return err
+	return !mustSwitchToStreamParse, err
 }
 
 func (sw *scrapeWork) pushData(at *auth.Token, wr *prompbmarshal.WriteRequest) {
diff --git a/lib/protoparser/csvimport/streamparser.go b/lib/protoparser/csvimport/streamparser.go
index 45a86a35d..50db7840a 100644
--- a/lib/protoparser/csvimport/streamparser.go
+++ b/lib/protoparser/csvimport/streamparser.go
@@ -12,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -26,13 +27,16 @@ var (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(req *http.Request, callback func(rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(req.Body)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r := io.Reader(wcr)
+
 	q := req.URL.Query()
 	format := q.Get("format")
 	cds, err := ParseColumnDescriptors(format)
 	if err != nil {
 		return fmt.Errorf("cannot parse the provided csv format: %w", err)
 	}
-	r := req.Body
 	if req.Header.Get("Content-Encoding") == "gzip" {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -51,6 +55,7 @@ func ParseStream(req *http.Request, callback func(rows []Row) error) error {
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/protoparser/datadog/streamparser.go b/lib/protoparser/datadog/streamparser.go
index ef1877071..bdc1ad4d6 100644
--- a/lib/protoparser/datadog/streamparser.go
+++ b/lib/protoparser/datadog/streamparser.go
@@ -13,6 +13,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -33,6 +34,10 @@ var (
 //
 // callback shouldn't hold series after returning.
 func ParseStream(r io.Reader, contentEncoding string, callback func(series []Series) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	switch contentEncoding {
 	case "gzip":
 		zr, err := common.GetGzipReader(r)
diff --git a/lib/protoparser/graphite/streamparser.go b/lib/protoparser/graphite/streamparser.go
index a4207b5df..1fb94fffd 100644
--- a/lib/protoparser/graphite/streamparser.go
+++ b/lib/protoparser/graphite/streamparser.go
@@ -12,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -26,6 +27,10 @@ var (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, callback func(rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	ctx := getStreamContext(r)
 	defer putStreamContext(ctx)
 
@@ -36,6 +41,7 @@ func ParseStream(r io.Reader, callback func(rows []Row) error) error {
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/protoparser/influx/streamparser.go b/lib/protoparser/influx/streamparser.go
index 2f21309a5..1d1a1057f 100644
--- a/lib/protoparser/influx/streamparser.go
+++ b/lib/protoparser/influx/streamparser.go
@@ -12,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -27,6 +28,10 @@ var (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, isGzipped bool, precision, db string, callback func(db string, rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -63,6 +68,7 @@ func ParseStream(r io.Reader, isGzipped bool, precision, db string, callback fun
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/protoparser/native/streamparser.go b/lib/protoparser/native/streamparser.go
index afb7d6924..e771ad554 100644
--- a/lib/protoparser/native/streamparser.go
+++ b/lib/protoparser/native/streamparser.go
@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -19,6 +20,10 @@ import (
 //
 // callback shouldn't hold block after returning.
 func ParseStream(r io.Reader, isGzip bool, callback func(block *Block) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	if isGzip {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -101,6 +106,7 @@ func ParseStream(r io.Reader, isGzip bool, callback func(block *Block) error) er
 
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 }
 
diff --git a/lib/protoparser/opentsdb/streamparser.go b/lib/protoparser/opentsdb/streamparser.go
index 94e7fc5b8..b71babdea 100644
--- a/lib/protoparser/opentsdb/streamparser.go
+++ b/lib/protoparser/opentsdb/streamparser.go
@@ -12,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -26,6 +27,10 @@ var (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, callback func(rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	ctx := getStreamContext(r)
 	defer putStreamContext(ctx)
 	for ctx.Read() {
@@ -35,6 +40,7 @@ func ParseStream(r io.Reader, callback func(rows []Row) error) error {
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/protoparser/opentsdbhttp/streamparser.go b/lib/protoparser/opentsdbhttp/streamparser.go
index edad74dd2..cfa0512a6 100644
--- a/lib/protoparser/opentsdbhttp/streamparser.go
+++ b/lib/protoparser/opentsdbhttp/streamparser.go
@@ -14,6 +14,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -29,8 +30,11 @@ var (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(req *http.Request, callback func(rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(req.Body)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r := io.Reader(req.Body)
+
 	readCalls.Inc()
-	r := req.Body
 	if req.Header.Get("Content-Encoding") == "gzip" {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
diff --git a/lib/protoparser/prometheus/streamparser.go b/lib/protoparser/prometheus/streamparser.go
index af093f82f..b7e8a2bff 100644
--- a/lib/protoparser/prometheus/streamparser.go
+++ b/lib/protoparser/prometheus/streamparser.go
@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -19,6 +20,10 @@ import (
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback func(rows []Row) error, errLogger func(string)) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -38,6 +43,7 @@ func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback f
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/protoparser/promremotewrite/streamparser.go b/lib/protoparser/promremotewrite/streamparser.go
index 12c118c0a..44a507260 100644
--- a/lib/protoparser/promremotewrite/streamparser.go
+++ b/lib/protoparser/promremotewrite/streamparser.go
@@ -11,6 +11,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/golang/snappy"
 )
@@ -21,6 +22,10 @@ var maxInsertRequestSize = flagutil.NewBytes("maxInsertRequestSize", 32*1024*102
 //
 // callback shouldn't hold tss after returning.
 func ParseStream(r io.Reader, callback func(tss []prompb.TimeSeries) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	ctx := getPushCtx(r)
 	defer putPushCtx(ctx)
 	if err := ctx.Read(); err != nil {
diff --git a/lib/protoparser/vmimport/streamparser.go b/lib/protoparser/vmimport/streamparser.go
index bc5a1570d..3d4ba19d6 100644
--- a/lib/protoparser/vmimport/streamparser.go
+++ b/lib/protoparser/vmimport/streamparser.go
@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -22,6 +23,10 @@ var maxLineLen = flagutil.NewBytes("import.maxLineLen", 100*1024*1024, "The maxi
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, isGzipped bool, callback func(rows []Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
 	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -39,6 +44,7 @@ func ParseStream(r io.Reader, isGzipped bool, callback func(rows []Row) error) e
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
 		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
 	}
 	ctx.wg.Wait()
 	if err := ctx.Error(); err != nil {
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index 8b59bbcd0..cdaaece1a 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -829,6 +829,7 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
 	mp := &is.mp
 	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
+	underscoreNameSeen := false
 	nsPrefixExpected := byte(nsPrefixDateTagToMetricIDs)
 	if date == 0 {
 		nsPrefixExpected = nsPrefixTagToMetricIDs
@@ -855,7 +856,7 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
 		}
 		labelName := mp.Tag.Key
 		if len(labelName) == 0 {
-			labelName = []byte("__name__")
+			underscoreNameSeen = true
 		}
 		if isArtificialTagKey(labelName) || string(labelName) == string(prevLabelName) {
 			// Search for the next tag key.
@@ -875,6 +876,9 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
 		lns[string(labelName)] = struct{}{}
 		prevLabelName = append(prevLabelName[:0], labelName...)
 	}
+	if underscoreNameSeen {
+		lns["__name__"] = struct{}{}
+	}
 	if err := ts.Error(); err != nil {
 		return fmt.Errorf("error during search for prefix %q: %w", prefix, err)
 	}
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index a3e8f19aa..f6d633467 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -18,7 +18,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
@@ -28,7 +27,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/snapshot"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storagepacelimiter"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/uint64set"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 	"github.com/VictoriaMetrics/fastcache"
@@ -46,10 +44,6 @@ type Storage struct {
 	tooSmallTimestampRows uint64
 	tooBigTimestampRows   uint64
 
-	addRowsConcurrencyLimitReached uint64
-	addRowsConcurrencyLimitTimeout uint64
-	addRowsConcurrencyDroppedRows  uint64
-
 	slowRowInserts         uint64
 	slowPerDayIndexInserts uint64
 	slowMetricNameLoads    uint64
@@ -451,12 +445,6 @@ type Metrics struct {
 	TooSmallTimestampRows uint64
 	TooBigTimestampRows   uint64
 
-	AddRowsConcurrencyLimitReached uint64
-	AddRowsConcurrencyLimitTimeout uint64
-	AddRowsConcurrencyDroppedRows  uint64
-	AddRowsConcurrencyCapacity     uint64
-	AddRowsConcurrencyCurrent      uint64
-
 	SearchDelays uint64
 
 	SlowRowInserts         uint64
@@ -528,12 +516,6 @@ func (s *Storage) UpdateMetrics(m *Metrics) {
 	m.TooSmallTimestampRows += atomic.LoadUint64(&s.tooSmallTimestampRows)
 	m.TooBigTimestampRows += atomic.LoadUint64(&s.tooBigTimestampRows)
 
-	m.AddRowsConcurrencyLimitReached += atomic.LoadUint64(&s.addRowsConcurrencyLimitReached)
-	m.AddRowsConcurrencyLimitTimeout += atomic.LoadUint64(&s.addRowsConcurrencyLimitTimeout)
-	m.AddRowsConcurrencyDroppedRows += atomic.LoadUint64(&s.addRowsConcurrencyDroppedRows)
-	m.AddRowsConcurrencyCapacity = uint64(cap(addRowsConcurrencyCh))
-	m.AddRowsConcurrencyCurrent = uint64(len(addRowsConcurrencyCh))
-
 	m.SearchDelays = storagepacelimiter.Search.DelaysTotal()
 
 	m.SlowRowInserts += atomic.LoadUint64(&s.slowRowInserts)
@@ -1521,38 +1503,14 @@ func (s *Storage) ForceMergePartitions(partitionNamePrefix string) error {
 var rowsAddedTotal uint64
 
 // AddRows adds the given mrs to s.
+//
+// The caller should limit the number of concurrent AddRows calls to the number
+// of available CPU cores in order to limit memory usage.
 func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 	if len(mrs) == 0 {
 		return nil
 	}
 
-	// Limit the number of concurrent goroutines that may add rows to the storage.
-	// This should prevent from out of memory errors and CPU thrashing when too many
-	// goroutines call AddRows.
-	select {
-	case addRowsConcurrencyCh <- struct{}{}:
-	default:
-		// Sleep for a while until giving up
-		atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1)
-		t := timerpool.Get(addRowsTimeout)
-
-		// Prioritize data ingestion over concurrent searches.
-		storagepacelimiter.Search.Inc()
-
-		select {
-		case addRowsConcurrencyCh <- struct{}{}:
-			timerpool.Put(t)
-			storagepacelimiter.Search.Dec()
-		case <-t.C:
-			timerpool.Put(t)
-			storagepacelimiter.Search.Dec()
-			atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1)
-			atomic.AddUint64(&s.addRowsConcurrencyDroppedRows, uint64(len(mrs)))
-			return fmt.Errorf("cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers; add more CPUs or reduce load",
-				len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh))
-		}
-	}
-
 	// Add rows to the storage in blocks with limited size in order to reduce memory usage.
 	var firstErr error
 	ic := getMetricRowsInsertCtx()
@@ -1575,8 +1533,6 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 	}
 	putMetricRowsInsertCtx(ic)
 
-	<-addRowsConcurrencyCh
-
 	return firstErr
 }
 
@@ -1608,14 +1564,6 @@ var metricRowsInsertCtxPool sync.Pool
 
 const maxMetricRowsPerBlock = 8000
 
-var (
-	// Limit the concurrency for data ingestion to GOMAXPROCS, since this operation
-	// is CPU bound, so there is no sense in running more than GOMAXPROCS concurrent
-	// goroutines on data ingestion path.
-	addRowsConcurrencyCh = make(chan struct{}, cgroup.AvailableCPUs())
-	addRowsTimeout       = 30 * time.Second
-)
-
 // RegisterMetricNames registers all the metric names from mns in the indexdb, so they can be queried later.
 //
 // The the MetricRow.Timestamp is used for registering the metric name starting from the given timestamp.
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index d963b86ff..5fcf4be7e 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -345,10 +346,18 @@ func (a *aggregator) runFlusher(interval time.Duration) {
 			return
 		case <-t.C:
 		}
+
+		// Globally limit the concurrency for metrics' flush
+		// in order to limit memory usage when big number of aggregators
+		// are flushed at the same time.
+		flushConcurrencyCh <- struct{}{}
 		a.flush()
+		<-flushConcurrencyCh
 	}
 }
 
+var flushConcurrencyCh = make(chan struct{}, cgroup.AvailableCPUs())
+
 func (a *aggregator) flush() {
 	ctx := &flushCtx{
 		suffix: a.suffix,
@@ -373,7 +382,7 @@ func (a *aggregator) flush() {
 			tss = dst
 		}
 
-		// Push the output metrics
+		// Push the output metrics.
 		a.pushFunc(tss)
 	}
 }
diff --git a/lib/writeconcurrencylimiter/concurrencylimiter.go b/lib/writeconcurrencylimiter/concurrencylimiter.go
index 0237f5ba1..eaf6fbc18 100644
--- a/lib/writeconcurrencylimiter/concurrencylimiter.go
+++ b/lib/writeconcurrencylimiter/concurrencylimiter.go
@@ -3,7 +3,9 @@ package writeconcurrencylimiter
 import (
 	"flag"
 	"fmt"
+	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
@@ -13,62 +15,120 @@ import (
 )
 
 var (
-	maxConcurrentInserts = flag.Int("maxConcurrentInserts", cgroup.AvailableCPUs()*4, "The maximum number of concurrent inserts. Default value should work for most cases, "+
-		"since it minimizes the overhead for concurrent inserts. This option is tigthly coupled with -insert.maxQueueDuration")
-	maxQueueDuration = flag.Duration("insert.maxQueueDuration", time.Minute, "The maximum duration for waiting in the queue for insert requests due to -maxConcurrentInserts")
+	maxConcurrentInserts = flag.Int("maxConcurrentInserts", 2*cgroup.AvailableCPUs(), "The maximum number of concurrent insert requests. "+
+		"Default value should work for most cases, since it minimizes the memory usage. The default value can be increased when clients send data over slow networks. "+
+		"See also -insert.maxQueueDuration")
+	maxQueueDuration = flag.Duration("insert.maxQueueDuration", time.Minute, "The maximum duration to wait in the queue when -maxConcurrentInserts "+
+		"concurrent insert requests are executed")
 )
 
-// ch is the channel for limiting concurrent calls to Do.
-var ch chan struct{}
+// Reader is a reader, which increases the concurrency after the first Read() call
+//
+// The concurrency can be reduced by calling DecConcurrency().
+// Then the concurrency is increased after the next Read() call.
+type Reader struct {
+	r                    io.Reader
+	increasedConcurrency bool
+}
+
+// GetReader returns the Reader for r.
+//
+// The PutReader() must be called when the returned Reader is no longer needed.
+func GetReader(r io.Reader) *Reader {
+	v := readerPool.Get()
+	if v == nil {
+		return &Reader{
+			r: r,
+		}
+	}
+	rr := v.(*Reader)
+	rr.r = r
+	return rr
+}
 
-// Init initializes concurrencylimiter.
+// PutReader returns the r to the pool.
 //
-// Init must be called after flag.Parse call.
-func Init() {
-	ch = make(chan struct{}, *maxConcurrentInserts)
+// It decreases the concurrency if r has increased concurrency.
+func PutReader(r *Reader) {
+	r.DecConcurrency()
+	r.r = nil
+	readerPool.Put(r)
+}
+
+var readerPool sync.Pool
+
+// Read implements io.Reader.
+//
+// It increases concurrency after the first call or after the next call after DecConcurrency() call.
+func (r *Reader) Read(p []byte) (int, error) {
+	n, err := r.r.Read(p)
+	if !r.increasedConcurrency {
+		if !incConcurrency() {
+			err = &httpserver.ErrorWithStatusCode{
+				Err: fmt.Errorf("cannot process insert request for %.3f seconds because %d concurrent insert requests are 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
+	}
+	return n, err
 }
 
-// Do calls f with the limited concurrency.
-func Do(f func() error) error {
-	// Limit the number of conurrent f calls in order to prevent from excess
-	// memory usage and CPU thrashing.
+// DecConcurrency decreases the concurrency, so it could be increased again after the next Read() call.
+func (r *Reader) DecConcurrency() {
+	if r.increasedConcurrency {
+		decConcurrency()
+		r.increasedConcurrency = false
+	}
+}
+
+func initConcurrencyLimitCh() {
+	concurrencyLimitCh = make(chan struct{}, *maxConcurrentInserts)
+}
+
+var (
+	concurrencyLimitCh     chan struct{}
+	concurrencyLimitChOnce sync.Once
+)
+
+func incConcurrency() bool {
+	concurrencyLimitChOnce.Do(initConcurrencyLimitCh)
+
 	select {
-	case ch <- struct{}{}:
-		err := f()
-		<-ch
-		return err
+	case concurrencyLimitCh <- struct{}{}:
+		return true
 	default:
 	}
 
-	// All the workers are busy.
-	// Sleep for up to *maxQueueDuration.
 	concurrencyLimitReached.Inc()
 	t := timerpool.Get(*maxQueueDuration)
 	select {
-	case ch <- struct{}{}:
+	case concurrencyLimitCh <- struct{}{}:
 		timerpool.Put(t)
-		err := f()
-		<-ch
-		return err
+		return true
 	case <-t.C:
 		timerpool.Put(t)
 		concurrencyLimitTimeout.Inc()
-		return &httpserver.ErrorWithStatusCode{
-			Err: fmt.Errorf("cannot handle more than %d concurrent inserts during %s; possible solutions: "+
-				"increase `-insert.maxQueueDuration`, increase `-maxConcurrentInserts`, increase server capacity", *maxConcurrentInserts, *maxQueueDuration),
-			StatusCode: http.StatusServiceUnavailable,
-		}
+		return false
 	}
 }
 
+func decConcurrency() {
+	<-concurrencyLimitCh
+}
+
 var (
 	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_insert_limit_reached_total`)
 	concurrencyLimitTimeout = metrics.NewCounter(`vm_concurrent_insert_limit_timeout_total`)
 
 	_ = metrics.NewGauge(`vm_concurrent_insert_capacity`, func() float64 {
-		return float64(cap(ch))
+		return float64(cap(concurrencyLimitCh))
 	})
 	_ = metrics.NewGauge(`vm_concurrent_insert_current`, func() float64 {
-		return float64(len(ch))
+		return float64(len(concurrencyLimitCh))
 	})
 )


Test patch:
diff --git a/lib/storage/index_db_test.go b/lib/storage/index_db_test.go
index db467213e..23189ed88 100644
--- a/lib/storage/index_db_test.go
+++ b/lib/storage/index_db_test.go
@@ -1585,7 +1585,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	perDayMetricIDs := make(map[uint64]*uint64set.Set)
 	var allMetricIDs uint64set.Set
 	labelNames := []string{
-		"__name__", "constant", "day", "uniqueid",
+		"__name__", "constant", "day", "UniqueId", "some_unique_id",
 	}
 	labelValues := []string{
 		"testMetric",
@@ -1606,9 +1606,13 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 				fmt.Sprintf("%v", day),
 			)
 			mn.AddTag(
-				"uniqueid",
+				"UniqueId",
 				fmt.Sprintf("%v", metric),
 			)
+			mn.AddTag(
+				"some_unique_id",
+				fmt.Sprintf("%v", day),
+			)
 			mn.sortTags()
 
 			metricNameBuf = mn.Marshal(metricNameBuf[:0])
@@ -1775,6 +1779,10 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 		t.Fatalf("unexpected SeriesCountByMetricName;\ngot\n%v\nwant\n%v", status.SeriesCountByMetricName, expectedSeriesCountByMetricName)
 	}
 	expectedSeriesCountByLabelName := []TopHeapEntry{
+		{
+			Name:  "UniqueId",
+			Count: 1000,
+		},
 		{
 			Name:  "__name__",
 			Count: 1000,
@@ -1788,7 +1796,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 			Count: 1000,
 		},
 		{
-			Name:  "uniqueid",
+			Name:  "some_unique_id",
 			Count: 1000,
 		},
 	}
@@ -1806,7 +1814,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	}
 	expectedLabelValueCountByLabelName := []TopHeapEntry{
 		{
-			Name:  "uniqueid",
+			Name:  "UniqueId",
 			Count: 1000,
 		},
 		{
@@ -1821,6 +1829,10 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 			Name:  "day",
 			Count: 1,
 		},
+		{
+			Name:  "some_unique_id",
+			Count: 1,
+		},
 	}
 	if !reflect.DeepEqual(status.LabelValueCountByLabelName, expectedLabelValueCountByLabelName) {
 		t.Fatalf("unexpected LabelValueCountByLabelName;\ngot\n%v\nwant\n%v", status.LabelValueCountByLabelName, expectedLabelValueCountByLabelName)
@@ -1839,11 +1851,11 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 			Count: 1000,
 		},
 		{
-			Name:  "uniqueid=0",
-			Count: 1,
+			Name:  "some_unique_id=0",
+			Count: 1000,
 		},
 		{
-			Name:  "uniqueid=1",
+			Name:  "UniqueId=1",
 			Count: 1,
 		},
 	}
@@ -1854,7 +1866,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	if status.TotalSeries != expectedTotalSeries {
 		t.Fatalf("unexpected TotalSeries; got %d; want %d", status.TotalSeries, expectedTotalSeries)
 	}
-	expectedLabelValuePairs := uint64(4000)
+	expectedLabelValuePairs := uint64(5000)
 	if status.TotalLabelValuePairs != expectedLabelValuePairs {
 		t.Fatalf("unexpected TotalLabelValuePairs; got %d; want %d", status.TotalLabelValuePairs, expectedLabelValuePairs)
 	}
@@ -1884,7 +1896,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	if status.TotalSeries != expectedTotalSeries {
 		t.Fatalf("unexpected TotalSeries; got %d; want %d", status.TotalSeries, expectedTotalSeries)
 	}
-	expectedLabelValuePairs = 4000
+	expectedLabelValuePairs = 5000
 	if status.TotalLabelValuePairs != expectedLabelValuePairs {
 		t.Fatalf("unexpected TotalLabelValuePairs; got %d; want %d", status.TotalLabelValuePairs, expectedLabelValuePairs)
 	}
@@ -1910,7 +1922,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	if status.TotalSeries != expectedTotalSeries {
 		t.Fatalf("unexpected TotalSeries; got %d; want %d", status.TotalSeries, expectedTotalSeries)
 	}
-	expectedLabelValuePairs = 20000
+	expectedLabelValuePairs = 25000
 	if status.TotalLabelValuePairs != expectedLabelValuePairs {
 		t.Fatalf("unexpected TotalLabelValuePairs; got %d; want %d", status.TotalLabelValuePairs, expectedLabelValuePairs)
 	}
@@ -1942,7 +1954,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 
 	// Check GetTSDBStatus with non-nil filter, which matches only 3 series
 	tfs = NewTagFilters()
-	if err := tfs.Add([]byte("uniqueid"), []byte("0|1|3"), false, true); err != nil {
+	if err := tfs.Add([]byte("UniqueId"), []byte("0|1|3"), false, true); err != nil {
 		t.Fatalf("cannot add filter: %s", err)
 	}
 	status, err = db.GetTSDBStatus(nil, []*TagFilters{tfs}, baseDate, "", 5, 1e6, noDeadline)
@@ -1965,7 +1977,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	if status.TotalSeries != expectedTotalSeries {
 		t.Fatalf("unexpected TotalSeries; got %d; want %d", status.TotalSeries, expectedTotalSeries)
 	}
-	expectedLabelValuePairs = 12
+	expectedLabelValuePairs = 15
 	if status.TotalLabelValuePairs != expectedLabelValuePairs {
 		t.Fatalf("unexpected TotalLabelValuePairs; got %d; want %d", status.TotalLabelValuePairs, expectedLabelValuePairs)
 	}
@@ -1991,7 +2003,7 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	if status.TotalSeries != expectedTotalSeries {
 		t.Fatalf("unexpected TotalSeries; got %d; want %d", status.TotalSeries, expectedTotalSeries)
 	}
-	expectedLabelValuePairs = 60
+	expectedLabelValuePairs = 75
 	if status.TotalLabelValuePairs != expectedLabelValuePairs {
 		t.Fatalf("unexpected TotalLabelValuePairs; got %d; want %d", status.TotalLabelValuePairs, expectedLabelValuePairs)
 	}