Keep track of offsets in the DAG builder and not the splitter.

This completely elements the need for AdvReader and cleans up a lot of
other code.

License: MIT
Signed-off-by: Kevin Atkinson <[email protected]>

commands/files/file.go

@@ -55,3 +55,14 @@ type SizeFile interface {
 
 	Size() (int64, error)
 }
+
+type FileInfo interface {
+	FullPath() string
+	Stat() os.FileInfo
+}
+
+type PosInfo struct {
+	Offset   uint64
+	FullPath string
+	Stat     os.FileInfo // can be nil
+}

commands/files/multipartfile.go

@@ -26,7 +26,6 @@ type MultipartFile struct {
 	Part      *multipart.Part
 	Reader    *multipart.Reader
 	Mediatype string
-	offset    int64
 }
 
 func NewFileFromPart(part *multipart.Part) (File, error) {
@@ -97,9 +96,7 @@ func (f *MultipartFile) Read(p []byte) (int, error) {
 	if f.IsDirectory() {
 		return 0, ErrNotReader
 	}
-	res, err := f.Part.Read(p)
-	f.offset += int64(res)
-	return res, err
+	return f.Part.Read(p)
 }
 
 func (f *MultipartFile) Close() error {

commands/files/readerfile.go

@@ -13,11 +13,10 @@ type ReaderFile struct {
 	fullpath string
 	reader   io.ReadCloser
 	stat     os.FileInfo
-	offset   uint64
 }
 
 func NewReaderFile(filename, path string, reader io.ReadCloser, stat os.FileInfo) *ReaderFile {
-	return &ReaderFile{filename, path, reader, stat, 0}
+	return &ReaderFile{filename, path, reader, stat}
 }
 
 func (f *ReaderFile) IsDirectory() bool {
@@ -36,14 +35,8 @@ func (f *ReaderFile) FullPath() string {
 	return f.fullpath
 }
 
-func (f *ReaderFile) PosInfo() *PosInfo {
-	return &PosInfo{f.offset,f.fullpath,f.stat}
-}
-
 func (f *ReaderFile) Read(p []byte) (int, error) {
-	res, err := f.reader.Read(p)
-	f.offset += uint64(res)
-	return res, err
+	return f.reader.Read(p)
 }
 
 func (f *ReaderFile) Close() error {

core/coreunix/add.go

@@ -107,7 +107,7 @@ type Adder struct {
 }
 
 // Perform the actual add & pin locally, outputting results to reader
-func (adder Adder) add(reader files.AdvReader) (*dag.Node, error) {
+func (adder Adder) add(reader io.Reader) (*dag.Node, error) {
 	chnk, err := chunk.FromString(reader, adder.Chunker)
 	if err != nil {
 		return nil, err
@@ -249,9 +249,7 @@ func Add(n *core.DataServices, r io.Reader) (string, error) {
 		return "", err
 	}
 
-	ar := files.AdvReaderAdapter(r)
-
-	node, err := fileAdder.add(ar)
+	node, err := fileAdder.add(r)
 	if err != nil {
 		return "", err
 	}
@@ -400,9 +398,14 @@ func (adder *Adder) addFile(file files.File) error {
 	// case for regular file
 	// if the progress flag was specified, wrap the file so that we can send
 	// progress updates to the client (over the output channel)
-	reader := files.AdvReaderAdapter(file)
+	var reader io.Reader = file
 	if adder.Progress {
-		reader = &progressReader{reader: reader, filename: file.FileName(), out: adder.out}
+		rdr := &progressReader{file: file, out: adder.out}
+		if fi, ok := file.(files.FileInfo); ok {
+			reader = &progressReader2{rdr, fi}
+		} else {
+			reader = rdr
+		}
 	}
 
 	dagnode, err := adder.add(reader)
@@ -513,28 +516,28 @@ func getOutput(dagnode *dag.Node) (*Object, error) {
 }
 
 type progressReader struct {
-	reader       files.AdvReader
-	filename     string
+	file         files.File
 	out          chan interface{}
 	bytes        int64
 	lastProgress int64
 }
 
 func (i *progressReader) Read(p []byte) (int, error) {
-	n, err := i.reader.Read(p)
+	n, err := i.file.Read(p)
 
 	i.bytes += int64(n)
 	if i.bytes-i.lastProgress >= progressReaderIncrement || err == io.EOF {
 		i.lastProgress = i.bytes
 		i.out <- &AddedObject{
-			Name:  i.filename,
+			Name:  i.file.FileName(),
 			Bytes: i.bytes,
 		}
 	}
 
 	return n, err
 }
 
-func (i *progressReader) PosInfo() *files.PosInfo {
-	return i.reader.PosInfo()
+type progressReader2 struct {
+	*progressReader
+	files.FileInfo
 }

filestore/support/misc.go

@@ -10,7 +10,7 @@ import (
 type FilestoreBlock struct {
 	blocks.BasicBlock
 	AltData  []byte
-	files.PosInfo
+	*files.PosInfo
 	Size     uint64
 }
 

importer/balanced/builder.go

@@ -9,10 +9,12 @@ import (
 )
 
 func BalancedLayout(db *h.DagBuilderHelper) (*dag.Node, error) {
+	var offset uint64 = 0
 	var root *h.UnixfsNode
 	for level := 0; !db.Done(); level++ {
-
+		
 		nroot := h.NewUnixfsNode()
+		db.SetPosInfo(nroot, 0)
 
 		// add our old root as a child of the new root.
 		if root != nil { // nil if it's the first node.
@@ -22,17 +24,18 @@ func BalancedLayout(db *h.DagBuilderHelper) (*dag.Node, error) {
 		}
 
 		// fill it up.
-		if err := fillNodeRec(db, nroot, level); err != nil {
+		if err := fillNodeRec(db, nroot, level, offset); err != nil {
 			return nil, err
 		}
 
+		offset = nroot.FileSize()
 		root = nroot
+
 	}
 	if root == nil {
 		root = h.NewUnixfsNode()
 	}
 
-	db.SetAsRoot(root)
 	out, err := db.Add(root)
 	if err != nil {
 		return nil, err
@@ -51,7 +54,7 @@ func BalancedLayout(db *h.DagBuilderHelper) (*dag.Node, error) {
 // it returns the total dataSize of the node, and a potential error
 //
 // warning: **children** pinned indirectly, but input node IS NOT pinned.
-func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
+func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int, offset uint64) error {
 	if depth < 0 {
 		return errors.New("attempt to fillNode at depth < 0")
 	}
@@ -64,14 +67,16 @@ func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
 	// while we have room AND we're not done
 	for node.NumChildren() < db.Maxlinks() && !db.Done() {
 		child := h.NewUnixfsNode()
+		db.SetPosInfo(child,offset)
 
-		if err := fillNodeRec(db, child, depth-1); err != nil {
+		if err := fillNodeRec(db, child, depth-1, offset); err != nil {
 			return err
 		}
 
 		if err := node.AddChild(child, db); err != nil {
 			return err
 		}
+		offset += child.FileSize()
 	}
 
 	return nil

importer/chunk/rabin.go

@@ -3,14 +3,15 @@ package chunk
 import (
 	"hash/fnv"
 	"io"
-
+	
 	"github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/whyrusleeping/chunker"
 )
 
 var IpfsRabinPoly = chunker.Pol(17437180132763653)
 
 type Rabin struct {
 	r *chunker.Chunker
+	reader io.Reader
 }
 
 func NewRabin(r io.Reader, avgBlkSize uint64) *Rabin {
@@ -26,14 +27,19 @@ func NewRabinMinMax(r io.Reader, min, avg, max uint64) *Rabin {
 
 	return &Rabin{
 		r: ch,
+		reader: r,
 	}
 }
 
-func (r *Rabin) NextBytes() (Bytes, error) {
+func (r *Rabin) NextBytes() ([]byte, error) {
 	ch, err := r.r.Next()
 	if err != nil {
-		return Bytes{}, err
+		return nil, err
 	}
 
-	return Bytes{nil, ch.Data}, nil
+	return ch.Data, nil
+}
+
+func (r *Rabin) Reader() io.Reader {
+	return r.reader
 }

importer/chunk/rabin_test.go

@@ -27,7 +27,7 @@ func TestRabinChunking(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		chunks = append(chunks, chunk.Data)
+		chunks = append(chunks, chunk)
 	}
 
 	fmt.Printf("average block size: %d\n", len(data)/len(chunks))
@@ -53,7 +53,7 @@ func chunkData(t *testing.T, data []byte) map[key.Key]blocks.Block {
 			t.Fatal(err)
 		}
 
-		b := blocks.NewBlock(blk.Data)
+		b := blocks.NewBlock(blk)
 		blkmap[b.Key()] = b
 	}
 

importer/chunk/splitting.go

@@ -4,21 +4,16 @@ package chunk
 import (
 	"io"
 
-	"github.com/ipfs/go-ipfs/commands/files"
 	logging "gx/ipfs/QmaDNZ4QMdBdku1YZWBysufYyoQt1negQGNav6PLYarbY8/go-log"
 )
 
 var log = logging.Logger("chunk")
 
 var DefaultBlockSize int64 = 1024 * 256
 
-type Bytes struct {
-	PosInfo *files.PosInfo
-	Data    []byte
-}
-
 type Splitter interface {
-	NextBytes() (Bytes, error)
+	Reader() io.Reader
+	NextBytes() ([]byte, error)
 }
 
 type SplitterGen func(r io.Reader) Splitter
@@ -48,29 +43,28 @@ func Chan(s Splitter) (<-chan []byte, <-chan error) {
 				return
 			}
 
-			out <- b.Data
+			out <- b
 		}
 	}()
 	return out, errs
 }
 
 type sizeSplitterv2 struct {
-	r    files.AdvReader
+	r    io.Reader
 	size int64
 	err  error
 }
 
 func NewSizeSplitter(r io.Reader, size int64) Splitter {
 	return &sizeSplitterv2{
-		r:    files.AdvReaderAdapter(r),
+		r:    r,
 		size: size,
 	}
 }
 
-func (ss *sizeSplitterv2) NextBytes() (Bytes, error) {
-	posInfo := ss.r.PosInfo()
+func (ss *sizeSplitterv2) NextBytes() ([]byte, error) {
 	if ss.err != nil {
-		return Bytes{posInfo, nil}, ss.err
+		return nil, ss.err
 	}
 	buf := make([]byte, ss.size)
 	n, err := io.ReadFull(ss.r, buf)
@@ -79,8 +73,12 @@ func (ss *sizeSplitterv2) NextBytes() (Bytes, error) {
 		err = nil
 	}
 	if err != nil {
-		return Bytes{posInfo, nil}, err
+		return nil, err
 	}
 
-	return Bytes{posInfo, buf[:n]}, nil
+	return buf[:n], nil
+}
+
+func (ss *sizeSplitterv2) Reader() io.Reader {
+	return ss.r
 }