func (c *conn) Commit() (err error) {
	if atomic.LoadInt32(&c.closed) != 0 {
		return driver.ErrBadConn
	}
	if !c.inTransaction {
		return sql.ErrTxDone
	}
	defer func() {
		c.queries = c.queries[:0]
		c.inTransaction = false
	}()
	if len(c.queries) > 0 {
		if _, _, _, err = c.sendQuery(context.Background(), types.WriteQuery, c.queries); err != nil {
			return
		}
	}
	return
}
Commit implements the driver.Tx.Commit method.
package client
import (
	"context"
	"database/sql"
	"database/sql/driver"
	"sync"
	"sync/atomic"
	"time"
	"github.com/pkg/errors"
	"github.com/SQLess/SQLess/crypto/asymmetric"
	"github.com/SQLess/SQLess/crypto/kms"
	"github.com/SQLess/SQLess/proto"
	"github.com/SQLess/SQLess/route"
	"github.com/SQLess/SQLess/rpc"
	"github.com/SQLess/SQLess/rpc/mux"
	"github.com/SQLess/SQLess/types"
	"github.com/SQLess/SQLess/utils/log"
	"github.com/SQLess/SQLess/utils/trace"
)
type conn struct {
	dbID proto.DatabaseID
	queries     []types.Query
	localNodeID proto.NodeID
	privKey     *asymmetric.PrivateKey
	inTransaction bool
	closed        int32
	leader   *pconn
	follower *pconn
}
type pconn struct {
	wg      *sync.WaitGroup
	parent  *conn
	ackCh   chan *types.Ack
	pCaller rpc.PCaller
}
const workerCount int = 2
func newConn(cfg *Config) (c *conn, err error) {
	var localNodeID proto.NodeID
	if localNodeID, err = kms.GetLocalNodeID(); err != nil {
		return
	}
	var privKey *asymmetric.PrivateKey
	if privKey, err = kms.GetLocalPrivateKey(); err != nil {
		return
	}
	c = &conn{
		dbID:        proto.DatabaseID(cfg.DatabaseID),
		localNodeID: localNodeID,
		privKey:     privKey,
		queries:     make([]types.Query, 0),
	}
	var peers *proto.Peers
	if peers, err = cacheGetPeers(c.dbID, c.privKey); err != nil {
		return nil, errors.WithMessage(err, "cacheGetPeers failed")
	}
	if cfg.UseLeader {
		var caller rpc.PCaller
		if cfg.UseDirectRPC {
			caller = rpc.NewPersistentCaller(peers.Leader)
		} else {
			caller = mux.NewPersistentCaller(peers.Leader)
		}
		c.leader = &pconn{
			wg:      &sync.WaitGroup{},
			ackCh:   make(chan *types.Ack, workerCount*4),
			parent:  c,
			pCaller: caller,
		}
	}
	if cfg.UseFollower && len(peers.Servers) > 1 {
		for {
			node := peers.Servers[randSource.Intn(len(peers.Servers))]
			if node != peers.Leader {
				var caller rpc.PCaller
				if cfg.UseDirectRPC {
					caller = rpc.NewPersistentCaller(node)
				} else {
					caller = mux.NewPersistentCaller(node)
				}
				c.follower = &pconn{
					wg:      &sync.WaitGroup{},
					ackCh:   make(chan *types.Ack, workerCount*4),
					parent:  c,
					pCaller: caller,
				}
				break
			}
		}
	}
	if c.leader == nil && c.follower == nil {
		return nil, errors.New("no follower peers found")
	}
	if c.leader != nil {
		if err := c.leader.startAckWorkers(); err != nil {
			return nil, errors.WithMessage(err, "leader startAckWorkers failed")
		}
	}
	if c.follower != nil {
		if err := c.follower.startAckWorkers(); err != nil {
			return nil, errors.WithMessage(err, "follower startAckWorkers failed")
		}
	}
	log.WithField("db", c.dbID).Debug("new connection to database")
	return
}
func (c *pconn) startAckWorkers() (err error) {
	for i := 0; i < workerCount; i++ {
		c.wg.Add(1)
		go c.ackWorker()
	}
	return
}
func (c *pconn) stopAckWorkers() {
	if c.ackCh != nil {
		close(c.ackCh)
	}
}
func (c *pconn) ackWorker() {
	defer c.wg.Done()
	var (
		oneTime sync.Once
		pc      rpc.PCaller
		err     error
	)
ackWorkerLoop:
	for {
		ack, got := <-c.ackCh
		if !got {  
			break ackWorkerLoop
		}
		oneTime.Do(func() {
			pc = c.pCaller.New()
		})
		if err = ack.Sign(c.parent.privKey); err != nil {
			log.WithField("target", pc.Target()).WithError(err).Error("failed to sign ack")
			continue
		}
		var ackRes types.AckResponse
		if err = pc.Call(route.DBSAck.String(), ack, &ackRes); err != nil {
			log.WithError(err).Debug("send ack failed")
			continue
		}
	}
	if pc != nil {
		pc.Close()
	}
	log.Debug("ack worker quiting")
}
func (c *pconn) close() error {
	c.stopAckWorkers()
	c.wg.Wait()
	if c.pCaller != nil {
		c.pCaller.Close()
	}
	return nil
}
func (c *conn) Prepare(query string) (driver.Stmt, error) {
	return c.PrepareContext(context.Background(), query)
}
func (c *conn) Close() error {
	if atomic.CompareAndSwapInt32(&c.closed, 0, 1) {
		log.WithField("db", c.dbID).Debug("closed connection")
	}
	if c.leader != nil {
		c.leader.close()
	}
	if c.follower != nil {
		c.follower.close()
	}
	return nil
}
func (c *conn) Begin() (driver.Tx, error) {
	return c.BeginTx(context.Background(), driver.TxOptions{})
}
func (c *conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
	if atomic.LoadInt32(&c.closed) != 0 {
		return nil, driver.ErrBadConn
	}
	log.WithField("inTx", c.inTransaction).Debug("begin transaction")
	if c.inTransaction {
		return nil, sql.ErrTxDone
	}
	c.inTransaction = true
	c.queries = c.queries[:0]
	return c, nil
}
func (c *conn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
	if atomic.LoadInt32(&c.closed) != 0 {
		return nil, driver.ErrBadConn
	}
	log.WithField("query", query).Debug("prepared statement")
	return newStmt(c, query), nil
}
func (c *conn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (result driver.Result, err error) {
	defer trace.StartRegion(ctx, "dbExec").End()
	if atomic.LoadInt32(&c.closed) != 0 {
		err = driver.ErrBadConn
		return
	}
	sq := convertQuery(query, args)
	var affectedRows, lastInsertID int64
	if affectedRows, lastInsertID, _, err = c.addQuery(ctx, types.WriteQuery, sq); err != nil {
		return
	}
	result = &execResult{
		affectedRows: affectedRows,
		lastInsertID: lastInsertID,
	}
	return
}
func (c *conn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (rows driver.Rows, err error) {
	defer trace.StartRegion(ctx, "dbQuery").End()
	if atomic.LoadInt32(&c.closed) != 0 {
		err = driver.ErrBadConn
		return
	}
	sq := convertQuery(query, args)
	_, _, rows, err = c.addQuery(ctx, types.ReadQuery, sq)
	return
}