20
moveGroupTaskCountMultiplier float64,
21
logger log.Logger,
23
>
return &actionMoveGroup{
24
>
maxReaderCount: maxReaderCount,
25
>
grouper: grouper,
26
>
moveGroupTaskCountBase: moveGroupTaskCountBase,
27
>
moveGroupTaskCountMultiplier: moveGroupTaskCountMultiplier,
28
>
logger: logger,
29
>
}
30
>
}
31
33
>
return "move-group"
34
>
}
35
37
>
38
>
// Move task groups from reader x to x+1 if the # of pending tasks for a group is higher than
39
>
// a threshold. The threshold is calculated as:
40
>
// moveGroupTaskCountBase * (moveGroupTaskCountMultiplier ^ x)
41
>
//
42
>
// If after moving a group to reader x+1, the # of pending tasks for that group becomes higher than
43
>
// the threshold for reader x+1, it will be moved to reader x+2 in the next iteration.
44
>
45
>
// TODO: instead of moving task groups down by just one reader, directly move it to the reader level
46
>
// based on the total number of pending tasks across all readers.
47
>
48
>
moved := false
49
>
moveGroupMinTaskCount := a.moveGroupTaskCountBase
50
>
for readerID := DefaultReaderId; readerID+1 < int64(a.maxReaderCount); readerID++ {
51
>
if readerID != DefaultReaderId {
52
>
moveGroupMinTaskCount = int(float64(moveGroupMinTaskCount) * a.moveGroupTaskCountMultiplier)
53
>
}
54
56
>
if !ok {
57
continue
58
}
59
61
>
reader.WalkSlices(func(s Slice) {
62
>
for key, pendingTaskCount := range s.TaskStats().PendingPerKey {
63
>
pendingTaskPerGroup[key] += pendingTaskCount
64
>
}
65
})
66
68
>
for key, pendingTaskCount := range pendingTaskPerGroup {
69
>
if pendingTaskCount >= moveGroupMinTaskCount {
70
>
groupsToMove = append(groupsToMove, key)
71
>
a.logger.Info("Too many pending tasks, moving group to next reader",
72
>
tag.QueueReaderID(readerID),
73
>
tag.Counter(pendingTaskCount),
74
>
tag.Value(key),
75
>
)
76
>
}
77
}
78
80
continue
81
}
82
84
>
85
>
var slicesToMove []Slice
86
>
reader.SplitSlices(func(s Slice) ([]Slice, bool) {
87
>
// Technically we don't need this empty scope check, but it helps avoid
88
>
// unnecessary allocation and task movement.
89
>
scope := s.Scope()
90
>
splitScope, _ := scope.SplitByPredicate(predicateForSplit)
91
>
if splitScope.IsEmpty() {
92
>
return nil, false
93
>
}
94
96
>
slicesToMove = append(slicesToMove, split)
97
>
return []Slice{remain}, true
98
})
99
101
>
nextReader.MergeSlices(slicesToMove...)
102
>
moved = true
103
}
104
106
}