optimize queue ack manager, in case the task IDs are not sequencial.
package history
import (
+ "sort"
"sync"
"github.com/uber-common/bark"
@@ -168,10 +169,17 @@ func (a *queueAckMgrImpl) updateQueueAckLevel() {
a.Lock()
ackLevel := a.ackLevel
+ // task ID is not sequancial, meaning there are a ton of missing chunks,
+ // so to optimize the performance, a sort is required
+ var taskIDs []int64
+ for k := range a.outstandingTasks {
+ taskIDs = append(taskIDs, k)
+ }
+ sort.Slice(taskIDs, func(i, j int) bool { return taskIDs[i] < taskIDs[j] })
+
MoveAckLevelLoop:
- for current := a.ackLevel + 1; current <= a.readLevel; current++ {
- // TODO: What happens if !ok?
- if acked, ok := a.outstandingTasks[current]; ok {
+ for _, current := range taskIDs {
+ acked := a.outstandingTasks[current]
if acked {
a.logger.Debugf("Updating ack level: %v", current)
ackLevel = current
@@ -181,7 +189,6 @@ MoveAckLevelLoop:
break MoveAckLevelLoop
}
}
- }
a.ackLevel = ackLevel
if a.isFailover && a.isReadFinished && len(a.outstandingTasks) == 0 {
