// Wait returns nil when mysqld is up and accepting connections.
func (mysqld *Mysqld) Wait(ctx context.Context) error {
	log.Infof("Waiting for mysqld socket file (%v) to be ready...", mysqld.config.SocketFile)
	for {
		select {
		case <-ctx.Done():
			return errors.New("deadline exceeded waiting for mysqld socket file to appear: " + mysqld.config.SocketFile)
		default:
		}

		_, statErr := os.Stat(mysqld.config.SocketFile)
		if statErr == nil {
			// Make sure the socket file isn't stale.
			// Use a user that exists even before we apply the init_db_sql_file.
			conn, connErr := mysql.Connect(sqldb.ConnParams{
				Uname:      "root",
				Charset:    "utf8",
				UnixSocket: mysqld.config.SocketFile,
			})
			if connErr == nil {
				conn.Close()
				return nil
			}
			log.Infof("mysqld socket file exists, but can't connect: %v", connErr)
		} else if !os.IsNotExist(statErr) {
			return fmt.Errorf("can't stat mysqld socket file: %v", statErr)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

func (mt *Mysqld) createDbaConnection() (*mysql.Connection, error) {
	params, err := dbconfigs.MysqlParams(mt.dbaParams)
	if err != nil {
		return nil, err
	}
	return mysql.Connect(params)
}

func TestMySQL(t *testing.T) {
	hdl, err := LaunchVitess(MySQLOnly("vttest"), Schema("create table a(id int, name varchar(128), primary key(id))"))
	if err != nil {
		t.Error(err)
		return
	}
	defer func() {
		err = hdl.TearDown()
		if err != nil {
			t.Error(err)
			return
		}
	}()
	params, err := hdl.MySQLConnParams()
	if err != nil {
		t.Error(err)
	}
	conn, err := mysql.Connect(params)
	if err != nil {
		t.Error(err)
	}
	_, err = conn.ExecuteFetch("insert into a values(1, 'name')", 10, false)
	if err != nil {
		t.Error(err)
	}
	qr, err := conn.ExecuteFetch("select * from a", 10, false)
	if err != nil {
		t.Error(err)
	}
	if qr.RowsAffected != 1 {
		t.Errorf("Rows affected: %d, want 1", qr.RowsAffected)
	}
}

func TestSchemaReload(t *testing.T) {
	conn, err := mysql.Connect(connParams)
	if err != nil {
		t.Error(err)
		return
	}
	_, err = conn.ExecuteFetch("create table vitess_temp(intval int)", 10, false)
	if err != nil {
		t.Error(err)
		return
	}
	defer func() {
		_, _ = conn.ExecuteFetch("drop table vitess_temp", 10, false)
		conn.Close()
	}()
	framework.Server.ReloadSchema()
	client := framework.NewClient()
	waitTime := 50 * time.Millisecond
	for i := 0; i < 10; i++ {
		time.Sleep(waitTime)
		waitTime += 50 * time.Millisecond
		_, err = client.Execute("select * from vitess_temp", nil)
		if err == nil {
			return
		}
		want := "error: table vitess_temp not found in schema"
		if err.Error() != want {
			t.Errorf("Error: %v, want %s", err, want)
			return
		}
	}
	t.Error("schema did not reload")
}

func CreateGenericConnection(info *mysql.ConnectionParams) (*DBConnection, error) {
	params, err := dbconfigs.MysqlParams(info)
	if err != nil {
		return nil, err
	}
	c, err := mysql.Connect(params)
	return &DBConnection{c}, err
}

// NewDBConnection returns a new DBConnection based on the ConnectionParams
// and will use the provided stats to collect timing.
func NewDBConnection(info *mysql.ConnectionParams, mysqlStats *stats.Timings) (*DBConnection, error) {
	params, err := dbconfigs.MysqlParams(info)
	if err != nil {
		return nil, err
	}
	c, err := mysql.Connect(params)
	return &DBConnection{c, mysqlStats}, err
}

func (dc *DBClient) Connect() error {
	var err error
	dc.dbConn, err = mysql.Connect(*dc.dbConfig)
	if err != nil {
		return fmt.Errorf("error in connecting to mysql db, err %v", err)
	}
	return nil
}

func (dc *DBClient) Connect() error {
	params, err := dbconfigs.MysqlParams(dc.dbConfig)
	if err != nil {
		return err
	}
	dc.dbConn, err = mysql.Connect(params)
	if err != nil {
		return fmt.Errorf("error in connecting to mysql db, err %v", err)
	}
	return nil
}

// NewSlaveConnection creates a new slave connection to the mysqld instance.
// It uses a pools.IDPool to ensure that the server IDs used to connect are
// unique within this process. This is done with the assumptions that:
//
// 1) No other processes are making fake slave connections to our mysqld.
// 2) No real slave servers will have IDs in the range 1-N where N is the peak
//    number of concurrent fake slave connections we will ever make.
func NewSlaveConnection(mysqld *Mysqld) (*SlaveConnection, error) {
	params, err := dbconfigs.MysqlParams(mysqld.dba)
	if err != nil {
		return nil, err
	}

	conn, err := mysql.Connect(params)
	if err != nil {
		return nil, err
	}

	sc := &SlaveConnection{
		Connection: conn,
		mysqld:     mysqld,
		slaveID:    slaveIDPool.Get(),
	}
	log.Infof("new slave connection: slaveID=%d", sc.slaveID)
	return sc, nil
}

// setupPrimerConnections creates the channel and the consumers for
// the sql statements
func (pc *PrimeCache) setupPrimerConnections() error {
	pc.workerChannel = make(chan string, 1000)
	for i := 0; i < pc.WorkerCount; i++ {
		// connect to the database using client for a replay connection
		params, err := dbconfigs.MysqlParams(&pc.dbcfgs.App.ConnectionParams)
		if err != nil {
			return fmt.Errorf("cannot get parameters to connect to MySQL: %v", err)
		}

		dbConn, err := mysql.Connect(params)
		if err != nil {
			return fmt.Errorf("mysql.Connect failed: %v", err)
		}

		// and launch the go routine that applies the statements
		go applyLoop(dbConn, pc.workerChannel)
	}
	return nil
}

func NewMysqld(config *Mycnf, dba, repl mysql.ConnectionParams) *Mysqld {
	if dba == DefaultDbaParams {
		dba.UnixSocket = config.SocketFile
	}

	// the super connection is not linked to a specific database
	// (allows us to create them)
	superParams := dba
	superParams.DbName = ""
	createSuperConnection := func() (*mysql.Connection, error) {
		return mysql.Connect(superParams)
	}
	return &Mysqld{config,
		dba,
		repl,
		createSuperConnection,
		TabletDir(config.ServerId),
		SnapshotDir(config.ServerId),
	}
}

func TestSidecarTables(t *testing.T) {
	conn, err := mysql.Connect(connParams)
	if err != nil {
		t.Error(err)
		return
	}
	defer conn.Close()
	for _, table := range []string{
		"redo_log_transaction",
		"redo_log_statement",
		"transaction",
		"participant",
	} {
		_, err = conn.ExecuteFetch(fmt.Sprintf("describe _vt.%s", table), 10, false)
		if err != nil {
			t.Error(err)
			return
		}
	}
}

func CreateGenericConnection(info mysql.ConnectionParams) (*DBConnection, error) {
	c, err := mysql.Connect(info)
	return &DBConnection{c}, err
}

func (sq *SqlQuery) allowQueries(dbconfig *dbconfigs.DBConfig, schemaOverrides []SchemaOverride, qrs *QueryRules) {
	sq.statemu.Lock()
	v := sq.state.Get()
	switch v {
	case CONNECTING, ABORT, SERVING:
		sq.statemu.Unlock()
		log.Infof("Ignoring allowQueries request, current state: %v", v)
		return
	case INITIALIZING, SHUTTING_DOWN:
		panic("unreachable")
	}
	// state is NOT_SERVING
	sq.setState(CONNECTING)
	sq.statemu.Unlock()

	// Try connecting. disallowQueries can change the state to ABORT during this time.
	waitTime := time.Second
	for {
		params, err := dbconfigs.MysqlParams(&dbconfig.ConnectionParams)
		if err == nil {
			c, err := mysql.Connect(params)
			if err == nil {
				c.Close()
				break
			}
			log.Errorf("mysql.Connect() error: %v", err)
		} else {
			log.Errorf("dbconfigs.MysqlParams error: %v", err)
		}
		time.Sleep(waitTime)
		// Cap at 32 seconds
		if waitTime < 30*time.Second {
			waitTime = waitTime * 2
		}
		if sq.state.Get() == ABORT {
			// Exclusive transition. No need to lock statemu.
			sq.setState(NOT_SERVING)
			log.Infof("allowQueries aborting")
			return
		}
	}

	// Connection successful. Keep statemu locked.
	sq.statemu.Lock()
	defer sq.statemu.Unlock()
	if sq.state.Get() == ABORT {
		sq.setState(NOT_SERVING)
		log.Infof("allowQueries aborting")
		return
	}
	sq.setState(INITIALIZING)

	defer func() {
		if x := recover(); x != nil {
			log.Errorf("%s", x.(*TabletError).Message)
			sq.setState(NOT_SERVING)
			return
		}
		sq.setState(SERVING)
	}()

	sq.qe.Open(&dbconfig.ConnectionParams, schemaOverrides, qrs)
	sq.dbconfig = dbconfig
	sq.sessionId = Rand()
	log.Infof("Session id: %d", sq.sessionId)
}

func (sq *SqlQuery) allowQueries(dbconfig *dbconfigs.DBConfig, schemaOverrides []SchemaOverride, qrs *QueryRules, mysqld *mysqlctl.Mysqld) {
	sq.statemu.Lock()
	defer sq.statemu.Unlock()

	v := sq.state.Get()
	switch v {
	case CONNECTING, ABORT, SERVING:
		log.Infof("Ignoring allowQueries request, current state: %v", v)
		return
	case INITIALIZING, SHUTTING_DOWN:
		panic("unreachable")
	}
	// state is NOT_SERVING
	sq.setState(CONNECTING)

	// When this function exits, state can be CONNECTING or ABORT
	func() {
		sq.statemu.Unlock()
		defer sq.statemu.Lock()

		waitTime := time.Second
		// disallowQueries can change the state to ABORT during this time.
		for sq.state.Get() != ABORT {
			params, err := dbconfigs.MysqlParams(&dbconfig.ConnectionParams)
			if err == nil {
				c, err := mysql.Connect(params)
				if err == nil {
					c.Close()
					break
				}
				log.Errorf("mysql.Connect() error: %v", err)
			} else {
				log.Errorf("dbconfigs.MysqlParams error: %v", err)
			}
			time.Sleep(waitTime)
			// Cap at 32 seconds
			if waitTime < 30*time.Second {
				waitTime = waitTime * 2
			}
		}
	}()

	if sq.state.Get() == ABORT {
		sq.setState(NOT_SERVING)
		log.Infof("allowQueries aborting")
		return
	}
	sq.setState(INITIALIZING)

	defer func() {
		if x := recover(); x != nil {
			log.Errorf("%s", x.(*TabletError).Message)
			sq.qe.Close()
			sq.rci.Close()
			sq.setState(NOT_SERVING)
			return
		}
		sq.setState(SERVING)
	}()

	sq.qe.Open(dbconfig, schemaOverrides, qrs)
	if dbconfig.EnableRowcache && dbconfig.EnableInvalidator {
		sq.rci.Open(dbconfig.DbName, mysqld)
	}
	sq.dbconfig = dbconfig
	sq.sessionId = Rand()
	log.Infof("Session id: %d", sq.sessionId)
}

// OneRun tries a single cycle connecting to MySQL, and if behind on
// replication, starts playing the logs ahead to prime the cache.
func (pc *PrimeCache) OneRun() {
	// connect to the database using dba for a control connection
	params, err := dbconfigs.MysqlParams(&pc.dbcfgs.Dba)
	if err != nil {
		log.Errorf("Cannot get parameters to connect to MySQL: %v", err)
		return
	}

	pc.dbConn, err = mysql.Connect(params)
	if err != nil {
		log.Errorf("mysql.Connect failed: %v", err)
		return
	}

	// get the slave status
	slavestat, err := pc.getSlaveStatus()
	if err != nil {
		log.Errorf("getSlaveStatus failed: %v", err)
		return
	}

	// if we're not replicating, we're done
	if !slavestat.slaveSQLRunning {
		log.Errorf("Slave is not replicating (SQL)")
		return
	}
	if !slavestat.slaveIORunning {
		log.Errorf("Slave is not replicating (IO)")
		return
	}
	if slavestat.secondsBehindMaster < 2 {
		log.Errorf("Slave lag is negligible - %v seconds", slavestat.secondsBehindMaster)
		return
	}

	// setup the connections to the db to apply the statements
	if err := pc.setupPrimerConnections(); err != nil {
		log.Errorf("setupPrimerConnections failed: %v", err)
		return
	}

	// open the binlogs from where we are on
	reader, err := pc.openBinlog(slavestat)
	if err != nil {
		log.Errorf("openBinlog failed: %v", err)
		return
	}

	maxLineCount := 10000
	var maxLeadBytes int64 = 5000000

	// and start the loop
	lineCount := 0
	deleteCount := 0
	appliedDeleteCount := 0
	updateCount := 0
	appliedUpdateCount := 0
	sleepCount := 0
	var logPos uint64
	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {
		line := scanner.Text()
		lowerLine := strings.ToLower(line)
		lineCount++

		if p, isComment := parseLogPos(line); isComment {
			// handle the comments with a log pos
			if p > logPos {
				logPos = p
			}

		} else if s, isDelete := parseDeleteStatement(line, lowerLine); isDelete {
			// handle delete statements
			deleteCount++
			if s != "" {
				appliedDeleteCount++
				pc.workerChannel <- s
			}

		} else if s, isUpdate := parseUpdateStatement(line, lowerLine); isUpdate {
			// handle update statements
			updateCount++
			if s != "" {
				appliedUpdateCount++
				pc.workerChannel <- s
			}
		}

		if lineCount%maxLineCount == 0 {
			var leadBytes int64
			for {
				slavestat, err = pc.getSlaveStatus()
				if err != nil {
					log.Errorf("getSlaveStatus failed: %v", err)
					return
				}

				// see how far ahead we are (it's signed because
//.........这里部分代码省略.........

func (dc DBClient) Connect() (*mysql.Connection, error) {
	return mysql.Connect(*dc.dbConfig)
}