@cryptotaxi247 / kubo / commits / 315849639

implement ipfs pin update

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Mar 30, 2017 at 18:20 UTC 315849639dab3ae56e8700c2cadc506acd92bfd0
8 files changed +373 -22
core/commands/object/diff.go
+1 -14
@@ -7,7 +7,6 @@ import (
7
8 cmds "github.com/ipfs/go-ipfs/commands"
9 core "github.com/ipfs/go-ipfs/core"
10 - dag "github.com/ipfs/go-ipfs/merkledag"
10 dagutils "github.com/ipfs/go-ipfs/merkledag/utils"
11 path "github.com/ipfs/go-ipfs/path"
12 )
@@ -86,19 +85,7 @@ Example:
85 return
86 }
87
89 - pbobj_a, ok := obj_a.(*dag.ProtoNode)
90 - if !ok {
91 - res.SetError(dag.ErrNotProtobuf, cmds.ErrNormal)
92 - return
93 - }
94 -
95 - pbobj_b, ok := obj_b.(*dag.ProtoNode)
96 - if !ok {
97 - res.SetError(dag.ErrNotProtobuf, cmds.ErrNormal)
98 - return
99 - }
100 -
101 - changes, err := dagutils.Diff(ctx, node.DAG, pbobj_a, pbobj_b)
88 + changes, err := dagutils.Diff(ctx, node.DAG, obj_a, obj_b)
89 if err != nil {
90 res.SetError(err, cmds.ErrNormal)
91 return
core/commands/pin.go
+81 -3
@@ -24,9 +24,10 @@ var PinCmd = &cmds.Command{
24 },
25
26 Subcommands: map[string]*cmds.Command{
27 - "add": addPinCmd,
28 - "rm": rmPinCmd,
29 - "ls": listPinCmd,
27 + "add": addPinCmd,
28 + "rm": rmPinCmd,
29 + "ls": listPinCmd,
30 + "update": updatePinCmd,
31 },
32 }
33
@@ -332,6 +333,83 @@ Example:
333 },
334 }
335
336 +var updatePinCmd = &cmds.Command{
337 + Helptext: cmds.HelpText{
338 + Tagline: "Update a recursive pin",
339 + ShortDescription: `
340 +Updates one pin to another, making sure that all objects in the new pin are
341 +local. Then removes the old pin. This is an optimized version of adding the
342 +new pin and removing the old one.
343 +`,
344 + },
345 +
346 + Arguments: []cmds.Argument{
347 + cmds.StringArg("from-path", true, false, "Path to old object."),
348 + cmds.StringArg("to-path", true, false, "Path to new object to be pinned."),
349 + },
350 + Options: []cmds.Option{
351 + cmds.BoolOption("unpin", "Remove the old pin.").Default(true),
352 + },
353 + Type: PinOutput{},
354 + Run: func(req cmds.Request, res cmds.Response) {
355 + n, err := req.InvocContext().GetNode()
356 + if err != nil {
357 + res.SetError(err, cmds.ErrNormal)
358 + return
359 + }
360 +
361 + unpin, _, err := req.Option("unpin").Bool()
362 + if err != nil {
363 + res.SetError(err, cmds.ErrNormal)
364 + return
365 + }
366 +
367 + from, err := path.ParsePath(req.Arguments()[0])
368 + if err != nil {
369 + res.SetError(err, cmds.ErrNormal)
370 + return
371 + }
372 +
373 + to, err := path.ParsePath(req.Arguments()[1])
374 + if err != nil {
375 + res.SetError(err, cmds.ErrNormal)
376 + return
377 + }
378 +
379 + fromc, err := core.ResolveToCid(req.Context(), n, from)
380 + if err != nil {
381 + res.SetError(err, cmds.ErrNormal)
382 + return
383 + }
384 +
385 + toc, err := core.ResolveToCid(req.Context(), n, to)
386 + if err != nil {
387 + res.SetError(err, cmds.ErrNormal)
388 + return
389 + }
390 +
391 + err = n.Pinning.Update(req.Context(), fromc, toc, unpin)
392 + if err != nil {
393 + res.SetError(err, cmds.ErrNormal)
394 + return
395 + }
396 +
397 + res.SetOutput(&PinOutput{Pins: []string{from.String(), to.String()}})
398 + },
399 + Marshalers: cmds.MarshalerMap{
400 + cmds.Text: func(res cmds.Response) (io.Reader, error) {
401 + added, ok := res.Output().(*PinOutput)
402 + if !ok {
403 + return nil, u.ErrCast()
404 + }
405 +
406 + buf := new(bytes.Buffer)
407 + fmt.Fprintf(buf, "updated %s to %s\n", added.Pins[0], added.Pins[1])
408 + return buf, nil
409 + },
410 + },
411 +}
412 +
413 type RefKeyObject struct {
414 Type string
415 }
merkledag/merkledag.go
+1 -1
@@ -153,7 +153,7 @@ func (n *dagService) Remove(nd node.Node) error {
153 // GetLinksDirect creates a function to get the links for a node, from
154 // the node, bypassing the LinkService. If the node does not exist
155 // locally (and can not be retrieved) an error will be returned.
156 -func GetLinksDirect(serv DAGService) GetLinks {
156 +func GetLinksDirect(serv node.NodeGetter) GetLinks {
157 return func(ctx context.Context, c *cid.Cid) ([]*node.Link, error) {
158 node, err := serv.Get(ctx, c)
159 if err != nil {
merkledag/utils/diff.go
+5 -3
@@ -1,12 +1,13 @@
1 package dagutils
2
3 import (
4 + "context"
5 "fmt"
6 "path"
7
8 dag "github.com/ipfs/go-ipfs/merkledag"
9
9 - context "context"
10 + node "github.com/ipfs/go-ipld-node"
11 cid "gx/ipfs/QmYhQaCYEcaPPjxJX7YcPcVKkQfRy6sJ7B3XmGFk82XYdQ/go-cid"
12 )
13
@@ -87,7 +88,8 @@ func ApplyChange(ctx context.Context, ds dag.DAGService, nd *dag.ProtoNode, cs [
88 return e.Finalize(ds)
89 }
90
90 -func Diff(ctx context.Context, ds dag.DAGService, a, b *dag.ProtoNode) ([]*Change, error) {
91 +// Diff returns a set of changes that transform node 'a' into node 'b'
92 +func Diff(ctx context.Context, ds dag.DAGService, a, b node.Node) ([]*Change, error) {
93 if len(a.Links()) == 0 && len(b.Links()) == 0 {
94 return []*Change{
95 &Change{
@@ -104,7 +106,7 @@ func Diff(ctx context.Context, ds dag.DAGService, a, b *dag.ProtoNode) ([]*Chang
106
107 // strip out unchanged stuff
108 for _, lnk := range a.Links() {
107 - l, err := b.GetNodeLink(lnk.Name)
109 + l, _, err := b.ResolveLink([]string{lnk.Name})
110 if err == nil {
111 if l.Cid.Equals(lnk.Cid) {
112 // no change... ignore it
merkledag/utils/diffenum.go new
+75
@@ -0,0 +1,75 @@
1 +package dagutils
2 +
3 +import (
4 + "context"
5 + "fmt"
6 +
7 + mdag "github.com/ipfs/go-ipfs/merkledag"
8 +
9 + node "github.com/ipfs/go-ipld-node"
10 + cid "gx/ipfs/QmYhQaCYEcaPPjxJX7YcPcVKkQfRy6sJ7B3XmGFk82XYdQ/go-cid"
11 +)
12 +
13 +// DiffEnumerate fetches every object in the graph pointed to by 'to' that is
14 +// not in 'from'. This can be used to more efficiently fetch a graph if you can
15 +// guarantee you already have the entirety of 'from'
16 +func DiffEnumerate(ctx context.Context, dserv node.NodeGetter, from, to *cid.Cid) error {
17 + fnd, err := dserv.Get(ctx, from)
18 + if err != nil {
19 + return fmt.Errorf("get %s: %s", from, err)
20 + }
21 +
22 + tnd, err := dserv.Get(ctx, to)
23 + if err != nil {
24 + return fmt.Errorf("get %s: %s", to, err)
25 + }
26 +
27 + diff := getLinkDiff(fnd, tnd)
28 +
29 + sset := cid.NewSet()
30 + for _, c := range diff {
31 + if c.a == nil {
32 + err := mdag.EnumerateChildrenAsync(ctx, mdag.GetLinksDirect(dserv), c.b, sset.Visit)
33 + if err != nil {
34 + return err
35 + }
36 + } else {
37 + err := DiffEnumerate(ctx, dserv, c.a, c.b)
38 + if err != nil {
39 + return err
40 + }
41 + }
42 + }
43 +
44 + return nil
45 +}
46 +
47 +type diffpair struct {
48 + a, b *cid.Cid
49 +}
50 +
51 +func getLinkDiff(a, b node.Node) []diffpair {
52 + have := make(map[string]*node.Link)
53 + names := make(map[string]*node.Link)
54 + for _, l := range a.Links() {
55 + have[l.Cid.KeyString()] = l
56 + names[l.Name] = l
57 + }
58 +
59 + var out []diffpair
60 +
61 + for _, l := range b.Links() {
62 + if have[l.Cid.KeyString()] != nil {
63 + continue
64 + }
65 +
66 + match, ok := names[l.Name]
67 + if !ok {
68 + out = append(out, diffpair{b: l.Cid})
69 + continue
70 + }
71 +
72 + out = append(out, diffpair{a: match.Cid, b: l.Cid})
73 + }
74 + return out
75 +}
merkledag/utils/diffenum_test.go new
+149
@@ -0,0 +1,149 @@
1 +package dagutils
2 +
3 +import (
4 + "context"
5 + "fmt"
6 + "testing"
7 +
8 + dag "github.com/ipfs/go-ipfs/merkledag"
9 + mdtest "github.com/ipfs/go-ipfs/merkledag/test"
10 +
11 + node "github.com/ipfs/go-ipld-node"
12 + cid "gx/ipfs/QmYhQaCYEcaPPjxJX7YcPcVKkQfRy6sJ7B3XmGFk82XYdQ/go-cid"
13 +)
14 +
15 +func buildNode(name string, desc map[string]ndesc, out map[string]node.Node) node.Node {
16 + this := desc[name]
17 + nd := new(dag.ProtoNode)
18 + nd.SetData([]byte(name))
19 + for k, v := range this {
20 + child, ok := out[v]
21 + if !ok {
22 + child = buildNode(v, desc, out)
23 + out[v] = child
24 + }
25 +
26 + if err := nd.AddNodeLink(k, child); err != nil {
27 + panic(err)
28 + }
29 + }
30 +
31 + return nd
32 +}
33 +
34 +type ndesc map[string]string
35 +
36 +func mkGraph(desc map[string]ndesc) map[string]node.Node {
37 + out := make(map[string]node.Node)
38 + for name := range desc {
39 + if _, ok := out[name]; ok {
40 + continue
41 + }
42 +
43 + out[name] = buildNode(name, desc, out)
44 + }
45 + return out
46 +}
47 +
48 +var tg1 = map[string]ndesc{
49 + "a1": ndesc{
50 + "foo": "b",
51 + },
52 + "b": ndesc{},
53 + "a2": ndesc{
54 + "foo": "b",
55 + "bar": "c",
56 + },
57 + "c": ndesc{},
58 +}
59 +
60 +var tg2 = map[string]ndesc{
61 + "a1": ndesc{
62 + "foo": "b",
63 + },
64 + "b": ndesc{},
65 + "a2": ndesc{
66 + "foo": "b",
67 + "bar": "c",
68 + },
69 + "c": ndesc{"baz": "d"},
70 + "d": ndesc{},
71 +}
72 +
73 +func TestDiffEnumBasic(t *testing.T) {
74 + ctx, cancel := context.WithCancel(context.Background())
75 + defer cancel()
76 + nds := mkGraph(tg1)
77 +
78 + ds := mdtest.Mock()
79 + lgds := &getLogger{ds: ds}
80 +
81 + for _, nd := range nds {
82 + _, err := ds.Add(nd)
83 + if err != nil {
84 + t.Fatal(err)
85 + }
86 + }
87 +
88 + err := DiffEnumerate(ctx, lgds, nds["a1"].Cid(), nds["a2"].Cid())
89 + if err != nil {
90 + t.Fatal(err)
91 + }
92 +
93 + err = assertCidList(lgds.log, []*cid.Cid{nds["a1"].Cid(), nds["a2"].Cid(), nds["c"].Cid()})
94 + if err != nil {
95 + t.Fatal(err)
96 + }
97 +}
98 +
99 +type getLogger struct {
100 + ds node.NodeGetter
101 + log []*cid.Cid
102 +}
103 +
104 +func (gl *getLogger) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
105 + nd, err := gl.ds.Get(ctx, c)
106 + if err != nil {
107 + return nil, err
108 + }
109 + gl.log = append(gl.log, c)
110 + return nd, nil
111 +}
112 +
113 +func assertCidList(a, b []*cid.Cid) error {
114 + if len(a) != len(b) {
115 + return fmt.Errorf("got different number of cids than expected")
116 + }
117 + for i, c := range a {
118 + if !c.Equals(b[i]) {
119 + return fmt.Errorf("expected %s, got %s", c, b[i])
120 + }
121 + }
122 + return nil
123 +}
124 +func TestDiffEnumFail(t *testing.T) {
125 + ctx, cancel := context.WithCancel(context.Background())
126 + defer cancel()
127 + nds := mkGraph(tg2)
128 +
129 + ds := mdtest.Mock()
130 + lgds := &getLogger{ds: ds}
131 +
132 + for _, s := range []string{"a1", "a2", "b", "c"} {
133 + _, err := ds.Add(nds[s])
134 + if err != nil {
135 + t.Fatal(err)
136 + }
137 + }
138 +
139 + err := DiffEnumerate(ctx, lgds, nds["a1"].Cid(), nds["a2"].Cid())
140 + if err != dag.ErrNotFound {
141 + t.Fatal("expected err not found")
142 + }
143 +
144 + err = assertCidList(lgds.log, []*cid.Cid{nds["a1"].Cid(), nds["a2"].Cid(), nds["c"].Cid()})
145 + if err != nil {
146 + t.Fatal(err)
147 + }
148 +
149 +}
pin/pin.go
+27
@@ -10,6 +10,7 @@ import (
10 "time"
11
12 mdag "github.com/ipfs/go-ipfs/merkledag"
13 + dutils "github.com/ipfs/go-ipfs/merkledag/utils"
14
15 ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
16 logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
@@ -86,6 +87,11 @@ type Pinner interface {
87 Pin(context.Context, node.Node, bool) error
88 Unpin(context.Context, *cid.Cid, bool) error
89
90 + // Update updates a recursive pin from one cid to another
91 + // this is more efficient than simply pinning the new one and unpinning the
92 + // old one
93 + Update(context.Context, *cid.Cid, *cid.Cid, bool) error
94 +
95 // Check if a set of keys are pinned, more efficient than
96 // calling IsPinned for each key
97 CheckIfPinned(cids ...*cid.Cid) ([]Pinned, error)
@@ -94,6 +100,7 @@ type Pinner interface {
100 // care! If used improperly, garbage collection may not be
101 // successful.
102 PinWithMode(*cid.Cid, PinMode)
103 +
104 // RemovePinWithMode is for manually editing the pin structure.
105 // Use with care! If used improperly, garbage collection may not
106 // be successful.
@@ -447,6 +454,26 @@ func (p *pinner) RecursiveKeys() []*cid.Cid {
454 return p.recursePin.Keys()
455 }
456
457 +func (p *pinner) Update(ctx context.Context, from, to *cid.Cid, unpin bool) error {
458 + p.lock.Lock()
459 + defer p.lock.Unlock()
460 +
461 + if !p.recursePin.Has(from) {
462 + return fmt.Errorf("'from' cid was not recursively pinned already")
463 + }
464 +
465 + err := dutils.DiffEnumerate(ctx, p.dserv, from, to)
466 + if err != nil {
467 + return err
468 + }
469 +
470 + p.recursePin.Add(to)
471 + if unpin {
472 + p.recursePin.Remove(from)
473 + }
474 + return nil
475 +}
476 +
477 // Flush encodes and writes pinner keysets to the datastore
478 func (p *pinner) Flush() error {
479 p.lock.Lock()
pin/pin_test.go
+34 -1
@@ -1,6 +1,7 @@
1 package pin
2
3 import (
4 + "context"
5 "testing"
6 "time"
7
@@ -9,7 +10,6 @@ import (
10 "github.com/ipfs/go-ipfs/exchange/offline"
11 mdag "github.com/ipfs/go-ipfs/merkledag"
12
12 - context "context"
13 ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
14 dssync "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/sync"
15 "gx/ipfs/QmWbjfz3u6HkAdPh34dgPchGbQjob6LXLhAeCGii2TX69n/go-ipfs-util"
@@ -367,3 +367,36 @@ func TestPinRecursiveFail(t *testing.T) {
367 t.Fatal(err)
368 }
369 }
370 +
371 +func TestPinUpdate(t *testing.T) {
372 + dstore := dssync.MutexWrap(ds.NewMapDatastore())
373 + bstore := blockstore.NewBlockstore(dstore)
374 + bserv := bs.New(bstore, offline.Exchange(bstore))
375 +
376 + dserv := mdag.NewDAGService(bserv)
377 + p := NewPinner(dstore, dserv, dserv)
378 + n1, c1 := randNode()
379 + n2, c2 := randNode()
380 +
381 + dserv.Add(n1)
382 + dserv.Add(n2)
383 +
384 + ctx := context.Background()
385 + if err := p.Pin(ctx, n1, true); err != nil {
386 + t.Fatal(err)
387 + }
388 +
389 + if err := p.Update(ctx, c1, c2, true); err != nil {
390 + t.Fatal(err)
391 + }
392 +
393 + assertPinned(t, p, c2, "c2 should be pinned now")
394 + assertUnpinned(t, p, c1, "c1 should no longer be pinned")
395 +
396 + if err := p.Update(ctx, c2, c1, false); err != nil {
397 + t.Fatal(err)
398 + }
399 +
400 + assertPinned(t, p, c2, "c2 should be pinned still")
401 + assertPinned(t, p, c1, "c1 should be pinned now")
402 +}