cli

Documentation

Index

• Variables
• func BatchLoadNode(ctx context.Context, unloadedLinks []*format.Link, bs bstore.Blockstore, ...) ([]format.Node, error)
• func BlockWalk(ctx context.Context, node format.Node, bs bstore.Blockstore, batchNum int, ...) error
• func GetAPI(c *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error)
• func PrintProgress(pb chan api.PBar) error
• func ReqContext(cctx *cli.Context) context.Context
• func WithCategory(cat string, cmd *cli.Command) *cli.Command
• func WritePieceV2(ctx context.Context, root cid.Cid, ppath string, bs bstore.Blockstore, ...) error

Variables

var Add2Cmd = &cli.Command{
	Name:  "add2",
	Usage: "add files",
	Flags: []cli.Flag{
		&cli.IntFlag{
			Name:    "batch-read",
			Aliases: []string{"br"},
			Usage:   "",
			Value:   32,
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)

		p, err := homedir.Expand(cctx.Args().First())
		if err != nil {
			return err
		}
		if !strings.HasPrefix(p, "/") {
			if dir, err := os.Getwd(); err == nil {
				p = filepath.Join(dir, p)
			}
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		pb, err := api.Add2(ctx, p, cctx.Int("batch-read"))
		if err != nil {
			return err
		}
		return PrintProgress(pb)

	},
}

var AddCmd = &cli.Command{
	Name:  "add",
	Usage: "add files",
	Flags: []cli.Flag{},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)

		p, err := homedir.Expand(cctx.Args().First())
		if err != nil {
			return err
		}
		if !strings.HasPrefix(p, "/") {
			if dir, err := os.Getwd(); err == nil {
				p = filepath.Join(dir, p)
			}
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		pb, err := api.Add(ctx, p)
		if err != nil {
			return err
		}
		return PrintProgress(pb)

	},
}

var Commands = []*cli.Command{
	AddCmd,
	Add2Cmd,
	GetCmd,
	SyncssCmd,
	importDatasetCmd,
	WithCategory("network", NetCmd),
	WithCategory("dag", DagCmd),
}

var DagCmd = &cli.Command{
	Name:  "dag",
	Usage: "Manage dag",
	Subcommands: []*cli.Command{
		DagStat,
		DagSync,
		DagExport,
		DagImport,
		DagImport2,
		DagHas,
		DagGenPieces,
	},
}

