Create a FilestoreNode object to carry PosInfo
When doing a filestore add, we wrap whatever nodes we create in a FilestoreNode object and add the PosInfo to it so that the filestore will be able to extract information as needed. Edited by whyrusleeping License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org>
Kevin Atkinson committed
Oct 16, 2016 at 21:06 UTC
73f9a90cbf439acc89d9740d6ed507861a6cd5b6
8 files changed
+116
-13
commands/files/file.go
+5
@@ -55,3 +55,8 @@ type SizeFile interface {
55
56
Size() (int64, error)
57
}
58
+
59
+type FileInfo interface {
60
+ FullPath() string
61
+ Stat() os.FileInfo
62
+}
core/coreunix/add.go
+11
-1
@@ -398,7 +398,12 @@ func (adder *Adder) addFile(file files.File) error {
398
// progress updates to the client (over the output channel)
399
var reader io.Reader = file
400
if adder.Progress {
401
- reader = &progressReader{file: file, out: adder.Out}
401
+ rdr := &progressReader{file: file, out: adder.Out}
402
+ if fi, ok := file.(files.FileInfo); ok {
403
+ reader = &progressReader2{rdr, fi}
404
+ } else {
405
+ reader = rdr
406
+ }
407
}
408
409
dagnode, err := adder.add(reader)
@@ -520,3 +525,8 @@ func (i *progressReader) Read(p []byte) (int, error) {
525
526
return n, err
527
}
528
+
529
+type progressReader2 struct {
530
+ *progressReader
531
+ files.FileInfo
532
+}
importer/balanced/builder.go
+9
-3
@@ -9,10 +9,12 @@ import (
9
)
10
11
func BalancedLayout(db *h.DagBuilderHelper) (node.Node, error) {
12
+ var offset uint64 = 0
13
var root *h.UnixfsNode
14
for level := 0; !db.Done(); level++ {
15
16
nroot := h.NewUnixfsNode()
17
+ db.SetPosInfo(nroot, 0)
18
19
// add our old root as a child of the new root.
20
if root != nil { // nil if it's the first node.
@@ -22,11 +24,13 @@ func BalancedLayout(db *h.DagBuilderHelper) (node.Node, error) {
24
}
25
26
// fill it up.
25
- if err := fillNodeRec(db, nroot, level); err != nil {
27
+ if err := fillNodeRec(db, nroot, level, offset); err != nil {
28
return nil, err
29
}
30
31
+ offset = nroot.FileSize()
32
root = nroot
33
+
34
}
35
if root == nil {
36
root = h.NewUnixfsNode()
@@ -50,7 +54,7 @@ func BalancedLayout(db *h.DagBuilderHelper) (node.Node, error) {
54
// it returns the total dataSize of the node, and a potential error
55
//
56
// warning: **children** pinned indirectly, but input node IS NOT pinned.
53
-func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
57
+func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int, offset uint64) error {
58
if depth < 0 {
59
return errors.New("attempt to fillNode at depth < 0")
60
}
@@ -69,8 +73,9 @@ func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
73
// while we have room AND we're not done
74
for node.NumChildren() < db.Maxlinks() && !db.Done() {
75
child := h.NewUnixfsNode()
76
+ db.SetPosInfo(child, offset)
77
73
- err := fillNodeRec(db, child, depth-1)
78
+ err := fillNodeRec(db, child, depth-1, offset)
79
if err != nil {
80
return err
81
}
@@ -78,6 +83,7 @@ func fillNodeRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
83
if err := node.AddChild(child, db); err != nil {
84
return err
85
}
86
+ offset += child.FileSize()
87
}
88
89
return nil
importer/chunk/rabin.go
+8
-2
@@ -10,7 +10,8 @@ import (
10
var IpfsRabinPoly = chunker.Pol(17437180132763653)
11
12
type Rabin struct {
13
- r *chunker.Chunker
13
+ r *chunker.Chunker
14
+ reader io.Reader
15
}
16
17
func NewRabin(r io.Reader, avgBlkSize uint64) *Rabin {
@@ -25,7 +26,8 @@ func NewRabinMinMax(r io.Reader, min, avg, max uint64) *Rabin {
26
ch := chunker.New(r, IpfsRabinPoly, h, avg, min, max)
27
28
return &Rabin{
28
- r: ch,
29
+ r: ch,
30
+ reader: r,
31
}
32
}
33
@@ -37,3 +39,7 @@ func (r *Rabin) NextBytes() ([]byte, error) {
39
40
return ch.Data, nil
41
}
42
+
43
+func (r *Rabin) Reader() io.Reader {
44
+ return r.reader
45
+}
importer/chunk/splitting.go
+5
@@ -12,6 +12,7 @@ var log = logging.Logger("chunk")
12
var DefaultBlockSize int64 = 1024 * 256
13
14
type Splitter interface {
15
+ Reader() io.Reader
16
NextBytes() ([]byte, error)
17
}
18
@@ -77,3 +78,7 @@ func (ss *sizeSplitterv2) NextBytes() ([]byte, error) {
78
79
return buf[:n], nil
80
}
81
+
82
+func (ss *sizeSplitterv2) Reader() io.Reader {
83
+ return ss.r
84
+}
importer/helpers/dagbuilder.go
+37
-7
@@ -1,6 +1,10 @@
1
package helpers
2
3
import (
4
+ "io"
5
+ "os"
6
+
7
+ "github.com/ipfs/go-ipfs/commands/files"
8
"github.com/ipfs/go-ipfs/importer/chunk"
9
dag "github.com/ipfs/go-ipfs/merkledag"
10
@@ -17,6 +21,8 @@ type DagBuilderHelper struct {
21
nextData []byte // the next item to return.
22
maxlinks int
23
batch *dag.Batch
24
+ fullPath string
25
+ stat os.FileInfo
26
}
27
28
type DagBuilderParams struct {
@@ -34,13 +40,18 @@ type DagBuilderParams struct {
40
// Generate a new DagBuilderHelper from the given params, which data source comes
41
// from chunks object
42
func (dbp *DagBuilderParams) New(spl chunk.Splitter) *DagBuilderHelper {
37
- return &DagBuilderHelper{
43
+ db := &DagBuilderHelper{
44
dserv: dbp.Dagserv,
45
spl: spl,
46
rawLeaves: dbp.RawLeaves,
47
maxlinks: dbp.Maxlinks,
48
batch: dbp.Dagserv.Batch(),
49
}
50
+ if fi, ok := spl.Reader().(files.FileInfo); ok {
51
+ db.fullPath = fi.FullPath()
52
+ db.stat = fi.Stat()
53
+ }
54
+ return db
55
}
56
57
// prepareNext consumes the next item from the splitter and puts it
@@ -48,12 +59,14 @@ func (dbp *DagBuilderParams) New(spl chunk.Splitter) *DagBuilderHelper {
59
// it will do nothing.
60
func (db *DagBuilderHelper) prepareNext() {
61
// if we already have data waiting to be consumed, we're ready
51
- if db.nextData != nil {
62
+ if db.nextData != nil || db.recvdErr != nil {
63
return
64
}
65
55
- // TODO: handle err (which wasn't handled either when the splitter was channeled)
56
- db.nextData, _ = db.spl.NextBytes()
66
+ db.nextData, db.recvdErr = db.spl.NextBytes()
67
+ if db.recvdErr == io.EOF {
68
+ db.recvdErr = nil
69
+ }
70
}
71
72
// Done returns whether or not we're done consuming the incoming data.
@@ -61,17 +74,24 @@ func (db *DagBuilderHelper) Done() bool {
74
// ensure we have an accurate perspective on data
75
// as `done` this may be called before `next`.
76
db.prepareNext() // idempotent
77
+ if db.recvdErr != nil {
78
+ return false
79
+ }
80
return db.nextData == nil
81
}
82
83
// Next returns the next chunk of data to be inserted into the dag
84
// if it returns nil, that signifies that the stream is at an end, and
85
// that the current building operation should finish
70
-func (db *DagBuilderHelper) Next() []byte {
86
+func (db *DagBuilderHelper) Next() ([]byte, error) {
87
db.prepareNext() // idempotent
88
d := db.nextData
89
db.nextData = nil // signal we've consumed it
74
- return d
90
+ if db.recvdErr != nil {
91
+ return nil, db.recvdErr
92
+ } else {
93
+ return d, nil
94
+ }
95
}
96
97
// GetDagServ returns the dagservice object this Helper is using
@@ -100,7 +120,11 @@ func (db *DagBuilderHelper) FillNodeLayer(node *UnixfsNode) error {
120
}
121
122
func (db *DagBuilderHelper) GetNextDataNode() (*UnixfsNode, error) {
103
- data := db.Next()
123
+ data, err := db.Next()
124
+ if err != nil {
125
+ return nil, err
126
+ }
127
+
128
if data == nil { // we're done!
129
return nil, nil
130
}
@@ -121,6 +145,12 @@ func (db *DagBuilderHelper) GetNextDataNode() (*UnixfsNode, error) {
145
}
146
}
147
148
+func (db *DagBuilderHelper) SetPosInfo(node *UnixfsNode, offset uint64) {
149
+ if db.stat != nil {
150
+ node.SetPosInfo(offset, db.fullPath, db.stat)
151
+ }
152
+}
153
+
154
func (db *DagBuilderHelper) Add(node *UnixfsNode) (node.Node, error) {
155
dn, err := node.GetDagNode()
156
if err != nil {
importer/helpers/helpers.go
+23
@@ -3,9 +3,11 @@ package helpers
3
import (
4
"context"
5
"fmt"
6
+ "os"
7
8
chunk "github.com/ipfs/go-ipfs/importer/chunk"
9
dag "github.com/ipfs/go-ipfs/merkledag"
10
+ pi "github.com/ipfs/go-ipfs/thirdparty/posinfo"
11
ft "github.com/ipfs/go-ipfs/unixfs"
12
13
node "gx/ipfs/QmZx42H5khbVQhV5odp66TApShV4XCujYazcvYduZ4TroB/go-ipld-node"
@@ -43,6 +45,7 @@ type UnixfsNode struct {
45
rawnode *dag.RawNode
46
node *dag.ProtoNode
47
ufmt *ft.FSNode
48
+ posInfo *pi.PosInfo
49
}
50
51
// NewUnixfsNode creates a new Unixfs node to represent a file
@@ -144,9 +147,29 @@ func (n *UnixfsNode) FileSize() uint64 {
147
return n.ufmt.FileSize()
148
}
149
150
+func (n *UnixfsNode) SetPosInfo(offset uint64, fullPath string, stat os.FileInfo) {
151
+ n.posInfo = &pi.PosInfo{offset, fullPath, stat}
152
+}
153
+
154
// getDagNode fills out the proper formatting for the unixfs node
155
// inside of a DAG node and returns the dag node
156
func (n *UnixfsNode) GetDagNode() (node.Node, error) {
157
+ nd, err := n.getBaseDagNode()
158
+ if err != nil {
159
+ return nil, err
160
+ }
161
+
162
+ if n.posInfo != nil {
163
+ return &pi.FilestoreNode{
164
+ Node: nd,
165
+ PosInfo: n.posInfo,
166
+ }, nil
167
+ }
168
+
169
+ return nd, nil
170
+}
171
+
172
+func (n *UnixfsNode) getBaseDagNode() (node.Node, error) {
173
if n.raw {
174
return n.rawnode, nil
175
}
thirdparty/posinfo/posinfo.go
new
+18
@@ -0,0 +1,18 @@
1
+package posinfo
2
+
3
+import (
4
+ "os"
5
+
6
+ node "gx/ipfs/QmZx42H5khbVQhV5odp66TApShV4XCujYazcvYduZ4TroB/go-ipld-node"
7
+)
8
+
9
+type PosInfo struct {
10
+ Offset uint64
11
+ FullPath string
12
+ Stat os.FileInfo // can be nil
13
+}
14
+
15
+type FilestoreNode struct {
16
+ node.Node
17
+ PosInfo *PosInfo
18
+}