add more comments!
Jeromy committed
Oct 8, 2014 at 16:51 UTC
916fc54633fcf028ab03396690556293e156e0ce
4 files changed
+51
-11
importer/dagwriter/dagmodifier.go
+8
-7
@@ -87,18 +87,16 @@ func (dm *DagModifier) WriteAt(b []byte, offset uint64) (int, error) {
87
if err != nil {
88
return 0, err
89
}
90
+
91
+ // We have to rewrite the data before our write in this block.
92
b = append(data[:offset-traversed], b...)
93
break
94
}
95
traversed += size
96
}
97
if startsubblk == len(dm.pbdata.Blocksizes) {
96
- // TODO: something?
97
- /*
98
- if traversed < offset {
99
- return 0, errors.New("Tried to start write outside bounds of file.")
100
- }
101
- */
98
+ // TODO: Im not sure if theres any case that isnt being handled here.
99
+ // leaving this note here as a future reference in case something breaks
100
}
101
}
102
@@ -157,7 +155,7 @@ func (dm *DagModifier) WriteAt(b []byte, offset uint64) (int, error) {
155
sizes = append(sizes, uint64(len(sb)))
156
}
157
160
- // This is disgusting
158
+ // This is disgusting (and can be rewritten if performance demands)
159
if len(changed) > 0 {
160
sechalflink := append(links, dm.curNode.Links[changed[len(changed)-1]+1:]...)
161
dm.curNode.Links = append(dm.curNode.Links[:changed[0]], sechalflink...)
@@ -172,6 +170,8 @@ func (dm *DagModifier) WriteAt(b []byte, offset uint64) (int, error) {
170
return origlen, nil
171
}
172
173
+// splitBytes uses a splitterFunc to turn a large array of bytes
174
+// into many smaller arrays of bytes
175
func splitBytes(b []byte, spl imp.BlockSplitter) [][]byte {
176
out := spl.Split(bytes.NewReader(b))
177
var arr [][]byte
@@ -181,6 +181,7 @@ func splitBytes(b []byte, spl imp.BlockSplitter) [][]byte {
181
return arr
182
}
183
184
+// GetNode gets the modified DAG Node
185
func (dm *DagModifier) GetNode() (*mdag.Node, error) {
186
b, err := proto.Marshal(dm.pbdata)
187
if err != nil {
importer/dagwriter/dagwriter.go
+17
@@ -29,14 +29,23 @@ func NewDagWriter(ds *dag.DAGService, splitter imp.BlockSplitter) *DagWriter {
29
return dw
30
}
31
32
+// startSplitter manages splitting incoming bytes and
33
+// creating dag nodes from them. Created nodes are stored
34
+// in the DAGService and then released to the GC.
35
func (dw *DagWriter) startSplitter() {
36
+
37
+ // Since the splitter functions take a reader (and should!)
38
+ // we wrap our byte chan input in a reader
39
r := util.NewByteChanReader(dw.splChan)
40
blkchan := dw.splitter.Split(r)
41
+
42
+ // First data block is reserved for storage in the root node
43
first := <-blkchan
44
mbf := new(ft.MultiBlock)
45
root := new(dag.Node)
46
47
for blkData := range blkchan {
48
+ // Store the block size in the root node
49
mbf.AddBlockSize(uint64(len(blkData)))
50
node := &dag.Node{Data: ft.WrapData(blkData)}
51
_, err := dw.dagserv.Add(node)
@@ -45,6 +54,8 @@ func (dw *DagWriter) startSplitter() {
54
log.Critical("Got error adding created node to dagservice: %s", err)
55
return
56
}
57
+
58
+ // Add a link to this node without storing a reference to the memory
59
err = root.AddNodeLinkClean("", node)
60
if err != nil {
61
dw.seterr = err
@@ -52,6 +63,8 @@ func (dw *DagWriter) startSplitter() {
63
return
64
}
65
}
66
+
67
+ // Generate the root node data
68
mbf.Data = first
69
data, err := mbf.GetBytes()
70
if err != nil {
@@ -61,6 +74,7 @@ func (dw *DagWriter) startSplitter() {
74
}
75
root.Data = data
76
77
+ // Add root node to the dagservice
78
_, err = dw.dagserv.Add(root)
79
if err != nil {
80
dw.seterr = err
@@ -79,6 +93,9 @@ func (dw *DagWriter) Write(b []byte) (int, error) {
93
return len(b), nil
94
}
95
96
+// Close the splitters input channel and wait for it to finish
97
+// Must be called to finish up splitting, otherwise split method
98
+// will never halt
99
func (dw *DagWriter) Close() error {
100
close(dw.splChan)
101
<-dw.done
importer/format/format.go
+10
-1
@@ -8,6 +8,10 @@ import (
8
"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
9
)
10
11
+var ErrMalformedFileFormat = errors.New("malformed data in file format")
12
+var ErrInvalidDirLocation = errors.New("found directory node in unexpected place")
13
+var ErrUnrecognizedType = errors.New("unrecognized node type")
14
+
15
func FromBytes(data []byte) (*PBData, error) {
16
pbdata := new(PBData)
17
err := proto.Unmarshal(data, pbdata)
@@ -26,12 +30,17 @@ func FilePBData(data []byte, totalsize uint64) []byte {
30
31
data, err := proto.Marshal(pbfile)
32
if err != nil {
29
- //this really shouldnt happen, i promise
33
+ // This really shouldnt happen, i promise
34
+ // The only failure case for marshal is if required fields
35
+ // are not filled out, and they all are. If the proto object
36
+ // gets changed and nobody updates this function, the code
37
+ // should panic due to programmer error
38
panic(err)
39
}
40
return data
41
}
42
43
+// Returns Bytes that represent a Directory
44
func FolderPBData() []byte {
45
pbfile := new(PBData)
46
typ := PBData_Directory
merkledag/dagreader.go
+16
-3
@@ -20,6 +20,8 @@ type DagReader struct {
20
buf *bytes.Buffer
21
}
22
23
+// NewDagReader creates a new reader object that reads the data represented by the given
24
+// node, using the passed in DAGService for data retreival
25
func NewDagReader(n *Node, serv *DAGService) (io.Reader, error) {
26
pb := new(ft.PBData)
27
err := proto.Unmarshal(n.Data, pb)
@@ -29,6 +31,7 @@ func NewDagReader(n *Node, serv *DAGService) (io.Reader, error) {
31
32
switch pb.GetType() {
33
case ft.PBData_Directory:
34
+ // Dont allow reading directories
35
return nil, ErrIsDir
36
case ft.PBData_File:
37
return &DagReader{
@@ -37,12 +40,15 @@ func NewDagReader(n *Node, serv *DAGService) (io.Reader, error) {
40
buf: bytes.NewBuffer(pb.GetData()),
41
}, nil
42
case ft.PBData_Raw:
43
+ // Raw block will just be a single level, return a byte buffer
44
return bytes.NewBuffer(pb.GetData()), nil
45
default:
42
- panic("Unrecognized node type!")
46
+ return nil, ft.ErrUnrecognizedType
47
}
48
}
49
50
+// Follows the next link in line and loads it from the DAGService,
51
+// setting the next buffer to read from
52
func (dr *DagReader) precalcNextBuf() error {
53
if dr.position >= len(dr.node.Links) {
54
return io.EOF
@@ -65,7 +71,7 @@ func (dr *DagReader) precalcNextBuf() error {
71
72
switch pb.GetType() {
73
case ft.PBData_Directory:
68
- panic("Why is there a directory under a file?")
74
+ return ft.ErrInvalidDirLocation
75
case ft.PBData_File:
76
//TODO: this *should* work, needs testing first
77
//return NewDagReader(nxt, dr.serv)
@@ -74,11 +80,12 @@ func (dr *DagReader) precalcNextBuf() error {
80
dr.buf = bytes.NewBuffer(pb.GetData())
81
return nil
82
default:
77
- panic("Unrecognized node type!")
83
+ return ft.ErrUnrecognizedType
84
}
85
}
86
87
func (dr *DagReader) Read(b []byte) (int, error) {
88
+ // If no cached buffer, load one
89
if dr.buf == nil {
90
err := dr.precalcNextBuf()
91
if err != nil {
@@ -87,16 +94,22 @@ func (dr *DagReader) Read(b []byte) (int, error) {
94
}
95
total := 0
96
for {
97
+ // Attempt to fill bytes from cached buffer
98
n, err := dr.buf.Read(b[total:])
99
total += n
100
if err != nil {
101
+ // EOF is expected
102
if err != io.EOF {
103
return total, err
104
}
105
}
106
+
107
+ // If weve read enough bytes, return
108
if total == len(b) {
109
return total, nil
110
}
111
+
112
+ // Otherwise, load up the next block
113
err = dr.precalcNextBuf()
114
if err != nil {
115
return total, err