Configurable queue
@@ -252,6 +252,10 @@ type Service struct {
KeyFile string // Path to SSL private key.
TLS1011 bool // Whether older, deprecated TLS should be supported.
+ DefaultQueueCap int
+ DefaultQueueBatchSz int
+ DefaultQueueTimeout time.Duration
+
credentialStore CredentialStore
Expvar bool
@@ -270,7 +274,9 @@ func New(addr string, store Store, cluster Cluster, credentials CredentialStore)
addr: addr,
store: store,
queueDone: make(chan struct{}),
- stmtQueue: queue.New(1024, 64, 100*time.Millisecond),
+ DefaultQueueCap: 1024,
+ DefaultQueueBatchSz: 128,
+ DefaultQueueTimeout: 100 * time.Millisecond,
cluster: cluster,
start: time.Now(),
statuses: make(map[string]StatusReporter),
@@ -305,8 +311,10 @@ func (s *Service) Start() error {
}
s.ln = ln
+ s.stmtQueue = queue.New(s.DefaultQueueCap, s.DefaultQueueBatchSz, s.DefaultQueueTimeout)
go s.runQueue()
- s.logger.Println("execute queue processing started")
+ s.logger.Printf("execute queue processing started with capacity %d, batch size %d, timeout %s",
+ s.DefaultQueueCap, s.DefaultQueueBatchSz, s.DefaultQueueTimeout.String())
go func() {
err := server.Serve(s.ln)
@@ -321,6 +329,8 @@ func (s *Service) Start() error {
// Close closes the service.
func (s *Service) Close() {
+ s.stmtQueue.Close()
+
close(s.closeCh)
<-s.queueDone
