coreapi: dag: Batching interface
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Aug 3, 2018 at 18:06 UTC
dfd4f94bf69d3e66587fbb82dd1a1d7b8a970fad
2 files changed
+81
-22
core/coreapi/dag.go
+63
-19
@@ -4,6 +4,7 @@ import (
4
"context"
5
"fmt"
6
"io"
7
+ "sync"
8
9
gopath "path"
10
@@ -17,34 +18,25 @@ import (
18
19
type DagAPI CoreAPI
20
21
+type dagBatch struct {
22
+ api *DagAPI
23
+ toPut []ipld.Node
24
+
25
+ lk sync.Mutex
26
+}
27
+
28
// Put inserts data using specified format and input encoding. Unless used with
29
// `WithCodes` or `WithHash`, the defaults "dag-cbor" and "sha256" are used.
30
// Returns the path of the inserted data.
31
func (api *DagAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPutOption) (coreiface.ResolvedPath, error) {
24
- settings, err := caopts.DagPutOptions(opts...)
25
- if err != nil {
26
- return nil, err
27
- }
32
+ nd, err := getNode(src, opts...)
33
29
- codec, ok := cid.CodecToStr[settings.Codec]
30
- if !ok {
31
- return nil, fmt.Errorf("invalid codec %d", settings.Codec)
32
- }
33
-
34
- nds, err := coredag.ParseInputs(settings.InputEnc, codec, src, settings.MhType, settings.MhLength)
34
+ err = api.node.DAG.Add(ctx, nd)
35
if err != nil {
36
return nil, err
37
}
38
- if len(nds) == 0 {
39
- return nil, fmt.Errorf("no node returned from ParseInputs")
40
- }
38
42
- err = api.node.DAG.Add(ctx, nds[0])
43
- if err != nil {
44
- return nil, err
45
- }
46
-
47
- return coreiface.IpldPath(nds[0].Cid()), nil
39
+ return coreiface.IpldPath(nd.Cid()), nil
40
}
41
42
// Get resolves `path` using Unixfs resolver, returns the resolved Node.
@@ -75,6 +67,58 @@ func (api *DagAPI) Tree(ctx context.Context, p coreiface.Path, opts ...caopts.Da
67
return out, nil
68
}
69
70
+func (api *DagAPI) Batch(ctx context.Context) coreiface.DagBatch {
71
+ return &dagBatch{api: api}
72
+}
73
+
74
+func (b *dagBatch) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPutOption) (coreiface.ResolvedPath, error) {
75
+ nd, err := getNode(src, opts...)
76
+ if err != nil {
77
+ return nil, err
78
+ }
79
+
80
+ b.lk.Lock()
81
+ b.toPut = append(b.toPut, nd)
82
+ b.lk.Unlock()
83
+
84
+ return coreiface.IpldPath(nd.Cid()), nil
85
+}
86
+
87
+func (b *dagBatch) Commit(ctx context.Context) error {
88
+ b.lk.Lock()
89
+ defer b.lk.Unlock()
90
+ defer func() {
91
+ b.toPut = nil
92
+ }()
93
+
94
+ return b.api.node.DAG.AddMany(ctx, b.toPut)
95
+}
96
+
97
+func getNode(src io.Reader, opts ...caopts.DagPutOption) (ipld.Node, error) {
98
+ settings, err := caopts.DagPutOptions(opts...)
99
+ if err != nil {
100
+ return nil, err
101
+ }
102
+
103
+ codec, ok := cid.CodecToStr[settings.Codec]
104
+ if !ok {
105
+ return nil, fmt.Errorf("invalid codec %d", settings.Codec)
106
+ }
107
+
108
+ nds, err := coredag.ParseInputs(settings.InputEnc, codec, src, settings.MhType, settings.MhLength)
109
+ if err != nil {
110
+ return nil, err
111
+ }
112
+ if len(nds) == 0 {
113
+ return nil, fmt.Errorf("no node returned from ParseInputs")
114
+ }
115
+ if len(nds) != 1 {
116
+ return nil, fmt.Errorf("got more that one node from ParseInputs")
117
+ }
118
+
119
+ return nds[0], nil
120
+}
121
+
122
func (api *DagAPI) core() coreiface.CoreAPI {
123
return (*CoreAPI)(api)
124
}
core/coreapi/interface/dag.go
+18
-3
@@ -4,21 +4,36 @@ import (
4
"context"
5
"io"
6
7
- options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
7
+ "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
8
9
ipld "gx/ipfs/QmZtNq8dArGfnpCZfx2pUNY7UcjGhVp5qqwQ4hH6mpTMRQ/go-ipld-format"
10
)
11
12
-// DagAPI specifies the interface to IPLD
13
-type DagAPI interface {
12
+// DagOps groups operations that can be batched together
13
+type DagOps interface {
14
// Put inserts data using specified format and input encoding.
15
// Unless used with WithCodec or WithHash, the defaults "dag-cbor" and
16
// "sha256" are used.
17
Put(ctx context.Context, src io.Reader, opts ...options.DagPutOption) (ResolvedPath, error)
18
+}
19
+
20
+// DagBatch is the batching version of DagAPI. All implementations of DagBatch
21
+// should be threadsafe
22
+type DagBatch interface {
23
+ DagOps
24
+
25
+ Commit(ctx context.Context) error
26
+}
27
+
28
+// DagAPI specifies the interface to IPLD
29
+type DagAPI interface {
30
+ DagOps
31
32
// Get attempts to resolve and get the node specified by the path
33
Get(ctx context.Context, path Path) (ipld.Node, error)
34
35
// Tree returns list of paths within a node specified by the path.
36
Tree(ctx context.Context, path Path, opts ...options.DagTreeOption) ([]Path, error)
37
+
38
+ Batch(ctx context.Context) DagBatch
39
}