-
Notifications
You must be signed in to change notification settings - Fork 336
/
command_content_rewrite.go
156 lines (129 loc) · 4.33 KB
/
command_content_rewrite.go
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
package cli
import (
"context"
"strings"
"sync"
"github.com/pkg/errors"
"github.com/kopia/kopia/repo"
"github.com/kopia/kopia/repo/content"
)
var (
contentRewriteCommand = contentCommands.Command("rewrite", "Rewrite content using most recent format")
contentRewriteIDs = contentRewriteCommand.Arg("contentID", "Identifiers of contents to rewrite").Strings()
contentRewritePrefixed = contentRewriteCommand.Flag("prefixed", "Rewrite contents with a non-empty prefix").Bool()
contentRewriteParallelism = contentRewriteCommand.Flag("parallelism", "Number of parallel workers").Default("16").Int()
contentRewriteShortPacks = contentRewriteCommand.Flag("short", "Rewrite contents from short packs").Bool()
contentRewriteFormatVersion = contentRewriteCommand.Flag("format-version", "Rewrite contents using the provided format version").Default("-1").Int()
contentRewritePackPrefix = contentRewriteCommand.Flag("pack-prefix", "Only rewrite contents from pack blobs with a given prefix").String()
contentRewriteDryRun = contentRewriteCommand.Flag("dry-run", "Do not actually rewrite, only print what would happen").Short('n').Bool()
)
type contentInfoOrError struct {
content.Info
err error
}
func runContentRewriteCommand(ctx context.Context, rep *repo.Repository) error {
cnt := getContentToRewrite(ctx, rep)
var (
mu sync.Mutex
totalBytes int64
failedCount int
)
var wg sync.WaitGroup
for i := 0; i < *contentRewriteParallelism; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for c := range cnt {
if c.err != nil {
log.Errorf("got error: %v", c.err)
mu.Lock()
failedCount++
mu.Unlock()
return
}
var optDeleted string
if c.Deleted {
optDeleted = " (deleted)"
}
printStderr("Rewriting content %v (%v bytes) from pack %v%v\n", c.ID, c.Length, c.PackBlobID, optDeleted)
mu.Lock()
totalBytes += int64(c.Length)
mu.Unlock()
if *contentRewriteDryRun {
continue
}
if err := rep.Content.RewriteContent(ctx, c.ID); err != nil {
log.Warningf("unable to rewrite content %q: %v", c.ID, err)
mu.Lock()
failedCount++
mu.Unlock()
}
}
}()
}
wg.Wait()
printStderr("Total bytes rewritten %v\n", totalBytes)
if failedCount == 0 {
return nil
}
return errors.Errorf("failed to rewrite %v contents", failedCount)
}
func getContentToRewrite(ctx context.Context, rep *repo.Repository) <-chan contentInfoOrError {
ch := make(chan contentInfoOrError)
go func() {
defer close(ch)
// get content IDs listed on command line
findContentInfos(ctx, rep, ch, toContentIDs(*contentRewriteIDs))
// add all content IDs from short packs
if *contentRewriteShortPacks {
threshold := int64(rep.Content.Format.MaxPackSize * 6 / 10)
findContentInShortPacks(rep, ch, threshold)
}
// add all blocks with given format version
if *contentRewriteFormatVersion != -1 {
findContentWithFormatVersion(rep, ch, *contentRewriteFormatVersion)
}
}()
return ch
}
func toContentIDs(s []string) []content.ID {
var result []content.ID
for _, cid := range s {
result = append(result, content.ID(cid))
}
return result
}
func findContentInfos(ctx context.Context, rep *repo.Repository, ch chan contentInfoOrError, contentIDs []content.ID) {
for _, contentID := range contentIDs {
i, err := rep.Content.ContentInfo(ctx, contentID)
if err != nil {
ch <- contentInfoOrError{err: errors.Wrapf(err, "unable to get info for content %q", contentID)}
} else {
ch <- contentInfoOrError{Info: i}
}
}
}
func findContentWithFormatVersion(rep *repo.Repository, ch chan contentInfoOrError, version int) {
_ = rep.Content.IterateContents(
content.IterateOptions{IncludeDeleted: true},
func(b content.Info) error {
if int(b.FormatVersion) == version && strings.HasPrefix(string(b.PackBlobID), *contentRewritePackPrefix) {
ch <- contentInfoOrError{Info: b}
}
return nil
})
}
func findContentInShortPacks(rep *repo.Repository, ch chan contentInfoOrError, threshold int64) {
if err := rep.Content.IterateContentInShortPacks(threshold, func(ci content.Info) error {
if ci.ID.HasPrefix() == *contentRewritePrefixed {
ch <- contentInfoOrError{Info: ci}
}
return nil
}); err != nil {
ch <- contentInfoOrError{err: err}
return
}
}
func init() {
contentRewriteCommand.Action(repositoryAction(runContentRewriteCommand))
}