func SetLimitReaderStderr(n int64) RunOption {
	return func(q *query) {
		q.limitReadErr = n
	}
}
SetLimitReaderStderr sets limit for stderr reader.
package vssh
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"log"
	"net"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"golang.org/x/crypto/ssh"
)
var (
	defaultMaxSessions  uint8  = 3
	maxErrRecent        uint64 = 10
	maxEstablishedRetry        = 20
	actionQueueSize            = 1000
	initNumProc                = 1000
	resetErrRecentDur          = time.Duration(300) * time.Second
	reConnDur                  = time.Duration(10) * time.Second
	errSSHConfig = errors.New("ssh config can not be nil")
	errNotExist  = errors.New("not exist")
)
type VSSH struct {
	clients clients
	logger  *log.Logger
	stats   stats
	mode    bool
	bufPool sync.Pool
	actionQ chan task
	procSig chan struct{}
	procCtl chan struct{}
}
type stats struct {
	queries   uint64
	connects  uint64
	processes uint64
}
type task interface {
	run(v *VSSH)
}
type ClientOption func(c *clientAttr)
type RunOption func(q *query)
func New() *VSSH {
	return &VSSH{
		clients: newClients(),
		actionQ: make(chan task, actionQueueSize),
		logger:  log.New(os.Stdout, "vssh: ", log.Lshortfile),
		procSig: make(chan struct{}, 1),
		procCtl: make(chan struct{}, 1),
		bufPool: sync.Pool{
			New: func() interface{} { return new(bytes.Buffer) },
		},
		mode: false,
	}
}
func (v *VSSH) OnDemand() *VSSH {
	v.mode = true
	return v
}
func (v *VSSH) AddClient(addr string, config *ssh.ClientConfig, opts ...ClientOption) error {
	client := &clientAttr{
		addr:        addr,
		config:      config,
		maxSessions: defaultMaxSessions,
		logger:      v.logger,
		pty: pty{
			enabled: true,
			term:    "xterm",
			modes: ssh.TerminalModes{
				ssh.ECHO:          0,
				ssh.TTY_OP_ISPEED: 14400,
				ssh.TTY_OP_OSPEED: 14400,
			},
			wide:   80,
			height: 40,
		},
	}
	for _, opt := range opts {
		opt(client)
	}
	if err := clientValidation(client); err != nil {
		return err
	}
	v.clients.add(client)
	if !v.mode {
		v.actionQ <- &connect{client}
	}
	return nil
}
func SetMaxSessions(n int) ClientOption {
	return func(c *clientAttr) {
		c.maxSessions = uint8(n)
	}
}
func RequestPty(term string, h, w uint, modes ssh.TerminalModes) ClientOption {
	return func(c *clientAttr) {
		c.pty = pty{
			enabled: true,
			term:    term,
			modes:   modes,
			wide:    w,
			height:  h,
		}
	}
}
func DisableRequestPty() ClientOption {
	return func(c *clientAttr) {
		c.pty.enabled = false
	}
}
func SetLabels(labels map[string]string) ClientOption {
	return func(c *clientAttr) {
		c.labels = labels
	}
}
func clientValidation(c *clientAttr) error {
	if c.config == nil {
		return errSSHConfig
	}
	_, _, err := net.SplitHostPort(c.addr)
	if err != nil {
		return err
	}
	return nil
}
func (v *VSSH) Start() *VSSH {
	ctx := context.Background()
	go v.process(ctx)
	go v.reConnect(ctx)
	for i := 0; i < initNumProc; i++ {
		v.procCtl <- struct{}{}
	}
	return v
}
func (v *VSSH) StartWithContext(ctx context.Context) *VSSH {
	go v.process(ctx)
	go v.reConnect(ctx)
	for i := 0; i < initNumProc; i++ {
		v.procCtl <- struct{}{}
	}
	return v
}
func (v *VSSH) process(ctx context.Context) {
	for {
		go func() {
			for {
				select {
				case a := <-v.actionQ:
					switch b := a.(type) {
					case *connect:
						atomic.AddUint64(&v.stats.connects, 1)
						b.run(v)
					case *query:
						atomic.AddUint64(&v.stats.queries, 1)
						b.run(v)
					}
				case <-v.procSig:
					atomic.AddUint64(&v.stats.processes, ^uint64(0))
					return
				case <-ctx.Done():
					return
				}
			}
		}()
		<-v.procCtl
		v.stats.processes++
	}
}
func (v *VSSH) IncreaseProc(n ...int) {
	num := 1
	if len(n) > 0 {
		num = n[0]
	}
	for i := 0; i < num; i++ {
		v.procCtl <- struct{}{}
	}
}
func (v *VSSH) DecreaseProc(n ...int) {
	num := 1
	if len(n) > 0 {
		num = n[0]
	}
	for i := 0; i < num; i++ {
		v.procSig <- struct{}{}
	}
}
func (v *VSSH) CurrentProc() uint64 {
	return v.stats.processes
}
func (v *VSSH) SetInitNumProc(n int) {
	initNumProc = n
}
func (v *VSSH) Run(ctx context.Context, cmd string, timeout time.Duration, opts ...RunOption) chan *Response {
	respChan := make(chan *Response, 100)
	q := &query{
		ctx:         ctx,
		cmd:         cmd,
		respChan:    respChan,
		respTimeout: timeout,
	}
	for _, opt := range opts {
		opt(q)
	}
	v.actionQ <- q
	return respChan
}
func (v *VSSH) RunWithLabel(ctx context.Context, cmd, queryStmt string, timeout time.Duration, opts ...RunOption) (chan *Response, error) {
	vis, err := parseExpr(queryStmt)
	if err != nil {
		return nil, err
	}
	respChan := make(chan *Response, 100)
	q := &query{
		ctx:           ctx,
		cmd:           cmd,
		stmt:          queryStmt,
		compiledQuery: vis,
		respChan:      respChan,
		respTimeout:   timeout,
	}
	for _, opt := range opts {
		opt(q)
	}
	v.actionQ <- q
	return respChan, nil
}
func SetLimitReaderStdout(n int64) RunOption {
	return func(q *query) {
		q.limitReadOut = n
	}
}