fix: merge schemaEngine connection pool into taskPool

go/vt/vttablet/tabletmanager/vreplication/external_connector.go

@@ -35,6 +35,7 @@ import (
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/queryservice"
 	"vitess.io/vitess/go/vt/vttablet/tabletconn"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/background"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/vstreamer"
@@ -92,12 +93,14 @@ func (ec *externalConnector) Get(name string) (*mysqlConnector, error) {
 	}
 	c := &mysqlConnector{}
 	c.env = tabletenv.NewEnv(config, name)
-	c.se = schema.NewEngine(c.env)
+	c.taskPool = background.NewTaskPool(c.env)
+	c.se = schema.NewEngine(c.env, c.taskPool)
 	c.vstreamer = vstreamer.NewEngine(c.env, nil, c.se, nil, "")
 	c.vstreamer.InitDBConfig("", "")
 	c.se.InitDBConfig(c.env.Config().DB.AllPrivsWithDB())
 
 	// Open
+	c.taskPool.Open()
 	if err := c.se.Open(); err != nil {
 		return nil, vterrors.Wrapf(err, "external mysqlConnector: %v", name)
 	}
@@ -114,11 +117,13 @@ type mysqlConnector struct {
 	env       tabletenv.Env
 	se        *schema.Engine
 	vstreamer *vstreamer.Engine
+	taskPool  *background.TaskPool
 }
 
 func (c *mysqlConnector) shutdown() {
 	c.vstreamer.Close()
 	c.se.Close()
+	c.taskPool.Close()
 }
 
 func (c *mysqlConnector) Open(ctx context.Context) error {

go/vt/vttablet/tabletmanager/vreplication/replica_connector.go

@@ -24,6 +24,7 @@ package vreplication
 import (
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/vt/dbconfigs"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/background"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 
 	"context"
@@ -54,13 +55,14 @@ func NewReplicaConnector(connParams *mysql.ConnParams) *ReplicaConnector {
 	config.DB = dbCfg
 	c := &ReplicaConnector{conn: connParams}
 	env := tabletenv.NewEnv(config, "source")
-	c.se = schema.NewEngine(env)
+	c.taskPool = background.NewTaskPool(env)
+	c.se = schema.NewEngine(env, c.taskPool)
 	c.se.SkipMetaCheck = true
 	c.vstreamer = vstreamer.NewEngine(env, nil, c.se, nil, "")
 	c.se.InitDBConfig(dbconfigs.New(connParams))
 
 	// Open
-
+	c.taskPool.Open()
 	c.vstreamer.Open()
 
 	return c
@@ -70,13 +72,15 @@ func NewReplicaConnector(connParams *mysql.ConnParams) *ReplicaConnector {
 
 type ReplicaConnector struct {
 	conn      *mysql.ConnParams
+	taskPool  *background.TaskPool
 	se        *schema.Engine
 	vstreamer *vstreamer.Engine
 }
 
 func (c *ReplicaConnector) shutdown() {
 	c.vstreamer.Close()
 	c.se.Close()
+	c.taskPool.Close()
 }
 
 func (c *ReplicaConnector) Open(ctx context.Context) error {

go/vt/vttablet/tabletserver/background/task_pool.go

@@ -13,7 +13,7 @@ import (
 )
 
 var backGroundTaskPoolConfig = tabletenv.ConnPoolConfig{
-	Size:               1,
+	Size:               5,
 	MaxSize:            50,
 	TimeoutSeconds:     1,
 	IdleTimeoutSeconds: 30 * 60,

go/vt/vttablet/tabletserver/messager/engine_test.go

@@ -25,6 +25,7 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/background"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 )
@@ -160,7 +161,8 @@ func newTestEngine(db *fakesqldb.DB) *Engine {
 	tsv := &fakeTabletServer{
 		Env: tabletenv.NewEnv(config, "MessagerTest"),
 	}
-	se := schema.NewEngine(tsv)
+	taskPool := background.NewTaskPool(tsv)
+	se := schema.NewEngine(tsv, taskPool)
 	te := NewEngine(tsv, se, newFakeVStreamer())
 	te.Open()
 	return te

go/vt/vttablet/tabletserver/query_engine_test.go

@@ -50,6 +50,7 @@ import (
 	"vitess.io/vitess/go/streamlog"
 	"vitess.io/vitess/go/vt/dbconfigs"
 	"vitess.io/vitess/go/vt/tableacl"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/background"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/planbuilder"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema/schematest"
@@ -60,7 +61,9 @@ func TestGetPlanPanicDuetoEmptyQuery(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
 	schematest.AddDefaultQueries(db)
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -95,7 +98,9 @@ func TestGetMessageStreamPlan(t *testing.T) {
 
 	addSchemaEngineQueries(db)
 
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -141,7 +146,9 @@ func TestQueryPlanCache(t *testing.T) {
 	db.AddQuery("select * from test_table_01 where 1 != 1", &sqltypes.Result{})
 	db.AddQuery("select * from test_table_02 where 1 != 1", &sqltypes.Result{})
 
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -178,7 +185,9 @@ func TestNoQueryPlanCache(t *testing.T) {
 	db.AddQuery("select * from test_table_01 where 1 != 1", &sqltypes.Result{})
 	db.AddQuery("select * from test_table_02 where 1 != 1", &sqltypes.Result{})
 
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -207,7 +216,9 @@ func TestNoQueryPlanCacheDirective(t *testing.T) {
 	db.AddQuery("select /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ * from test_table_01 where 1 != 1", &sqltypes.Result{})
 	db.AddQuery("select /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ * from test_table_02 where 1 != 1", &sqltypes.Result{})
 
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -232,7 +243,9 @@ func TestStatsURL(t *testing.T) {
 	schematest.AddDefaultQueries(db)
 	query := "select * from test_table_01"
 	db.AddQuery("select * from test_table_01 where 1 != 1", &sqltypes.Result{})
-	qe := newTestQueryEngine(1*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(1*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -254,24 +267,27 @@ func TestStatsURL(t *testing.T) {
 	qe.handleHTTPQueryRules(response, request)
 }
 
-func newTestQueryEngine(idleTimeout time.Duration, _ bool, dbcfgs *dbconfigs.DBConfigs) *QueryEngine {
+func newTestQueryEngine(idleTimeout time.Duration, _ bool, dbcfgs *dbconfigs.DBConfigs) (*QueryEngine, *background.TaskPool) {
 	config := tabletenv.NewDefaultConfig()
 	config.DB = dbcfgs
 	config.OltpReadPool.IdleTimeoutSeconds.Set(idleTimeout)
 	config.OlapReadPool.IdleTimeoutSeconds.Set(idleTimeout)
 	config.TxPool.IdleTimeoutSeconds.Set(idleTimeout)
 	env := tabletenv.NewEnv(config, "TabletServerTest")
-	se := schema.NewEngine(env)
+	taskPool := background.NewTaskPool(env)
+	se := schema.NewEngine(env, taskPool)
 	qe := NewQueryEngine(env, se)
 	se.InitDBConfig(dbcfgs.DbaWithDB())
-	return qe
+	return qe, taskPool
 }
 
 func runConsolidatedQuery(t *testing.T, sql string) *QueryEngine {
 	db := fakesqldb.New(t)
 	defer db.Close()
 
-	qe := newTestQueryEngine(1*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(1*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -334,7 +350,9 @@ func BenchmarkPlanCacheThroughput(b *testing.B) {
 
 	db.AddQueryPattern(".*", &sqltypes.Result{})
 
-	qe := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	qe, taskPool := newTestQueryEngine(10*time.Second, true, newDBConfigs(db))
+	taskPool.Open()
+	defer taskPool.Close()
 	qe.se.Open()
 	qe.Open()
 	defer qe.Close()
@@ -360,9 +378,12 @@ func benchmarkPlanCache(b *testing.B, db *fakesqldb.DB, lfu bool, par int) {
 	config.QueryCacheLFU = lfu
 
 	env := tabletenv.NewEnv(config, "TabletServerTest")
-	se := schema.NewEngine(env)
+	taskPool := background.NewTaskPool(env)
+	se := schema.NewEngine(env, taskPool)
 	qe := NewQueryEngine(env, se)
 
+	taskPool.Open()
+	defer taskPool.Close()
 	se.InitDBConfig(dbcfgs.DbaWithDB())
 	require.NoError(b, se.Open())
 	require.NoError(b, qe.Open())
@@ -422,9 +443,12 @@ func TestPlanCachePollution(t *testing.T) {
 	// config.LFUQueryCacheSizeBytes = 3 * 1024 * 1024
 
 	env := tabletenv.NewEnv(config, "TabletServerTest")
-	se := schema.NewEngine(env)
+	taskPool := background.NewTaskPool(env)
+	se := schema.NewEngine(env, taskPool)
 	qe := NewQueryEngine(env, se)
 
+	taskPool.Open()
+	defer taskPool.Close()
 	se.InitDBConfig(dbcfgs.DbaWithDB())
 	se.Open()
 
@@ -614,7 +638,8 @@ func TestAddQueryStats(t *testing.T) {
 			config := tabletenv.NewDefaultConfig()
 			config.DB = newDBConfigs(fakesqldb.New(t))
 			env := tabletenv.NewEnv(config, "TestAddQueryStats_"+testcase.name)
-			se := schema.NewEngine(env)
+			taskPool := background.NewTaskPool(env)
+			se := schema.NewEngine(env, taskPool)
 			qe := NewQueryEngine(env, se)
 			qe.AddStats(testcase.planType, testcase.tableName, testcase.queryCount, testcase.duration, testcase.mysqlTime, testcase.rowsAffected, testcase.rowsReturned, testcase.errorCount)
 			assert.Equal(t, testcase.expectedQueryCounts, qe.queryCounts.String())

go/vt/vttablet/tabletserver/queryz_test.go

@@ -35,7 +35,7 @@ import (
 func TestQueryzHandler(t *testing.T) {
 	resp := httptest.NewRecorder()
 	req, _ := http.NewRequest("GET", "/schemaz", nil)
-	qe := newTestQueryEngine(10*time.Second, true, &dbconfigs.DBConfigs{})
+	qe, _ := newTestQueryEngine(10*time.Second, true, &dbconfigs.DBConfigs{})
 
 	const query1 = "select name from test_table"
 	plan1 := &TabletPlan{

go/vt/vttablet/tabletserver/schema/engine.go

@@ -49,6 +49,7 @@ import (
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/background"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 
@@ -83,8 +84,8 @@ type Engine struct {
 
 	historian *historian
 
-	conns *connpool.Pool
-	ticks *timer.Timer
+	taskPool *background.TaskPool
+	ticks    *timer.Timer
 
 	// dbCreationFailed is for preventing log spam.
 	dbCreationFailed bool
@@ -96,16 +97,11 @@ type Engine struct {
 }
 
 // NewEngine creates a new Engine.
-func NewEngine(env tabletenv.Env) *Engine {
+func NewEngine(env tabletenv.Env, taskPool *background.TaskPool) *Engine {
 	reloadTime := env.Config().SchemaReloadIntervalSeconds.Get()
 	se := &Engine{
-		env: env,
-		// We need three connections: one for the reloader, one for
-		// the historian, and one for the tracker.
-		conns: connpool.NewPool(env, "SchemaEngine", tabletenv.ConnPoolConfig{
-			Size:               3,
-			IdleTimeoutSeconds: env.Config().OltpReadPool.IdleTimeoutSeconds,
-		}),
+		env:        env,
+		taskPool:   taskPool,
 		ticks:      timer.NewTimer(reloadTime),
 		reloadTime: reloadTime,
 	}
@@ -127,7 +123,7 @@ func NewEngine(env tabletenv.Env) *Engine {
 
 		schemazHandler(se.GetFullyQualifiedTables(), w, r)
 	})
-	se.historian = newHistorian(env.Config().TrackSchemaVersions, se.conns)
+	se.historian = newHistorian(env.Config().TrackSchemaVersions, se.taskPool)
 	return se
 }
 
@@ -231,17 +227,6 @@ func (se *Engine) Open() error {
 
 	ctx := tabletenv.LocalContext()
 
-	// The function we're in is supposed to be idempotent, but this conns.Open()
-	// call is not itself idempotent. Therefore, if we return for any reason
-	// without marking ourselves as open, we need to call conns.Close() so the
-	// pools aren't leaked the next time we call Open().
-	se.conns.Open(se.cp, se.cp, se.cp)
-	defer func() {
-		if !se.isOpen {
-			se.conns.Close()
-		}
-	}()
-
 	se.tables = map[sqlparser.TableSchemaAndName]*Table{
 		sqlparser.NewTableSchemaAndName("", "dual"): NewTable("dual"),
 	}
@@ -301,7 +286,6 @@ func (se *Engine) closeLocked() {
 		wg.Done()
 	}()
 	se.historian.Close()
-	se.conns.Close()
 
 	se.tables = make(map[sqlparser.TableSchemaAndName]*Table)
 	se.lastChange = 0
@@ -384,7 +368,7 @@ func (se *Engine) reload(ctx context.Context, includeStats bool) error {
 		se.SchemaReloadTimings.Record("SchemaReload", start)
 	}()
 
-	conn, err := se.conns.Get(ctx, nil)
+	conn, err := se.taskPool.BorrowConn(ctx, nil)
 	if err != nil {
 		return err
 	}
@@ -583,7 +567,7 @@ func (se *Engine) GetTableFromSchema(tableSchema string, tableName string) (*Tab
 	setting.SetQuery(fmt.Sprintf("use %s", tableSchema))
 	setting.SetResetQuery(fmt.Sprintf("use %s", se.cp.DBName()))
 
-	conn, err := se.conns.Get(ctx, &setting)
+	conn, err := se.taskPool.BorrowConn(ctx, &setting)
 	if err != nil {
 		return nil, err
 	}
@@ -738,7 +722,7 @@ func (se *Engine) GetFullyQualifiedTables() map[string]*Table {
 
 // GetConnection returns a connection from the pool
 func (se *Engine) GetConnection(ctx context.Context) (*connpool.DBConn, error) {
-	return se.conns.Get(ctx, nil)
+	return se.taskPool.BorrowConn(ctx, nil)
 }
 
 func (se *Engine) handleDebugSchema(response http.ResponseWriter, request *http.Request) {