Set channel work pool size in datanode (#27728)

Signed-off-by: lixinguo <xinguo.li@zilliz.com>
Co-authored-by: lixinguo <xinguo.li@zilliz.com>

configs/milvus.yaml

@@ -418,6 +418,11 @@ dataNode:
     watermarkCluster: 0.5 # memory watermark for cluster, upon reaching this watermark, segments will be synced.
   timetick:
     byRPC: true
+  channel:
+    # specify the size of global work pool of all channels
+    # if this parameter <= 0, will set it as the maximum number of CPUs that can be executing
+    # suggest to set it bigger on large collection numbers to avoid blocking
+    workPoolSize: -1
 
 # Configures the system log output.
 log:

internal/datanode/io_pool.go

@@ -32,7 +32,11 @@ func getOrCreateIOPool() *conc.Pool[any] {
 }
 
 func initStatsPool() {
-	statsPool = conc.NewPool[any](runtime.GOMAXPROCS(0), conc.WithPreAlloc(false), conc.WithNonBlocking(false))
+	poolSize := Params.DataNodeCfg.ChannelWorkPoolSize.GetAsInt()
+	if poolSize <= 0 {
+		poolSize = runtime.GOMAXPROCS(0)
+	}
+	statsPool = conc.NewPool[any](poolSize, conc.WithPreAlloc(false), conc.WithNonBlocking(false))
 }
 
 func getOrCreateStatsPool() *conc.Pool[any] {

pkg/util/paramtable/component_param.go

@@ -2391,6 +2391,9 @@ type dataNodeConfig struct {
 
 	// Skip BF
 	SkipBFStatsLoad ParamItem `refreshable:"true"`
+
+	// channel
+	ChannelWorkPoolSize ParamItem `refreshable:"true"`
 }
 
 func (p *dataNodeConfig) init(base *BaseTable) {
@@ -2545,6 +2548,14 @@ func (p *dataNodeConfig) init(base *BaseTable) {
 		DefaultValue: "18000",
 	}
 	p.BulkInsertTimeoutSeconds.Init(base.mgr)
+
+	p.ChannelWorkPoolSize = ParamItem{
+		Key:          "datanode.channel.workPoolSize",
+		Version:      "2.3.2",
+		PanicIfEmpty: false,
+		DefaultValue: "-1",
+	}
+	p.ChannelWorkPoolSize.Init(base.mgr)
 }
 
 // /////////////////////////////////////////////////////////////////////////////

pkg/util/paramtable/component_param_test.go

@@ -380,6 +380,10 @@ func TestComponentParam(t *testing.T) {
 		bulkinsertTimeout := &Params.BulkInsertTimeoutSeconds
 		t.Logf("BulkInsertTimeoutSeconds: %v", bulkinsertTimeout)
 		assert.Equal(t, "18000", Params.BulkInsertTimeoutSeconds.GetValue())
+
+		channelWorkPoolSize := Params.ChannelWorkPoolSize.GetAsInt()
+		t.Logf("channelWorkPoolSize: %d", channelWorkPoolSize)
+		assert.Equal(t, -1, Params.ChannelWorkPoolSize.GetAsInt())
 	})
 
 	t.Run("test indexNodeConfig", func(t *testing.T) {