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

func (mysqld *Mysqld) StartReplicationCommands(status *proto.ReplicationStatus) ([]string, error) {
	params, err := dbconfigs.MysqlParams(mysqld.replParams)
	if err != nil {
		return nil, err
	}
	return mysqld.flavor.StartReplicationCommands(&params, status)
}

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

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
}

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

// SetMasterCommands returns the commands to run to make the provided
// host / port the master.
func (mysqld *Mysqld) SetMasterCommands(masterHost string, masterPort int) ([]string, error) {
	flavor, err := mysqld.flavor()
	if err != nil {
		return nil, fmt.Errorf("SetMasterCommands needs flavor: %v", err)
	}
	params, err := dbconfigs.MysqlParams(mysqld.replParams)
	if err != nil {
		return nil, err
	}
	return flavor.SetMasterCommands(&params, masterHost, masterPort, int(masterConnectRetry.Seconds()))
}

// StartReplicationCommands returns the commands used to start
// replication to the provided master using the provided starting
// position.  The provided MasterConnectRetry will be ignored and
// replaced by the command line parameter.
func (mysqld *Mysqld) StartReplicationCommands(status *proto.ReplicationStatus) ([]string, error) {
	flavor, err := mysqld.flavor()
	if err != nil {
		return nil, fmt.Errorf("StartReplicationCommands needs flavor: %v", err)
	}
	params, err := dbconfigs.MysqlParams(mysqld.replParams)
	if err != nil {
		return nil, err
	}
	status.MasterConnectRetry = int(masterConnectRetry.Seconds())
	return flavor.StartReplicationCommands(&params, status)
}

// 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
}

// connectForReplication create a MySQL connection ready to use for replication.
func (mysqld *Mysqld) connectForReplication() (sqldb.Conn, error) {
	params, err := dbconfigs.MysqlParams(mysqld.dba)
	if err != nil {
		return nil, err
	}

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

	// Tell the server that we understand the format of events
	// that will be used if binlog_checksum is enabled on the server.
	if _, err := conn.ExecuteFetch("SET @[email protected]@global.binlog_checksum", 0, false); err != nil {
		return nil, fmt.Errorf("failed to set @[email protected]@global.binlog_checksum: %v", err)
	}

	return conn, 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 StartReplicationCommands(mysqld *Mysqld, replState *proto.ReplicationState) ([]string, error) {
	params, err := dbconfigs.MysqlParams(mysqld.replParams)
	if err != nil {
		return nil, err
	}
	nmd := &newMasterData{
		ReplicationState: replState,
		MasterUser:       params.Uname,
		MasterPassword:   params.Pass,
	}
	cmc, err := fillStringTemplate(changeMasterCmd, nmd)
	if err != nil {
		return nil, err
	}
	if params.SslEnabled() {
		cmc += ",\n  MASTER_SSL = 1"
	}
	if params.SslCa != "" {
		cmc += ",\n  MASTER_SSL_CA = '" + params.SslCa + "'"
	}
	if params.SslCaPath != "" {
		cmc += ",\n  MASTER_SSL_CAPATH = '" + params.SslCaPath + "'"
	}
	if params.SslCert != "" {
		cmc += ",\n  MASTER_SSL_CERT = '" + params.SslCert + "'"
	}
	if params.SslKey != "" {
		cmc += ",\n  MASTER_SSL_KEY = '" + params.SslKey + "'"
	}

	return []string{
		"STOP SLAVE",
		"RESET SLAVE",
		cmc,
		"START SLAVE"}, nil
}

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
//.........这里部分代码省略.........