master
go 631 lines 14 KB
Raw
1 package coreunix
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "os"
9 gopath "path"
10 "strconv"
11 "time"
12
13 bstore "github.com/ipfs/boxo/blockstore"
14 chunker "github.com/ipfs/boxo/chunker"
15 "github.com/ipfs/boxo/files"
16 posinfo "github.com/ipfs/boxo/filestore/posinfo"
17 dag "github.com/ipfs/boxo/ipld/merkledag"
18 "github.com/ipfs/boxo/ipld/unixfs"
19 "github.com/ipfs/boxo/ipld/unixfs/importer/balanced"
20 ihelper "github.com/ipfs/boxo/ipld/unixfs/importer/helpers"
21 "github.com/ipfs/boxo/ipld/unixfs/importer/trickle"
22 uio "github.com/ipfs/boxo/ipld/unixfs/io"
23 "github.com/ipfs/boxo/mfs"
24 "github.com/ipfs/boxo/path"
25 pin "github.com/ipfs/boxo/pinning/pinner"
26 "github.com/ipfs/go-cid"
27 ipld "github.com/ipfs/go-ipld-format"
28 logging "github.com/ipfs/go-log/v2"
29 "github.com/ipfs/kubo/config"
30 coreiface "github.com/ipfs/kubo/core/coreiface"
31
32 "github.com/ipfs/kubo/tracing"
33 )
34
35 var log = logging.Logger("coreunix")
36
37 // how many bytes of progress to wait before sending a progress update message
38 const progressReaderIncrement = 1024 * 256
39
40 var liveCacheSize = uint64(256 << 10)
41
42 type Link struct {
43 Name, Hash string
44 Size uint64
45 }
46
47 type syncer interface {
48 Sync() error
49 }
50
51 // NewAdder Returns a new Adder used for a file add operation.
52 func NewAdder(ctx context.Context, p pin.Pinner, bs bstore.GCLocker, ds ipld.DAGService) (*Adder, error) {
53 bufferedDS := ipld.NewBufferedDAG(ctx, ds)
54
55 return &Adder{
56 ctx: ctx,
57 pinning: p,
58 gcLocker: bs,
59 dagService: ds,
60 bufferedDS: bufferedDS,
61 Progress: false,
62 Pin: true,
63 Trickle: false,
64 MaxLinks: ihelper.DefaultLinksPerBlock,
65 MaxHAMTFanout: uio.DefaultShardWidth,
66 Chunker: "",
67 IncludeEmptyDirs: config.DefaultUnixFSIncludeEmptyDirs,
68 }, nil
69 }
70
71 // Adder holds the switches passed to the `add` command.
72 type Adder struct {
73 ctx context.Context
74 pinning pin.Pinner
75 gcLocker bstore.GCLocker
76 dagService ipld.DAGService
77 bufferedDS *ipld.BufferedDAG
78 Out chan<- any
79 Progress bool
80 Pin bool
81 PinName string
82 Trickle bool
83 RawLeaves bool
84 MaxLinks int
85 MaxDirectoryLinks int
86 MaxHAMTFanout int
87 SizeEstimationMode *uio.SizeEstimationMode
88 Silent bool
89 NoCopy bool
90 Chunker string
91 mroot *mfs.Root
92 unlocker bstore.Unlocker
93 tempRoot cid.Cid
94 CidBuilder cid.Builder
95 liveNodes uint64
96
97 PreserveMode bool
98 PreserveMtime bool
99 FileMode os.FileMode
100 FileMtime time.Time
101 IncludeEmptyDirs bool
102 }
103
104 func (adder *Adder) mfsRoot() (*mfs.Root, error) {
105 if adder.mroot != nil {
106 return adder.mroot, nil
107 }
108
109 // Note, this adds it to DAGService already.
110 mr, err := mfs.NewEmptyRoot(adder.ctx, adder.dagService, nil, nil, adder.mkdirOpts()...)
111 if err != nil {
112 return nil, err
113 }
114 adder.mroot = mr
115 return adder.mroot, nil
116 }
117
118 // SetMfsRoot sets `r` as the root for Adder.
119 func (adder *Adder) SetMfsRoot(r *mfs.Root) {
120 adder.mroot = r
121 }
122
123 // mkdirOpts returns MFS options derived from the adder's config,
124 // with any additional options appended.
125 func (adder *Adder) mkdirOpts(extra ...mfs.Option) []mfs.Option {
126 opts := []mfs.Option{
127 mfs.WithCidBuilder(adder.CidBuilder),
128 mfs.WithMaxLinks(adder.MaxDirectoryLinks),
129 mfs.WithMaxHAMTFanout(adder.MaxHAMTFanout),
130 }
131 if adder.SizeEstimationMode != nil {
132 opts = append(opts, mfs.WithSizeEstimationMode(*adder.SizeEstimationMode))
133 }
134 return append(opts, extra...)
135 }
136
137 // Constructs a node from reader's data, and adds it. Doesn't pin.
138 func (adder *Adder) add(reader io.Reader) (ipld.Node, error) {
139 chnk, err := chunker.FromString(reader, adder.Chunker)
140 if err != nil {
141 return nil, err
142 }
143
144 maxLinks := ihelper.DefaultLinksPerBlock
145 if adder.MaxLinks > 0 {
146 maxLinks = adder.MaxLinks
147 }
148
149 params := ihelper.DagBuilderParams{
150 Dagserv: adder.bufferedDS,
151 RawLeaves: adder.RawLeaves,
152 Maxlinks: maxLinks,
153 NoCopy: adder.NoCopy,
154 CidBuilder: adder.CidBuilder,
155 FileMode: adder.FileMode,
156 FileModTime: adder.FileMtime,
157 }
158
159 db, err := params.New(chnk)
160 if err != nil {
161 return nil, err
162 }
163 var nd ipld.Node
164 if adder.Trickle {
165 nd, err = trickle.Layout(db)
166 } else {
167 nd, err = balanced.Layout(db)
168 }
169 if err != nil {
170 return nil, err
171 }
172
173 return nd, adder.bufferedDS.Commit()
174 }
175
176 // RootNode returns the mfs root node
177 func (adder *Adder) curRootNode() (ipld.Node, error) {
178 mr, err := adder.mfsRoot()
179 if err != nil {
180 return nil, err
181 }
182 root, err := mr.GetDirectory().GetNode()
183 if err != nil {
184 return nil, err
185 }
186
187 // if one root file, use that hash as root.
188 if len(root.Links()) == 1 {
189 nd, err := root.Links()[0].GetNode(adder.ctx, adder.dagService)
190 if err != nil {
191 return nil, err
192 }
193
194 root = nd
195 }
196
197 return root, err
198 }
199
200 // PinRoot recursively pins the root node of Adder with an optional name and
201 // writes the pin state to the backing datastore. If name is empty, the pin
202 // will be created without a name.
203 func (adder *Adder) PinRoot(ctx context.Context, root ipld.Node, name string) error {
204 ctx, span := tracing.Span(ctx, "CoreUnix.Adder", "PinRoot")
205 defer span.End()
206
207 if !adder.Pin {
208 return nil
209 }
210
211 rnk := root.Cid()
212
213 err := adder.dagService.Add(ctx, root)
214 if err != nil {
215 return err
216 }
217
218 if adder.tempRoot.Defined() {
219 err := adder.pinning.Unpin(ctx, adder.tempRoot, true)
220 if err != nil {
221 return err
222 }
223 adder.tempRoot = rnk
224 }
225
226 err = adder.pinning.PinWithMode(ctx, rnk, pin.Recursive, name)
227 if err != nil {
228 return err
229 }
230
231 return adder.pinning.Flush(ctx)
232 }
233
234 func (adder *Adder) outputDirs(path string, fsn mfs.FSNode) error {
235 switch fsn := fsn.(type) {
236 case *mfs.File:
237 return nil
238 case *mfs.Directory:
239 names, err := fsn.ListNames(adder.ctx)
240 if err != nil {
241 return err
242 }
243
244 for _, name := range names {
245 child, err := fsn.Child(name)
246 if err != nil {
247 return err
248 }
249
250 childpath := gopath.Join(path, name)
251 err = adder.outputDirs(childpath, child)
252 if err != nil {
253 return err
254 }
255
256 fsn.Uncache(name)
257 }
258 nd, err := fsn.GetNode()
259 if err != nil {
260 return err
261 }
262
263 return outputDagnode(adder.Out, path, nd)
264 default:
265 return fmt.Errorf("unrecognized fsn type: %#v", fsn)
266 }
267 }
268
269 func (adder *Adder) addNode(node ipld.Node, path string) error {
270 // patch it into the root
271 if path == "" {
272 path = node.Cid().String()
273 }
274
275 if pi, ok := node.(*posinfo.FilestoreNode); ok {
276 node = pi.Node
277 }
278
279 mr, err := adder.mfsRoot()
280 if err != nil {
281 return err
282 }
283
284 dir := gopath.Dir(path)
285 if dir != "." {
286 mkdirOpts := adder.mkdirOpts()
287 if err := mfs.Mkdir(mr, dir, mfs.MkdirOpts{Mkparents: true, Flush: false}, mkdirOpts...); err != nil {
288 return err
289 }
290 }
291
292 if err := mfs.PutNode(mr, path, node); err != nil {
293 return err
294 }
295
296 if !adder.Silent {
297 return outputDagnode(adder.Out, path, node)
298 }
299 return nil
300 }
301
302 // AddAllAndPin adds the given request's files and pin them.
303 func (adder *Adder) AddAllAndPin(ctx context.Context, file files.Node) (ipld.Node, error) {
304 ctx, span := tracing.Span(ctx, "CoreUnix.Adder", "AddAllAndPin")
305 defer span.End()
306
307 if adder.Pin {
308 adder.unlocker = adder.gcLocker.PinLock(ctx)
309 }
310 defer func() {
311 if adder.unlocker != nil {
312 adder.unlocker.Unlock(ctx)
313 }
314 }()
315
316 if err := adder.addFileNode(ctx, "", file, true); err != nil {
317 return nil, err
318 }
319
320 // get root
321 mr, err := adder.mfsRoot()
322 if err != nil {
323 return nil, err
324 }
325 var root mfs.FSNode
326 rootdir := mr.GetDirectory()
327 root = rootdir
328
329 err = root.Flush()
330 if err != nil {
331 return nil, err
332 }
333
334 // if adding a file without wrapping, swap the root to it (when adding a
335 // directory, mfs root is the directory)
336 _, dir := file.(files.Directory)
337 var name string
338 if !dir {
339 children, err := rootdir.ListNames(adder.ctx)
340 if err != nil {
341 return nil, err
342 }
343
344 if len(children) == 0 {
345 return nil, fmt.Errorf("expected at least one child dir, got none")
346 }
347
348 // Replace root with the first child
349 name = children[0]
350 root, err = rootdir.Child(name)
351 if err != nil {
352 return nil, err
353 }
354 }
355
356 err = mr.Close()
357 if err != nil {
358 return nil, err
359 }
360
361 nd, err := root.GetNode()
362 if err != nil {
363 return nil, err
364 }
365
366 // output directory events
367 err = adder.outputDirs(name, root)
368 if err != nil {
369 return nil, err
370 }
371
372 if asyncDagService, ok := adder.dagService.(syncer); ok {
373 err = asyncDagService.Sync()
374 if err != nil {
375 return nil, err
376 }
377 }
378
379 if !adder.Pin {
380 return nd, nil
381 }
382
383 if err := adder.PinRoot(ctx, nd, adder.PinName); err != nil {
384 return nil, err
385 }
386
387 return nd, nil
388 }
389
390 func (adder *Adder) addFileNode(ctx context.Context, path string, file files.Node, toplevel bool) error {
391 ctx, span := tracing.Span(ctx, "CoreUnix.Adder", "AddFileNode")
392 defer span.End()
393
394 defer file.Close()
395
396 err := adder.maybePauseForGC(ctx)
397 if err != nil {
398 return err
399 }
400
401 if adder.PreserveMtime {
402 adder.FileMtime = file.ModTime()
403 }
404
405 if adder.PreserveMode {
406 adder.FileMode = file.Mode()
407 }
408
409 if adder.liveNodes >= liveCacheSize {
410 // TODO: A smarter cache that uses some sort of lru cache with an eviction handler
411 mr, err := adder.mfsRoot()
412 if err != nil {
413 return err
414 }
415 if err := mr.FlushMemFree(adder.ctx); err != nil {
416 return err
417 }
418
419 adder.liveNodes = 0
420 }
421 adder.liveNodes++
422
423 switch f := file.(type) {
424 case files.Directory:
425 return adder.addDir(ctx, path, f, toplevel)
426 case *files.Symlink:
427 return adder.addSymlink(ctx, path, f)
428 case files.File:
429 return adder.addFile(path, f)
430 default:
431 return errors.New("unknown file type")
432 }
433 }
434
435 func (adder *Adder) addSymlink(ctx context.Context, path string, l *files.Symlink) error {
436 sdata, err := unixfs.SymlinkData(l.Target)
437 if err != nil {
438 return err
439 }
440
441 if !adder.FileMtime.IsZero() {
442 fsn, err := unixfs.FSNodeFromBytes(sdata)
443 if err != nil {
444 return err
445 }
446
447 fsn.SetModTime(adder.FileMtime)
448 if sdata, err = fsn.GetBytes(); err != nil {
449 return err
450 }
451 }
452
453 dagnode := dag.NodeWithData(sdata)
454 err = dagnode.SetCidBuilder(adder.CidBuilder)
455 if err != nil {
456 return err
457 }
458 err = adder.dagService.Add(adder.ctx, dagnode)
459 if err != nil {
460 return err
461 }
462
463 return adder.addNode(dagnode, path)
464 }
465
466 func (adder *Adder) addFile(path string, file files.File) error {
467 // if the progress flag was specified, wrap the file so that we can send
468 // progress updates to the client (over the output channel)
469 var reader io.Reader = file
470 if adder.Progress {
471 rdr := &progressReader{file: reader, path: path, out: adder.Out}
472 if fi, ok := file.(files.FileInfo); ok {
473 reader = &progressReader2{rdr, fi}
474 } else {
475 reader = rdr
476 }
477 }
478
479 dagnode, err := adder.add(reader)
480 if err != nil {
481 return err
482 }
483
484 // patch it into the root
485 return adder.addNode(dagnode, path)
486 }
487
488 func (adder *Adder) addDir(ctx context.Context, path string, dir files.Directory, toplevel bool) error {
489 log.Infof("adding directory: %s", path)
490
491 // Peek at first entry to check if directory is empty.
492 // We advance the iterator once here and continue from this position
493 // in the processing loop below. This avoids allocating a slice to
494 // collect all entries just to check for emptiness.
495 it := dir.Entries()
496 hasEntry := it.Next()
497 if !hasEntry {
498 if err := it.Err(); err != nil {
499 return err
500 }
501 // Directory is empty. Skip it unless IncludeEmptyDirs is set or
502 // this is the toplevel directory (we always include the root).
503 if !adder.IncludeEmptyDirs && !toplevel {
504 log.Debugf("skipping empty directory: %s", path)
505 return nil
506 }
507 }
508
509 // if we need to store mode or modification time then create a new root which includes that data
510 if toplevel && (adder.FileMode != 0 || !adder.FileMtime.IsZero()) {
511 opts := adder.mkdirOpts(mfs.WithMode(adder.FileMode), mfs.WithModTime(adder.FileMtime))
512 mr, err := mfs.NewEmptyRoot(ctx, adder.dagService, nil, nil, opts...)
513 if err != nil {
514 return err
515 }
516 adder.SetMfsRoot(mr)
517 }
518
519 if !(toplevel && path == "") {
520 mr, err := adder.mfsRoot()
521 if err != nil {
522 return err
523 }
524 mkdirOpts := adder.mkdirOpts(mfs.WithMode(adder.FileMode), mfs.WithModTime(adder.FileMtime))
525 err = mfs.Mkdir(mr, path, mfs.MkdirOpts{Mkparents: true, Flush: false}, mkdirOpts...)
526 if err != nil {
527 return err
528 }
529 }
530
531 // Process directory entries. The iterator was already advanced once above
532 // to peek for emptiness, so we start from that position.
533 for hasEntry {
534 fpath := gopath.Join(path, it.Name())
535 if err := adder.addFileNode(ctx, fpath, it.Node(), false); err != nil {
536 return err
537 }
538 hasEntry = it.Next()
539 }
540
541 return it.Err()
542 }
543
544 func (adder *Adder) maybePauseForGC(ctx context.Context) error {
545 ctx, span := tracing.Span(ctx, "CoreUnix.Adder", "MaybePauseForGC")
546 defer span.End()
547
548 if adder.unlocker != nil && adder.gcLocker.GCRequested(ctx) {
549 rn, err := adder.curRootNode()
550 if err != nil {
551 return err
552 }
553
554 err = adder.PinRoot(ctx, rn, "")
555 if err != nil {
556 return err
557 }
558
559 adder.unlocker.Unlock(ctx)
560 adder.unlocker = adder.gcLocker.PinLock(ctx)
561 }
562 return nil
563 }
564
565 // outputDagnode sends dagnode info over the output channel
566 func outputDagnode(out chan<- any, name string, dn ipld.Node) error {
567 if out == nil {
568 return nil
569 }
570
571 o, err := getOutput(dn)
572 if err != nil {
573 return err
574 }
575
576 out <- &coreiface.AddEvent{
577 Path: o.Path,
578 Name: name,
579 Size: o.Size,
580 }
581
582 return nil
583 }
584
585 // from core/commands/object.go
586 func getOutput(dagnode ipld.Node) (*coreiface.AddEvent, error) {
587 c := dagnode.Cid()
588 s, err := dagnode.Size()
589 if err != nil {
590 return nil, err
591 }
592
593 output := &coreiface.AddEvent{
594 Path: path.FromCid(c),
595 Size: strconv.FormatUint(s, 10),
596 }
597
598 return output, nil
599 }
600
601 type progressReader struct {
602 file io.Reader
603 path string
604 out chan<- any
605 bytes int64
606 lastProgress int64
607 }
608
609 func (i *progressReader) Read(p []byte) (int, error) {
610 n, err := i.file.Read(p)
611
612 i.bytes += int64(n)
613 if i.bytes-i.lastProgress >= progressReaderIncrement || err == io.EOF {
614 i.lastProgress = i.bytes
615 i.out <- &coreiface.AddEvent{
616 Name: i.path,
617 Bytes: i.bytes,
618 }
619 }
620
621 return n, err
622 }
623
624 type progressReader2 struct {
625 *progressReader
626 files.FileInfo
627 }
628
629 func (i *progressReader2) Read(p []byte) (int, error) {
630 return i.progressReader.Read(p)
631 }