pin cmd: stream recursive pins
Add a --stream flag to stream the results instead of accumulating the final result in memory. This is a rework of https://github.com/ipfs/go-ipfs/pull/5005
Michael Muré committed
Jun 26, 2019 at 23:07 UTC
bb601845268fa7ffcf90cfe21d5eca9c8b42f590
1 file changed
+107
-44
core/commands/pin.go
+107
-44
@@ -259,8 +259,9 @@ collected if needed. (By default, recursively. Use -r=false for direct pins.)
259
}
260
261
const (
262
- pinTypeOptionName = "type"
263
- pinQuietOptionName = "quiet"
262
+ pinTypeOptionName = "type"
263
+ pinQuietOptionName = "quiet"
264
+ pinStreamOptionName = "stream"
265
)
266
267
var listPinCmd = &cmds.Command{
@@ -313,6 +314,7 @@ Example:
314
Options: []cmds.Option{
315
cmds.StringOption(pinTypeOptionName, "t", "The type of pinned keys to list. Can be \"direct\", \"indirect\", \"recursive\", or \"all\".").WithDefault("all"),
316
cmds.BoolOption(pinQuietOptionName, "q", "Write just hashes of objects."),
317
+ cmds.BoolOption(pinStreamOptionName, "s", "Don't buffer pins before sending."),
318
},
319
Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
320
n, err := cmdenv.GetNode(env)
@@ -326,9 +328,7 @@ Example:
328
}
329
330
typeStr, _ := req.Options[pinTypeOptionName].(string)
329
- if err != nil {
330
- return err
331
- }
331
+ stream, _ := req.Options[pinStreamOptionName].(bool)
332
333
switch typeStr {
334
case "all", "direct", "indirect", "recursive":
@@ -337,34 +337,50 @@ Example:
337
return err
338
}
339
340
- enc, err := cmdenv.GetCidEncoder(req)
341
- if err != nil {
342
- return err
340
+ // For backward compatibility, we accumulate the pins in the same output type as before.
341
+ emit := res.Emit
342
+ lgcList := map[string]RefObject{}
343
+ if !stream {
344
+ emit = func(v interface{}) error {
345
+ obj := v.(*PinLsOutputWrapper)
346
+ lgcList[obj.RefKeyObject.Cid] = RefObject{Type: obj.RefKeyObject.Type}
347
+ return nil
348
+ }
349
}
350
345
- var keys map[cid.Cid]RefKeyObject
351
if len(req.Arguments) > 0 {
347
- keys, err = pinLsKeys(req.Context, req.Arguments, typeStr, n, api)
352
+ err = pinLsKeys(req.Context, req.Arguments, typeStr, n, api, emit)
353
} else {
349
- keys, err = pinLsAll(req.Context, typeStr, n)
354
+ err = pinLsAll(req.Context, typeStr, n, emit)
355
}
356
if err != nil {
357
return err
358
}
359
355
- refKeys := make(map[string]RefKeyObject, len(keys))
356
- for k, v := range keys {
357
- refKeys[enc.Encode(k)] = v
360
+ if !stream {
361
+ return cmds.EmitOnce(res, &PinLsOutputWrapper{
362
+ RefKeyList: RefKeyList{Keys: lgcList},
363
+ })
364
}
365
360
- return cmds.EmitOnce(res, &RefKeyList{Keys: refKeys})
366
+ return nil
367
},
362
- Type: RefKeyList{},
368
+ Type: &PinLsOutputWrapper{},
369
Encoders: cmds.EncoderMap{
364
- cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *RefKeyList) error {
370
+ cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *PinLsOutputWrapper) error {
371
quiet, _ := req.Options[pinQuietOptionName].(bool)
372
+ stream, _ := req.Options[pinStreamOptionName].(bool)
373
+
374
+ if stream {
375
+ if quiet {
376
+ fmt.Fprintf(w, "%s\n", out.RefKeyObject.Cid)
377
+ } else {
378
+ fmt.Fprintf(w, "%s %s\n", out.RefKeyObject.Cid, out.RefKeyObject.Type)
379
+ }
380
+ return nil
381
+ }
382
367
- for k, v := range out.Keys {
383
+ for k, v := range out.RefKeyList.Keys {
384
if quiet {
385
fmt.Fprintf(w, "%s\n", k)
386
} else {
@@ -492,35 +508,44 @@ var verifyPinCmd = &cmds.Command{
508
}
509
510
type RefKeyObject struct {
511
+ Cid string
512
+ Type string
513
+}
514
+
515
+type RefObject struct {
516
Type string
517
}
518
519
type RefKeyList struct {
499
- Keys map[string]RefKeyObject
520
+ Keys map[string]RefObject
521
}
522
502
-func pinLsKeys(ctx context.Context, args []string, typeStr string, n *core.IpfsNode, api coreiface.CoreAPI) (map[cid.Cid]RefKeyObject, error) {
523
+// Pin ls needs to output two different type depending on if it's streamed or not.
524
+// We use this to bypass the cmds lib refusing to have interface{}
525
+type PinLsOutputWrapper struct {
526
+ RefKeyList
527
+ RefKeyObject
528
+}
529
530
+func pinLsKeys(ctx context.Context, args []string, typeStr string, n *core.IpfsNode, api coreiface.CoreAPI, emit func(value interface{}) error) error {
531
mode, ok := pin.StringToMode(typeStr)
532
if !ok {
506
- return nil, fmt.Errorf("invalid pin mode '%s'", typeStr)
533
+ return fmt.Errorf("invalid pin mode '%s'", typeStr)
534
}
535
509
- keys := make(map[cid.Cid]RefKeyObject)
510
-
536
for _, p := range args {
537
c, err := api.ResolvePath(ctx, path.New(p))
538
if err != nil {
514
- return nil, err
539
+ return err
540
}
541
542
pinType, pinned, err := n.Pinning.IsPinnedWithType(c.Cid(), mode)
543
if err != nil {
519
- return nil, err
544
+ return err
545
}
546
547
if !pinned {
523
- return nil, fmt.Errorf("path '%s' is not pinned", p)
548
+ return fmt.Errorf("path '%s' is not pinned", p)
549
}
550
551
switch pinType {
@@ -528,44 +553,82 @@ func pinLsKeys(ctx context.Context, args []string, typeStr string, n *core.IpfsN
553
default:
554
pinType = "indirect through " + pinType
555
}
531
- keys[c.Cid()] = RefKeyObject{
532
- Type: pinType,
556
+
557
+ err = emit(&PinLsOutputWrapper{
558
+ RefKeyObject: RefKeyObject{
559
+ Type: pinType,
560
+ Cid: c.Cid().String(),
561
+ },
562
+ })
563
+ if err != nil {
564
+ return err
565
}
566
}
567
536
- return keys, nil
568
+ return nil
569
}
570
539
-func pinLsAll(ctx context.Context, typeStr string, n *core.IpfsNode) (map[cid.Cid]RefKeyObject, error) {
540
-
541
- keys := make(map[cid.Cid]RefKeyObject)
571
+func pinLsAll(ctx context.Context, typeStr string, n *core.IpfsNode, emit func(value interface{}) error) error {
572
+ keys := cid.NewSet()
573
543
- AddToResultKeys := func(keyList []cid.Cid, typeStr string) {
574
+ AddToResultKeys := func(keyList []cid.Cid, typeStr string) error {
575
for _, c := range keyList {
545
- keys[c] = RefKeyObject{
546
- Type: typeStr,
576
+ if keys.Visit(c) {
577
+ err := emit(&PinLsOutputWrapper{
578
+ RefKeyObject: RefKeyObject{
579
+ Type: typeStr,
580
+ Cid: c.String(),
581
+ },
582
+ })
583
+ if err != nil {
584
+ return err
585
+ }
586
}
587
}
588
+ return nil
589
}
590
591
if typeStr == "direct" || typeStr == "all" {
552
- AddToResultKeys(n.Pinning.DirectKeys(), "direct")
592
+ err := AddToResultKeys(n.Pinning.DirectKeys(), "direct")
593
+ if err != nil {
594
+ return err
595
+ }
596
+ }
597
+ if typeStr == "recursive" || typeStr == "all" {
598
+ err := AddToResultKeys(n.Pinning.RecursiveKeys(), "recursive")
599
+ if err != nil {
600
+ return err
601
+ }
602
}
603
if typeStr == "indirect" || typeStr == "all" {
555
- set := cid.NewSet()
604
for _, k := range n.Pinning.RecursiveKeys() {
557
- err := dag.EnumerateChildren(ctx, dag.GetLinksWithDAG(n.DAG), k, set.Visit)
605
+ var visitErr error
606
+ err := dag.EnumerateChildren(ctx, dag.GetLinksWithDAG(n.DAG), k, func(c cid.Cid) bool {
607
+ r := keys.Visit(c)
608
+ if r {
609
+ err := emit(&PinLsOutputWrapper{
610
+ RefKeyObject: RefKeyObject{
611
+ Type: "indirect",
612
+ Cid: c.String(),
613
+ },
614
+ })
615
+ if err != nil {
616
+ visitErr = err
617
+ }
618
+ }
619
+ return r
620
+ })
621
+
622
+ if visitErr != nil {
623
+ return visitErr
624
+ }
625
if err != nil {
559
- return nil, err
626
+ return err
627
}
628
}
562
- AddToResultKeys(set.Keys(), "indirect")
563
- }
564
- if typeStr == "recursive" || typeStr == "all" {
565
- AddToResultKeys(n.Pinning.RecursiveKeys(), "recursive")
629
}
630
568
- return keys, nil
631
+ return nil
632
}
633
634
// PinVerifyRes is the result returned for each pin checked in "pin verify"