1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
package main
import (
"os"
"sync"
"time"
)
func getBatchSize(parallel int) (int, error) {
batchSize := parallel
if batchSize < 1 {
// Detect the number of parallel actions
cpus, err := os.ReadDir("/sys/bus/cpu/devices")
if err != nil {
return -1, err
}
batchSize = len(cpus)
}
return batchSize, nil
}
func processBatch(count int, batchSize int, process func(index int, wg *sync.WaitGroup)) time.Duration {
batches := count / batchSize
remainder := count % batchSize
processed := 0
wg := sync.WaitGroup{}
nextStat := batchSize
logf("Batch processing start")
timeStart := time.Now()
for i := 0; i < batches; i++ {
for j := 0; j < batchSize; j++ {
wg.Add(1)
go process(processed, &wg)
processed++
}
wg.Wait()
if processed >= nextStat {
interval := time.Since(timeStart).Seconds()
logf("Processed %d containers in %.3fs (%.3f/s)", processed, interval, float64(processed)/interval)
nextStat = nextStat * 2
}
}
for k := 0; k < remainder; k++ {
wg.Add(1)
go process(processed, &wg)
processed++
}
wg.Wait()
timeEnd := time.Now()
duration := timeEnd.Sub(timeStart)
logf("Batch processing completed in %.3fs", duration.Seconds())
return duration
}
|
