gc: output all errors to a channel
Errors from ColoredSet are now reported as encountered and errors encountered when deleting blocks are no longer ignored. License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org> gc: report errors from ColoredSet as encountered License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org>
Kevin Atkinson committed
Feb 20, 2017 at 18:00 UTC
f1938f3cbd0f4af0fa4761db96175285613006ae
4 files changed
+144
-69
core/commands/repo.go
+27
-7
@@ -58,19 +58,39 @@ order to reclaim hard disk space.
58
return
59
}
60
61
- gcOutChan, err := corerepo.GarbageCollectAsync(n, req.Context())
62
- if err != nil {
63
- res.SetError(err, cmds.ErrNormal)
64
- return
65
- }
61
+ gcOutChan, erro := corerepo.GarbageCollectAsync(n, req.Context())
62
63
outChan := make(chan interface{})
64
res.SetOutput((<-chan interface{})(outChan))
65
66
go func() {
67
defer close(outChan)
72
- for k := range gcOutChan {
73
- outChan <- k
68
+ var errors []error
69
+ for gcOutChan != nil || erro != nil {
70
+ select {
71
+ case k, ok := <-gcOutChan:
72
+ if ok {
73
+ outChan <- k
74
+ } else {
75
+ gcOutChan = nil
76
+ }
77
+ case err, ok := <-erro:
78
+ if ok {
79
+ errors = append(errors, err)
80
+ } else {
81
+ erro = nil
82
+ }
83
+ }
84
+ }
85
+ switch len(errors) {
86
+ case 0:
87
+ return
88
+ case 1:
89
+ res.SetError(errors[0], cmds.ErrNormal)
90
+ return
91
+ default:
92
+ res.SetError(corerepo.NewMultiError(errors...), cmds.ErrNormal)
93
+ return
94
}
95
}()
96
},
core/corerepo/gc.go
+47
-16
@@ -1,6 +1,7 @@
1
package corerepo
2
3
import (
4
+ "bytes"
5
"context"
6
"errors"
7
"time"
@@ -89,35 +90,65 @@ func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
90
if err != nil {
91
return err
92
}
92
- rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
93
- if err != nil {
94
- return err
95
- }
93
+ rmed, erro := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
94
97
- for {
95
+ var errors []error
96
+ for rmed != nil || erro != nil {
97
select {
98
case _, ok := <-rmed:
99
if !ok {
101
- return nil
100
+ rmed = nil
101
+ }
102
+ case err, ok := <-erro:
103
+ if ok {
104
+ errors = append(errors, err)
105
+ } else {
106
+ erro = nil
107
}
103
- case <-ctx.Done():
104
- return ctx.Err()
108
}
109
}
110
+ switch len(errors) {
111
+ case 0:
112
+ return nil
113
+ case 1:
114
+ return errors[0]
115
+ default:
116
+ return NewMultiError(errors...)
117
+ }
118
+}
119
120
+func NewMultiError(errs ...error) *MultiError {
121
+ return &MultiError{errs[:len(errs)-1], errs[len(errs)-1]}
122
}
123
110
-func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemoved, error) {
111
- roots, err := BestEffortRoots(n.FilesRoot)
112
- if err != nil {
113
- return nil, err
124
+type MultiError struct {
125
+ Errors []error
126
+ Summary error
127
+}
128
+
129
+func (e *MultiError) Error() string {
130
+ var buf bytes.Buffer
131
+ for _, err := range e.Errors {
132
+ buf.WriteString(err.Error())
133
+ buf.WriteString("\n")
134
}
115
- rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
135
+ buf.WriteString(e.Summary.Error())
136
+ return buf.String()
137
+}
138
+
139
+func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemoved, <-chan error) {
140
+ out := make(chan *KeyRemoved)
141
+
142
+ roots, err := BestEffortRoots(n.FilesRoot)
143
if err != nil {
117
- return nil, err
144
+ erro := make(chan error)
145
+ erro <- err
146
+ close(erro)
147
+ close(out)
148
+ return out, erro
149
}
150
+ rmed, erro := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
151
120
- out := make(chan *KeyRemoved)
152
go func() {
153
defer close(out)
154
for k := range rmed {
@@ -128,7 +159,7 @@ func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemo
159
}
160
}
161
}()
131
- return out, nil
162
+ return out, erro
163
}
164
165
func PeriodicGC(ctx context.Context, node *core.IpfsNode) error {
core/coreunix/add_test.go
+3
-8
@@ -59,7 +59,6 @@ func TestAddGCLive(t *testing.T) {
59
t.Fatal(err)
60
}
61
62
- errs := make(chan error)
62
out := make(chan interface{})
63
adder, err := NewAdder(context.Background(), node.Pinning, node.Blockstore, node.DAG)
64
if err != nil {
@@ -100,16 +99,12 @@ func TestAddGCLive(t *testing.T) {
99
}
100
101
var gcout <-chan *cid.Cid
102
+ var errs <-chan error
103
gcstarted := make(chan struct{})
104
go func() {
105
defer close(gcstarted)
106
- gcchan, err := gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
107
- if err != nil {
108
- log.Error("GC ERROR:", err)
109
- errs <- err
110
- return
111
- }
112
-
106
+ gcchan, erro := gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
107
+ errs = erro
108
gcout = gcchan
109
}()
110
pin/gc/gc.go
+67
-38
@@ -1,8 +1,9 @@
1
package gc
2
3
import (
4
- "bytes"
4
"context"
5
+ "errors"
6
+ "fmt"
7
8
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
9
dag "github.com/ipfs/go-ipfs/merkledag"
@@ -24,50 +25,65 @@ var log = logging.Logger("gc")
25
//
26
// The routine then iterates over every block in the blockstore and
27
// deletes any block that is not found in the marked set.
27
-func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.Pinner, bestEffortRoots []*cid.Cid) (<-chan *cid.Cid, error) {
28
+//
29
+func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.Pinner, bestEffortRoots []*cid.Cid) (<-chan *cid.Cid, <-chan error) {
30
unlocker := bs.GCLock()
29
-
31
ls = ls.GetOfflineLinkService()
32
32
- gcs, errs := ColoredSet(ctx, pn, ls, bestEffortRoots)
33
- if errs != nil {
34
- return nil, &UnsafeToContinueError{errs}
35
- }
36
-
37
- keychan, err := bs.AllKeysChan(ctx)
38
- if err != nil {
39
- return nil, err
40
- }
41
-
33
output := make(chan *cid.Cid)
34
+ errOutput := make(chan error)
35
+
36
go func() {
37
+ defer close(errOutput)
38
defer close(output)
39
defer unlocker.Unlock()
40
+
41
+ gcs, err := ColoredSet(ctx, pn, ls, bestEffortRoots, errOutput)
42
+ if err != nil {
43
+ errOutput <- err
44
+ return
45
+ }
46
+
47
+ keychan, err := bs.AllKeysChan(ctx)
48
+ if err != nil {
49
+ errOutput <- err
50
+ return
51
+ }
52
+
53
+ errors := false
54
+
55
+ loop:
56
for {
57
select {
58
case k, ok := <-keychan:
59
if !ok {
50
- return
60
+ break loop
61
}
62
if !gcs.Has(k) {
63
err := bs.DeleteBlock(k)
64
if err != nil {
55
- log.Errorf("Error removing key from blockstore: %s", err)
56
- return
65
+ errors = true
66
+ errOutput <- &CouldNotDeleteBlockError{k, err}
67
+ //log.Errorf("Error removing key from blockstore: %s", err)
68
+ // continue as error is non-fatal
69
+ continue loop
70
}
71
select {
72
case output <- k:
73
case <-ctx.Done():
61
- return
74
+ break loop
75
}
76
}
77
case <-ctx.Done():
65
- return
78
+ break loop
79
}
80
}
81
+ if errors {
82
+ errOutput <- ErrCouldNotDeleteSomeBlocks
83
+ }
84
}()
85
70
- return output, nil
86
+ return output, errOutput
87
}
88
89
func Descendants(ctx context.Context, getLinks dag.GetLinks, set *cid.Set, roots []*cid.Cid) error {
@@ -84,33 +100,37 @@ func Descendants(ctx context.Context, getLinks dag.GetLinks, set *cid.Set, roots
100
return nil
101
}
102
87
-func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffortRoots []*cid.Cid) (*cid.Set, []error) {
103
+func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffortRoots []*cid.Cid, errOutput chan<- error) (*cid.Set, error) {
104
// KeySet currently implemented in memory, in the future, may be bloom filter or
105
// disk backed to conserve memory.
106
+ errors := false
107
gcs := cid.NewSet()
91
- var errors []error
108
getLinks := func(ctx context.Context, cid *cid.Cid) ([]*node.Link, error) {
109
links, err := ls.GetLinks(ctx, cid)
110
if err != nil {
95
- errors = append(errors, err)
111
+ errors = true
112
+ errOutput <- &CouldNotFetchLinksError{cid, err}
113
}
114
return links, nil
115
}
116
err := Descendants(ctx, getLinks, gcs, pn.RecursiveKeys())
117
if err != nil {
101
- errors = append(errors, err)
118
+ errors = true
119
+ errOutput <- err
120
}
121
122
bestEffortGetLinks := func(ctx context.Context, cid *cid.Cid) ([]*node.Link, error) {
123
links, err := ls.GetLinks(ctx, cid)
124
if err != nil && err != dag.ErrNotFound {
107
- errors = append(errors, err)
125
+ errors = true
126
+ errOutput <- &CouldNotFetchLinksError{cid, err}
127
}
128
return links, nil
129
}
130
err = Descendants(ctx, bestEffortGetLinks, gcs, bestEffortRoots)
131
if err != nil {
113
- errors = append(errors, err)
132
+ errors = true
133
+ errOutput <- err
134
}
135
136
for _, k := range pn.DirectKeys() {
@@ -119,26 +139,35 @@ func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffo
139
140
err = Descendants(ctx, getLinks, gcs, pn.InternalPins())
141
if err != nil {
122
- errors = append(errors, err)
142
+ errors = true
143
+ errOutput <- err
144
}
145
125
- if errors != nil {
126
- return nil, errors
146
+ if errors {
147
+ return nil, ErrCouldNotFetchAllLinks
148
} else {
149
return gcs, nil
150
}
151
}
152
132
-type UnsafeToContinueError struct {
133
- Errors []error
153
+var ErrCouldNotFetchAllLinks = errors.New("garbage collection aborted: could not retrieve some links")
154
+
155
+var ErrCouldNotDeleteSomeBlocks = errors.New("garbage collection incomplete: could not delete some blocks")
156
+
157
+type CouldNotFetchLinksError struct {
158
+ Key *cid.Cid
159
+ Err error
160
+}
161
+
162
+func (e *CouldNotFetchLinksError) Error() string {
163
+ return fmt.Sprintf("could not retrieve links for %s: %s", e.Key, e.Err)
164
}
165
136
-func (e *UnsafeToContinueError) Error() string {
137
- var buf bytes.Buffer
138
- for _, err := range e.Errors {
139
- buf.WriteString(err.Error())
140
- buf.WriteString("\n")
141
- }
142
- buf.WriteString("aborting due to previous errors")
143
- return buf.String()
166
+type CouldNotDeleteBlockError struct {
167
+ Key *cid.Cid
168
+ Err error
169
+}
170
+
171
+func (e *CouldNotDeleteBlockError) Error() string {
172
+ return fmt.Sprintf("could not remove %s: %s", e.Key, e.Err)
173
}