enhance: optimize plan parser pool to avoid unnessary recycle (milvus…

…-io#32869)

fix milvus-io#32868
plan parser takes too much cpu on high qps,this pr try to avoid create
lexer and parser too freequent

Signed-off-by: xiaofanluan <[email protected]>

go.mod

@@ -137,6 +137,7 @@ require (
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/ianlancetaylor/cgosymbolizer v0.0.0-20221217025313-27d3c9f66b6a // indirect
+	github.com/jolestar/go-commons-pool/v2 v2.1.2 // indirect
 	github.com/jonboulle/clockwork v0.2.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/asmfmt v1.3.2 // indirect

go.sum

@@ -250,6 +250,7 @@ github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/
 github.com/fogleman/gg v1.3.0/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
 github.com/form3tech-oss/jwt-go v3.2.3+incompatible h1:7ZaBxOI7TMoYBfyA3cQHErNNyAWIKUMIwqxEtgHOs5c=
 github.com/form3tech-oss/jwt-go v3.2.3+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
+github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/frankban/quicktest v1.2.2/go.mod h1:Qh/WofXFeiAFII1aEBu529AtJo6Zg2VHscnEsbBnJ20=
 github.com/frankban/quicktest v1.7.2/go.mod h1:jaStnuzAqU1AJdCO0l53JDCJrVDKcS03DbaAcR7Ks/o=
 github.com/frankban/quicktest v1.10.0/go.mod h1:ui7WezCLWMWxVWr1GETZY3smRy0G4KWq9vcPtJmFl7Y=
@@ -288,6 +289,7 @@ github.com/go-latex/latex v0.0.0-20210118124228-b3d85cf34e07/go.mod h1:CO1AlKB2C
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
+github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
 github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -488,6 +490,8 @@ github.com/jhump/goprotoc v0.5.0/go.mod h1:VrbvcYrQOrTi3i0Vf+m+oqQWk9l72mjkJCYo7
 github.com/jhump/protoreflect v1.11.0/go.mod h1:U7aMIjN0NWq9swDP7xDdoMfRHb35uiuTd3Z9nFXJf5E=
 github.com/jhump/protoreflect v1.12.0/go.mod h1:JytZfP5d0r8pVNLZvai7U/MCuTWITgrI4tTg7puQFKI=
 github.com/jmespath/go-jmespath v0.3.0/go.mod h1:9QtRXoHjLGCJ5IBSaohpXITPlowMeeYCZ7fLUTSywik=
+github.com/jolestar/go-commons-pool/v2 v2.1.2 h1:E+XGo58F23t7HtZiC/W6jzO2Ux2IccSH/yx4nD+J1CM=
+github.com/jolestar/go-commons-pool/v2 v2.1.2/go.mod h1:r4NYccrkS5UqP1YQI1COyTZ9UjPJAAGTUxzcsK1kqhY=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
 github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
@@ -512,6 +516,7 @@ github.com/kataras/iris/v12 v12.1.8/go.mod h1:LMYy4VlP67TQ3Zgriz8RE2h2kMZV2SgMYb
 github.com/kataras/neffos v0.0.14/go.mod h1:8lqADm8PnbeFfL7CLXh1WHw53dG27MC3pgi2R1rmoTE=
 github.com/kataras/pio v0.0.2/go.mod h1:hAoW0t9UmXi4R5Oyq5Z4irTbaTsOemSrDGUtaTl7Dro=
 github.com/kataras/sitemap v0.0.5/go.mod h1:KY2eugMKiPwsJgx7+U103YZehfvNGOXURubcGyk0Bz8=
+github.com/keybase/go-keychain v0.0.0-20190712205309-48d3d31d256d/go.mod h1:JJNrCn9otv/2QP4D7SMJBgaleKpOf66PnW6F5WGNRIc=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=

internal/parser/planparserv2/pool.go

@@ -1,28 +1,50 @@
 package planparserv2
 
 import (
-	"sync"
+	"context"
 
 	"github.com/antlr/antlr4/runtime/Go/antlr"
+	pool "github.com/jolestar/go-commons-pool/v2"
 
 	antlrparser "github.com/milvus-io/milvus/internal/parser/planparserv2/generated"
+	"github.com/milvus-io/milvus/pkg/util/hardware"
 )
 
 var (
-	lexerPool = sync.Pool{
-		New: func() interface{} {
-			return antlrparser.NewPlanLexer(nil)
-		},
-	}
-	parserPool = sync.Pool{
-		New: func() interface{} {
-			return antlrparser.NewPlanParser(nil)
-		},
+	config = &pool.ObjectPoolConfig{
+		LIFO:                     pool.DefaultLIFO,
+		MaxTotal:                 hardware.GetCPUNum() * 8,
+		MaxIdle:                  hardware.GetCPUNum() * 8,
+		MinIdle:                  pool.DefaultMinIdle,
+		MinEvictableIdleTime:     pool.DefaultMinEvictableIdleTime,
+		SoftMinEvictableIdleTime: pool.DefaultSoftMinEvictableIdleTime,
+		NumTestsPerEvictionRun:   pool.DefaultNumTestsPerEvictionRun,
+		EvictionPolicyName:       pool.DefaultEvictionPolicyName,
+		EvictionContext:          context.Background(),
+		TestOnCreate:             pool.DefaultTestOnCreate,
+		TestOnBorrow:             pool.DefaultTestOnBorrow,
+		TestOnReturn:             pool.DefaultTestOnReturn,
+		TestWhileIdle:            pool.DefaultTestWhileIdle,
+		TimeBetweenEvictionRuns:  pool.DefaultTimeBetweenEvictionRuns,
+		BlockWhenExhausted:       false,
 	}
+	ctx              = context.Background()
+	lexerPoolFactory = pool.NewPooledObjectFactorySimple(
+		func(context.Context) (interface{}, error) {
+			return antlrparser.NewPlanLexer(nil), nil
+		})
+	lexerPool = pool.NewObjectPool(ctx, lexerPoolFactory, config)
+
+	parserPoolFactory = pool.NewPooledObjectFactorySimple(
+		func(context.Context) (interface{}, error) {
+			return antlrparser.NewPlanParser(nil), nil
+		})
+	parserPool = pool.NewObjectPool(ctx, parserPoolFactory, config)
 )
 
 func getLexer(stream *antlr.InputStream, listeners ...antlr.ErrorListener) *antlrparser.PlanLexer {
-	lexer, ok := lexerPool.Get().(*antlrparser.PlanLexer)
+	cached, _ := lexerPool.BorrowObject(context.Background())
+	lexer, ok := cached.(*antlrparser.PlanLexer)
 	if !ok {
 		lexer = antlrparser.NewPlanLexer(nil)
 	}
@@ -35,7 +57,8 @@ func getLexer(stream *antlr.InputStream, listeners ...antlr.ErrorListener) *antl
 
 func getParser(lexer *antlrparser.PlanLexer, listeners ...antlr.ErrorListener) *antlrparser.PlanParser {
 	tokenStream := antlr.NewCommonTokenStream(lexer, antlr.TokenDefaultChannel)
-	parser, ok := parserPool.Get().(*antlrparser.PlanParser)
+	cached, _ := parserPool.BorrowObject(context.Background())
+	parser, ok := cached.(*antlrparser.PlanParser)
 	if !ok {
 		parser = antlrparser.NewPlanParser(nil)
 	}
@@ -49,10 +72,38 @@ func getParser(lexer *antlrparser.PlanLexer, listeners ...antlr.ErrorListener) *
 
 func putLexer(lexer *antlrparser.PlanLexer) {
 	lexer.SetInputStream(nil)
-	lexerPool.Put(lexer)
+	lexerPool.ReturnObject(context.TODO(), lexer)
 }
 
 func putParser(parser *antlrparser.PlanParser) {
 	parser.SetInputStream(nil)
-	parserPool.Put(parser)
+	parserPool.ReturnObject(context.TODO(), parser)
+}
+
+func getLexerPool() *pool.ObjectPool {
+	return lexerPool
+}
+
+// only for test
+func resetLexerPool() {
+	ctx = context.Background()
+	lexerPoolFactory = pool.NewPooledObjectFactorySimple(
+		func(context.Context) (interface{}, error) {
+			return antlrparser.NewPlanLexer(nil), nil
+		})
+	lexerPool = pool.NewObjectPool(ctx, lexerPoolFactory, config)
+}
+
+func getParserPool() *pool.ObjectPool {
+	return parserPool
+}
+
+// only for test
+func resetParserPool() {
+	ctx = context.Background()
+	parserPoolFactory = pool.NewPooledObjectFactorySimple(
+		func(context.Context) (interface{}, error) {
+			return antlrparser.NewPlanParser(nil), nil
+		})
+	parserPool = pool.NewObjectPool(ctx, parserPoolFactory, config)
 }

internal/parser/planparserv2/pool_test.go

@@ -15,18 +15,27 @@ func genNaiveInputStream() *antlr.InputStream {
 
 func Test_getLexer(t *testing.T) {
 	var lexer *antlrparser.PlanLexer
-
+	resetLexerPool()
 	lexer = getLexer(genNaiveInputStream(), &errorListener{})
 	assert.NotNil(t, lexer)
 
 	lexer = getLexer(genNaiveInputStream(), &errorListener{})
 	assert.NotNil(t, lexer)
+
+	pool := getLexerPool()
+	assert.Equal(t, pool.GetNumActive(), 2)
+	assert.Equal(t, pool.GetNumIdle(), 0)
+
+	putLexer(lexer)
+	assert.Equal(t, pool.GetNumActive(), 1)
+	assert.Equal(t, pool.GetNumIdle(), 1)
 }
 
 func Test_getParser(t *testing.T) {
 	var lexer *antlrparser.PlanLexer
 	var parser *antlrparser.PlanParser
 
+	resetParserPool()
 	lexer = getLexer(genNaiveInputStream(), &errorListener{})
 	assert.NotNil(t, lexer)
 
@@ -35,4 +44,12 @@ func Test_getParser(t *testing.T) {
 
 	parser = getParser(lexer, &errorListener{})
 	assert.NotNil(t, parser)
+
+	pool := getParserPool()
+	assert.Equal(t, pool.GetNumActive(), 2)
+	assert.Equal(t, pool.GetNumIdle(), 0)
+
+	putParser(parser)
+	assert.Equal(t, pool.GetNumActive(), 1)
+	assert.Equal(t, pool.GetNumIdle(), 1)
 }