retrieve file from CAR

Author: eric

.gitignore

@@ -15,3 +15,5 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
+
+.idea

Makefile

@@ -0,0 +1,3 @@
+.PHONY: build
+build:
+	go build -ldflags "-s -w" -o graphsplit graphsplit.go utils.go retrieve.go

README.md

@@ -11,7 +11,7 @@ Graphsplit has solved the problem we faced above. It takes advantage of IPLD con
 
 ## Build
 ```sh
-go build -o graphsplit graphsplit.go utils.go
+make
 ```
 
 ## Usage
@@ -43,6 +43,17 @@ Import car file to IPFS:
 ipfs dag import /path/to/car-dir/car-file
 ```
 
+Retrieve files:
+```sh
+# car-path: directory or file, in form of .car
+# output-dir: usually just be the same as /path/to/output-dir
+# parallel: number goroutines run when retrieving
+./graphsplit retrieve \
+--car-path=/path/to/car-path \
+--output-dir=/path/to/output-dir \
+--parallel=2
+```
+
 ## Contribute
 
 PRs are welcome!

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/ipfs/go-ipfs-blockstore v1.0.3
 	github.com/ipfs/go-ipfs-chunker v0.0.5
 	github.com/ipfs/go-ipfs-exchange-offline v0.0.1
+	github.com/ipfs/go-ipfs-files v0.0.8
 	github.com/ipfs/go-ipld-cbor v0.0.5-0.20200204214505-252690b78669 // indirect
 	github.com/ipfs/go-ipld-format v0.2.0
 	github.com/ipfs/go-log/v2 v2.1.2

graphsplit.go

@@ -3,12 +3,11 @@ package main
 import (
 	"context"
 	"fmt"
-	"os"
-
 	logging "github.com/ipfs/go-log/v2"
 	_ "github.com/jinzhu/gorm/dialects/mysql"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
+	"os"
 )
 
 var log = logging.Logger("graphsplit")
@@ -17,6 +16,7 @@ func main() {
 	logging.SetLogLevel("*", "INFO")
 	local := []*cli.Command{
 		chunkCmd,
+		retrieveCmd,
 	}
 
 	app := &cli.App{
@@ -33,7 +33,7 @@ func main() {
 
 var chunkCmd = &cli.Command{
 	Name:  "chunk",
-	Usage: "",
+	Usage: "Generate CAR files of the specified size",
 	Flags: []cli.Flag{
 		&cli.Int64Flag{
 			Name:  "slice-size",
@@ -58,6 +58,7 @@ var chunkCmd = &cli.Command{
 		&cli.StringFlag{
 			Name:     "car-dir",
 			Required: true,
+			Usage:    fmt.Sprintf("specify output CAR directory"),
 		},
 	},
 	Action: func(c *cli.Context) error {
@@ -154,7 +155,6 @@ var chunkCmd = &cli.Command{
 				}
 
 			}
-
 		}
 		if cumuSize > 0 {
 			// todo build ipld from graphFiles
@@ -166,3 +166,36 @@ var chunkCmd = &cli.Command{
 		return nil
 	},
 }
+
+var retrieveCmd = &cli.Command{
+	Name:  "retrieve",
+	Usage: "Retrieve files from CAR files",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "car-path",
+			Required: true,
+			Usage:    fmt.Sprintf("specify source car path, directory or file"),
+		},
+		&cli.StringFlag{
+			Name:     "output-dir",
+			Required: true,
+			Usage:    fmt.Sprintf("specify output directory"),
+		},
+		&cli.IntFlag{
+			Name:  "parallel",
+			Value: 4,
+			Usage: fmt.Sprintf("specify how many number of goroutines runs when generate file node"),
+		},
+	},
+	Action: func(c *cli.Context) error {
+		parallel := c.Int("parallel")
+		outputDir := c.String("output-dir")
+		carPath := c.String("car-path")
+
+		CarTo(carPath, outputDir, parallel)
+		Merge(outputDir, parallel)
+
+		fmt.Println("completed!")
+		return nil
+	},
+}

retrieve.go

@@ -0,0 +1,239 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/ipfs/go-blockservice"
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-datastore"
+	dss "github.com/ipfs/go-datastore/sync"
+	bstore "github.com/ipfs/go-ipfs-blockstore"
+	offline "github.com/ipfs/go-ipfs-exchange-offline"
+	files "github.com/ipfs/go-ipfs-files"
+	"github.com/ipfs/go-merkledag"
+	unixfile "github.com/ipfs/go-unixfs/file"
+	"github.com/ipld/go-car"
+	"golang.org/x/xerrors"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+)
+
+func Import(path string, st car.Store) (cid.Cid, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return cid.Undef, err
+	}
+	defer f.Close() //nolint:errcheck
+
+	stat, err := f.Stat()
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	file, err := files.NewReaderPathFile(path, f, stat)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	result, err := car.LoadCar(st, file)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	if len(result.Roots) != 1 {
+		return cid.Undef, xerrors.New("cannot import car with more than one root")
+	}
+
+	return result.Roots[0], nil
+}
+
+func NodeWriteTo(nd files.Node, fpath string) error {
+	switch nd := nd.(type) {
+	case *files.Symlink:
+		return os.Symlink(nd.Target, fpath)
+	case files.File:
+		f, err := os.Create(fpath)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		_, err = io.Copy(f, nd)
+		if err != nil {
+			return err
+		}
+		return nil
+	case files.Directory:
+		if !ExistDir(fpath) {
+			err := os.Mkdir(fpath, 0777)
+			if err != nil {
+				return err
+			}
+		}
+
+		entries := nd.Entries()
+		for entries.Next() {
+			child := filepath.Join(fpath, entries.Name())
+			if err := NodeWriteTo(entries.Node(), child); err != nil {
+				return err
+			}
+		}
+		return entries.Err()
+	default:
+		return fmt.Errorf("file type %T at %q is not supported", nd, fpath)
+	}
+}
+
+func ExistDir(path string) bool {
+	s, err := os.Stat(path)
+	if err != nil {
+		return false
+	}
+	return s.IsDir()
+}
+
+func CarTo(carPath, outputDir string, parallel int) {
+	ctx := context.Background()
+	bs2 := bstore.NewBlockstore(dss.MutexWrap(datastore.NewMapDatastore()))
+	rdag := merkledag.NewDAGService(blockservice.New(bs2, offline.Exchange(bs2)))
+
+	workerCh := make(chan func())
+	go func() {
+		defer close(workerCh)
+		err := filepath.Walk(carPath, func(path string, fi os.FileInfo, err error) error {
+			if err != nil {
+				return err
+			}
+			if fi.IsDir() {
+				return nil
+			}
+			workerCh <- func() {
+				log.Info(path)
+				root, err := Import(path, bs2)
+				if err != nil {
+					log.Error("import error, ", err)
+					return
+				}
+				nd, err := rdag.Get(ctx, root)
+				if err != nil {
+					log.Error("dagService.Get error, ", err)
+					return
+				}
+				file, err := unixfile.NewUnixfsFile(ctx, rdag, nd)
+				if err != nil {
+					log.Error("NewUnixfsFile error, ", err)
+					return
+				}
+				err = NodeWriteTo(file, outputDir)
+				if err != nil {
+					log.Error("NodeWriteTo error, ", err)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			log.Error("Walk path failed, ", err)
+		}
+	}()
+
+	limitCh := make(chan struct{}, parallel)
+	wg := sync.WaitGroup{}
+	func() {
+		for {
+			select {
+			case taskFunc, ok := <-workerCh:
+				if !ok {
+					return
+				}
+				limitCh <- struct{}{}
+				wg.Add(1)
+				go func() {
+					defer func() {
+						<-limitCh
+						wg.Done()
+					}()
+					taskFunc()
+				}()
+			}
+		}
+	}()
+	wg.Wait()
+}
+
+func Merge(dir string, parallel int) {
+	wg := sync.WaitGroup{}
+	limitCh := make(chan struct{}, parallel)
+	mergeCh := make(chan string)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case fpath, ok := <-mergeCh:
+				if !ok {
+					return
+				}
+				limitCh <- struct{}{}
+				wg.Add(1)
+				go func() {
+					defer func() {
+						<-limitCh
+						wg.Done()
+					}()
+					log.Info("merge to ", fpath)
+					f, err := os.Create(fpath)
+					if err != nil {
+						log.Error("Create file failed, ", err)
+						return
+					}
+					defer f.Close()
+					for i := 0; ; i++ {
+						chunkPath := fmt.Sprintf("%s.%08d", fpath, i)
+						err := func(path string) error {
+							chunkF, err := os.Open(path)
+							if err != nil {
+								if os.IsExist(err) {
+									log.Error("Open file failed, ", err)
+								}
+								return err
+							}
+							defer chunkF.Close()
+							_, err = io.Copy(f, chunkF)
+							if err != nil {
+								log.Error("io.Copy failed, ", err)
+							}
+							return err
+						}(chunkPath)
+						os.Remove(chunkPath)
+						if err != nil {
+							break
+						}
+					}
+				}()
+			}
+		}
+	}()
+	err := filepath.Walk(dir, func(path string, fi os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if fi.IsDir() {
+			return nil
+		}
+		matched, err := filepath.Match("*.00000000", fi.Name())
+		if err != nil {
+			log.Error("filepath.Match failed, ", err)
+			return nil
+		} else if matched {
+			mergeCh <- strings.TrimSuffix(path, ".00000000")
+		}
+		return nil
+	})
+	if err != nil {
+		log.Error("Walk path failed, ", err)
+	}
+	close(mergeCh)
+	wg.Wait()
+}