var DagExport = &cli.Command{
	Name:  "export",
	Usage: "export car or padded car file",
	Flags: []cli.Flag{
		&cli.BoolFlag{
			Name:  "swarm",
			Usage: "",
		},
		&cli.BoolFlag{
			Name:    "pad",
			Aliases: []string{"p"},
			Usage:   "filecoin piece pad",
		},
		&cli.IntFlag{
			Name:  "batch",
			Usage: "",
			Value: 32,
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		if len(args) < 2 {
			log.Info("usage: filejoy dag export [cid] [path]")
			return nil
		}
		cid, err := cid.Decode(args[0])
		if err != nil {
			return err
		}
		p, err := homedir.Expand(args[1])
		if err != nil {
			return err
		}
		if !strings.HasPrefix(p, "/") {
			if dir, err := os.Getwd(); err == nil {
				p = filepath.Join(dir, p)
			}
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		pb, err := api.DagExport(ctx, cid, p, cctx.Bool("pad"), cctx.Int("batch"), cctx.Bool("swarm"))
		if err != nil {
			return err
		}

		return PrintProgress(pb)
	},
}

var DagGenPieces = &cli.Command{
	Name:  "gen-pieces",
	Usage: "gen pieces from cid list",
	Flags: []cli.Flag{
		&cli.StringFlag{
			Name:    "dscluster",
			Aliases: []string{"dsc"},
			Usage:   "",
		},
		&cli.IntFlag{
			Name:  "batch",
			Value: 32,
			Usage: "",
		},
		&cli.BoolFlag{
			Name: "flat-path",
		},
		&cli.BoolFlag{
			Name:  "pad",
			Value: true,
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		if len(args) < 2 {
			log.Info("usage: filejoy dag gen-pieces [input-file] [path-to-filestore]")
			return nil
		}
		fileStore := args[1]
		if isDir, err := isDir(fileStore); err != nil || !isDir {
			log.Info("usage: filejoy dag gen-pieces [input-file] [path-to-filestore]")
			return err
		}
		var shouldPad = cctx.Bool("pad")
		var flatPath = cctx.Bool("flat-path")
		var batchNum = cctx.Int("batch")
		var cds datastore.Datastore
		dsclustercfgpath := cctx.String("dscluster")
		dcfg, err := dsccfg.ReadConfig(dsclustercfgpath)
		if err != nil {
			return err
		}

		cds, err = clusterclient.NewClusterClient(ctx, dcfg)
		if err != nil {
			return err
		}
		cds = dsmount.New([]dsmount.Mount{
			{
				Prefix:    bstore.BlockPrefix,
				Datastore: cds,
			},
		})
		blkst := bstore.NewBlockstore(cds.(*dsmount.Datastore))

		cidListFile := args[0]
		f, err := os.Open(cidListFile)
		if err != nil {
			return err
		}
		defer f.Close()
		liner := bufio.NewReader(f)
		for {
			line, err := liner.ReadString('\n')
			if err != nil {
				if err == io.EOF {
					fmt.Println("end with input file")
				} else {
					log.Error(err)
				}
				break
			}
			arr, err := splitSSLine(line)
			if err != nil {
				return err
			}
			cid, err := cid.Decode(arr[0])
			if err != nil {
				return err
			}
			expectSize, err := strconv.ParseInt(strings.TrimSpace(arr[2]), 10, 64)
			if err != nil {
				return err
			}
			var pdir, ppath string
			if flatPath {
				pdir, ppath = flatPiecePath(arr[1], fileStore)
			} else {
				pdir, ppath = piecePath(arr[1], fileStore)
			}
			fmt.Printf("will gen: %s\n", ppath)
			if finfo, err := os.Stat(ppath); err == nil && finfo.Size() == expectSize {
				fmt.Printf("aleady has %s, size: %d, esize: %s; will ignore\n", ppath, finfo.Size(), arr[2])
				continue
			}

			if err = os.MkdirAll(pdir, 0755); err != nil {
				return err
			}

			if err = writePieceV3(ctx, cid, ppath, blkst, batchNum, shouldPad); err != nil {
				log.Error("%s write piece failed: %s", cid, err)
				continue
			}

		}
		return nil

	},
}

var DagHas = &cli.Command{
	Name:  "has",
	Usage: "check if local block store has dag",
	Flags: []cli.Flag{},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)

		cid, err := cid.Decode(cctx.Args().First())
		if err != nil {
			return err
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		has, err := api.DagHas(ctx, cid)
		if err != nil {
			return err
		}
		if has {
			fmt.Printf("has %s\n", cid)
		} else {
			fmt.Printf("hasn't %s\n", cid)
		}

		return nil
	},
}

var DagImport = &cli.Command{
	Name:  "import",
	Usage: "import car file",
	Flags: []cli.Flag{
		&cli.StringFlag{
			Name:  "f",
			Usage: "",
		},
		&cli.StringFlag{
			Name:  "filestore",
			Usage: "",
		},
		&cli.BoolFlag{
			Name:  "delete-source",
			Value: false,
			Usage: "delete the car file been imported",
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		cidsFilePath := cctx.String("f")
		filestorePath := cctx.String("filestore")
		deleteSource := cctx.Bool("delete-source")

		curdir, err := os.Getwd()
		if err != nil {
			return err
		}
		if cidsFilePath != "" && filestorePath != "" {
			if bs, err := ioutil.ReadFile(cidsFilePath); err == nil {
				cidlist := strings.Split(string(bs), "\n")
				for _, cidstr := range cidlist {
					cidstr = strings.TrimSpace(cidstr)
					if cidstr != "" {
						_, cidPath := piecePath(cidstr, filestorePath)

						if !fileExist(cidPath) {
							cidPath = cidPath + ".car"
						}
						if !fileExist(cidPath) {
							log.Infof("piece not exist: %s", cidPath)
							continue
						}
						args = append(args, cidPath)
					}
				}
			}
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		for _, carPath := range args {
			if !strings.HasPrefix(carPath, "/") {
				carPath = filepath.Join(curdir, carPath)
			}
			log.Infof("start to import %s", carPath)
			pb, err := api.DagImport(ctx, carPath)
			if err != nil {
				log.Error(err)
				continue
			}
			err = PrintProgress(pb)
			if err != nil {
				log.Error(err)
				continue
			}

			if deleteSource {
				if err = os.Remove(carPath); err != nil {
					log.Warn(err)
				}
			}
			log.Infof("end with import %s", carPath)
		}

		return err
	},
}

var DagImport2 = &cli.Command{
	Name:  "import2",
	Usage: "import car file",
	Flags: []cli.Flag{
		&cli.StringFlag{
			Name:  "f",
			Usage: "",
		},
		&cli.StringFlag{
			Name:  "filestore",
			Usage: "",
		},
		&cli.BoolFlag{
			Name:  "delete-source",
			Value: false,
			Usage: "delete the car file been imported",
		},
		&cli.StringFlag{
			Name:    "dscluster",
			Aliases: []string{"dsc"},
			Usage:   "",
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		cidsFilePath := cctx.String("f")
		filestorePath := cctx.String("filestore")
		deleteSource := cctx.Bool("delete-source")

		curdir, err := os.Getwd()
		if err != nil {
			return err
		}
		if cidsFilePath != "" && filestorePath != "" {
			if bs, err := ioutil.ReadFile(cidsFilePath); err == nil {
				cidlist := strings.Split(string(bs), "\n")
				for _, cidstr := range cidlist {
					cidstr = strings.TrimSpace(cidstr)
					if cidstr != "" {
						_, cidPath := piecePath(cidstr, filestorePath)

						if !fileExist(cidPath) {
							cidPath = cidPath + ".car"
						}
						if !fileExist(cidPath) {
							log.Infof("piece not exist: %s", cidPath)
							continue
						}
						args = append(args, cidPath)
					}
				}
			}
		}
		var cds datastore.Datastore
		dsclustercfgpath := cctx.String("dscluster")
		dcfg, err := dsccfg.ReadConfig(dsclustercfgpath)
		if err != nil {
			return err
		}

		cds, err = clusterclient.NewClusterClient(ctx, dcfg)
		if err != nil {
			return err
		}
		cds = dsmount.New([]dsmount.Mount{
			{
				Prefix:    bstore.BlockPrefix,
				Datastore: cds,
			},
		})
		blkst := bstore.NewBlockstore(cds.(*dsmount.Datastore))

		for _, carPath := range args {
			if !strings.HasPrefix(carPath, "/") {
				carPath = filepath.Join(curdir, carPath)
			}
			err = func(carPath string, blkst bstore.Blockstore, deleteSource bool) error {
				log.Infof("start to import %s", carPath)
				f, err := os.OpenFile(carPath, os.O_RDWR, 0644)
				if err != nil {
					return err
				}
				defer f.Close()
				finfo, err := f.Stat()
				if err != nil {
					return err
				}
				bar := progressbar.DefaultBytes(
					finfo.Size(),
					"importing",
				)

				br := bufio.NewReader(io.TeeReader(f, bar))
				_, err = gocar.ReadHeader(br)
				if err != nil {
					return err
				}
				for {
					cid, data, err := readDagNode(br)
					if err != nil {
						if err == io.EOF {
							log.Error(err)
						}
						break
					}
					bn, err := blocks.NewBlockWithCid(data, cid)
					if err != nil {
						return err
					}
					if err = blkst.Put(bn); err != nil {
						return err
					}
				}

				if deleteSource {
					if err = os.Remove(carPath); err != nil {
						log.Warn(err)
					}
				}
				log.Infof("end with import %s", carPath)
				return err
			}(carPath, blkst, deleteSource)
			if err != nil {
				log.Error(err)
			}
		}

		return err
	},
}

var DagStat = &cli.Command{
	Name:  "stat",
	Usage: "print dag info",
	Flags: []cli.Flag{
		&cli.IntFlag{
			Name:  "timeout",
			Usage: "",
			Value: 15,
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)

		cid, err := cid.Decode(cctx.Args().First())
		if err != nil {
			return err
		}

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		stat, err := api.DagStat(ctx, cid, cctx.Uint("timeout"))
		if err != nil {
			return err
		}
		fmt.Printf("%v\n", stat)

		return nil
	},
}

var DagSync = &cli.Command{
	Name:  "sync",
	Usage: "sync dags",
	Flags: []cli.Flag{
		&cli.IntFlag{
			Name:    "concurrent",
			Aliases: []string{"c"},
			Usage:   "",
			Value:   32,
		},
		&cli.StringFlag{
			Name:  "f",
			Usage: "",
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		cidsFilePath := cctx.String("f")

		if cidsFilePath != "" {
			if bs, err := ioutil.ReadFile(cidsFilePath); err == nil {
				cidlist := strings.Split(string(bs), "\n")
				for _, cidstr := range cidlist {
					cidstr = strings.TrimSpace(cidstr)
					if cidstr != "" {
						args = append(args, cidstr)
					}
				}
			}
		}

		var cids = make([]cid.Cid, 0)
		for _, cidstr := range args {
			cid, err := cid.Decode(cidstr)
			if err != nil {
				return err
			}
			cids = append(cids, cid)
		}
		fmt.Println(cids)

		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()
		for _, item := range cids {
			msgch, err := api.DagSync(ctx, []cid.Cid{item}, cctx.Int("concurrent"))
			if err != nil {
				return err
			}
			for msg := range msgch {
				fmt.Println(msg)
			}
		}

		return nil
	},
}

var GetCmd = &cli.Command{
	Name:  "get",
	Usage: "get file by cid",
	Flags: []cli.Flag{},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()
		if len(args) < 2 {
			log.Info("usage: filejoy get [cid] [path]")
			return nil
		}
		cid, err := cid.Decode(args[0])
		if err != nil {
			return err
		}
		p, err := homedir.Expand(args[1])
		if err != nil {
			return err
		}
		if !strings.HasPrefix(p, "/") {
			if dir, err := os.Getwd(); err == nil {
				p = filepath.Join(dir, p)
			}
		}
		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()

		pb, err := api.Get(ctx, cid, p)
		if err != nil {
			return err
		}

		return PrintProgress(pb)
	},
}

var NetCmd = &cli.Command{
	Name:  "net",
	Usage: "Manage P2P Network",
	Subcommands: []*cli.Command{
		NetPeers,
		NetConnect,
		NetListen,
		NetId,
	},
}

var NetConnect = &cli.Command{
	Name:      "connect",
	Usage:     "Connect to a peer",
	ArgsUsage: "[peerMultiaddr]",
	Action: func(cctx *cli.Context) error {
		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()
		ctx := ReqContext(cctx)
		args := cctx.Args().Slice()

		pis := make([]*peer.AddrInfo, 0, len(args))
		for _, addr := range args {
			if maddr, err := peer.AddrInfoFromString(addr); err == nil {
				pis = append(pis, maddr)
			}
		}

		for _, pi := range pis {
			fmt.Printf("connect %s: ", pi.ID.Pretty())
			err := api.NetConnect(ctx, *pi)
			if err != nil {
				fmt.Println("failure")
				return err
			}
			fmt.Println("success")
		}

		return nil
	},
}

var NetId = &cli.Command{
	Name:  "id",
	Usage: "Get node identity",
	Action: func(cctx *cli.Context) error {
		api, closer, err := GetAPI(cctx)
		if err != nil {
			log.Errorf("%#v", err)
			return err
		}
		defer closer()

		ctx := ReqContext(cctx)

		pid, err := api.ID(ctx)
		if err != nil {
			return err
		}
		fmt.Println(pid.Pretty())
		return nil
	},
}

var NetListen = &cli.Command{
	Name:  "listen",
	Usage: "List listen addresses",
	Action: func(cctx *cli.Context) error {
		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()
		ctx := ReqContext(cctx)

		addrs, err := api.NetAddrsListen(ctx)
		if err != nil {
			return err
		}

		for _, peer := range addrs.Addrs {
			fmt.Printf("%s/p2p/%s\n", peer, addrs.ID)
		}
		return nil
	},
}

var NetPeers = &cli.Command{
	Name:  "peers",
	Usage: "Print peers",
	Flags: []cli.Flag{},
	Action: func(cctx *cli.Context) error {
		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()
		ctx := ReqContext(cctx)
		peers, err := api.NetPeers(ctx)
		if err != nil {
			return err
		}

		sort.Slice(peers, func(i, j int) bool {
			return strings.Compare(string(peers[i].ID), string(peers[j].ID)) > 0
		})

		for _, peer := range peers {
			fmt.Printf("%s, %s\n", peer.ID, peer.Addrs)
		}

		return nil
	},
}

var SyncssCmd = &cli.Command{
	Name:  "syncss",
	Usage: "sync dataset with snapshot",
	Flags: []cli.Flag{
		&cli.BoolFlag{
			Name:    "only-dag",
			Aliases: []string{"od"},
			Usage:   "",
			Value:   false,
		},
		&cli.BoolFlag{
			Name:    "only-check",
			Aliases: []string{"oc"},
			Usage:   "",
			Value:   false,
		},
		&cli.BoolFlag{
			Name:    "save-snapshot",
			Aliases: []string{"ss"},
			Usage:   "",
			Value:   false,
		},
		&cli.StringFlag{
			Name:    "file-list",
			Aliases: []string{"fl"},
			Usage:   "sync directly from lists in the text file",
		},
		&cli.Int64Flag{
			Name:  "sssize",
			Value: 0,
			Usage: "split snapshot file into slice according to sssize",
		},
	},
	Action: func(cctx *cli.Context) error {
		ctx := ReqContext(cctx)
		sssize := cctx.Int64("sssize")
		onlyDag := cctx.Bool("only-dag")
		onlyCheck := cctx.Bool("only-check")
		args := cctx.Args().Slice()
		if len(args) < 2 && !(onlyCheck || onlyDag || sssize > 0) {
			log.Info("usage: filejoy syncss [snapshot-cid] [target-path]")
			return nil
		}
		var err error
		api, closer, err := GetAPI(cctx)
		if err != nil {
			return err
		}
		defer closer()
		var sscidstr string
		if len(args) > 0 {
			sscidstr = args[0]
		}
		var p string

		var totalLine, checkedLine, errLine int
		var sr *bufio.Reader
		fromFileList := cctx.String("file-list")
		if fromFileList != "" {
			f, err := os.Open(fromFileList)
			if err != nil {
				return err
			}
			defer f.Close()
			sr = bufio.NewReader(f)
		} else {
			if len(args) > 1 {
				p, err = homedir.Expand(args[1])
				if err != nil {
					return err
				}
			}
			if !strings.HasPrefix(p, "/") {
				if dir, err := os.Getwd(); err == nil {
					p = filepath.Join(dir, p)
				}
			}
			sscid, err := cid.Decode(sscidstr)
			if err != nil {
				return err
			}

			ssfn := filepath.Join(os.TempDir(), args[0])

			log.Infof("loading snapshot file to %s", ssfn)
			{
				pb, err := api.Get(ctx, sscid, ssfn)
				if err != nil {
					return err
				}
				if err := PrintProgress(pb); err != nil {
					return err
				}
			}

			f, err := os.Open(ssfn)
			if err != nil {
				return err
			}
			defer f.Close()
			var iorder io.Reader
			if cctx.Bool("save-snapshot") {
				cw, err := os.Getwd()
				if err != nil {
					return err
				}
				targetpath := filepath.Join(cw, args[0]+".txt")
				sf, err := os.Create(targetpath)
				if err != nil {
					return err
				}
				defer sf.Close()
				iorder = io.TeeReader(lz4.NewReader(f), sf)
			} else {
				iorder = lz4.NewReader(f)
			}
			sr = bufio.NewReader(iorder)
		}

		slice_index := 0
		slice_line_cache := make([]string, 0)
		slice_size := int64(0)
		for {
			line, err := sr.ReadString('\n')
			if err != nil {
				if err == io.EOF {
					fmt.Println("end with snapshot file")
				} else {
					log.Error(err)
				}
				break
			}
			log.Info(line)
			arr, err := splitSSLine(line)
			if err != nil {
				return err
			}
			if sssize > 0 {

				fsize, err := strconv.ParseInt(strings.TrimSuffix(arr[2], "\n"), 10, 64)
				if err != nil {
					return err
				}
				slice_size += fsize
				slice_line_cache = append(slice_line_cache, line)
				if slice_size >= sssize {
					if err := writeSlice(slice_line_cache, fmt.Sprintf("%s_%d", sscidstr, slice_index)); err != nil {
						return err
					}
					slice_size = 0
					slice_line_cache = make([]string, 0)
					slice_index++
				}
				continue
			}
			fcid, err := cid.Decode(arr[1])
			if err != nil {
				return err
			}
			if onlyCheck {
				totalLine++
				if has, err := api.DagHas(ctx, fcid); has {
					checkedLine++
				} else {
					fmt.Printf("%s: %s\n", fcid, err)
					errLine++
				}
				fmt.Printf("sum: %d; has: %d; not has: %d\n", totalLine, checkedLine, errLine)
				continue
			}
			if onlyDag {
				info, err := api.DagSync(ctx, []cid.Cid{fcid}, 32)
				if err != nil {
					return err
				}
				for msg := range info {
					fmt.Println(msg)
				}
				continue
			}

			pb, err := api.Get(ctx, fcid, filepath.Join(p, arr[0]))
			if err != nil {
				return err
			}

			if err = PrintProgress(pb); err != nil {
				return err
			}
		}
		if onlyCheck {
			fmt.Printf("total: %d\n", totalLine)
			fmt.Printf("checked: %d\n", checkedLine)
			fmt.Printf("err: %d\n", errLine)
		}
		if sssize > 0 && len(slice_line_cache) > 0 {
			if err := writeSlice(slice_line_cache, fmt.Sprintf("%s_%d", sscidstr, slice_index)); err != nil {
				return err
			}
		}

		return nil
	},
}

Functions

func BatchLoadNode

func BatchLoadNode(ctx context.Context, unloadedLinks []*format.Link, bs bstore.Blockstore, batchNum int) ([]format.Node, error)

func BlockWalk

func BlockWalk(ctx context.Context, node format.Node, bs bstore.Blockstore, batchNum int, cb func(nd format.Node) error) error

func GetAPI

func GetAPI(c *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error)

func PrintProgress

func PrintProgress(pb chan api.PBar) error

func ReqContext

func ReqContext(cctx *cli.Context) context.Context

func WithCategory

func WithCategory(cat string, cmd *cli.Command) *cli.Command

func WritePieceV2

func WritePieceV2(ctx context.Context, root cid.Cid, ppath string, bs bstore.Blockstore, batchNum int) error

广度优先算法