remove dagwriter in favor of new importer function
Jeromy committed
Oct 31, 2014 at 03:20 UTC
ae339a032097e705ea1975e777b060f46b12fe7b
9 files changed
+83
-314
Godeps/Godeps.json
+1
-1
@@ -1,6 +1,6 @@
1
{
2
"ImportPath": "github.com/jbenet/go-ipfs",
3
- "GoVersion": "go1.3",
3
+ "GoVersion": "go1.3.3",
4
"Packages": [
5
"./..."
6
],
core/commands/add.go
+1
-5
@@ -10,11 +10,9 @@ import (
10
11
"github.com/jbenet/go-ipfs/core"
12
"github.com/jbenet/go-ipfs/importer"
13
- "github.com/jbenet/go-ipfs/importer/chunk"
13
dag "github.com/jbenet/go-ipfs/merkledag"
14
"github.com/jbenet/go-ipfs/pin"
15
ft "github.com/jbenet/go-ipfs/unixfs"
17
- uio "github.com/jbenet/go-ipfs/unixfs/io"
16
)
17
18
// Error indicating the max depth has been exceded.
@@ -90,14 +88,12 @@ func addDir(n *core.IpfsNode, fpath string, depth int, out io.Writer) (*dag.Node
88
}
89
90
func addFile(n *core.IpfsNode, fpath string, depth int, out io.Writer) (*dag.Node, error) {
93
- dw := uio.NewDagWriter(n.DAG, chunk.DefaultSplitter)
91
mp, ok := n.Pinning.(pin.ManualPinner)
92
if !ok {
93
return nil, errors.New("invalid pinner type! expected manual pinner")
94
}
98
- dw.Pinner = mp
95
100
- root, err := importer.ImportFileDag(fpath, dw)
96
+ root, err := importer.BuildDagFromFile(fpath, n.DAG, mp)
97
if err != nil {
98
return nil, err
99
}
importer/importer.go
+40
-7
@@ -7,8 +7,8 @@ import (
7
8
"github.com/jbenet/go-ipfs/importer/chunk"
9
dag "github.com/jbenet/go-ipfs/merkledag"
10
+ "github.com/jbenet/go-ipfs/pin"
11
ft "github.com/jbenet/go-ipfs/unixfs"
11
- uio "github.com/jbenet/go-ipfs/unixfs/io"
12
"github.com/jbenet/go-ipfs/util"
13
)
14
@@ -74,7 +74,7 @@ func NewDagFromFile(fpath string) (*dag.Node, error) {
74
return NewDagFromReader(f)
75
}
76
77
-func ImportFileDag(fpath string, dw *uio.DagWriter) (*dag.Node, error) {
77
+func BuildDagFromFile(fpath string, ds dag.DAGService, mp pin.ManualPinner) (*dag.Node, error) {
78
stat, err := os.Stat(fpath)
79
if err != nil {
80
return nil, err
@@ -90,17 +90,50 @@ func ImportFileDag(fpath string, dw *uio.DagWriter) (*dag.Node, error) {
90
}
91
defer f.Close()
92
93
- return ImportReaderDag(f, dw)
93
+ return BuildDagFromReader(f, ds, mp, chunk.DefaultSplitter)
94
}
95
96
-func ImportReaderDag(r io.Reader, dw *uio.DagWriter) (*dag.Node, error) {
97
- _, err := io.Copy(dw, r)
96
+func BuildDagFromReader(r io.Reader, ds dag.DAGService, mp pin.ManualPinner, spl chunk.BlockSplitter) (*dag.Node, error) {
97
+ blkChan := spl.Split(r)
98
+ first := <-blkChan
99
+ root := &dag.Node{}
100
+
101
+ mbf := new(ft.MultiBlock)
102
+ for blk := range blkChan {
103
+ // Store the block size in the root node
104
+ mbf.AddBlockSize(uint64(len(blk)))
105
+ node := &dag.Node{Data: ft.WrapData(blk)}
106
+ nk, err := ds.Add(node)
107
+ if mp != nil {
108
+ mp.PinWithMode(nk, pin.Indirect)
109
+ }
110
+ if err != nil {
111
+ return nil, err
112
+ }
113
+
114
+ // Add a link to this node without storing a reference to the memory
115
+ err = root.AddNodeLinkClean("", node)
116
+ if err != nil {
117
+ return nil, err
118
+ }
119
+ }
120
+
121
+ // Generate the root node data
122
+ mbf.Data = first
123
+ data, err := mbf.GetBytes()
124
if err != nil {
125
return nil, err
126
}
101
- err = dw.Close()
127
+ root.Data = data
128
+
129
+ // Add root node to the dagservice
130
+ rootk, err := ds.Add(root)
131
if err != nil {
132
return nil, err
133
}
105
- return dw.GetNode(), nil
134
+ if mp != nil {
135
+ mp.PinWithMode(rootk, pin.Recursive)
136
+ }
137
+
138
+ return root, nil
139
}
importer/importer_test.go
+34
@@ -9,7 +9,10 @@ import (
9
"os"
10
"testing"
11
12
+ ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
13
+ bsrv "github.com/jbenet/go-ipfs/blockservice"
14
"github.com/jbenet/go-ipfs/importer/chunk"
15
+ dag "github.com/jbenet/go-ipfs/merkledag"
16
uio "github.com/jbenet/go-ipfs/unixfs/io"
17
)
18
@@ -77,6 +80,37 @@ func testFileConsistency(t *testing.T, bs chunk.BlockSplitter, nbytes int) {
80
}
81
}
82
83
+func TestBuilderConsistency(t *testing.T) {
84
+ nbytes := 100000
85
+ dstore := ds.NewMapDatastore()
86
+ bserv, err := bsrv.NewBlockService(dstore, nil)
87
+ if err != nil {
88
+ t.Fatal(err)
89
+ }
90
+ dagserv := dag.NewDAGService(bserv)
91
+ buf := new(bytes.Buffer)
92
+ io.CopyN(buf, rand.Reader, int64(nbytes))
93
+ should := dup(buf.Bytes())
94
+ nd, err := BuildDagFromReader(buf, dagserv, nil, chunk.DefaultSplitter)
95
+ if err != nil {
96
+ t.Fatal(err)
97
+ }
98
+ r, err := uio.NewDagReader(nd, dagserv)
99
+ if err != nil {
100
+ t.Fatal(err)
101
+ }
102
+
103
+ out, err := ioutil.ReadAll(r)
104
+ if err != nil {
105
+ t.Fatal(err)
106
+ }
107
+
108
+ err = arrComp(out, should)
109
+ if err != nil {
110
+ t.Fatal(err)
111
+ }
112
+}
113
+
114
func arrComp(a, b []byte) error {
115
if len(a) != len(b) {
116
return fmt.Errorf("Arrays differ in length. %d != %d", len(a), len(b))
routing/dht/query.go
+2
-2
@@ -207,7 +207,7 @@ func (r *dhtQueryRunner) spawnWorkers() {
207
}
208
209
func (r *dhtQueryRunner) queryPeer(p peer.Peer) {
210
- log.Debugf("spawned worker for: %v\n", p)
210
+ log.Debugf("spawned worker for: %v", p)
211
212
// make sure we rate limit concurrency.
213
select {
@@ -256,7 +256,7 @@ func (r *dhtQueryRunner) queryPeer(p peer.Peer) {
256
r.cancel() // signal to everyone that we're done.
257
258
} else if res.closerPeers != nil {
259
- log.Debugf("PEERS CLOSER -- worker for: %v\n", p)
259
+ log.Debugf("PEERS CLOSER -- worker for: %v", p)
260
for _, next := range res.closerPeers {
261
r.addPeerToQuery(next, p)
262
}
unixfs/io/dagmodifier.go
+2
@@ -13,6 +13,8 @@ import (
13
u "github.com/jbenet/go-ipfs/util"
14
)
15
16
+var log = u.Logger("dagio")
17
+
18
// DagModifier is the only struct licensed and able to correctly
19
// perform surgery on a DAG 'file'
20
// Dear god, please rename this to something more pleasant
unixfs/io/dagmodifier_test.go
+3
-13
@@ -7,6 +7,7 @@ import (
7
"testing"
8
9
bs "github.com/jbenet/go-ipfs/blockservice"
10
+ imp "github.com/jbenet/go-ipfs/importer"
11
"github.com/jbenet/go-ipfs/importer/chunk"
12
mdag "github.com/jbenet/go-ipfs/merkledag"
13
ft "github.com/jbenet/go-ipfs/unixfs"
@@ -26,22 +27,11 @@ func getMockDagServ(t *testing.T) mdag.DAGService {
27
}
28
29
func getNode(t *testing.T, dserv mdag.DAGService, size int64) ([]byte, *mdag.Node) {
29
- dw := NewDagWriter(dserv, &chunk.SizeSplitter{500})
30
-
31
- n, err := io.CopyN(dw, u.NewTimeSeededRand(), size)
30
+ in := io.LimitReader(u.NewTimeSeededRand(), size)
31
+ node, err := imp.BuildDagFromReader(in, dserv, nil, &chunk.SizeSplitter{500})
32
if err != nil {
33
t.Fatal(err)
34
}
35
- if n != size {
36
- t.Fatal("Incorrect copy amount!")
37
- }
38
-
39
- err = dw.Close()
40
- if err != nil {
41
- t.Fatal("DagWriter failed to close,", err)
42
- }
43
-
44
- node := dw.GetNode()
35
36
dr, err := NewDagReader(node, dserv)
37
if err != nil {
unixfs/io/dagwriter.go
deleted
-115
@@ -1,115 +0,0 @@
1
-package io
2
-
3
-import (
4
- "github.com/jbenet/go-ipfs/importer/chunk"
5
- dag "github.com/jbenet/go-ipfs/merkledag"
6
- "github.com/jbenet/go-ipfs/pin"
7
- ft "github.com/jbenet/go-ipfs/unixfs"
8
- "github.com/jbenet/go-ipfs/util"
9
-)
10
-
11
-var log = util.Logger("dagwriter")
12
-
13
-type DagWriter struct {
14
- dagserv dag.DAGService
15
- node *dag.Node
16
- totalSize int64
17
- splChan chan []byte
18
- done chan struct{}
19
- splitter chunk.BlockSplitter
20
- seterr error
21
- Pinner pin.ManualPinner
22
-}
23
-
24
-func NewDagWriter(ds dag.DAGService, splitter chunk.BlockSplitter) *DagWriter {
25
- dw := new(DagWriter)
26
- dw.dagserv = ds
27
- dw.splChan = make(chan []byte, 8)
28
- dw.splitter = splitter
29
- dw.done = make(chan struct{})
30
- go dw.startSplitter()
31
- return dw
32
-}
33
-
34
-// startSplitter manages splitting incoming bytes and
35
-// creating dag nodes from them. Created nodes are stored
36
-// in the DAGService and then released to the GC.
37
-func (dw *DagWriter) startSplitter() {
38
-
39
- // Since the splitter functions take a reader (and should!)
40
- // we wrap our byte chan input in a reader
41
- r := util.NewByteChanReader(dw.splChan)
42
- blkchan := dw.splitter.Split(r)
43
-
44
- // First data block is reserved for storage in the root node
45
- first := <-blkchan
46
- mbf := new(ft.MultiBlock)
47
- root := new(dag.Node)
48
-
49
- for blkData := range blkchan {
50
- // Store the block size in the root node
51
- mbf.AddBlockSize(uint64(len(blkData)))
52
- node := &dag.Node{Data: ft.WrapData(blkData)}
53
- nk, err := dw.dagserv.Add(node)
54
- if dw.Pinner != nil {
55
- dw.Pinner.PinWithMode(nk, pin.Indirect)
56
- }
57
- if err != nil {
58
- dw.seterr = err
59
- log.Critical("Got error adding created node to dagservice: %s", err)
60
- return
61
- }
62
-
63
- // Add a link to this node without storing a reference to the memory
64
- err = root.AddNodeLinkClean("", node)
65
- if err != nil {
66
- dw.seterr = err
67
- log.Critical("Got error adding created node to root node: %s", err)
68
- return
69
- }
70
- }
71
-
72
- // Generate the root node data
73
- mbf.Data = first
74
- data, err := mbf.GetBytes()
75
- if err != nil {
76
- dw.seterr = err
77
- log.Critical("Failed generating bytes for multiblock file: %s", err)
78
- return
79
- }
80
- root.Data = data
81
-
82
- // Add root node to the dagservice
83
- rootk, err := dw.dagserv.Add(root)
84
- if err != nil {
85
- dw.seterr = err
86
- log.Critical("Got error adding created node to dagservice: %s", err)
87
- return
88
- }
89
- if dw.Pinner != nil {
90
- dw.Pinner.PinWithMode(rootk, pin.Recursive)
91
- }
92
- dw.node = root
93
- dw.done <- struct{}{}
94
-}
95
-
96
-func (dw *DagWriter) Write(b []byte) (int, error) {
97
- if dw.seterr != nil {
98
- return 0, dw.seterr
99
- }
100
- dw.splChan <- b
101
- return len(b), nil
102
-}
103
-
104
-// Close the splitters input channel and wait for it to finish
105
-// Must be called to finish up splitting, otherwise split method
106
-// will never halt
107
-func (dw *DagWriter) Close() error {
108
- close(dw.splChan)
109
- <-dw.done
110
- return nil
111
-}
112
-
113
-func (dw *DagWriter) GetNode() *dag.Node {
114
- return dw.node
115
-}
unixfs/io/dagwriter_test.go
deleted
-171
@@ -1,171 +0,0 @@
1
-package io_test
2
-
3
-import (
4
- "testing"
5
-
6
- "io"
7
-
8
- ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
9
- bs "github.com/jbenet/go-ipfs/blockservice"
10
- importer "github.com/jbenet/go-ipfs/importer"
11
- chunk "github.com/jbenet/go-ipfs/importer/chunk"
12
- mdag "github.com/jbenet/go-ipfs/merkledag"
13
- dagio "github.com/jbenet/go-ipfs/unixfs/io"
14
-)
15
-
16
-type datasource struct {
17
- i int
18
-}
19
-
20
-func (d *datasource) Read(b []byte) (int, error) {
21
- for i, _ := range b {
22
- b[i] = byte(d.i % 256)
23
- d.i++
24
- }
25
- return len(b), nil
26
-}
27
-
28
-func (d *datasource) Matches(t *testing.T, r io.Reader, length int) bool {
29
- b := make([]byte, 100)
30
- i := 0
31
- for {
32
- n, err := r.Read(b)
33
- if err != nil && err != io.EOF {
34
- t.Fatal(err)
35
- }
36
- for _, v := range b[:n] {
37
- if v != byte(i%256) {
38
- t.Fatalf("Buffers differed at byte: %d (%d != %d)", i, v, (i % 256))
39
- }
40
- i++
41
- }
42
- if err == io.EOF {
43
- break
44
- }
45
- }
46
- if i != length {
47
- t.Fatalf("Incorrect length. (%d != %d)", i, length)
48
- }
49
- return true
50
-}
51
-
52
-func TestDagWriter(t *testing.T) {
53
- dstore := ds.NewMapDatastore()
54
- bserv, err := bs.NewBlockService(dstore, nil)
55
- if err != nil {
56
- t.Fatal(err)
57
- }
58
- dag := mdag.NewDAGService(bserv)
59
- dw := dagio.NewDagWriter(dag, &chunk.SizeSplitter{Size: 4096})
60
-
61
- nbytes := int64(1024 * 1024 * 2)
62
- n, err := io.CopyN(dw, &datasource{}, nbytes)
63
- if err != nil {
64
- t.Fatal(err)
65
- }
66
-
67
- if n != nbytes {
68
- t.Fatal("Copied incorrect amount of bytes!")
69
- }
70
-
71
- dw.Close()
72
-
73
- node := dw.GetNode()
74
- read, err := dagio.NewDagReader(node, dag)
75
- if err != nil {
76
- t.Fatal(err)
77
- }
78
-
79
- d := &datasource{}
80
- if !d.Matches(t, read, int(nbytes)) {
81
- t.Fatal("Failed to validate!")
82
- }
83
-}
84
-
85
-func TestMassiveWrite(t *testing.T) {
86
- t.SkipNow()
87
- dstore := ds.NewNullDatastore()
88
- bserv, err := bs.NewBlockService(dstore, nil)
89
- if err != nil {
90
- t.Fatal(err)
91
- }
92
- dag := mdag.NewDAGService(bserv)
93
- dw := dagio.NewDagWriter(dag, &chunk.SizeSplitter{Size: 4096})
94
-
95
- nbytes := int64(1024 * 1024 * 1024 * 16)
96
- n, err := io.CopyN(dw, &datasource{}, nbytes)
97
- if err != nil {
98
- t.Fatal(err)
99
- }
100
- if n != nbytes {
101
- t.Fatal("Incorrect copy size.")
102
- }
103
- dw.Close()
104
-}
105
-
106
-func BenchmarkDagWriter(b *testing.B) {
107
- dstore := ds.NewNullDatastore()
108
- bserv, err := bs.NewBlockService(dstore, nil)
109
- if err != nil {
110
- b.Fatal(err)
111
- }
112
- dag := mdag.NewDAGService(bserv)
113
-
114
- b.ResetTimer()
115
- nbytes := int64(100000)
116
- for i := 0; i < b.N; i++ {
117
- b.SetBytes(nbytes)
118
- dw := dagio.NewDagWriter(dag, &chunk.SizeSplitter{Size: 4096})
119
- n, err := io.CopyN(dw, &datasource{}, nbytes)
120
- if err != nil {
121
- b.Fatal(err)
122
- }
123
- if n != nbytes {
124
- b.Fatal("Incorrect copy size.")
125
- }
126
- dw.Close()
127
- }
128
-
129
-}
130
-
131
-func TestAgainstImporter(t *testing.T) {
132
- dstore := ds.NewMapDatastore()
133
- bserv, err := bs.NewBlockService(dstore, nil)
134
- if err != nil {
135
- t.Fatal(err)
136
- }
137
- dag := mdag.NewDAGService(bserv)
138
-
139
- nbytes := int64(1024 * 1024 * 2)
140
-
141
- // DagWriter
142
- dw := dagio.NewDagWriter(dag, &chunk.SizeSplitter{4096})
143
- n, err := io.CopyN(dw, &datasource{}, nbytes)
144
- if err != nil {
145
- t.Fatal(err)
146
- }
147
- if n != nbytes {
148
- t.Fatal("Copied incorrect amount of bytes!")
149
- }
150
-
151
- dw.Close()
152
- dwNode := dw.GetNode()
153
- dwKey, err := dwNode.Key()
154
- if err != nil {
155
- t.Fatal(err)
156
- }
157
-
158
- // DagFromFile
159
- rl := &io.LimitedReader{&datasource{}, nbytes}
160
-
161
- dffNode, err := importer.NewDagFromReaderWithSplitter(rl, &chunk.SizeSplitter{4096})
162
- dffKey, err := dffNode.Key()
163
- if err != nil {
164
- t.Fatal(err)
165
- }
166
- if dwKey.String() != dffKey.String() {
167
- t.Errorf("\nDagWriter produced %s\n"+
168
- "DagFromReader produced %s",
169
- dwKey, dffKey)
170
- }
171
-}