feat: improve dag/import (#9721)
- don't bypass the CoreApi - don't use a goroutine and return channel for `importWorker`, when what's happening is really just a synchronous call - only `PinLock()` when we are going to pin - use `cid.Set` instead of an explicit map - fail the request early if any pinning fail, no need to try to pin more if the request failed already
Michael Muré committed
Mar 22, 2023 at 03:06 UTC
1457b4fd4abff0bdcdccbacc26934cb7fa8e8a43
2 files changed
+57
-128
core/commands/dag/dag.go
-7
@@ -168,13 +168,6 @@ var DagResolveCmd = &cmds.Command{
168
Type: ResolveOutput{},
169
}
170
171
-type importResult struct {
172
- blockCount uint64
173
- blockBytesCount uint64
174
- roots map[cid.Cid]struct{}
175
- err error
176
-}
177
-
171
// DagImportCmd is a command for importing a car to ipfs
172
var DagImportCmd = &cmds.Command{
173
Helptext: cmds.HelpText{
core/commands/dag/import.go
+57
-121
@@ -2,24 +2,22 @@ package dagcmd
2
3
import (
4
"errors"
5
- "fmt"
5
"io"
6
7
cid "github.com/ipfs/go-cid"
8
+ cmds "github.com/ipfs/go-ipfs-cmds"
9
ipld "github.com/ipfs/go-ipld-format"
10
ipldlegacy "github.com/ipfs/go-ipld-legacy"
11
"github.com/ipfs/go-libipfs/files"
12
- iface "github.com/ipfs/interface-go-ipfs-core"
12
"github.com/ipfs/interface-go-ipfs-core/options"
13
+ "github.com/ipfs/interface-go-ipfs-core/path"
14
+ gocarv2 "github.com/ipld/go-car/v2"
15
+
16
"github.com/ipfs/kubo/core/commands/cmdenv"
17
"github.com/ipfs/kubo/core/commands/cmdutils"
16
-
17
- cmds "github.com/ipfs/go-ipfs-cmds"
18
- gocarv2 "github.com/ipld/go-car/v2"
18
)
19
20
func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
22
-
21
node, err := cmdenv.GetNode(env)
22
if err != nil {
23
return err
@@ -38,127 +36,42 @@ func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment
36
return err
37
}
38
39
+ doPinRoots, _ := req.Options[pinRootsOptionName].(bool)
40
+
41
// grab a pinlock ( which doubles as a GC lock ) so that regardless of the
42
// size of the streamed-in cars nothing will disappear on us before we had
43
// a chance to roots that may show up at the very end
44
// This is especially important for use cases like dagger:
45
// ipfs dag import $( ... | ipfs-dagger --stdout=carfifos )
46
//
47
- unlocker := node.Blockstore.PinLock(req.Context)
48
- defer unlocker.Unlock(req.Context)
49
-
50
- doPinRoots, _ := req.Options[pinRootsOptionName].(bool)
51
-
52
- retCh := make(chan importResult, 1)
53
- go importWorker(req, res, api, retCh)
54
-
55
- done := <-retCh
56
- if done.err != nil {
57
- return done.err
58
- }
59
-
60
- // It is not guaranteed that a root in a header is actually present in the same ( or any )
61
- // .car file. This is the case in version 1, and ideally in further versions too
62
- // Accumulate any root CID seen in a header, and supplement its actual node if/when encountered
63
- // We will attempt a pin *only* at the end in case all car files were well formed
64
- //
65
- // The boolean value indicates whether we have encountered the root within the car file's
66
- roots := done.roots
67
-
68
- // opportunistic pinning: try whatever sticks
47
if doPinRoots {
70
-
71
- var failedPins int
72
- for c := range roots {
73
-
74
- // We need to re-retrieve a block, convert it to ipld, and feed it
75
- // to the Pinning interface, sigh...
76
- //
77
- // If we didn't have the problem of inability to take multiple pinlocks,
78
- // we could use the api directly like so (though internally it does the same):
79
- //
80
- // // not ideal, but the pinning api takes only paths :(
81
- // rp := path.NewResolvedPath(
82
- // ipfspath.FromCid(c),
83
- // c,
84
- // c,
85
- // "",
86
- // )
87
- //
88
- // if err := api.Pin().Add(req.Context, rp, options.Pin.Recursive(true)); err != nil {
89
-
90
- ret := RootMeta{Cid: c}
91
-
92
- if block, err := node.Blockstore.Get(req.Context, c); err != nil {
93
- ret.PinErrorMsg = err.Error()
94
- } else if nd, err := ipldlegacy.DecodeNode(req.Context, block); err != nil {
95
- ret.PinErrorMsg = err.Error()
96
- } else if err := node.Pinning.Pin(req.Context, nd, true); err != nil {
97
- ret.PinErrorMsg = err.Error()
98
- } else if err := node.Pinning.Flush(req.Context); err != nil {
99
- ret.PinErrorMsg = err.Error()
100
- }
101
-
102
- if ret.PinErrorMsg != "" {
103
- failedPins++
104
- }
105
-
106
- if err := res.Emit(&CarImportOutput{Root: &ret}); err != nil {
107
- return err
108
- }
109
- }
110
-
111
- if failedPins > 0 {
112
- return fmt.Errorf(
113
- "unable to pin all roots: %d out of %d failed",
114
- failedPins,
115
- len(roots),
116
- )
117
- }
118
- }
119
-
120
- stats, _ := req.Options[statsOptionName].(bool)
121
- if stats {
122
- err = res.Emit(&CarImportOutput{
123
- Stats: &CarImportStats{
124
- BlockCount: done.blockCount,
125
- BlockBytesCount: done.blockBytesCount,
126
- },
127
- })
128
- if err != nil {
129
- return err
130
- }
48
+ unlocker := node.Blockstore.PinLock(req.Context)
49
+ defer unlocker.Unlock(req.Context)
50
}
51
133
- return nil
134
-}
135
-
136
-func importWorker(req *cmds.Request, re cmds.ResponseEmitter, api iface.CoreAPI, ret chan importResult) {
137
-
52
// this is *not* a transaction
53
// it is simply a way to relieve pressure on the blockstore
54
// similar to pinner.Pin/pinner.Flush
55
batch := ipld.NewBatch(req.Context, api.Dag())
56
143
- roots := make(map[cid.Cid]struct{})
57
+ roots := cid.NewSet()
58
var blockCount, blockBytesCount uint64
59
60
it := req.Files.Entries()
61
for it.Next() {
148
-
62
file := files.FileFromEntry(it)
63
if file == nil {
151
- ret <- importResult{err: errors.New("expected a file handle")}
152
- return
64
+ return errors.New("expected a file handle")
65
}
66
155
- // wrap a defer-closer-scope
156
- //
157
- // every single file in it() is already open before we start
158
- // just close here sooner rather than later for neatness
159
- // and to surface potential errors writing on closed fifos
160
- // this won't/can't help with not running out of handles
161
- err := func() error {
67
+ // import blocks
68
+ err = func() error {
69
+ // wrap a defer-closer-scope
70
+ //
71
+ // every single file in it() is already open before we start
72
+ // just close here sooner rather than later for neatness
73
+ // and to surface potential errors writing on closed fifos
74
+ // this won't/can't help with not running out of handles
75
defer file.Close()
76
77
car, err := gocarv2.NewBlockReader(file)
@@ -167,7 +80,7 @@ func importWorker(req *cmds.Request, re cmds.ResponseEmitter, api iface.CoreAPI,
80
}
81
82
for _, c := range car.Roots {
170
- roots[c] = struct{}{}
83
+ roots.Add(c)
84
}
85
86
for {
@@ -193,28 +106,51 @@ func importWorker(req *cmds.Request, re cmds.ResponseEmitter, api iface.CoreAPI,
106
blockCount++
107
blockBytesCount += uint64(len(block.RawData()))
108
}
196
-
109
return nil
110
}()
111
+ }
112
200
- if err != nil {
201
- ret <- importResult{err: err}
202
- return
203
- }
113
+ if err := batch.Commit(); err != nil {
114
+ return err
115
}
116
206
- if err := it.Err(); err != nil {
207
- ret <- importResult{err: err}
208
- return
117
+ // It is not guaranteed that a root in a header is actually present in the same ( or any )
118
+ // .car file. This is the case in version 1, and ideally in further versions too.
119
+ // Accumulate any root CID seen in a header, and supplement its actual node if/when encountered
120
+ // We will attempt a pin *only* at the end in case all car files were well-formed.
121
+
122
+ // opportunistic pinning: try whatever sticks
123
+ if doPinRoots {
124
+ err = roots.ForEach(func(c cid.Cid) error {
125
+ ret := RootMeta{Cid: c}
126
+
127
+ // This will trigger a full read of the DAG in the pinner, to make sure we have all blocks.
128
+ // Ideally we would have a lighter merkledag.Walk() instead of the underlying merkledag.FetchDag,
129
+ // then pinner.PinWithMode().
130
+ err = api.Pin().Add(req.Context, path.IpldPath(c), options.Pin.Recursive(true))
131
+ if err != nil {
132
+ return err
133
+ }
134
+
135
+ return res.Emit(&CarImportOutput{Root: &ret})
136
+ })
137
+ if err != nil {
138
+ return err
139
+ }
140
}
141
211
- if err := batch.Commit(); err != nil {
212
- ret <- importResult{err: err}
213
- return
142
+ stats, _ := req.Options[statsOptionName].(bool)
143
+ if stats {
144
+ err = res.Emit(&CarImportOutput{
145
+ Stats: &CarImportStats{
146
+ BlockCount: blockCount,
147
+ BlockBytesCount: blockBytesCount,
148
+ },
149
+ })
150
+ if err != nil {
151
+ return err
152
+ }
153
}
154
216
- ret <- importResult{
217
- blockCount: blockCount,
218
- blockBytesCount: blockBytesCount,
219
- roots: roots}
155
+ return nil
156
}