move GC code into core/repo, and add sharness test
Jeromy committed
Jan 20, 2015 at 21:20 UTC
4de881a185447a68604876c0d612901b44d77959
4 files changed
+119
-69
core/commands/ping.go
+46
-43
@@ -4,6 +4,7 @@ import (
4
"bytes"
5
"fmt"
6
"io"
7
+ "reflect"
8
"strings"
9
"time"
10
@@ -46,6 +47,7 @@ trip latency information.
47
cmds.Text: func(res cmds.Response) (io.Reader, error) {
48
outChan, ok := res.Output().(<-chan interface{})
49
if !ok {
50
+ fmt.Println(reflect.TypeOf(res.Output()))
51
return nil, u.ErrCast()
52
}
53
@@ -103,63 +105,64 @@ trip latency information.
105
numPings = val
106
}
107
106
- outChan := make(chan interface{})
107
-
108
- go pingPeer(ctx, n, peerID, numPings, outChan)
109
-
108
+ outChan := pingPeer(ctx, n, peerID, numPings)
109
return outChan, nil
110
},
111
Type: PingResult{},
112
}
113
115
-func pingPeer(ctx context.Context, n *core.IpfsNode, pid peer.ID, numPings int, outChan chan interface{}) {
116
- defer close(outChan)
114
+func pingPeer(ctx context.Context, n *core.IpfsNode, pid peer.ID, numPings int) <-chan interface{} {
115
+ outChan := make(chan interface{})
116
+ go func() {
117
+ defer close(outChan)
118
118
- if len(n.Peerstore.Addresses(pid)) == 0 {
119
- // Make sure we can find the node in question
120
- outChan <- &PingResult{
121
- Text: fmt.Sprintf("Looking up peer %s", pid.Pretty()),
122
- }
119
+ if len(n.Peerstore.Addresses(pid)) == 0 {
120
+ // Make sure we can find the node in question
121
+ outChan <- &PingResult{
122
+ Text: fmt.Sprintf("Looking up peer %s", pid.Pretty()),
123
+ }
124
124
- ctx, _ := context.WithTimeout(ctx, kPingTimeout)
125
- p, err := n.Routing.FindPeer(ctx, pid)
126
- if err != nil {
127
- outChan <- &PingResult{Text: fmt.Sprintf("Peer lookup error: %s", err)}
128
- return
125
+ ctx, _ := context.WithTimeout(ctx, kPingTimeout)
126
+ p, err := n.Routing.FindPeer(ctx, pid)
127
+ if err != nil {
128
+ outChan <- &PingResult{Text: fmt.Sprintf("Peer lookup error: %s", err)}
129
+ return
130
+ }
131
+ n.Peerstore.AddPeerInfo(p)
132
}
130
- n.Peerstore.AddPeerInfo(p)
131
- }
133
133
- outChan <- &PingResult{Text: fmt.Sprintf("PING %s.", pid.Pretty())}
134
+ outChan <- &PingResult{Text: fmt.Sprintf("PING %s.", pid.Pretty())}
135
135
- var done bool
136
- var total time.Duration
137
- for i := 0; i < numPings && !done; i++ {
138
- select {
139
- case <-ctx.Done():
140
- done = true
141
- continue
142
- default:
143
- }
136
+ var done bool
137
+ var total time.Duration
138
+ for i := 0; i < numPings && !done; i++ {
139
+ select {
140
+ case <-ctx.Done():
141
+ done = true
142
+ continue
143
+ default:
144
+ }
145
145
- ctx, _ := context.WithTimeout(ctx, kPingTimeout)
146
- took, err := n.Routing.Ping(ctx, pid)
147
- if err != nil {
148
- log.Errorf("Ping error: %s", err)
149
- outChan <- &PingResult{Text: fmt.Sprintf("Ping error: %s", err)}
150
- break
146
+ ctx, _ := context.WithTimeout(ctx, kPingTimeout)
147
+ took, err := n.Routing.Ping(ctx, pid)
148
+ if err != nil {
149
+ log.Errorf("Ping error: %s", err)
150
+ outChan <- &PingResult{Text: fmt.Sprintf("Ping error: %s", err)}
151
+ break
152
+ }
153
+ outChan <- &PingResult{
154
+ Success: true,
155
+ Time: took,
156
+ }
157
+ total += took
158
+ time.Sleep(time.Second)
159
}
160
+ averagems := total.Seconds() * 1000 / float64(numPings)
161
outChan <- &PingResult{
153
- Success: true,
154
- Time: took,
162
+ Text: fmt.Sprintf("Average latency: %.2fms", averagems),
163
}
156
- total += took
157
- time.Sleep(time.Second)
158
- }
159
- averagems := total.Seconds() * 1000 / float64(numPings)
160
- outChan <- &PingResult{
161
- Text: fmt.Sprintf("Average latency: %.2fms", averagems),
162
- }
164
+ }()
165
+ return outChan
166
}
167
168
func ParsePeerParam(text string) (ma.Multiaddr, peer.ID, error) {
core/commands/repo.go
+7
-25
@@ -3,10 +3,11 @@ package commands
3
import (
4
"bytes"
5
"fmt"
6
+ "io"
7
+
8
cmds "github.com/jbenet/go-ipfs/commands"
7
- "github.com/jbenet/go-ipfs/core"
9
+ corerepo "github.com/jbenet/go-ipfs/core/repo"
10
u "github.com/jbenet/go-ipfs/util"
9
- "io"
11
)
12
13
var RepoCmd = &cmds.Command{
@@ -22,10 +23,6 @@ var RepoCmd = &cmds.Command{
23
},
24
}
25
25
-type KeyRemoved struct {
26
- Key u.Key
27
-}
28
-
26
var repoGcCmd = &cmds.Command{
27
Helptext: cmds.HelpText{
28
Tagline: "Perform a garbage collection sweep on the repo",
@@ -45,19 +42,17 @@ order to reclaim hard disk space.
42
return nil, err
43
}
44
48
- keychan, err := n.Blockstore.AllKeysChan(req.Context().Context, 0, 1<<16)
45
+ outChan, err := corerepo.GarbageCollectBlockstore(n, req.Context().Context)
46
if err != nil {
47
return nil, err
48
}
49
53
- outChan := make(chan interface{})
54
- go GarbageCollectBlockstore(n, keychan, outChan)
50
return outChan, nil
51
},
57
- Type: KeyRemoved{},
52
+ Type: corerepo.KeyRemoved{},
53
Marshalers: cmds.MarshalerMap{
54
cmds.Text: func(res cmds.Response) (io.Reader, error) {
60
- outChan, ok := res.Output().(chan interface{})
55
+ outChan, ok := res.Output().(<-chan interface{})
56
if !ok {
57
return nil, u.ErrCast()
58
}
@@ -68,7 +63,7 @@ order to reclaim hard disk space.
63
}
64
65
marshal := func(v interface{}) (io.Reader, error) {
71
- obj, ok := v.(*KeyRemoved)
66
+ obj, ok := v.(*corerepo.KeyRemoved)
67
if !ok {
68
return nil, u.ErrCast()
69
}
@@ -89,16 +84,3 @@ order to reclaim hard disk space.
84
},
85
},
86
}
92
-
93
-func GarbageCollectBlockstore(n *core.IpfsNode, keychan <-chan u.Key, output chan interface{}) {
94
- defer close(output)
95
- for k := range keychan {
96
- if !n.Pinning.IsPinned(k) {
97
- err := n.Blockstore.DeleteBlock(k)
98
- if err != nil {
99
- log.Errorf("Error removing key from blockstore: %s", err)
100
- }
101
- output <- &KeyRemoved{k}
102
- }
103
- }
104
-}
core/repo/gc.go
new
+49
@@ -0,0 +1,49 @@
1
+package corerepo
2
+
3
+import (
4
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
+ "github.com/jbenet/go-ipfs/core"
6
+ u "github.com/jbenet/go-ipfs/util"
7
+
8
+ eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
9
+)
10
+
11
+var log = eventlog.Logger("corerepo")
12
+
13
+type KeyRemoved struct {
14
+ Key u.Key
15
+}
16
+
17
+func GarbageCollectBlockstore(n *core.IpfsNode, ctx context.Context) (<-chan interface{}, error) {
18
+
19
+ keychan, err := n.Blockstore.AllKeysChan(ctx, 0, 1<<16)
20
+ if err != nil {
21
+ return nil, err
22
+ }
23
+
24
+ output := make(chan interface{})
25
+ go func() {
26
+ defer close(output)
27
+ for {
28
+ select {
29
+ case k, ok := <-keychan:
30
+ if !ok {
31
+ return
32
+ }
33
+ if !n.Pinning.IsPinned(k) {
34
+ err := n.Blockstore.DeleteBlock(k)
35
+ if err != nil {
36
+ log.Errorf("Error removing key from blockstore: %s", err)
37
+ }
38
+ select {
39
+ case output <- &KeyRemoved{k}:
40
+ case <-ctx.Done():
41
+ }
42
+ }
43
+ case <-ctx.Done():
44
+ return
45
+ }
46
+ }
47
+ }()
48
+ return output, nil
49
+}
test/sharness/t0080-repo.sh
+17
-1
@@ -21,7 +21,11 @@ test_expect_success "added file was pinned" '
21
ipfs pin ls -type=recursive | grep `cat hashfile`
22
'
23
24
-# TODO: run gc, then ipfs cat file, should still be there
24
+test_expect_success "'ipfs repo gc' doesnt remove file" '
25
+ ipfs repo gc
26
+ ipfs cat `cat hashfile` > out
27
+ test_cmp out afile
28
+'
29
30
test_expect_success "'ipfs pin rm' succeeds" '
31
echo unpinned `cat hashfile` > expected1
@@ -64,6 +68,18 @@ test_expect_success "remove direct pin" '
68
test_cmp expected6 actual6
69
'
70
71
+test_expect_success "'ipfs repo gc' removes file" '
72
+ echo removed `cat hashfile` > expected7
73
+ ipfs repo gc > actual7
74
+ test_cmp expected7 actual7
75
+'
76
+
77
+test_expect_success "'ipfs refs local' no longer shows file" '
78
+ echo -n "" > expected8
79
+ ipfs refs local > actual8
80
+ test_cmp expected8 actual8
81
+'
82
+
83
84
test_kill_ipfs_daemon
85