@cryptotaxi247 / kubo / commits / b12ee40ab

implement mark and sweep GC

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> dont GC blocks used by pinner License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> comment GC algo License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> add lock to blockstore to prevent GC from eating wanted blocks License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> improve FetchGraph License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> separate interfaces for blockstore and GCBlockstore License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> reintroduce indirect pinning, add enumerateChildren dag method License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Jun 23, 2015 at 16:01 UTC b12ee40abaee684979198dbcaa39b8e31267648b
18 files changed +200 -313
core/commands/add.go
+1 -6
@@ -169,7 +169,6 @@ remains to be implemented.
169 return err
170 }
171
172 - n.Pinning.RemovePinWithMode(rnk, pin.Indirect)
172 n.Pinning.PinWithMode(rnk, pin.Recursive)
173 return n.Pinning.Flush()
174 }
@@ -325,13 +324,11 @@ func add(n *core.IpfsNode, reader io.Reader, useTrickle bool, chunker string) (*
324 node, err = importer.BuildTrickleDagFromReader(
325 n.DAG,
326 chnk,
328 - importer.PinIndirectCB(n.Pinning),
327 )
328 } else {
329 node, err = importer.BuildDagFromReader(
330 n.DAG,
331 chnk,
334 - importer.PinIndirectCB(n.Pinning),
332 )
333 }
334
@@ -458,13 +455,11 @@ func (params *adder) addDir(file files.File) (*dag.Node, error) {
455 return nil, err
456 }
457
461 - k, err := params.node.DAG.Add(tree)
458 + _, err := params.node.DAG.Add(tree)
459 if err != nil {
460 return nil, err
461 }
462
466 - params.node.Pinning.PinWithMode(k, pin.Indirect)
467 -
463 return tree, nil
464 }
465
core/commands/pin.go
+33 -34
@@ -8,6 +8,7 @@ import (
8 key "github.com/ipfs/go-ipfs/blocks/key"
9 cmds "github.com/ipfs/go-ipfs/commands"
10 corerepo "github.com/ipfs/go-ipfs/core/corerepo"
11 + dag "github.com/ipfs/go-ipfs/merkledag"
12 u "github.com/ipfs/go-ipfs/util"
13 )
14
@@ -160,8 +161,16 @@ Returns a list of objects that are pinned locally.
161 By default, only recursively pinned returned, but others may be shown via the '--type' flag.
162 `,
163 LongDescription: `
164 +<<<<<<< HEAD
165 Returns a list of objects that are pinned locally.
166 By default, only recursively pinned returned, but others may be shown via the '--type' flag.
167 +
168 +Use --type=<type> to specify the type of pinned keys to list. Valid values are:
169 + * "direct": pin that specific object.
170 + * "recursive": pin that specific object, and indirectly pin all its decendants
171 + * "indirect": pinned indirectly by an ancestor (like a refcount)
172 + * "all"
173 +
174 Example:
175 $ echo "hello" | ipfs add -q
176 QmZULkCELmmk5XNfCgTnCyFgAVxBRBXyDHGGMVoLFLiXEN
@@ -207,24 +216,35 @@ Example:
216 if typeStr == "direct" || typeStr == "all" {
217 for _, k := range n.Pinning.DirectKeys() {
218 keys[k.B58String()] = RefKeyObject{
210 - Type: "direct",
211 - Count: 1,
219 + Type: "direct",
220 }
221 }
222 }
223 if typeStr == "indirect" || typeStr == "all" {
216 - for k, v := range n.Pinning.IndirectKeys() {
224 + ks := key.NewKeySet()
225 + for _, k := range n.Pinning.RecursiveKeys() {
226 + nd, err := n.DAG.Get(n.Context(), k)
227 + if err != nil {
228 + res.SetError(err, cmds.ErrNormal)
229 + return
230 + }
231 + err = dag.EnumerateChildren(n.Context(), n.DAG, nd, ks)
232 + if err != nil {
233 + res.SetError(err, cmds.ErrNormal)
234 + return
235 + }
236 +
237 + }
238 + for _, k := range ks.Keys() {
239 keys[k.B58String()] = RefKeyObject{
218 - Type: "indirect",
219 - Count: v,
240 + Type: "indirect",
241 }
242 }
243 }
244 if typeStr == "recursive" || typeStr == "all" {
245 for _, k := range n.Pinning.RecursiveKeys() {
246 keys[k.B58String()] = RefKeyObject{
226 - Type: "recursive",
227 - Count: 1,
247 + Type: "recursive",
248 }
249 }
250 }
@@ -234,16 +254,6 @@ Example:
254 Type: RefKeyList{},
255 Marshalers: cmds.MarshalerMap{
256 cmds.Text: func(res cmds.Response) (io.Reader, error) {
237 - typeStr, _, err := res.Request().Option("type").String()
238 - if err != nil {
239 - return nil, err
240 - }
241 -
242 - count, _, err := res.Request().Option("count").Bool()
243 - if err != nil {
244 - return nil, err
245 - }
246 -
257 quiet, _, err := res.Request().Option("quiet").Bool()
258 if err != nil {
259 return nil, err
@@ -254,21 +264,11 @@ Example:
264 return nil, u.ErrCast()
265 }
266 out := new(bytes.Buffer)
257 - if typeStr == "indirect" && count {
258 - for k, v := range keys.Keys {
259 - if quiet {
260 - fmt.Fprintf(out, "%s %d\n", k, v.Count)
261 - } else {
262 - fmt.Fprintf(out, "%s %s %d\n", k, v.Type, v.Count)
263 - }
264 - }
265 - } else {
266 - for k, v := range keys.Keys {
267 - if quiet {
268 - fmt.Fprintf(out, "%s\n", k)
269 - } else {
270 - fmt.Fprintf(out, "%s %s\n", k, v.Type)
271 - }
267 + for k, v := range keys.Keys {
268 + if quiet {
269 + fmt.Fprintf(out, "%s\n", k)
270 + } else {
271 + fmt.Fprintf(out, "%s %s\n", k, v.Type)
272 }
273 }
274 return out, nil
@@ -277,8 +277,7 @@ Example:
277 }
278
279 type RefKeyObject struct {
280 - Type string
281 - Count uint64
280 + Type string
281 }
282
283 type RefKeyList struct {
core/corehttp/gateway_handler.go
+1 -2
@@ -50,8 +50,7 @@ func (i *gatewayHandler) newDagFromReader(r io.Reader) (*dag.Node, error) {
50 // return ufs.AddFromReader(i.node, r.Body)
51 return importer.BuildDagFromReader(
52 i.node.DAG,
53 - chunk.DefaultSplitter(r),
54 - importer.BasicPinnerCB(i.node.Pinning))
53 + chunk.DefaultSplitter(r))
54 }
55
56 // TODO(btc): break this apart into separate handlers using a more expressive muxer
core/corerepo/gc.go
+17 -27
@@ -8,6 +8,7 @@ import (
8 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
9 key "github.com/ipfs/go-ipfs/blocks/key"
10 "github.com/ipfs/go-ipfs/core"
11 + gc "github.com/ipfs/go-ipfs/pin/gc"
12 repo "github.com/ipfs/go-ipfs/repo"
13 logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
14 )
@@ -73,53 +74,42 @@ func NewGC(n *core.IpfsNode) (*GC, error) {
74 func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
75 ctx, cancel := context.WithCancel(ctx)
76 defer cancel() // in case error occurs during operation
76 - keychan, err := n.Blockstore.AllKeysChan(ctx)
77 + rmed, err := gc.GC(ctx, n.Blockstore, n.Pinning)
78 if err != nil {
79 return err
80 }
80 - for k := range keychan { // rely on AllKeysChan to close chan
81 - if !n.Pinning.IsPinned(k) {
82 - if err := n.Blockstore.DeleteBlock(k); err != nil {
83 - return err
81 +
82 + for {
83 + select {
84 + case _, ok := <-rmed:
85 + if !ok {
86 + return nil
87 }
88 + case <-ctx.Done():
89 + return ctx.Err()
90 }
91 }
87 - return nil
92 +
93 }
94
95 func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemoved, error) {
91 -
92 - keychan, err := n.Blockstore.AllKeysChan(ctx)
96 + rmed, err := gc.GC(ctx, n.Blockstore, n.Pinning)
97 if err != nil {
98 return nil, err
99 }
100
97 - output := make(chan *KeyRemoved)
101 + out := make(chan *KeyRemoved)
102 go func() {
99 - defer close(output)
100 - for {
103 + defer close(out)
104 + for k := range rmed {
105 select {
102 - case k, ok := <-keychan:
103 - if !ok {
104 - return
105 - }
106 - if !n.Pinning.IsPinned(k) {
107 - err := n.Blockstore.DeleteBlock(k)
108 - if err != nil {
109 - log.Debugf("Error removing key from blockstore: %s", err)
110 - continue
111 - }
112 - select {
113 - case output <- &KeyRemoved{k}:
114 - case <-ctx.Done():
115 - }
116 - }
106 + case out <- &KeyRemoved{k}:
107 case <-ctx.Done():
108 return
109 }
110 }
111 }()
122 - return output, nil
112 + return out, nil
113 }
114
115 func PeriodicGC(ctx context.Context, node *core.IpfsNode) error {
core/coreunix/add.go
-8
@@ -13,7 +13,6 @@ import (
13 importer "github.com/ipfs/go-ipfs/importer"
14 chunk "github.com/ipfs/go-ipfs/importer/chunk"
15 merkledag "github.com/ipfs/go-ipfs/merkledag"
16 - "github.com/ipfs/go-ipfs/pin"
16 unixfs "github.com/ipfs/go-ipfs/unixfs"
17 logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
18 )
@@ -31,7 +30,6 @@ func Add(n *core.IpfsNode, r io.Reader) (string, error) {
30 dagNode, err := importer.BuildDagFromReader(
31 n.DAG,
32 chunk.NewSizeSplitter(r, chunk.DefaultBlockSize),
34 - importer.BasicPinnerCB(n.Pinning),
33 )
34 if err != nil {
35 return "", err
@@ -70,11 +68,6 @@ func AddR(n *core.IpfsNode, root string) (key string, err error) {
68 return "", err
69 }
70
73 - n.Pinning.RemovePinWithMode(k, pin.Indirect)
74 - if err := n.Pinning.Flush(); err != nil {
75 - return "", err
76 - }
77 -
71 return k.String(), nil
72 }
73
@@ -103,7 +96,6 @@ func add(n *core.IpfsNode, reader io.Reader) (*merkledag.Node, error) {
96 return importer.BuildDagFromReader(
97 n.DAG,
98 chunk.DefaultSplitter(reader),
106 - importer.PinIndirectCB(n.Pinning),
99 )
100 }
101
core/coreunix/metadata_test.go
+1 -1
@@ -36,7 +36,7 @@ func TestMetadata(t *testing.T) {
36 data := make([]byte, 1000)
37 u.NewTimeSeededRand().Read(data)
38 r := bytes.NewReader(data)
39 - nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r), nil)
39 + nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r))
40 if err != nil {
41 t.Fatal(err)
42 }
importer/helpers/dagbuilder.go
+1 -29
@@ -2,30 +2,18 @@ package helpers
2
3 import (
4 dag "github.com/ipfs/go-ipfs/merkledag"
5 - "github.com/ipfs/go-ipfs/pin"
5 )
6
8 -// NodeCB is callback function for dag generation
9 -// the `last` flag signifies whether or not this is the last
10 -// (top-most root) node being added. useful for things like
11 -// only pinning the first node recursively.
12 -type NodeCB func(node *dag.Node, last bool) error
13 -
14 -var nilFunc NodeCB = func(_ *dag.Node, _ bool) error { return nil }
15 -
7 // DagBuilderHelper wraps together a bunch of objects needed to
8 // efficiently create unixfs dag trees
9 type DagBuilderHelper struct {
10 dserv dag.DAGService
20 - mp pin.Pinner
11 in <-chan []byte
12 errs <-chan error
13 recvdErr error
14 nextData []byte // the next item to return.
15 maxlinks int
26 - ncb NodeCB
27 -
28 - batch *dag.Batch
16 + batch *dag.Batch
17 }
18
19 type DagBuilderParams struct {
@@ -34,25 +22,16 @@ type DagBuilderParams struct {
22
23 // DAGService to write blocks to (required)
24 Dagserv dag.DAGService
37 -
38 - // Callback for each block added
39 - NodeCB NodeCB
25 }
26
27 // Generate a new DagBuilderHelper from the given params, using 'in' as a
28 // data source
29 func (dbp *DagBuilderParams) New(in <-chan []byte, errs <-chan error) *DagBuilderHelper {
45 - ncb := dbp.NodeCB
46 - if ncb == nil {
47 - ncb = nilFunc
48 - }
49 -
30 return &DagBuilderHelper{
31 dserv: dbp.Dagserv,
32 in: in,
33 errs: errs,
34 maxlinks: dbp.Maxlinks,
55 - ncb: ncb,
35 batch: dbp.Dagserv.Batch(),
36 }
37 }
@@ -106,7 +85,6 @@ func (db *DagBuilderHelper) GetDagServ() dag.DAGService {
85 // FillNodeLayer will add datanodes as children to the give node until
86 // at most db.indirSize ndoes are added
87 //
109 -// warning: **children** pinned indirectly, but input node IS NOT pinned.
88 func (db *DagBuilderHelper) FillNodeLayer(node *UnixfsNode) error {
89
90 // while we have room AND we're not done
@@ -150,12 +128,6 @@ func (db *DagBuilderHelper) Add(node *UnixfsNode) (*dag.Node, error) {
128 return nil, err
129 }
130
153 - // node callback
154 - err = db.ncb(dn, true)
155 - if err != nil {
156 - return nil, err
157 - }
158 -
131 return dn, nil
132 }
133
importer/helpers/helpers.go
-12
@@ -4,10 +4,8 @@ import (
4 "fmt"
5
6 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
7 - key "github.com/ipfs/go-ipfs/blocks/key"
7 chunk "github.com/ipfs/go-ipfs/importer/chunk"
8 dag "github.com/ipfs/go-ipfs/merkledag"
10 - "github.com/ipfs/go-ipfs/pin"
9 ft "github.com/ipfs/go-ipfs/unixfs"
10 )
11
@@ -108,21 +106,11 @@ func (n *UnixfsNode) AddChild(child *UnixfsNode, db *DagBuilderHelper) error {
106 return err
107 }
108
111 - // Pin the child node indirectly
112 - err = db.ncb(childnode, false)
113 - if err != nil {
114 - return err
115 - }
116 -
109 return nil
110 }
111
112 // Removes the child node at the given index
113 func (n *UnixfsNode) RemoveChild(index int, dbh *DagBuilderHelper) {
122 - k := key.Key(n.node.Links[index].Hash)
123 - if dbh.mp != nil {
124 - dbh.mp.RemovePinWithMode(k, pin.Indirect)
125 - }
114 n.ufmt.RemoveBlockSize(index)
115 n.node.Links = append(n.node.Links[:index], n.node.Links[index+1:]...)
116 }
importer/importer.go
+4 -36
@@ -12,7 +12,6 @@ import (
12 h "github.com/ipfs/go-ipfs/importer/helpers"
13 trickle "github.com/ipfs/go-ipfs/importer/trickle"
14 dag "github.com/ipfs/go-ipfs/merkledag"
15 - "github.com/ipfs/go-ipfs/pin"
15 logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
16 )
17
@@ -20,7 +19,7 @@ var log = logging.Logger("importer")
19
20 // Builds a DAG from the given file, writing created blocks to disk as they are
21 // created
23 -func BuildDagFromFile(fpath string, ds dag.DAGService, mp pin.Pinner) (*dag.Node, error) {
22 +func BuildDagFromFile(fpath string, ds dag.DAGService) (*dag.Node, error) {
23 stat, err := os.Lstat(fpath)
24 if err != nil {
25 return nil, err
@@ -36,60 +35,29 @@ func BuildDagFromFile(fpath string, ds dag.DAGService, mp pin.Pinner) (*dag.Node
35 }
36 defer f.Close()
37
39 - return BuildDagFromReader(ds, chunk.NewSizeSplitter(f, chunk.DefaultBlockSize), BasicPinnerCB(mp))
38 + return BuildDagFromReader(ds, chunk.NewSizeSplitter(f, chunk.DefaultBlockSize))
39 }
40
42 -func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter, ncb h.NodeCB) (*dag.Node, error) {
41 +func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) {
42 // Start the splitter
43 blkch, errch := chunk.Chan(spl)
44
45 dbp := h.DagBuilderParams{
46 Dagserv: ds,
47 Maxlinks: h.DefaultLinksPerBlock,
49 - NodeCB: ncb,
48 }
49
50 return bal.BalancedLayout(dbp.New(blkch, errch))
51 }
52
55 -func BuildTrickleDagFromReader(ds dag.DAGService, spl chunk.Splitter, ncb h.NodeCB) (*dag.Node, error) {
53 +func BuildTrickleDagFromReader(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) {
54 // Start the splitter
55 blkch, errch := chunk.Chan(spl)
56
57 dbp := h.DagBuilderParams{
58 Dagserv: ds,
59 Maxlinks: h.DefaultLinksPerBlock,
62 - NodeCB: ncb,
60 }
61
62 return trickle.TrickleLayout(dbp.New(blkch, errch))
63 }
67 -
68 -func BasicPinnerCB(p pin.Pinner) h.NodeCB {
69 - return func(n *dag.Node, last bool) error {
70 - k, err := n.Key()
71 - if err != nil {
72 - return err
73 - }
74 -
75 - if last {
76 - p.PinWithMode(k, pin.Recursive)
77 - return p.Flush()
78 - } else {
79 - p.PinWithMode(k, pin.Indirect)
80 - return nil
81 - }
82 - }
83 -}
84 -
85 -func PinIndirectCB(p pin.Pinner) h.NodeCB {
86 - return func(n *dag.Node, last bool) error {
87 - k, err := n.Key()
88 - if err != nil {
89 - return err
90 - }
91 -
92 - p.PinWithMode(k, pin.Indirect)
93 - return nil
94 - }
95 -}
importer/importer_test.go
+3 -3
@@ -17,7 +17,7 @@ import (
17 func getBalancedDag(t testing.TB, size int64, blksize int64) (*dag.Node, dag.DAGService) {
18 ds := mdtest.Mock()
19 r := io.LimitReader(u.NewTimeSeededRand(), size)
20 - nd, err := BuildDagFromReader(ds, chunk.NewSizeSplitter(r, blksize), nil)
20 + nd, err := BuildDagFromReader(ds, chunk.NewSizeSplitter(r, blksize))
21 if err != nil {
22 t.Fatal(err)
23 }
@@ -27,7 +27,7 @@ func getBalancedDag(t testing.TB, size int64, blksize int64) (*dag.Node, dag.DAG
27 func getTrickleDag(t testing.TB, size int64, blksize int64) (*dag.Node, dag.DAGService) {
28 ds := mdtest.Mock()
29 r := io.LimitReader(u.NewTimeSeededRand(), size)
30 - nd, err := BuildTrickleDagFromReader(ds, chunk.NewSizeSplitter(r, blksize), nil)
30 + nd, err := BuildTrickleDagFromReader(ds, chunk.NewSizeSplitter(r, blksize))
31 if err != nil {
32 t.Fatal(err)
33 }
@@ -40,7 +40,7 @@ func TestBalancedDag(t *testing.T) {
40 u.NewTimeSeededRand().Read(buf)
41 r := bytes.NewReader(buf)
42
43 - nd, err := BuildDagFromReader(ds, chunk.DefaultSplitter(r), nil)
43 + nd, err := BuildDagFromReader(ds, chunk.DefaultSplitter(r))
44 if err != nil {
45 t.Fatal(err)
46 }
merkledag/merkledag_test.go
+3 -3
@@ -164,7 +164,7 @@ func runBatchFetchTest(t *testing.T, read io.Reader) {
164
165 spl := chunk.NewSizeSplitter(read, 512)
166
167 - root, err := imp.BuildDagFromReader(dagservs[0], spl, nil)
167 + root, err := imp.BuildDagFromReader(dagservs[0], spl)
168 if err != nil {
169 t.Fatal(err)
170 }
@@ -306,7 +306,7 @@ func TestFetchGraph(t *testing.T) {
306 }
307
308 read := io.LimitReader(u.NewTimeSeededRand(), 1024*32)
309 - root, err := imp.BuildDagFromReader(dservs[0], chunk.NewSizeSplitter(read, 512), nil)
309 + root, err := imp.BuildDagFromReader(dservs[0], chunk.NewSizeSplitter(read, 512))
310 if err != nil {
311 t.Fatal(err)
312 }
@@ -333,7 +333,7 @@ func TestEnumerateChildren(t *testing.T) {
333 ds := NewDAGService(bsi[0])
334
335 read := io.LimitReader(u.NewTimeSeededRand(), 1024*1024)
336 - root, err := imp.BuildDagFromReader(ds, chunk.NewSizeSplitter(read, 512), nil)
336 + root, err := imp.BuildDagFromReader(ds, chunk.NewSizeSplitter(read, 512))
337 if err != nil {
338 t.Fatal(err)
339 }
pin/gc/gc.go new
+99
@@ -0,0 +1,99 @@
1 +package gc
2 +
3 +import (
4 + bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
5 + key "github.com/ipfs/go-ipfs/blocks/key"
6 + bserv "github.com/ipfs/go-ipfs/blockservice"
7 + offline "github.com/ipfs/go-ipfs/exchange/offline"
8 + dag "github.com/ipfs/go-ipfs/merkledag"
9 + pin "github.com/ipfs/go-ipfs/pin"
10 +
11 + context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
12 + logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
13 +)
14 +
15 +var log = logging.Logger("gc")
16 +
17 +// GC performs a mark and sweep garbage collection of the blocks in the blockstore
18 +// first, it creates a 'marked' set and adds to it the following:
19 +// - all recursively pinned blocks, plus all of their descendants (recursively)
20 +// - all directly pinned blocks
21 +// - all blocks utilized internally by the pinner
22 +//
23 +// The routine then iterates over every block in the blockstore and
24 +// deletes any block that is not found in the marked set.
25 +func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner) (<-chan key.Key, error) {
26 + unlock := bs.GCLock()
27 + defer unlock()
28 +
29 + bsrv := bserv.New(bs, offline.Exchange(bs))
30 + ds := dag.NewDAGService(bsrv)
31 +
32 + // KeySet currently implemented in memory, in the future, may be bloom filter or
33 + // disk backed to conserve memory.
34 + gcs := key.NewKeySet()
35 + for _, k := range pn.RecursiveKeys() {
36 + gcs.Add(k)
37 + nd, err := ds.Get(ctx, k)
38 + if err != nil {
39 + return nil, err
40 + }
41 +
42 + // EnumerateChildren recursively walks the dag and adds the keys to the given set
43 + err = dag.EnumerateChildren(ctx, ds, nd, gcs)
44 + if err != nil {
45 + return nil, err
46 + }
47 + }
48 + for _, k := range pn.DirectKeys() {
49 + gcs.Add(k)
50 + }
51 + for _, k := range pn.InternalPins() {
52 + gcs.Add(k)
53 +
54 + nd, err := ds.Get(ctx, k)
55 + if err != nil {
56 + return nil, err
57 + }
58 +
59 + // EnumerateChildren recursively walks the dag and adds the keys to the given set
60 + err = dag.EnumerateChildren(ctx, ds, nd, gcs)
61 + if err != nil {
62 + return nil, err
63 + }
64 + }
65 +
66 + keychan, err := bs.AllKeysChan(ctx)
67 + if err != nil {
68 + return nil, err
69 + }
70 +
71 + output := make(chan key.Key)
72 + go func() {
73 + defer close(output)
74 + for {
75 + select {
76 + case k, ok := <-keychan:
77 + if !ok {
78 + return
79 + }
80 + if !gcs.Has(k) {
81 + err := bs.DeleteBlock(k)
82 + if err != nil {
83 + log.Debugf("Error removing key from blockstore: %s", err)
84 + return
85 + }
86 + select {
87 + case output <- k:
88 + case <-ctx.Done():
89 + return
90 + }
91 + }
92 + case <-ctx.Done():
93 + return
94 + }
95 + }
96 + }()
97 +
98 + return output, nil
99 +}
pin/pin.go
+20 -87
@@ -24,7 +24,6 @@ var emptyKey = key.B58KeyDecode("QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n"
24 const (
25 linkDirect = "direct"
26 linkRecursive = "recursive"
27 - linkIndirect = "indirect"
27 )
28
29 type PinMode int
@@ -32,7 +31,6 @@ type PinMode int
31 const (
32 Recursive PinMode = iota
33 Direct
35 - Indirect
34 NotPinned
35 )
36
@@ -52,8 +50,8 @@ type Pinner interface {
50
51 Flush() error
52 DirectKeys() []key.Key
55 - IndirectKeys() map[key.Key]uint64
53 RecursiveKeys() []key.Key
54 + InternalPins() []key.Key
55 }
56
57 // pinner implements the Pinner interface
@@ -61,7 +59,7 @@ type pinner struct {
59 lock sync.RWMutex
60 recursePin set.BlockSet
61 directPin set.BlockSet
64 - indirPin *indirectPin
62 +
63 // Track the keys used for storing the pinning state, so gc does
64 // not delete them.
65 internalPin map[key.Key]struct{}
@@ -80,7 +78,6 @@ func NewPinner(dstore ds.ThreadSafeDatastore, serv mdag.DAGService) Pinner {
78 return &pinner{
79 recursePin: rcset,
80 directPin: dirset,
83 - indirPin: newIndirectPin(),
81 dserv: serv,
82 dstore: dstore,
83 }
@@ -104,7 +101,8 @@ func (p *pinner) Pin(ctx context.Context, node *mdag.Node, recurse bool) error {
101 p.directPin.RemoveBlock(k)
102 }
103
107 - err := p.pinLinks(ctx, node)
104 + // fetch entire graph
105 + err := mdag.FetchGraph(ctx, node, p.dserv)
106 if err != nil {
107 return err
108 }
@@ -131,72 +129,18 @@ func (p *pinner) Unpin(ctx context.Context, k key.Key, recursive bool) error {
129 if p.recursePin.HasKey(k) {
130 if recursive {
131 p.recursePin.RemoveBlock(k)
134 - node, err := p.dserv.Get(ctx, k)
135 - if err != nil {
136 - return err
137 - }
138 -
139 - return p.unpinLinks(ctx, node)
132 + return nil
133 } else {
134 return fmt.Errorf("%s is pinned recursively", k)
135 }
136 } else if p.directPin.HasKey(k) {
137 p.directPin.RemoveBlock(k)
138 return nil
146 - } else if p.indirPin.HasKey(k) {
147 - return fmt.Errorf("%s is pinned indirectly. indirect pins cannot be removed directly", k)
139 } else {
140 return fmt.Errorf("%s is not pinned", k)
141 }
142 }
143
153 -func (p *pinner) unpinLinks(ctx context.Context, node *mdag.Node) error {
154 - for _, l := range node.Links {
155 - node, err := l.GetNode(ctx, p.dserv)
156 - if err != nil {
157 - return err
158 - }
159 -
160 - k, err := node.Key()
161 - if err != nil {
162 - return err
163 - }
164 -
165 - p.indirPin.Decrement(k)
166 -
167 - err = p.unpinLinks(ctx, node)
168 - if err != nil {
169 - return err
170 - }
171 - }
172 - return nil
173 -}
174 -
175 -func (p *pinner) pinIndirectRecurse(ctx context.Context, node *mdag.Node) error {
176 - k, err := node.Key()
177 - if err != nil {
178 - return err
179 - }
180 -
181 - p.indirPin.Increment(k)
182 - return p.pinLinks(ctx, node)
183 -}
184 -
185 -func (p *pinner) pinLinks(ctx context.Context, node *mdag.Node) error {
186 - for _, ng := range p.dserv.GetDAG(ctx, node) {
187 - subnode, err := ng.Get(ctx)
188 - if err != nil {
189 - // TODO: Maybe just log and continue?
190 - return err
191 - }
192 - err = p.pinIndirectRecurse(ctx, subnode)
193 - if err != nil {
194 - return err
195 - }
196 - }
197 - return nil
198 -}
199 -
144 func (p *pinner) isInternalPin(key key.Key) bool {
145 _, ok := p.internalPin[key]
146 return ok
@@ -208,7 +152,6 @@ func (p *pinner) IsPinned(key key.Key) bool {
152 defer p.lock.RUnlock()
153 return p.recursePin.HasKey(key) ||
154 p.directPin.HasKey(key) ||
211 - p.indirPin.HasKey(key) ||
155 p.isInternalPin(key)
156 }
157
@@ -218,8 +161,6 @@ func (p *pinner) RemovePinWithMode(key key.Key, mode PinMode) {
161 switch mode {
162 case Direct:
163 p.directPin.RemoveBlock(key)
221 - case Indirect:
222 - p.indirPin.Decrement(key)
164 case Recursive:
165 p.recursePin.RemoveBlock(key)
166 default:
@@ -274,14 +215,6 @@ func LoadPinner(d ds.ThreadSafeDatastore, dserv mdag.DAGService) (Pinner, error)
215 p.directPin = set.SimpleSetFromKeys(directKeys)
216 }
217
277 - { // load indirect set
278 - refcnt, err := loadMultiset(ctx, dserv, root, linkIndirect, recordInternal)
279 - if err != nil {
280 - return nil, fmt.Errorf("cannot load indirect pins: %v", err)
281 - }
282 - p.indirPin = &indirectPin{refCounts: refcnt}
283 - }
284 -
218 p.internalPin = internalPin
219
220 // assign services
@@ -296,11 +229,6 @@ func (p *pinner) DirectKeys() []key.Key {
229 return p.directPin.GetKeys()
230 }
231
299 -// IndirectKeys returns a slice containing the indirectly pinned keys
300 -func (p *pinner) IndirectKeys() map[key.Key]uint64 {
301 - return p.indirPin.GetRefs()
302 -}
303 -
232 // RecursiveKeys returns a slice containing the recursively pinned keys
233 func (p *pinner) RecursiveKeys() []key.Key {
234 return p.recursePin.GetKeys()
@@ -339,20 +267,17 @@ func (p *pinner) Flush() error {
267 }
268 }
269
342 - {
343 - n, err := storeMultiset(ctx, p.dserv, p.indirPin.GetRefs(), recordInternal)
344 - if err != nil {
345 - return err
346 - }
347 - if err := root.AddNodeLink(linkIndirect, n); err != nil {
348 - return err
349 - }
270 + // add the empty node, its referenced by the pin sets but never created
271 + _, err := p.dserv.Add(new(mdag.Node))
272 + if err != nil {
273 + return err
274 }
275
276 k, err := p.dserv.Add(root)
277 if err != nil {
278 return err
279 }
280 +
281 internalPin[k] = struct{}{}
282 if err := p.dstore.Put(pinDatastoreKey, []byte(k)); err != nil {
283 return fmt.Errorf("cannot store pin state: %v", err)
@@ -361,6 +286,16 @@ func (p *pinner) Flush() error {
286 return nil
287 }
288
289 +func (p *pinner) InternalPins() []key.Key {
290 + p.lock.Lock()
291 + defer p.lock.Unlock()
292 + var out []key.Key
293 + for k, _ := range p.internalPin {
294 + out = append(out, k)
295 + }
296 + return out
297 +}
298 +
299 // PinWithMode allows the user to have fine grained control over pin
300 // counts
301 func (p *pinner) PinWithMode(k key.Key, mode PinMode) {
@@ -371,7 +306,5 @@ func (p *pinner) PinWithMode(k key.Key, mode PinMode) {
306 p.recursePin.AddBlock(k)
307 case Direct:
308 p.directPin.AddBlock(k)
374 - case Indirect:
375 - p.indirPin.Increment(k)
309 }
310 }
pin/pin_test.go
+3 -21
@@ -53,7 +53,7 @@ func TestPinnerBasic(t *testing.T) {
53 }
54
55 // create new node c, to be indirectly pinned through b
56 - c, ck := randNode()
56 + c, _ := randNode()
57 _, err = dserv.Add(c)
58 if err != nil {
59 t.Fatal(err)
@@ -82,10 +82,6 @@ func TestPinnerBasic(t *testing.T) {
82 t.Fatal(err)
83 }
84
85 - if !p.IsPinned(ck) {
86 - t.Fatal("Child of recursively pinned node not found")
87 - }
88 -
85 bk, _ := b.Key()
86 if !p.IsPinned(bk) {
87 t.Fatal("Recursively pinned node not found..")
@@ -95,7 +91,7 @@ func TestPinnerBasic(t *testing.T) {
91 d.AddNodeLink("a", a)
92 d.AddNodeLink("c", c)
93
98 - e, ek := randNode()
94 + e, _ := randNode()
95 d.AddNodeLink("e", e)
96
97 // Must be in dagserv for unpin to work
@@ -110,10 +106,6 @@ func TestPinnerBasic(t *testing.T) {
106 t.Fatal(err)
107 }
108
113 - if !p.IsPinned(ek) {
114 - t.Fatal(err)
115 - }
116 -
109 dk, _ := d.Key()
110 if !p.IsPinned(dk) {
111 t.Fatal("pinned node not found.")
@@ -125,11 +117,6 @@ func TestPinnerBasic(t *testing.T) {
117 t.Fatal(err)
118 }
119
128 - // c should still be pinned under b
129 - if !p.IsPinned(ck) {
130 - t.Fatal("Recursive / indirect unpin fail.")
131 - }
132 -
120 err = p.Flush()
121 if err != nil {
122 t.Fatal(err)
@@ -145,11 +132,6 @@ func TestPinnerBasic(t *testing.T) {
132 t.Fatal("Could not find pinned node!")
133 }
134
148 - // Test indirectly pinned
149 - if !np.IsPinned(ck) {
150 - t.Fatal("could not find indirectly pinned node")
151 - }
152 -
135 // Test recursively pinned
136 if !np.IsPinned(bk) {
137 t.Fatal("could not find recursively pinned node")
@@ -201,7 +183,7 @@ func TestFlush(t *testing.T) {
183 p := NewPinner(dstore, dserv)
184 _, k := randNode()
185
204 - p.PinWithMode(k, Indirect)
186 + p.PinWithMode(k, Recursive)
187 if err := p.Flush(); err != nil {
188 t.Fatal(err)
189 }
tar/format.go
+1 -1
@@ -68,7 +68,7 @@ func ImportTar(r io.Reader, ds dag.DAGService) (*dag.Node, error) {
68
69 if h.Size > 0 {
70 spl := chunk.NewRabin(tr, uint64(chunk.DefaultBlockSize))
71 - nd, err := importer.BuildDagFromReader(ds, spl, nil)
71 + nd, err := importer.BuildDagFromReader(ds, spl)
72 if err != nil {
73 return nil, err
74 }
test/sharness/t0080-repo.sh
+6 -15
@@ -15,11 +15,6 @@ test_expect_success "'ipfs repo gc' succeeds" '
15 ipfs repo gc >gc_out_actual
16 '
17
18 -test_expect_success "'ipfs repo gc' looks good (empty)" '
19 - true >empty &&
20 - test_cmp empty gc_out_actual
21 -'
22 -
18 test_expect_success "'ipfs add afile' succeeds" '
19 echo "some text" >afile &&
20 HASH=`ipfs add -q afile`
@@ -36,8 +31,7 @@ test_expect_success "'ipfs repo gc' succeeds" '
31
32 test_expect_success "'ipfs repo gc' looks good (patch root)" '
33 PATCH_ROOT=QmQXirSbubiySKnqaFyfs5YzziXRB5JEVQVjU6xsd7innr &&
39 - echo "removed $PATCH_ROOT" >patch_root &&
40 - test_cmp patch_root gc_out_actual
34 + grep "removed $PATCH_ROOT" gc_out_actual
35 '
36
37 test_expect_success "'ipfs repo gc' doesnt remove file" '
@@ -66,13 +60,13 @@ test_expect_failure "ipfs repo gc fully reverse ipfs add" '
60 '
61
62 test_expect_success "file no longer pinned" '
69 - # we expect the welcome files to show up here
63 + # we expect the welcome files and gw assets to show up here
64 echo "$HASH_WELCOME_DOCS" >expected2 &&
65 ipfs refs -r "$HASH_WELCOME_DOCS" >>expected2 &&
66 EMPTY_DIR=QmUNLLsPACCz1vLxQVkXqqLX5R1X345qqfHbsf67hvA3Nn &&
67 echo "$EMPTY_DIR" >>expected2 &&
68 ipfs pin ls --type=recursive --quiet >actual2 &&
75 - test_sort_cmp expected2 actual2
69 + test_expect_code 1 grep $HASH actual2
70 '
71
72 test_expect_success "recursively pin afile(default action)" '
@@ -114,10 +108,9 @@ test_expect_success "remove direct pin" '
108 '
109
110 test_expect_success "'ipfs repo gc' removes file" '
117 - echo "removed $HASH" >expected7 &&
118 - echo "removed $PATCH_ROOT" >>expected7 &&
111 ipfs repo gc >actual7 &&
120 - test_sort_cmp expected7 actual7
112 + grep "removed $HASH" actual7 &&
113 + grep "removed $PATCH_ROOT" actual7
114 '
115
116 # TODO: there seems to be a serious bug with leveldb not returning a key.
@@ -135,8 +128,7 @@ test_expect_success "adding multiblock random file succeeds" '
128 MBLOCKHASH=`ipfs add -q multiblock`
129 '
130
138 -# TODO: this starts to fail with the pinning rewrite, for unclear reasons
139 -test_expect_failure "'ipfs pin ls --type=indirect' is correct" '
131 +test_expect_success "'ipfs pin ls --type=indirect' is correct" '
132 ipfs refs "$MBLOCKHASH" >refsout &&
133 ipfs refs -r "$HASH_WELCOME_DOCS" >>refsout &&
134 sed -i"~" "s/\(.*\)/\1 indirect/g" refsout &&
@@ -166,7 +158,6 @@ test_expect_success "'ipfs pin ls --type=recursive' is correct" '
158 echo "$MBLOCKHASH" >rp_expected &&
159 echo "$HASH_WELCOME_DOCS" >>rp_expected &&
160 echo "$EMPTY_DIR" >>rp_expected &&
169 - ipfs refs -r "$HASH_WELCOME_DOCS" >>rp_expected &&
161 sed -i"~" "s/\(.*\)/\1 recursive/g" rp_expected &&
162 ipfs pin ls --type=recursive >rp_actual &&
163 test_sort_cmp rp_expected rp_actual
unixfs/mod/dagmodifier.go
-9
@@ -11,7 +11,6 @@ import (
11 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
12
13 key "github.com/ipfs/go-ipfs/blocks/key"
14 - imp "github.com/ipfs/go-ipfs/importer"
14 chunk "github.com/ipfs/go-ipfs/importer/chunk"
15 help "github.com/ipfs/go-ipfs/importer/helpers"
16 trickle "github.com/ipfs/go-ipfs/importer/trickle"
@@ -266,10 +265,6 @@ func (dm *DagModifier) modifyDag(node *mdag.Node, offset uint64, data io.Reader)
265 for i, bs := range f.GetBlocksizes() {
266 // We found the correct child to write into
267 if cur+bs > offset {
269 - // Unpin block
270 - ckey := key.Key(node.Links[i].Hash)
271 - dm.mp.RemovePinWithMode(ckey, pin.Indirect)
272 -
268 child, err := node.Links[i].GetNode(dm.ctx, dm.dagserv)
269 if err != nil {
270 return "", false, err
@@ -279,9 +274,6 @@ func (dm *DagModifier) modifyDag(node *mdag.Node, offset uint64, data io.Reader)
274 return "", false, err
275 }
276
282 - // pin the new node
283 - dm.mp.PinWithMode(k, pin.Indirect)
284 -
277 offset += bs
278 node.Links[i].Hash = mh.Multihash(k)
279
@@ -310,7 +302,6 @@ func (dm *DagModifier) appendData(node *mdag.Node, blks <-chan []byte, errs <-ch
302 dbp := &help.DagBuilderParams{
303 Dagserv: dm.dagserv,
304 Maxlinks: help.DefaultLinksPerBlock,
313 - NodeCB: imp.BasicPinnerCB(dm.mp),
305 }
306
307 return trickle.TrickleAppend(dm.ctx, node, dbp.New(blks, errs))
unixfs/mod/dagmodifier_test.go
+7 -19
@@ -19,6 +19,7 @@ import (
19 trickle "github.com/ipfs/go-ipfs/importer/trickle"
20 mdag "github.com/ipfs/go-ipfs/merkledag"
21 pin "github.com/ipfs/go-ipfs/pin"
22 + gc "github.com/ipfs/go-ipfs/pin/gc"
23 ft "github.com/ipfs/go-ipfs/unixfs"
24 uio "github.com/ipfs/go-ipfs/unixfs/io"
25 u "github.com/ipfs/go-ipfs/util"
@@ -36,7 +37,7 @@ func getMockDagServ(t testing.TB) (mdag.DAGService, pin.Pinner) {
37 return dserv, pin.NewPinner(tsds, dserv)
38 }
39
39 -func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.Blockstore, pin.Pinner) {
40 +func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.GCBlockstore, pin.Pinner) {
41 dstore := ds.NewMapDatastore()
42 tsds := sync.MutexWrap(dstore)
43 bstore := blockstore.NewBlockstore(tsds)
@@ -47,7 +48,7 @@ func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.Blocksto
48
49 func getNode(t testing.TB, dserv mdag.DAGService, size int64, pinner pin.Pinner) ([]byte, *mdag.Node) {
50 in := io.LimitReader(u.NewTimeSeededRand(), size)
50 - node, err := imp.BuildTrickleDagFromReader(dserv, sizeSplitterGen(500)(in), imp.BasicPinnerCB(pinner))
51 + node, err := imp.BuildTrickleDagFromReader(dserv, sizeSplitterGen(500)(in))
52 if err != nil {
53 t.Fatal(err)
54 }
@@ -469,22 +470,17 @@ func TestSparseWrite(t *testing.T) {
470 }
471 }
472
472 -func basicGC(t *testing.T, bs blockstore.Blockstore, pins pin.Pinner) {
473 +func basicGC(t *testing.T, bs blockstore.GCBlockstore, pins pin.Pinner) {
474 ctx, cancel := context.WithCancel(context.Background())
475 defer cancel() // in case error occurs during operation
475 - keychan, err := bs.AllKeysChan(ctx)
476 + out, err := gc.GC(ctx, bs, pins)
477 if err != nil {
478 t.Fatal(err)
479 }
479 - for k := range keychan { // rely on AllKeysChan to close chan
480 - if !pins.IsPinned(k) {
481 - err := bs.DeleteBlock(k)
482 - if err != nil {
483 - t.Fatal(err)
484 - }
485 - }
480 + for range out {
481 }
482 }
483 +
484 func TestCorrectPinning(t *testing.T) {
485 dserv, bstore, pins := getMockDagServAndBstore(t)
486 b, n := getNode(t, dserv, 50000, pins)
@@ -566,14 +562,6 @@ func TestCorrectPinning(t *testing.T) {
562 t.Fatal("Incorrect node recursively pinned")
563 }
564
569 - indirpins := pins.IndirectKeys()
570 - children := enumerateChildren(t, nd, dserv)
571 - // TODO this is not true if the contents happen to be identical
572 - if len(indirpins) != len(children) {
573 - t.Log(len(indirpins), len(children))
574 - t.Fatal("Incorrect number of indirectly pinned blocks")
575 - }
576 -
565 }
566
567 func enumerateChildren(t *testing.T, nd *mdag.Node, ds mdag.DAGService) []key.Key {