Clearer runQueue logic
@@ -178,6 +178,7 @@ const (
numExecutions = "executions"
numQueuedExecutions = "queued_executions"
numQueuedExecutionsOK = "queued_executions_ok"
+ numQueuedExecutionsNoLeader = "queued_executions_no_leader"
numQueuedExecutionsFailed = "queued_executions_failed"
numQueuedExecutionsWait = "queued_executions_wait"
numQueries = "queries"
@@ -213,6 +214,7 @@ func init() {
stats.Add(numExecutions, 0)
stats.Add(numQueuedExecutions, 0)
stats.Add(numQueuedExecutionsOK, 0)
+ stats.Add(numQueuedExecutionsNoLeader, 0)
stats.Add(numQueuedExecutionsFailed, 0)
stats.Add(numQueuedExecutionsWait, 0)
stats.Add(numQueries, 0)
@@ -1561,10 +1563,11 @@ func (s *Service) runQueue() {
Transaction: s.DefaultQueueTx,
},
}
- for {
+
// Nil statements are valid, as clients may want to just send
// a "checkpoint" through the queue.
if er.Request.Statements != nil {
+ for {
_, err = s.store.Execute(er)
if err != nil {
if err == store.ErrNotLeader {
@@ -1572,7 +1575,7 @@ func (s *Service) runQueue() {
if err != nil || addr == "" {
s.logger.Printf("execute queue can't find leader for sequence number %d",
req.SequenceNumber)
- stats.Add(numQueuedExecutionsFailed, 1)
+ stats.Add(numQueuedExecutionsNoLeader, 1)
time.Sleep(retryDelay)
continue
}
@@ -1580,6 +1583,7 @@ func (s *Service) runQueue() {
if err != nil {
s.logger.Printf("execute queue write failed for sequence number %d: %s",
req.SequenceNumber, err.Error())
+ stats.Add(numQueuedExecutionsFailed, 1)
time.Sleep(retryDelay)
continue
}
@@ -1587,14 +1591,14 @@ func (s *Service) runQueue() {
}
}
}
+ }
+
+ // Perform post-write processing.
s.seqNumMu.Lock()
s.seqNum = req.SequenceNumber
s.seqNumMu.Unlock()
-
req.Close()
stats.Add(numQueuedExecutionsOK, 1)
- break
- }
}
}
}
