@cryptotaxi247 / kubo / commits / 39a23392c

use rabin fingerprinting for a chunker

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> implement rabin fingerprinting as a chunker for ipfs License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> vendor correctly License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> refactor chunking interface a little License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> work chunking interface changes up into importer License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> move chunker type parsing into its own file in chunk License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Jul 29, 2015 at 13:08 UTC 39a23392c184a288db29201c5f6535fb3454a12f
27 files changed +1829 -321
Godeps/Godeps.json
+4
@@ -278,6 +278,10 @@
278 "ImportPath": "github.com/texttheater/golang-levenshtein/levenshtein",
279 "Rev": "dfd657628c58d3eeaa26391097853b2473c8b94e"
280 },
281 + {
282 + "ImportPath": "github.com/whyrusleeping/chunker",
283 + "Rev": "537e901819164627ca4bb5ce4e3faa8ce7956564"
284 + },
285 {
286 "ImportPath": "github.com/whyrusleeping/go-metrics",
287 "Rev": "1cd8009604ec2238b5a71305a0ecd974066e0e16"
Godeps/_workspace/src/github.com/whyrusleeping/chunker/.travis.yml new
+10
@@ -0,0 +1,10 @@
1 +language: go
2 +sudo: false
3 +
4 +go:
5 + - 1.3.3
6 + - 1.4.2
7 +
8 +os:
9 + - linux
10 + - osx
Godeps/_workspace/src/github.com/whyrusleeping/chunker/LICENSE new
+23
@@ -0,0 +1,23 @@
1 +Copyright (c) 2014, Alexander Neumann <alexander@bumpern.de>
2 +All rights reserved.
3 +
4 +Redistribution and use in source and binary forms, with or without
5 +modification, are permitted provided that the following conditions are met:
6 +
7 +1. Redistributions of source code must retain the above copyright notice, this
8 + list of conditions and the following disclaimer.
9 +
10 +2. Redistributions in binary form must reproduce the above copyright notice,
11 + this list of conditions and the following disclaimer in the documentation
12 + and/or other materials provided with the distribution.
13 +
14 +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
15 +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
16 +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
17 +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
18 +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
19 +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
20 +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
21 +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
22 +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
23 +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Godeps/_workspace/src/github.com/whyrusleeping/chunker/README.md new
+7
@@ -0,0 +1,7 @@
1 +[![Build Status](https://travis-ci.org/restic/chunker.svg?branch=master)](https://travis-ci.org/restic/chunker)
2 +
3 +Content Defined Chunking (CDC) based on a rolling Rabin Checksum.
4 +
5 +Part of https://github.com/restic/restic.
6 +
7 +Better README will follow soon.
Godeps/_workspace/src/github.com/whyrusleeping/chunker/chunker.go new
+370
@@ -0,0 +1,370 @@
1 +package chunker
2 +
3 +import (
4 + "errors"
5 + "hash"
6 + "io"
7 + "math"
8 + "sync"
9 +)
10 +
11 +const (
12 + KiB = 1024
13 + MiB = 1024 * KiB
14 +
15 + // WindowSize is the size of the sliding window.
16 + windowSize = 16
17 +
18 + chunkerBufSize = 512 * KiB
19 +)
20 +
21 +var bufPool = sync.Pool{
22 + New: func() interface{} { return make([]byte, chunkerBufSize) },
23 +}
24 +
25 +type tables struct {
26 + out [256]Pol
27 + mod [256]Pol
28 +}
29 +
30 +// cache precomputed tables, these are read-only anyway
31 +var cache struct {
32 + entries map[Pol]*tables
33 + sync.Mutex
34 +}
35 +
36 +func init() {
37 + cache.entries = make(map[Pol]*tables)
38 +}
39 +
40 +// Chunk is one content-dependent chunk of bytes whose end was cut when the
41 +// Rabin Fingerprint had the value stored in Cut.
42 +type Chunk struct {
43 + Start uint64
44 + Length uint64
45 + Cut uint64
46 + Digest []byte
47 + Data []byte
48 +}
49 +
50 +func (c Chunk) Reader(r io.ReaderAt) io.Reader {
51 + return io.NewSectionReader(r, int64(c.Start), int64(c.Length))
52 +}
53 +
54 +// Chunker splits content with Rabin Fingerprints.
55 +type Chunker struct {
56 + pol Pol
57 + polShift uint64
58 + tables *tables
59 +
60 + rd io.Reader
61 + closed bool
62 +
63 + chunkbuf []byte
64 +
65 + window [windowSize]byte
66 + wpos int
67 +
68 + buf []byte
69 + bpos uint64
70 + bmax uint64
71 +
72 + start uint64
73 + count uint64
74 + pos uint64
75 +
76 + pre uint64 // wait for this many bytes before start calculating an new chunk
77 +
78 + digest uint64
79 + h hash.Hash
80 +
81 + sizeMask uint64
82 +
83 + // minimal and maximal size of the outputted blocks
84 + MinSize uint64
85 + MaxSize uint64
86 +}
87 +
88 +// New returns a new Chunker based on polynomial p that reads from rd
89 +// with bufsize and pass all data to hash along the way.
90 +func New(rd io.Reader, pol Pol, h hash.Hash, avSize, min, max uint64) *Chunker {
91 +
92 + sizepow := uint(math.Log2(float64(avSize)))
93 +
94 + c := &Chunker{
95 + buf: bufPool.Get().([]byte),
96 + h: h,
97 + pol: pol,
98 + rd: rd,
99 + chunkbuf: make([]byte, 0, max),
100 + sizeMask: (1 << sizepow) - 1,
101 +
102 + MinSize: min,
103 + MaxSize: max,
104 + }
105 +
106 + c.reset()
107 +
108 + return c
109 +}
110 +
111 +func (c *Chunker) reset() {
112 + c.polShift = uint64(c.pol.Deg() - 8)
113 + c.fillTables()
114 +
115 + for i := 0; i < windowSize; i++ {
116 + c.window[i] = 0
117 + }
118 +
119 + c.closed = false
120 + c.digest = 0
121 + c.wpos = 0
122 + c.count = 0
123 + c.slide(1)
124 + c.start = c.pos
125 +
126 + if c.h != nil {
127 + c.h.Reset()
128 + }
129 +
130 + // do not start a new chunk unless at least MinSize bytes have been read
131 + c.pre = c.MinSize - windowSize
132 +}
133 +
134 +// Calculate out_table and mod_table for optimization. Must be called only
135 +// once. This implementation uses a cache in the global variable cache.
136 +func (c *Chunker) fillTables() {
137 + // if polynomial hasn't been specified, do not compute anything for now
138 + if c.pol == 0 {
139 + return
140 + }
141 +
142 + // test if the tables are cached for this polynomial
143 + cache.Lock()
144 + defer cache.Unlock()
145 + if t, ok := cache.entries[c.pol]; ok {
146 + c.tables = t
147 + return
148 + }
149 +
150 + // else create a new entry
151 + c.tables = &tables{}
152 + cache.entries[c.pol] = c.tables
153 +
154 + // calculate table for sliding out bytes. The byte to slide out is used as
155 + // the index for the table, the value contains the following:
156 + // out_table[b] = Hash(b || 0 || ... || 0)
157 + // \ windowsize-1 zero bytes /
158 + // To slide out byte b_0 for window size w with known hash
159 + // H := H(b_0 || ... || b_w), it is sufficient to add out_table[b_0]:
160 + // H(b_0 || ... || b_w) + H(b_0 || 0 || ... || 0)
161 + // = H(b_0 + b_0 || b_1 + 0 || ... || b_w + 0)
162 + // = H( 0 || b_1 || ... || b_w)
163 + //
164 + // Afterwards a new byte can be shifted in.
165 + for b := 0; b < 256; b++ {
166 + var h Pol
167 +
168 + h = appendByte(h, byte(b), c.pol)
169 + for i := 0; i < windowSize-1; i++ {
170 + h = appendByte(h, 0, c.pol)
171 + }
172 + c.tables.out[b] = h
173 + }
174 +
175 + // calculate table for reduction mod Polynomial
176 + k := c.pol.Deg()
177 + for b := 0; b < 256; b++ {
178 + // mod_table[b] = A | B, where A = (b(x) * x^k mod pol) and B = b(x) * x^k
179 + //
180 + // The 8 bits above deg(Polynomial) determine what happens next and so
181 + // these bits are used as a lookup to this table. The value is split in
182 + // two parts: Part A contains the result of the modulus operation, part
183 + // B is used to cancel out the 8 top bits so that one XOR operation is
184 + // enough to reduce modulo Polynomial
185 + c.tables.mod[b] = Pol(uint64(b)<<uint64(k)).Mod(c.pol) | (Pol(b) << uint64(k))
186 + }
187 +}
188 +
189 +func (c *Chunker) nextBytes() []byte {
190 + data := dupBytes(c.chunkbuf[:c.count])
191 + n := copy(c.chunkbuf, c.chunkbuf[c.count:])
192 + c.chunkbuf = c.chunkbuf[:n]
193 +
194 + return data
195 +}
196 +
197 +// Next returns the position and length of the next chunk of data. If an error
198 +// occurs while reading, the error is returned with a nil chunk. The state of
199 +// the current chunk is undefined. When the last chunk has been returned, all
200 +// subsequent calls yield a nil chunk and an io.EOF error.
201 +func (c *Chunker) Next() (*Chunk, error) {
202 + if c.tables == nil {
203 + return nil, errors.New("polynomial is not set")
204 + }
205 +
206 + for {
207 + if c.bpos >= c.bmax {
208 + n, err := io.ReadFull(c.rd, c.buf[:])
209 + c.chunkbuf = append(c.chunkbuf, c.buf[:n]...)
210 +
211 + if err == io.ErrUnexpectedEOF {
212 + err = nil
213 + }
214 +
215 + // io.ReadFull only returns io.EOF when no bytes could be read. If
216 + // this is the case and we're in this branch, there are no more
217 + // bytes to buffer, so this was the last chunk. If a different
218 + // error has occurred, return that error and abandon the current
219 + // chunk.
220 + if err == io.EOF && !c.closed {
221 + c.closed = true
222 +
223 + // return the buffer to the pool
224 + bufPool.Put(c.buf)
225 +
226 + data := c.nextBytes()
227 +
228 + // return current chunk, if any bytes have been processed
229 + if c.count > 0 {
230 + return &Chunk{
231 + Start: c.start,
232 + Length: c.count,
233 + Cut: c.digest,
234 + Digest: c.hashDigest(),
235 + Data: data,
236 + }, nil
237 + }
238 + }
239 +
240 + if err != nil {
241 + return nil, err
242 + }
243 +
244 + c.bpos = 0
245 + c.bmax = uint64(n)
246 + }
247 +
248 + // check if bytes have to be dismissed before starting a new chunk
249 + if c.pre > 0 {
250 + n := c.bmax - c.bpos
251 + if c.pre > uint64(n) {
252 + c.pre -= uint64(n)
253 + c.updateHash(c.buf[c.bpos:c.bmax])
254 +
255 + c.count += uint64(n)
256 + c.pos += uint64(n)
257 + c.bpos = c.bmax
258 +
259 + continue
260 + }
261 +
262 + c.updateHash(c.buf[c.bpos : c.bpos+c.pre])
263 +
264 + c.bpos += c.pre
265 + c.count += c.pre
266 + c.pos += c.pre
267 + c.pre = 0
268 + }
269 +
270 + add := c.count
271 + for _, b := range c.buf[c.bpos:c.bmax] {
272 + // inline c.slide(b) and append(b) to increase performance
273 + out := c.window[c.wpos]
274 + c.window[c.wpos] = b
275 + c.digest ^= uint64(c.tables.out[out])
276 + c.wpos = (c.wpos + 1) % windowSize
277 +
278 + // c.append(b)
279 + index := c.digest >> c.polShift
280 + c.digest <<= 8
281 + c.digest |= uint64(b)
282 +
283 + c.digest ^= uint64(c.tables.mod[index])
284 + // end inline
285 +
286 + add++
287 + if add < c.MinSize {
288 + continue
289 + }
290 +
291 + if (c.digest&c.sizeMask) == 0 || add >= c.MaxSize {
292 + i := add - c.count - 1
293 + c.updateHash(c.buf[c.bpos : c.bpos+uint64(i)+1])
294 + c.count = add
295 + c.pos += uint64(i) + 1
296 + c.bpos += uint64(i) + 1
297 +
298 + data := c.nextBytes()
299 +
300 + chunk := &Chunk{
301 + Start: c.start,
302 + Length: c.count,
303 + Cut: c.digest,
304 + Digest: c.hashDigest(),
305 + Data: data,
306 + }
307 +
308 + c.reset()
309 +
310 + return chunk, nil
311 + }
312 + }
313 +
314 + steps := c.bmax - c.bpos
315 + if steps > 0 {
316 + c.updateHash(c.buf[c.bpos : c.bpos+steps])
317 + }
318 + c.count += steps
319 + c.pos += steps
320 + c.bpos = c.bmax
321 + }
322 +}
323 +
324 +func dupBytes(b []byte) []byte {
325 + out := make([]byte, len(b))
326 + copy(out, b)
327 + return out
328 +}
329 +
330 +func (c *Chunker) updateHash(data []byte) {
331 + if c.h != nil {
332 + // the hashes from crypto/sha* do not return an error
333 + _, err := c.h.Write(data)
334 + if err != nil {
335 + panic(err)
336 + }
337 + }
338 +}
339 +
340 +func (c *Chunker) hashDigest() []byte {
341 + if c.h == nil {
342 + return nil
343 + }
344 +
345 + return c.h.Sum(nil)
346 +}
347 +
348 +func (c *Chunker) append(b byte) {
349 + index := c.digest >> c.polShift
350 + c.digest <<= 8
351 + c.digest |= uint64(b)
352 +
353 + c.digest ^= uint64(c.tables.mod[index])
354 +}
355 +
356 +func (c *Chunker) slide(b byte) {
357 + out := c.window[c.wpos]
358 + c.window[c.wpos] = b
359 + c.digest ^= uint64(c.tables.out[out])
360 + c.wpos = (c.wpos + 1) % windowSize
361 +
362 + c.append(b)
363 +}
364 +
365 +func appendByte(hash Pol, b byte, pol Pol) Pol {
366 + hash <<= 8
367 + hash |= Pol(b)
368 +
369 + return hash.Mod(pol)
370 +}
Godeps/_workspace/src/github.com/whyrusleeping/chunker/chunker_test.go new
+298
@@ -0,0 +1,298 @@
1 +package chunker_test
2 +
3 +import (
4 + "bytes"
5 + "crypto/md5"
6 + "crypto/sha256"
7 + "encoding/hex"
8 + "hash"
9 + "io"
10 + "io/ioutil"
11 + "math/rand"
12 + "testing"
13 + "time"
14 +
15 + "github.com/restic/chunker"
16 + . "github.com/restic/restic/test"
17 +)
18 +
19 +func parseDigest(s string) []byte {
20 + d, err := hex.DecodeString(s)
21 + if err != nil {
22 + panic(err)
23 + }
24 +
25 + return d
26 +}
27 +
28 +type chunk struct {
29 + Length uint
30 + CutFP uint64
31 + Digest []byte
32 +}
33 +
34 +// polynomial used for all the tests below
35 +const testPol = chunker.Pol(0x3DA3358B4DC173)
36 +
37 +// created for 32MB of random data out of math/rand's Uint32() seeded by
38 +// constant 23
39 +//
40 +// chunking configuration:
41 +// window size 64, avg chunksize 1<<20, min chunksize 1<<19, max chunksize 1<<23
42 +// polynom 0x3DA3358B4DC173
43 +var chunks1 = []chunk{
44 + chunk{2163460, 0x000b98d4cdf00000, parseDigest("4b94cb2cf293855ea43bf766731c74969b91aa6bf3c078719aabdd19860d590d")},
45 + chunk{643703, 0x000d4e8364d00000, parseDigest("5727a63c0964f365ab8ed2ccf604912f2ea7be29759a2b53ede4d6841e397407")},
46 + chunk{1528956, 0x0015a25c2ef00000, parseDigest("a73759636a1e7a2758767791c69e81b69fb49236c6929e5d1b654e06e37674ba")},
47 + chunk{1955808, 0x00102a8242e00000, parseDigest("c955fb059409b25f07e5ae09defbbc2aadf117c97a3724e06ad4abd2787e6824")},
48 + chunk{2222372, 0x00045da878000000, parseDigest("6ba5e9f7e1b310722be3627716cf469be941f7f3e39a4c3bcefea492ec31ee56")},
49 + chunk{2538687, 0x00198a8179900000, parseDigest("8687937412f654b5cfe4a82b08f28393a0c040f77c6f95e26742c2fc4254bfde")},
50 + chunk{609606, 0x001d4e8d17100000, parseDigest("5da820742ff5feb3369112938d3095785487456f65a8efc4b96dac4be7ebb259")},
51 + chunk{1205738, 0x000a7204dd600000, parseDigest("cc70d8fad5472beb031b1aca356bcab86c7368f40faa24fe5f8922c6c268c299")},
52 + chunk{959742, 0x00183e71e1400000, parseDigest("4065bdd778f95676c92b38ac265d361f81bff17d76e5d9452cf985a2ea5a4e39")},
53 + chunk{4036109, 0x001fec043c700000, parseDigest("b9cf166e75200eb4993fc9b6e22300a6790c75e6b0fc8f3f29b68a752d42f275")},
54 + chunk{1525894, 0x000b1574b1500000, parseDigest("2f238180e4ca1f7520a05f3d6059233926341090f9236ce677690c1823eccab3")},
55 + chunk{1352720, 0x00018965f2e00000, parseDigest("afd12f13286a3901430de816e62b85cc62468c059295ce5888b76b3af9028d84")},
56 + chunk{811884, 0x00155628aa100000, parseDigest("42d0cdb1ee7c48e552705d18e061abb70ae7957027db8ae8db37ec756472a70a")},
57 + chunk{1282314, 0x001909a0a1400000, parseDigest("819721c2457426eb4f4c7565050c44c32076a56fa9b4515a1c7796441730eb58")},
58 + chunk{1318021, 0x001cceb980000000, parseDigest("842eb53543db55bacac5e25cb91e43cc2e310fe5f9acc1aee86bdf5e91389374")},
59 + chunk{948640, 0x0011f7a470a00000, parseDigest("b8e36bf7019bb96ac3fb7867659d2167d9d3b3148c09fe0de45850b8fe577185")},
60 + chunk{645464, 0x00030ce2d9400000, parseDigest("5584bd27982191c3329f01ed846bfd266e96548dfa87018f745c33cfc240211d")},
61 + chunk{533758, 0x0004435c53c00000, parseDigest("4da778a25b72a9a0d53529eccfe2e5865a789116cb1800f470d8df685a8ab05d")},
62 + chunk{1128303, 0x0000c48517800000, parseDigest("08c6b0b38095b348d80300f0be4c5184d2744a17147c2cba5cc4315abf4c048f")},
63 + chunk{800374, 0x000968473f900000, parseDigest("820284d2c8fd243429674c996d8eb8d3450cbc32421f43113e980f516282c7bf")},
64 + chunk{2453512, 0x001e197c92600000, parseDigest("5fa870ed107c67704258e5e50abe67509fb73562caf77caa843b5f243425d853")},
65 + chunk{2651975, 0x000ae6c868000000, parseDigest("181347d2bbec32bef77ad5e9001e6af80f6abcf3576549384d334ee00c1988d8")},
66 + chunk{237392, 0x0000000000000001, parseDigest("fcd567f5d866357a8e299fd5b2359bb2c8157c30395229c4e9b0a353944a7978")},
67 +}
68 +
69 +// test if nullbytes are correctly split, even if length is a multiple of MinSize.
70 +var chunks2 = []chunk{
71 + chunk{chunker.MinSize, 0, parseDigest("07854d2fef297a06ba81685e660c332de36d5d18d546927d30daad6d7fda1541")},
72 + chunk{chunker.MinSize, 0, parseDigest("07854d2fef297a06ba81685e660c332de36d5d18d546927d30daad6d7fda1541")},
73 + chunk{chunker.MinSize, 0, parseDigest("07854d2fef297a06ba81685e660c332de36d5d18d546927d30daad6d7fda1541")},
74 + chunk{chunker.MinSize, 0, parseDigest("07854d2fef297a06ba81685e660c332de36d5d18d546927d30daad6d7fda1541")},
75 +}
76 +
77 +func testWithData(t *testing.T, chnker *chunker.Chunker, testChunks []chunk) []*chunker.Chunk {
78 + chunks := []*chunker.Chunk{}
79 +
80 + pos := uint(0)
81 + for i, chunk := range testChunks {
82 + c, err := chnker.Next()
83 +
84 + if err != nil {
85 + t.Fatalf("Error returned with chunk %d: %v", i, err)
86 + }
87 +
88 + if c == nil {
89 + t.Fatalf("Nil chunk returned")
90 + }
91 +
92 + if c != nil {
93 + if c.Start != pos {
94 + t.Fatalf("Start for chunk %d does not match: expected %d, got %d",
95 + i, pos, c.Start)
96 + }
97 +
98 + if c.Length != chunk.Length {
99 + t.Fatalf("Length for chunk %d does not match: expected %d, got %d",
100 + i, chunk.Length, c.Length)
101 + }
102 +
103 + if c.Cut != chunk.CutFP {
104 + t.Fatalf("Cut fingerprint for chunk %d/%d does not match: expected %016x, got %016x",
105 + i, len(chunks)-1, chunk.CutFP, c.Cut)
106 + }
107 +
108 + if c.Digest != nil && !bytes.Equal(c.Digest, chunk.Digest) {
109 + t.Fatalf("Digest fingerprint for chunk %d/%d does not match: expected %02x, got %02x",
110 + i, len(chunks)-1, chunk.Digest, c.Digest)
111 + }
112 +
113 + pos += c.Length
114 + chunks = append(chunks, c)
115 + }
116 + }
117 +
118 + c, err := chnker.Next()
119 +
120 + if c != nil {
121 + t.Fatal("additional non-nil chunk returned")
122 + }
123 +
124 + if err != io.EOF {
125 + t.Fatal("wrong error returned after last chunk")
126 + }
127 +
128 + return chunks
129 +}
130 +
131 +func getRandom(seed, count int) []byte {
132 + buf := make([]byte, count)
133 +
134 + rnd := rand.New(rand.NewSource(23))
135 + for i := 0; i < count; i += 4 {
136 + r := rnd.Uint32()
137 + buf[i] = byte(r)
138 + buf[i+1] = byte(r >> 8)
139 + buf[i+2] = byte(r >> 16)
140 + buf[i+3] = byte(r >> 24)
141 + }
142 +
143 + return buf
144 +}
145 +
146 +func TestChunker(t *testing.T) {
147 + // setup data source
148 + buf := getRandom(23, 32*1024*1024)
149 + ch := chunker.New(bytes.NewReader(buf), testPol, sha256.New())
150 + chunks := testWithData(t, ch, chunks1)
151 +
152 + // test reader
153 + for i, c := range chunks {
154 + rd := c.Reader(bytes.NewReader(buf))
155 +
156 + h := sha256.New()
157 + n, err := io.Copy(h, rd)
158 + if err != nil {
159 + t.Fatalf("io.Copy(): %v", err)
160 + }
161 +
162 + if uint(n) != chunks1[i].Length {
163 + t.Fatalf("reader returned wrong number of bytes: expected %d, got %d",
164 + chunks1[i].Length, n)
165 + }
166 +
167 + d := h.Sum(nil)
168 + if !bytes.Equal(d, chunks1[i].Digest) {
169 + t.Fatalf("wrong hash returned: expected %02x, got %02x",
170 + chunks1[i].Digest, d)
171 + }
172 + }
173 +
174 + // setup nullbyte data source
175 + buf = bytes.Repeat([]byte{0}, len(chunks2)*chunker.MinSize)
176 + ch = chunker.New(bytes.NewReader(buf), testPol, sha256.New())
177 +
178 + testWithData(t, ch, chunks2)
179 +}
180 +
181 +func TestChunkerWithRandomPolynomial(t *testing.T) {
182 + // setup data source
183 + buf := getRandom(23, 32*1024*1024)
184 +
185 + // generate a new random polynomial
186 + start := time.Now()
187 + p, err := chunker.RandomPolynomial()
188 + OK(t, err)
189 + t.Logf("generating random polynomial took %v", time.Since(start))
190 +
191 + start = time.Now()
192 + ch := chunker.New(bytes.NewReader(buf), p, sha256.New())
193 + t.Logf("creating chunker took %v", time.Since(start))
194 +
195 + // make sure that first chunk is different
196 + c, err := ch.Next()
197 +
198 + Assert(t, c.Cut != chunks1[0].CutFP,
199 + "Cut point is the same")
200 + Assert(t, c.Length != chunks1[0].Length,
201 + "Length is the same")
202 + Assert(t, !bytes.Equal(c.Digest, chunks1[0].Digest),
203 + "Digest is the same")
204 +}
205 +
206 +func TestChunkerWithoutHash(t *testing.T) {
207 + // setup data source
208 + buf := getRandom(23, 32*1024*1024)
209 +
210 + ch := chunker.New(bytes.NewReader(buf), testPol, nil)
211 + chunks := testWithData(t, ch, chunks1)
212 +
213 + // test reader
214 + for i, c := range chunks {
215 + rd := c.Reader(bytes.NewReader(buf))
216 +
217 + buf2, err := ioutil.ReadAll(rd)
218 + if err != nil {
219 + t.Fatalf("io.Copy(): %v", err)
220 + }
221 +
222 + if uint(len(buf2)) != chunks1[i].Length {
223 + t.Fatalf("reader returned wrong number of bytes: expected %d, got %d",
224 + chunks1[i].Length, uint(len(buf2)))
225 + }
226 +
227 + if uint(len(buf2)) != chunks1[i].Length {
228 + t.Fatalf("wrong number of bytes returned: expected %02x, got %02x",
229 + chunks[i].Length, len(buf2))
230 + }
231 +
232 + if !bytes.Equal(buf[c.Start:c.Start+c.Length], buf2) {
233 + t.Fatalf("invalid data for chunk returned: expected %02x, got %02x",
234 + buf[c.Start:c.Start+c.Length], buf2)
235 + }
236 + }
237 +
238 + // setup nullbyte data source
239 + buf = bytes.Repeat([]byte{0}, len(chunks2)*chunker.MinSize)
240 + ch = chunker.New(bytes.NewReader(buf), testPol, sha256.New())
241 +
242 + testWithData(t, ch, chunks2)
243 +}
244 +
245 +func benchmarkChunker(b *testing.B, hash hash.Hash) {
246 + size := 10 * 1024 * 1024
247 + rd := bytes.NewReader(getRandom(23, size))
248 +
249 + b.ResetTimer()
250 + b.SetBytes(int64(size))
251 +
252 + var chunks int
253 + for i := 0; i < b.N; i++ {
254 + chunks = 0
255 +
256 + rd.Seek(0, 0)
257 + ch := chunker.New(rd, testPol, hash)
258 +
259 + for {
260 + _, err := ch.Next()
261 +
262 + if err == io.EOF {
263 + break
264 + }
265 +
266 + if err != nil {
267 + b.Fatalf("Unexpected error occurred: %v", err)
268 + }
269 +
270 + chunks++
271 + }
272 + }
273 +
274 + b.Logf("%d chunks, average chunk size: %d bytes", chunks, size/chunks)
275 +}
276 +
277 +func BenchmarkChunkerWithSHA256(b *testing.B) {
278 + benchmarkChunker(b, sha256.New())
279 +}
280 +
281 +func BenchmarkChunkerWithMD5(b *testing.B) {
282 + benchmarkChunker(b, md5.New())
283 +}
284 +
285 +func BenchmarkChunker(b *testing.B) {
286 + benchmarkChunker(b, nil)
287 +}
288 +
289 +func BenchmarkNewChunker(b *testing.B) {
290 + p, err := chunker.RandomPolynomial()
291 + OK(b, err)
292 +
293 + b.ResetTimer()
294 +
295 + for i := 0; i < b.N; i++ {
296 + chunker.New(bytes.NewBuffer(nil), p, nil)
297 + }
298 +}
Godeps/_workspace/src/github.com/whyrusleeping/chunker/doc.go new
+82
@@ -0,0 +1,82 @@
1 +// Copyright 2014 Alexander Neumann. All rights reserved.
2 +// Use of this source code is governed by a BSD-style
3 +// license that can be found in the LICENSE file.
4 +
5 +/*
6 +Package chunker implements Content Defined Chunking (CDC) based on a rolling
7 +Rabin Checksum.
8 +
9 +Choosing a Random Irreducible Polynomial
10 +
11 +The function RandomPolynomial() returns a new random polynomial of degree 53
12 +for use with the chunker. The degree 53 is chosen because it is the largest
13 +prime below 64-8 = 56, so that the top 8 bits of an uint64 can be used for
14 +optimising calculations in the chunker.
15 +
16 +A random polynomial is chosen selecting 64 random bits, masking away bits
17 +64..54 and setting bit 53 to one (otherwise the polynomial is not of the
18 +desired degree) and bit 0 to one (otherwise the polynomial is trivially
19 +reducible), so that 51 bits are chosen at random.
20 +
21 +This process is repeated until Irreducible() returns true, then this
22 +polynomials is returned. If this doesn't happen after 1 million tries, the
23 +function returns an error. The probability for selecting an irreducible
24 +polynomial at random is about 7.5% ( (2^53-2)/53 / 2^51), so the probability
25 +that no irreducible polynomial has been found after 100 tries is lower than
26 +0.04%.
27 +
28 +Verifying Irreducible Polynomials
29 +
30 +During development the results have been verified using the computational
31 +discrete algebra system GAP, which can be obtained from the website at
32 +http://www.gap-system.org/.
33 +
34 +For filtering a given list of polynomials in hexadecimal coefficient notation,
35 +the following script can be used:
36 +
37 + # create x over F_2 = GF(2)
38 + x := Indeterminate(GF(2), "x");
39 +
40 + # test if polynomial is irreducible, i.e. the number of factors is one
41 + IrredPoly := function (poly)
42 + return (Length(Factors(poly)) = 1);
43 + end;;
44 +
45 + # create a polynomial in x from the hexadecimal representation of the
46 + # coefficients
47 + Hex2Poly := function (s)
48 + return ValuePol(CoefficientsQadic(IntHexString(s), 2), x);
49 + end;;
50 +
51 + # list of candidates, in hex
52 + candidates := [ "3DA3358B4DC173" ];
53 +
54 + # create real polynomials
55 + L := List(candidates, Hex2Poly);
56 +
57 + # filter and display the list of irreducible polynomials contained in L
58 + Display(Filtered(L, x -> (IrredPoly(x))));
59 +
60 +All irreducible polynomials from the list are written to the output.
61 +
62 +Background Literature
63 +
64 +An introduction to Rabin Fingerprints/Checksums can be found in the following articles:
65 +
66 +Michael O. Rabin (1981): "Fingerprinting by Random Polynomials"
67 +http://www.xmailserver.org/rabin.pdf
68 +
69 +Ross N. Williams (1993): "A Painless Guide to CRC Error Detection Algorithms"
70 +http://www.zlib.net/crc_v3.txt
71 +
72 +Andrei Z. Broder (1993): "Some Applications of Rabin's Fingerprinting Method"
73 +http://www.xmailserver.org/rabin_apps.pdf
74 +
75 +Shuhong Gao and Daniel Panario (1997): "Tests and Constructions of Irreducible Polynomials over Finite Fields"
76 +http://www.math.clemson.edu/~sgao/papers/GP97a.pdf
77 +
78 +Andrew Kadatch, Bob Jenkins (2007): "Everything we know about CRC but afraid to forget"
79 +http://crcutil.googlecode.com/files/crc-doc.1.0.pdf
80 +
81 +*/
82 +package chunker
Godeps/_workspace/src/github.com/whyrusleeping/chunker/polynomials.go new
+278
@@ -0,0 +1,278 @@
1 +package chunker
2 +
3 +import (
4 + "crypto/rand"
5 + "encoding/binary"
6 + "errors"
7 + "fmt"
8 + "strconv"
9 +)
10 +
11 +// Pol is a polynomial from F_2[X].
12 +type Pol uint64
13 +
14 +// Add returns x+y.
15 +func (x Pol) Add(y Pol) Pol {
16 + r := Pol(uint64(x) ^ uint64(y))
17 + return r
18 +}
19 +
20 +// mulOverflows returns true if the multiplication would overflow uint64.
21 +// Code by Rob Pike, see
22 +// https://groups.google.com/d/msg/golang-nuts/h5oSN5t3Au4/KaNQREhZh0QJ
23 +func mulOverflows(a, b Pol) bool {
24 + if a <= 1 || b <= 1 {
25 + return false
26 + }
27 + c := a.mul(b)
28 + d := c.Div(b)
29 + if d != a {
30 + return true
31 + }
32 +
33 + return false
34 +}
35 +
36 +func (x Pol) mul(y Pol) Pol {
37 + if x == 0 || y == 0 {
38 + return 0
39 + }
40 +
41 + var res Pol
42 + for i := 0; i <= y.Deg(); i++ {
43 + if (y & (1 << uint(i))) > 0 {
44 + res = res.Add(x << uint(i))
45 + }
46 + }
47 +
48 + return res
49 +}
50 +
51 +// Mul returns x*y. When an overflow occurs, Mul panics.
52 +func (x Pol) Mul(y Pol) Pol {
53 + if mulOverflows(x, y) {
54 + panic("multiplication would overflow uint64")
55 + }
56 +
57 + return x.mul(y)
58 +}
59 +
60 +// Deg returns the degree of the polynomial x. If x is zero, -1 is returned.
61 +func (x Pol) Deg() int {
62 + // the degree of 0 is -1
63 + if x == 0 {
64 + return -1
65 + }
66 +
67 + var mask Pol = (1 << 63)
68 + for i := 63; i >= 0; i-- {
69 + // test if bit i is set
70 + if x&mask > 0 {
71 + // this is the degree of x
72 + return i
73 + }
74 + mask >>= 1
75 + }
76 +
77 + // fall-through, return -1
78 + return -1
79 +}
80 +
81 +// String returns the coefficients in hex.
82 +func (x Pol) String() string {
83 + return "0x" + strconv.FormatUint(uint64(x), 16)
84 +}
85 +
86 +// Expand returns the string representation of the polynomial x.
87 +func (x Pol) Expand() string {
88 + if x == 0 {
89 + return "0"
90 + }
91 +
92 + s := ""
93 + for i := x.Deg(); i > 1; i-- {
94 + if x&(1<<uint(i)) > 0 {
95 + s += fmt.Sprintf("+x^%d", i)
96 + }
97 + }
98 +
99 + if x&2 > 0 {
100 + s += "+x"
101 + }
102 +
103 + if x&1 > 0 {
104 + s += "+1"
105 + }
106 +
107 + return s[1:]
108 +}
109 +
110 +// DivMod returns x / d = q, and remainder r,
111 +// see https://en.wikipedia.org/wiki/Division_algorithm
112 +func (x Pol) DivMod(d Pol) (Pol, Pol) {
113 + if x == 0 {
114 + return 0, 0
115 + }
116 +
117 + if d == 0 {
118 + panic("division by zero")
119 + }
120 +
121 + D := d.Deg()
122 + diff := x.Deg() - D
123 + if diff < 0 {
124 + return 0, x
125 + }
126 +
127 + var q Pol
128 + for diff >= 0 {
129 + m := d << uint(diff)
130 + q |= (1 << uint(diff))
131 + x = x.Add(m)
132 +
133 + diff = x.Deg() - D
134 + }
135 +
136 + return q, x
137 +}
138 +
139 +// Div returns the integer division result x / d.
140 +func (x Pol) Div(d Pol) Pol {
141 + q, _ := x.DivMod(d)
142 + return q
143 +}
144 +
145 +// Mod returns the remainder of x / d
146 +func (x Pol) Mod(d Pol) Pol {
147 + _, r := x.DivMod(d)
148 + return r
149 +}
150 +
151 +// I really dislike having a function that does not terminate, so specify a
152 +// really large upper bound for finding a new irreducible polynomial, and
153 +// return an error when no irreducible polynomial has been found within
154 +// randPolMaxTries.
155 +const randPolMaxTries = 1e6
156 +
157 +// RandomPolynomial returns a new random irreducible polynomial of degree 53
158 +// (largest prime number below 64-8). There are (2^53-2/53) irreducible
159 +// polynomials of degree 53 in F_2[X], c.f. Michael O. Rabin (1981):
160 +// "Fingerprinting by Random Polynomials", page 4. If no polynomial could be
161 +// found in one million tries, an error is returned.
162 +func RandomPolynomial() (Pol, error) {
163 + for i := 0; i < randPolMaxTries; i++ {
164 + var f Pol
165 +
166 + // choose polynomial at random
167 + err := binary.Read(rand.Reader, binary.LittleEndian, &f)
168 + if err != nil {
169 + return 0, err
170 + }
171 +
172 + // mask away bits above bit 53
173 + f &= Pol((1 << 54) - 1)
174 +
175 + // set highest and lowest bit so that the degree is 53 and the
176 + // polynomial is not trivially reducible
177 + f |= (1 << 53) | 1
178 +
179 + // test if f is irreducible
180 + if f.Irreducible() {
181 + return f, nil
182 + }
183 + }
184 +
185 + // If this is reached, we haven't found an irreducible polynomial in
186 + // randPolMaxTries. This error is very unlikely to occur.
187 + return 0, errors.New("unable to find new random irreducible polynomial")
188 +}
189 +
190 +// GCD computes the Greatest Common Divisor x and f.
191 +func (x Pol) GCD(f Pol) Pol {
192 + if f == 0 {
193 + return x
194 + }
195 +
196 + if x == 0 {
197 + return f
198 + }
199 +
200 + if x.Deg() < f.Deg() {
201 + x, f = f, x
202 + }
203 +
204 + return f.GCD(x.Mod(f))
205 +}
206 +
207 +// Irreducible returns true iff x is irreducible over F_2. This function
208 +// uses Ben Or's reducibility test.
209 +//
210 +// For details see "Tests and Constructions of Irreducible Polynomials over
211 +// Finite Fields".
212 +func (x Pol) Irreducible() bool {
213 + for i := 1; i <= x.Deg()/2; i++ {
214 + if x.GCD(qp(uint(i), x)) != 1 {
215 + return false
216 + }
217 + }
218 +
219 + return true
220 +}
221 +
222 +// MulMod computes x*f mod g
223 +func (x Pol) MulMod(f, g Pol) Pol {
224 + if x == 0 || f == 0 {
225 + return 0
226 + }
227 +
228 + var res Pol
229 + for i := 0; i <= f.Deg(); i++ {
230 + if (f & (1 << uint(i))) > 0 {
231 + a := x
232 + for j := 0; j < i; j++ {
233 + a = a.Mul(2).Mod(g)
234 + }
235 + res = res.Add(a).Mod(g)
236 + }
237 + }
238 +
239 + return res
240 +}
241 +
242 +// qp computes the polynomial (x^(2^p)-x) mod g. This is needed for the
243 +// reducibility test.
244 +func qp(p uint, g Pol) Pol {
245 + num := (1 << p)
246 + i := 1
247 +
248 + // start with x
249 + res := Pol(2)
250 +
251 + for i < num {
252 + // repeatedly square res
253 + res = res.MulMod(res, g)
254 + i *= 2
255 + }
256 +
257 + // add x
258 + return res.Add(2).Mod(g)
259 +}
260 +
261 +func (p Pol) MarshalJSON() ([]byte, error) {
262 + buf := strconv.AppendUint([]byte{'"'}, uint64(p), 16)
263 + buf = append(buf, '"')
264 + return buf, nil
265 +}
266 +
267 +func (p *Pol) UnmarshalJSON(data []byte) error {
268 + if len(data) < 2 {
269 + return errors.New("invalid string for polynomial")
270 + }
271 + n, err := strconv.ParseUint(string(data[1:len(data)-1]), 16, 64)
272 + if err != nil {
273 + return err
274 + }
275 + *p = Pol(n)
276 +
277 + return nil
278 +}
Godeps/_workspace/src/github.com/whyrusleeping/chunker/polynomials_test.go new
+385
@@ -0,0 +1,385 @@
1 +package chunker_test
2 +
3 +import (
4 + "strconv"
5 + "testing"
6 +
7 + "github.com/restic/chunker"
8 + . "github.com/restic/restic/test"
9 +)
10 +
11 +var polAddTests = []struct {
12 + x, y chunker.Pol
13 + sum chunker.Pol
14 +}{
15 + {23, 16, 23 ^ 16},
16 + {0x9a7e30d1e855e0a0, 0x670102a1f4bcd414, 0xfd7f32701ce934b4},
17 + {0x9a7e30d1e855e0a0, 0x9a7e30d1e855e0a0, 0},
18 +}
19 +
20 +func TestPolAdd(t *testing.T) {
21 + for _, test := range polAddTests {
22 + Equals(t, test.sum, test.x.Add(test.y))
23 + Equals(t, test.sum, test.y.Add(test.x))
24 + }
25 +}
26 +
27 +func parseBin(s string) chunker.Pol {
28 + i, err := strconv.ParseUint(s, 2, 64)
29 + if err != nil {
30 + panic(err)
31 + }
32 +
33 + return chunker.Pol(i)
34 +}
35 +
36 +var polMulTests = []struct {
37 + x, y chunker.Pol
38 + res chunker.Pol
39 +}{
40 + {1, 2, 2},
41 + {
42 + parseBin("1101"),
43 + parseBin("10"),
44 + parseBin("11010"),
45 + },
46 + {
47 + parseBin("1101"),
48 + parseBin("11"),
49 + parseBin("10111"),
50 + },
51 + {
52 + 0x40000000,
53 + 0x40000000,
54 + 0x1000000000000000,
55 + },
56 + {
57 + parseBin("1010"),
58 + parseBin("100100"),
59 + parseBin("101101000"),
60 + },
61 + {
62 + parseBin("100"),
63 + parseBin("11"),
64 + parseBin("1100"),
65 + },
66 + {
67 + parseBin("11"),
68 + parseBin("110101"),
69 + parseBin("1011111"),
70 + },
71 + {
72 + parseBin("10011"),
73 + parseBin("110101"),
74 + parseBin("1100001111"),
75 + },
76 +}
77 +
78 +func TestPolMul(t *testing.T) {
79 + for i, test := range polMulTests {
80 + m := test.x.Mul(test.y)
81 + Assert(t, test.res == m,
82 + "TestPolMul failed for test %d: %v * %v: want %v, got %v",
83 + i, test.x, test.y, test.res, m)
84 + m = test.y.Mul(test.x)
85 + Assert(t, test.res == test.y.Mul(test.x),
86 + "TestPolMul failed for %d: %v * %v: want %v, got %v",
87 + i, test.x, test.y, test.res, m)
88 + }
89 +}
90 +
91 +func TestPolMulOverflow(t *testing.T) {
92 + defer func() {
93 + // try to recover overflow error
94 + err := recover()
95 +
96 + if e, ok := err.(string); ok && e == "multiplication would overflow uint64" {
97 + return
98 + } else {
99 + t.Logf("invalid error raised: %v", err)
100 + // re-raise error if not overflow
101 + panic(err)
102 + }
103 + }()
104 +
105 + x := chunker.Pol(1 << 63)
106 + x.Mul(2)
107 + t.Fatal("overflow test did not panic")
108 +}
109 +
110 +var polDivTests = []struct {
111 + x, y chunker.Pol
112 + res chunker.Pol
113 +}{
114 + {10, 50, 0},
115 + {0, 1, 0},
116 + {
117 + parseBin("101101000"), // 0x168
118 + parseBin("1010"), // 0xa
119 + parseBin("100100"), // 0x24
120 + },
121 + {2, 2, 1},
122 + {
123 + 0x8000000000000000,
124 + 0x8000000000000000,
125 + 1,
126 + },
127 + {
128 + parseBin("1100"),
129 + parseBin("100"),
130 + parseBin("11"),
131 + },
132 + {
133 + parseBin("1100001111"),
134 + parseBin("10011"),
135 + parseBin("110101"),
136 + },
137 +}
138 +
139 +func TestPolDiv(t *testing.T) {
140 + for i, test := range polDivTests {
141 + m := test.x.Div(test.y)
142 + Assert(t, test.res == m,
143 + "TestPolDiv failed for test %d: %v * %v: want %v, got %v",
144 + i, test.x, test.y, test.res, m)
145 + }
146 +}
147 +
148 +var polModTests = []struct {
149 + x, y chunker.Pol
150 + res chunker.Pol
151 +}{
152 + {10, 50, 10},
153 + {0, 1, 0},
154 + {
155 + parseBin("101101001"),
156 + parseBin("1010"),
157 + parseBin("1"),
158 + },
159 + {2, 2, 0},
160 + {
161 + 0x8000000000000000,
162 + 0x8000000000000000,
163 + 0,
164 + },
165 + {
166 + parseBin("1100"),
167 + parseBin("100"),
168 + parseBin("0"),
169 + },
170 + {
171 + parseBin("1100001111"),
172 + parseBin("10011"),
173 + parseBin("0"),
174 + },
175 +}
176 +
177 +func TestPolModt(t *testing.T) {
178 + for _, test := range polModTests {
179 + Equals(t, test.res, test.x.Mod(test.y))
180 + }
181 +}
182 +
183 +func BenchmarkPolDivMod(t *testing.B) {
184 + f := chunker.Pol(0x2482734cacca49)
185 + g := chunker.Pol(0x3af4b284899)
186 +
187 + for i := 0; i < t.N; i++ {
188 + g.DivMod(f)
189 + }
190 +}
191 +
192 +func BenchmarkPolDiv(t *testing.B) {
193 + f := chunker.Pol(0x2482734cacca49)
194 + g := chunker.Pol(0x3af4b284899)
195 +
196 + for i := 0; i < t.N; i++ {
197 + g.Div(f)
198 + }
199 +}
200 +
201 +func BenchmarkPolMod(t *testing.B) {
202 + f := chunker.Pol(0x2482734cacca49)
203 + g := chunker.Pol(0x3af4b284899)
204 +
205 + for i := 0; i < t.N; i++ {
206 + g.Mod(f)
207 + }
208 +}
209 +
210 +func BenchmarkPolDeg(t *testing.B) {
211 + f := chunker.Pol(0x3af4b284899)
212 + d := f.Deg()
213 + if d != 41 {
214 + t.Fatalf("BenchmalPolDeg: Wrong degree %d returned, expected %d",
215 + d, 41)
216 + }
217 +
218 + for i := 0; i < t.N; i++ {
219 + f.Deg()
220 + }
221 +}
222 +
223 +func TestRandomPolynomial(t *testing.T) {
224 + _, err := chunker.RandomPolynomial()
225 + OK(t, err)
226 +}
227 +
228 +func BenchmarkRandomPolynomial(t *testing.B) {
229 + for i := 0; i < t.N; i++ {
230 + _, err := chunker.RandomPolynomial()
231 + OK(t, err)
232 + }
233 +}
234 +
235 +func TestExpandPolynomial(t *testing.T) {
236 + pol := chunker.Pol(0x3DA3358B4DC173)
237 + s := pol.Expand()
238 + Equals(t, "x^53+x^52+x^51+x^50+x^48+x^47+x^45+x^41+x^40+x^37+x^36+x^34+x^32+x^31+x^27+x^25+x^24+x^22+x^19+x^18+x^16+x^15+x^14+x^8+x^6+x^5+x^4+x+1", s)
239 +}
240 +
241 +var polIrredTests = []struct {
242 + f chunker.Pol
243 + irred bool
244 +}{
245 + {0x38f1e565e288df, false},
246 + {0x3DA3358B4DC173, true},
247 + {0x30a8295b9d5c91, false},
248 + {0x255f4350b962cb, false},
249 + {0x267f776110a235, false},
250 + {0x2f4dae10d41227, false},
251 + {0x2482734cacca49, true},
252 + {0x312daf4b284899, false},
253 + {0x29dfb6553d01d1, false},
254 + {0x3548245eb26257, false},
255 + {0x3199e7ef4211b3, false},
256 + {0x362f39017dae8b, false},
257 + {0x200d57aa6fdacb, false},
258 + {0x35e0a4efa1d275, false},
259 + {0x2ced55b026577f, false},
260 + {0x260b012010893d, false},
261 + {0x2df29cbcd59e9d, false},
262 + {0x3f2ac7488bd429, false},
263 + {0x3e5cb1711669fb, false},
264 + {0x226d8de57a9959, false},
265 + {0x3c8de80aaf5835, false},
266 + {0x2026a59efb219b, false},
267 + {0x39dfa4d13fb231, false},
268 + {0x3143d0464b3299, false},
269 +}
270 +
271 +func TestPolIrreducible(t *testing.T) {
272 + for _, test := range polIrredTests {
273 + Assert(t, test.f.Irreducible() == test.irred,
274 + "Irreducibility test for Polynomial %v failed: got %v, wanted %v",
275 + test.f, test.f.Irreducible(), test.irred)
276 + }
277 +}
278 +
279 +func BenchmarkPolIrreducible(b *testing.B) {
280 + // find first irreducible polynomial
281 + var pol chunker.Pol
282 + for _, test := range polIrredTests {
283 + if test.irred {
284 + pol = test.f
285 + break
286 + }
287 + }
288 +
289 + for i := 0; i < b.N; i++ {
290 + Assert(b, pol.Irreducible(),
291 + "Irreducibility test for Polynomial %v failed", pol)
292 + }
293 +}
294 +
295 +var polGCDTests = []struct {
296 + f1 chunker.Pol
297 + f2 chunker.Pol
298 + gcd chunker.Pol
299 +}{
300 + {10, 50, 2},
301 + {0, 1, 1},
302 + {
303 + parseBin("101101001"),
304 + parseBin("1010"),
305 + parseBin("1"),
306 + },
307 + {2, 2, 2},
308 + {
309 + parseBin("1010"),
310 + parseBin("11"),
311 + parseBin("11"),
312 + },
313 + {
314 + 0x8000000000000000,
315 + 0x8000000000000000,
316 + 0x8000000000000000,
317 + },
318 + {
319 + parseBin("1100"),
320 + parseBin("101"),
321 + parseBin("11"),
322 + },
323 + {
324 + parseBin("1100001111"),
325 + parseBin("10011"),
326 + parseBin("10011"),
327 + },
328 + {
329 + 0x3DA3358B4DC173,
330 + 0x3DA3358B4DC173,
331 + 0x3DA3358B4DC173,
332 + },
333 + {
334 + 0x3DA3358B4DC173,
335 + 0x230d2259defd,
336 + 1,
337 + },
338 + {
339 + 0x230d2259defd,
340 + 0x51b492b3eff2,
341 + parseBin("10011"),
342 + },
343 +}
344 +
345 +func TestPolGCD(t *testing.T) {
346 + for i, test := range polGCDTests {
347 + gcd := test.f1.GCD(test.f2)
348 + Assert(t, test.gcd == gcd,
349 + "GCD test %d (%+v) failed: got %v, wanted %v",
350 + i, test, gcd, test.gcd)
351 + gcd = test.f2.GCD(test.f1)
352 + Assert(t, test.gcd == gcd,
353 + "GCD test %d (%+v) failed: got %v, wanted %v",
354 + i, test, gcd, test.gcd)
355 + }
356 +}
357 +
358 +var polMulModTests = []struct {
359 + f1 chunker.Pol
360 + f2 chunker.Pol
361 + g chunker.Pol
362 + mod chunker.Pol
363 +}{
364 + {
365 + 0x1230,
366 + 0x230,
367 + 0x55,
368 + 0x22,
369 + },
370 + {
371 + 0x0eae8c07dbbb3026,
372 + 0xd5d6db9de04771de,
373 + 0xdd2bda3b77c9,
374 + 0x425ae8595b7a,
375 + },
376 +}
377 +
378 +func TestPolMulMod(t *testing.T) {
379 + for i, test := range polMulModTests {
380 + mod := test.f1.MulMod(test.f2, test.g)
381 + Assert(t, mod == test.mod,
382 + "MulMod test %d (%+v) failed: got %v, wanted %v",
383 + i, test, mod, test.mod)
384 + }
385 +}
core/commands/add.go
+14 -7
@@ -31,6 +31,7 @@ const (
31 wrapOptionName = "wrap-with-directory"
32 hiddenOptionName = "hidden"
33 onlyHashOptionName = "only-hash"
34 + chunkerOptionName = "chunker"
35 )
36
37 type AddedObject struct {
@@ -61,6 +62,7 @@ remains to be implemented.
62 cmds.BoolOption(onlyHashOptionName, "n", "Only chunk and hash - do not write to disk"),
63 cmds.BoolOption(wrapOptionName, "w", "Wrap files with a directory object"),
64 cmds.BoolOption(hiddenOptionName, "Include files that are hidden"),
65 + cmds.StringOption(chunkerOptionName, "s", "chunking algorithm to use"),
66 },
67 PreRun: func(req cmds.Request) error {
68 if quiet, _, _ := req.Option(quietOptionName).Bool(); quiet {
@@ -97,6 +99,7 @@ remains to be implemented.
99 wrap, _, _ := req.Option(wrapOptionName).Bool()
100 hash, _, _ := req.Option(onlyHashOptionName).Bool()
101 hidden, _, _ := req.Option(hiddenOptionName).Bool()
102 + chunker, _, _ := req.Option(chunkerOptionName).String()
103
104 if hash {
105 nilnode, err := core.NewNodeBuilder().NilRepo().Build(n.Context())
@@ -118,6 +121,7 @@ remains to be implemented.
121 progress: progress,
122 hidden: hidden,
123 trickle: trickle,
124 + chunker: chunker,
125 }
126
127 rootnd, err := addParams.addFile(file)
@@ -265,24 +269,27 @@ type adder struct {
269 progress bool
270 hidden bool
271 trickle bool
272 + chunker string
273 }
274
275 // Perform the actual add & pin locally, outputting results to reader
271 -func add(n *core.IpfsNode, reader io.Reader, useTrickle bool) (*dag.Node, error) {
276 +func add(n *core.IpfsNode, reader io.Reader, useTrickle bool, chunker string) (*dag.Node, error) {
277 + chnk, err := chunk.FromString(reader, chunker)
278 + if err != nil {
279 + return nil, err
280 + }
281 +
282 var node *dag.Node
273 - var err error
283 if useTrickle {
284 node, err = importer.BuildTrickleDagFromReader(
276 - reader,
285 n.DAG,
278 - chunk.DefaultSplitter,
286 + chnk,
287 importer.PinIndirectCB(n.Pinning.GetManual()),
288 )
289 } else {
290 node, err = importer.BuildDagFromReader(
283 - reader,
291 n.DAG,
285 - chunk.DefaultSplitter,
292 + chnk,
293 importer.PinIndirectCB(n.Pinning.GetManual()),
294 )
295 }
@@ -314,7 +321,7 @@ func (params *adder) addFile(file files.File) (*dag.Node, error) {
321 reader = &progressReader{file: file, out: params.out}
322 }
323
317 - dagnode, err := add(params.node, reader, params.trickle)
324 + dagnode, err := add(params.node, reader, params.trickle, params.chunker)
325 if err != nil {
326 return nil, err
327 }
core/corehttp/gateway_handler.go
+3 -1
@@ -46,7 +46,9 @@ func (i *gatewayHandler) newDagFromReader(r io.Reader) (*dag.Node, error) {
46 // TODO(cryptix): change and remove this helper once PR1136 is merged
47 // return ufs.AddFromReader(i.node, r.Body)
48 return importer.BuildDagFromReader(
49 - r, i.node.DAG, chunk.DefaultSplitter, importer.BasicPinnerCB(i.node.Pinning.GetManual()))
49 + i.node.DAG,
50 + chunk.DefaultSplitter(r),
51 + importer.BasicPinnerCB(i.node.Pinning.GetManual()))
52 }
53
54 // TODO(btc): break this apart into separate handlers using a more expressive muxer
core/coreunix/add.go
+3 -4
@@ -25,10 +25,10 @@ var log = eventlog.Logger("coreunix")
25 // datastore. Returns a key representing the root node.
26 func Add(n *core.IpfsNode, r io.Reader) (string, error) {
27 // TODO more attractive function signature importer.BuildDagFromReader
28 +
29 dagNode, err := importer.BuildDagFromReader(
29 - r,
30 n.DAG,
31 - chunk.DefaultSplitter,
31 + chunk.NewSizeSplitter(r, chunk.DefaultBlockSize),
32 importer.BasicPinnerCB(n.Pinning.GetManual()),
33 )
34 if err != nil {
@@ -96,9 +96,8 @@ func add(n *core.IpfsNode, reader io.Reader) (*merkledag.Node, error) {
96 mp := n.Pinning.GetManual()
97
98 node, err := importer.BuildDagFromReader(
99 - reader,
99 n.DAG,
101 - chunk.DefaultSplitter,
100 + chunk.DefaultSplitter(reader),
101 importer.PinIndirectCB(mp),
102 )
103 if err != nil {
core/coreunix/metadata_test.go
+1 -1
@@ -38,7 +38,7 @@ func TestMetadata(t *testing.T) {
38 data := make([]byte, 1000)
39 u.NewTimeSeededRand().Read(data)
40 r := bytes.NewReader(data)
41 - nd, err := importer.BuildDagFromReader(r, ds, chunk.DefaultSplitter, nil)
41 + nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r), nil)
42 if err != nil {
43 t.Fatal(err)
44 }
fuse/readonly/ipfs_test.go
+1 -1
@@ -36,7 +36,7 @@ func randObj(t *testing.T, nd *core.IpfsNode, size int64) (*dag.Node, []byte) {
36 buf := make([]byte, size)
37 u.NewTimeSeededRand().Read(buf)
38 read := bytes.NewReader(buf)
39 - obj, err := importer.BuildTrickleDagFromReader(read, nd.DAG, chunk.DefaultSplitter, nil)
39 + obj, err := importer.BuildTrickleDagFromReader(nd.DAG, chunk.DefaultSplitter(read), nil)
40 if err != nil {
41 t.Fatal(err)
42 }
importer/balanced/balanced_test.go
+35 -112
@@ -12,23 +12,38 @@ import (
12 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
13 chunk "github.com/ipfs/go-ipfs/importer/chunk"
14 h "github.com/ipfs/go-ipfs/importer/helpers"
15 - merkledag "github.com/ipfs/go-ipfs/merkledag"
15 + dag "github.com/ipfs/go-ipfs/merkledag"
16 mdtest "github.com/ipfs/go-ipfs/merkledag/test"
17 pin "github.com/ipfs/go-ipfs/pin"
18 uio "github.com/ipfs/go-ipfs/unixfs/io"
19 u "github.com/ipfs/go-ipfs/util"
20 )
21
22 -func buildTestDag(r io.Reader, ds merkledag.DAGService, spl chunk.BlockSplitter) (*merkledag.Node, error) {
22 +// TODO: extract these tests and more as a generic layout test suite
23 +
24 +func buildTestDag(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) {
25 // Start the splitter
24 - blkch := spl.Split(r)
26 + blkch, errs := chunk.Chan(spl)
27
28 dbp := h.DagBuilderParams{
29 Dagserv: ds,
30 Maxlinks: h.DefaultLinksPerBlock,
31 }
32
31 - return BalancedLayout(dbp.New(blkch))
33 + return BalancedLayout(dbp.New(blkch, errs))
34 +}
35 +
36 +func getTestDag(t *testing.T, ds dag.DAGService, size int64, blksize int64) (*dag.Node, []byte) {
37 + data := make([]byte, size)
38 + u.NewTimeSeededRand().Read(data)
39 + r := bytes.NewReader(data)
40 +
41 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(r, blksize))
42 + if err != nil {
43 + t.Fatal(err)
44 + }
45 +
46 + return nd, data
47 }
48
49 //Test where calls to read are smaller than the chunk size
@@ -36,9 +51,10 @@ func TestSizeBasedSplit(t *testing.T) {
51 if testing.Short() {
52 t.SkipNow()
53 }
39 - bs := &chunk.SizeSplitter{Size: 512}
54 +
55 + bs := chunk.SizeSplitterGen(512)
56 testFileConsistency(t, bs, 32*512)
41 - bs = &chunk.SizeSplitter{Size: 4096}
57 + bs = chunk.SizeSplitterGen(4096)
58 testFileConsistency(t, bs, 32*4096)
59
60 // Uneven offset
@@ -51,13 +67,13 @@ func dup(b []byte) []byte {
67 return o
68 }
69
54 -func testFileConsistency(t *testing.T, bs chunk.BlockSplitter, nbytes int) {
70 +func testFileConsistency(t *testing.T, bs chunk.SplitterGen, nbytes int) {
71 should := make([]byte, nbytes)
72 u.NewTimeSeededRand().Read(should)
73
74 read := bytes.NewReader(should)
75 ds := mdtest.Mock(t)
60 - nd, err := buildTestDag(read, ds, bs)
76 + nd, err := buildTestDag(ds, bs(read))
77 if err != nil {
78 t.Fatal(err)
79 }
@@ -79,15 +95,9 @@ func testFileConsistency(t *testing.T, bs chunk.BlockSplitter, nbytes int) {
95 }
96
97 func TestBuilderConsistency(t *testing.T) {
82 - nbytes := 100000
83 - buf := new(bytes.Buffer)
84 - io.CopyN(buf, u.NewTimeSeededRand(), int64(nbytes))
85 - should := dup(buf.Bytes())
98 dagserv := mdtest.Mock(t)
87 - nd, err := buildTestDag(buf, dagserv, chunk.DefaultSplitter)
88 - if err != nil {
89 - t.Fatal(err)
90 - }
99 + nd, should := getTestDag(t, dagserv, 100000, chunk.DefaultBlockSize)
100 +
101 r, err := uio.NewDagReader(context.Background(), nd, dagserv)
102 if err != nil {
103 t.Fatal(err)
@@ -116,50 +126,14 @@ func arrComp(a, b []byte) error {
126 return nil
127 }
128
119 -func TestMaybeRabinConsistency(t *testing.T) {
120 - if testing.Short() {
121 - t.SkipNow()
122 - }
123 - testFileConsistency(t, chunk.NewMaybeRabin(4096), 256*4096)
124 -}
125 -
126 -func TestRabinBlockSize(t *testing.T) {
127 - if testing.Short() {
128 - t.SkipNow()
129 - }
130 - buf := new(bytes.Buffer)
131 - nbytes := 1024 * 1024
132 - io.CopyN(buf, u.NewTimeSeededRand(), int64(nbytes))
133 - rab := chunk.NewMaybeRabin(4096)
134 - blkch := rab.Split(buf)
135 -
136 - var blocks [][]byte
137 - for b := range blkch {
138 - blocks = append(blocks, b)
139 - }
140 -
141 - fmt.Printf("Avg block size: %d\n", nbytes/len(blocks))
142 -
143 -}
144 -
129 type dagservAndPinner struct {
146 - ds merkledag.DAGService
130 + ds dag.DAGService
131 mp pin.ManualPinner
132 }
133
134 func TestIndirectBlocks(t *testing.T) {
151 - splitter := &chunk.SizeSplitter{512}
152 - nbytes := 1024 * 1024
153 - buf := make([]byte, nbytes)
154 - u.NewTimeSeededRand().Read(buf)
155 -
156 - read := bytes.NewReader(buf)
157 -
135 ds := mdtest.Mock(t)
159 - dag, err := buildTestDag(read, ds, splitter)
160 - if err != nil {
161 - t.Fatal(err)
162 - }
136 + dag, buf := getTestDag(t, ds, 1024*1024, 512)
137
138 reader, err := uio.NewDagReader(context.Background(), dag, ds)
139 if err != nil {
@@ -178,15 +152,8 @@ func TestIndirectBlocks(t *testing.T) {
152
153 func TestSeekingBasic(t *testing.T) {
154 nbytes := int64(10 * 1024)
181 - should := make([]byte, nbytes)
182 - u.NewTimeSeededRand().Read(should)
183 -
184 - read := bytes.NewReader(should)
155 ds := mdtest.Mock(t)
186 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
187 - if err != nil {
188 - t.Fatal(err)
189 - }
156 + nd, should := getTestDag(t, ds, nbytes, 500)
157
158 rs, err := uio.NewDagReader(context.Background(), nd, ds)
159 if err != nil {
@@ -214,16 +181,8 @@ func TestSeekingBasic(t *testing.T) {
181 }
182
183 func TestSeekToBegin(t *testing.T) {
217 - nbytes := int64(10 * 1024)
218 - should := make([]byte, nbytes)
219 - u.NewTimeSeededRand().Read(should)
220 -
221 - read := bytes.NewReader(should)
184 ds := mdtest.Mock(t)
223 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
224 - if err != nil {
225 - t.Fatal(err)
226 - }
185 + nd, should := getTestDag(t, ds, 10*1024, 500)
186
187 rs, err := uio.NewDagReader(context.Background(), nd, ds)
188 if err != nil {
@@ -258,16 +217,8 @@ func TestSeekToBegin(t *testing.T) {
217 }
218
219 func TestSeekToAlmostBegin(t *testing.T) {
261 - nbytes := int64(10 * 1024)
262 - should := make([]byte, nbytes)
263 - u.NewTimeSeededRand().Read(should)
264 -
265 - read := bytes.NewReader(should)
220 ds := mdtest.Mock(t)
267 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
268 - if err != nil {
269 - t.Fatal(err)
270 - }
221 + nd, should := getTestDag(t, ds, 10*1024, 500)
222
223 rs, err := uio.NewDagReader(context.Background(), nd, ds)
224 if err != nil {
@@ -303,15 +254,8 @@ func TestSeekToAlmostBegin(t *testing.T) {
254
255 func TestSeekEnd(t *testing.T) {
256 nbytes := int64(50 * 1024)
306 - should := make([]byte, nbytes)
307 - u.NewTimeSeededRand().Read(should)
308 -
309 - read := bytes.NewReader(should)
257 ds := mdtest.Mock(t)
311 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
312 - if err != nil {
313 - t.Fatal(err)
314 - }
258 + nd, _ := getTestDag(t, ds, nbytes, 500)
259
260 rs, err := uio.NewDagReader(context.Background(), nd, ds)
261 if err != nil {
@@ -329,15 +273,8 @@ func TestSeekEnd(t *testing.T) {
273
274 func TestSeekEndSingleBlockFile(t *testing.T) {
275 nbytes := int64(100)
332 - should := make([]byte, nbytes)
333 - u.NewTimeSeededRand().Read(should)
334 -
335 - read := bytes.NewReader(should)
276 ds := mdtest.Mock(t)
337 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{5000})
338 - if err != nil {
339 - t.Fatal(err)
340 - }
277 + nd, _ := getTestDag(t, ds, nbytes, 5000)
278
279 rs, err := uio.NewDagReader(context.Background(), nd, ds)
280 if err != nil {
@@ -355,15 +292,8 @@ func TestSeekEndSingleBlockFile(t *testing.T) {
292
293 func TestSeekingStress(t *testing.T) {
294 nbytes := int64(1024 * 1024)
358 - should := make([]byte, nbytes)
359 - u.NewTimeSeededRand().Read(should)
360 -
361 - read := bytes.NewReader(should)
295 ds := mdtest.Mock(t)
363 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{1000})
364 - if err != nil {
365 - t.Fatal(err)
366 - }
296 + nd, should := getTestDag(t, ds, nbytes, 1000)
297
298 rs, err := uio.NewDagReader(context.Background(), nd, ds)
299 if err != nil {
@@ -400,15 +330,8 @@ func TestSeekingStress(t *testing.T) {
330
331 func TestSeekingConsistency(t *testing.T) {
332 nbytes := int64(128 * 1024)
403 - should := make([]byte, nbytes)
404 - u.NewTimeSeededRand().Read(should)
405 -
406 - read := bytes.NewReader(should)
333 ds := mdtest.Mock(t)
408 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
409 - if err != nil {
410 - t.Fatal(err)
411 - }
334 + nd, should := getTestDag(t, ds, nbytes, 500)
335
336 rs, err := uio.NewDagReader(context.Background(), nd, ds)
337 if err != nil {
importer/chunk/parse.go new
+76
@@ -0,0 +1,76 @@
1 +package chunk
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "io"
7 + "strconv"
8 + "strings"
9 +)
10 +
11 +func FromString(r io.Reader, chunker string) (Splitter, error) {
12 + switch {
13 + case chunker == "" || chunker == "default":
14 + return NewSizeSplitter(r, DefaultBlockSize), nil
15 +
16 + case strings.HasPrefix(chunker, "size-"):
17 + sizeStr := strings.Split(chunker, "-")[1]
18 + size, err := strconv.Atoi(sizeStr)
19 + if err != nil {
20 + return nil, err
21 + }
22 + return NewSizeSplitter(r, int64(size)), nil
23 +
24 + case strings.HasPrefix(chunker, "rabin"):
25 + return parseRabinString(r, chunker)
26 +
27 + default:
28 + return nil, fmt.Errorf("unrecognized chunker option: %s", chunker)
29 + }
30 +}
31 +
32 +func parseRabinString(r io.Reader, chunker string) (Splitter, error) {
33 + parts := strings.Split(chunker, "-")
34 + switch len(parts) {
35 + case 1:
36 + return NewRabin(r, uint64(DefaultBlockSize)), nil
37 + case 2:
38 + size, err := strconv.Atoi(parts[1])
39 + if err != nil {
40 + return nil, err
41 + }
42 + return NewRabin(r, uint64(size)), nil
43 + case 4:
44 + sub := strings.Split(parts[1], ":")
45 + if len(sub) > 1 && sub[0] != "min" {
46 + return nil, errors.New("first label must be min")
47 + }
48 + min, err := strconv.Atoi(sub[len(sub)-1])
49 + if err != nil {
50 + return nil, err
51 + }
52 +
53 + sub = strings.Split(parts[2], ":")
54 + if len(sub) > 1 && sub[0] != "avg" {
55 + log.Error("sub == ", sub)
56 + return nil, errors.New("second label must be avg")
57 + }
58 + avg, err := strconv.Atoi(sub[len(sub)-1])
59 + if err != nil {
60 + return nil, err
61 + }
62 +
63 + sub = strings.Split(parts[3], ":")
64 + if len(sub) > 1 && sub[0] != "max" {
65 + return nil, errors.New("final label must be max")
66 + }
67 + max, err := strconv.Atoi(sub[len(sub)-1])
68 + if err != nil {
69 + return nil, err
70 + }
71 +
72 + return NewRabinMinMax(r, uint64(min), uint64(avg), uint64(max)), nil
73 + default:
74 + return nil, errors.New("incorrect format (expected 'rabin' 'rabin-[avg]' or 'rabin-[min]-[avg]-[max]'")
75 + }
76 +}
importer/chunk/rabin.go
+24 -79
@@ -1,94 +1,39 @@
1 package chunk
2
3 import (
4 - "bufio"
5 - "bytes"
6 - "fmt"
4 + "hash/fnv"
5 "io"
8 - "math"
6 +
7 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/whyrusleeping/chunker"
8 )
9
11 -type MaybeRabin struct {
12 - mask int
13 - windowSize int
14 - MinBlockSize int
15 - MaxBlockSize int
16 -}
10 +var IpfsRabinPoly = chunker.Pol(17437180132763653)
11
18 -func NewMaybeRabin(avgBlkSize int) *MaybeRabin {
19 - blkbits := uint(math.Log2(float64(avgBlkSize)))
20 - rb := new(MaybeRabin)
21 - rb.mask = (1 << blkbits) - 1
22 - rb.windowSize = 16 // probably a good number...
23 - rb.MinBlockSize = avgBlkSize / 2
24 - rb.MaxBlockSize = (avgBlkSize / 2) * 3
25 - return rb
12 +type Rabin struct {
13 + r *chunker.Chunker
14 }
15
28 -func (mr *MaybeRabin) Split(r io.Reader) chan []byte {
29 - out := make(chan []byte, 16)
30 - go func() {
31 - inbuf := bufio.NewReader(r)
32 - blkbuf := new(bytes.Buffer)
33 -
34 - // some bullshit numbers i made up
35 - a := 10 // honestly, no idea what this is
36 - MOD := 33554383 // randomly chosen (seriously)
37 - an := 1
38 - rollingHash := 0
16 +func NewRabin(r io.Reader, avgBlkSize uint64) *Rabin {
17 + min := avgBlkSize / 3
18 + max := avgBlkSize + (avgBlkSize / 2)
19
40 - // Window is a circular buffer
41 - window := make([]byte, mr.windowSize)
42 - push := func(i int, val byte) (outval int) {
43 - outval = int(window[i%len(window)])
44 - window[i%len(window)] = val
45 - return
46 - }
20 + return NewRabinMinMax(r, avgBlkSize, min, max)
21 +}
22
48 - // Duplicate byte slice
49 - dup := func(b []byte) []byte {
50 - d := make([]byte, len(b))
51 - copy(d, b)
52 - return d
53 - }
23 +func NewRabinMinMax(r io.Reader, min, avg, max uint64) *Rabin {
24 + h := fnv.New32a()
25 + ch := chunker.New(r, IpfsRabinPoly, h, avg, min, max)
26
55 - // Fill up the window
56 - i := 0
57 - for ; i < mr.windowSize; i++ {
58 - b, err := inbuf.ReadByte()
59 - if err != nil {
60 - fmt.Println(err)
61 - return
62 - }
63 - blkbuf.WriteByte(b)
64 - push(i, b)
65 - rollingHash = (rollingHash*a + int(b)) % MOD
66 - an = (an * a) % MOD
67 - }
27 + return &Rabin{
28 + r: ch,
29 + }
30 +}
31
69 - for ; true; i++ {
70 - b, err := inbuf.ReadByte()
71 - if err != nil {
72 - break
73 - }
74 - outval := push(i, b)
75 - blkbuf.WriteByte(b)
76 - rollingHash = (rollingHash*a + int(b) - an*outval) % MOD
77 - if (rollingHash&mr.mask == mr.mask && blkbuf.Len() > mr.MinBlockSize) ||
78 - blkbuf.Len() >= mr.MaxBlockSize {
79 - out <- dup(blkbuf.Bytes())
80 - blkbuf.Reset()
81 - }
32 +func (r *Rabin) NextBytes() ([]byte, error) {
33 + ch, err := r.r.Next()
34 + if err != nil {
35 + return nil, err
36 + }
37
83 - // Check if there are enough remaining
84 - peek, err := inbuf.Peek(mr.windowSize)
85 - if err != nil || len(peek) != mr.windowSize {
86 - break
87 - }
88 - }
89 - io.Copy(blkbuf, inbuf)
90 - out <- blkbuf.Bytes()
91 - close(out)
92 - }()
93 - return out
38 + return ch.Data, nil
39 }
importer/chunk/rabin_test.go new
+84
@@ -0,0 +1,84 @@
1 +package chunk
2 +
3 +import (
4 + "bytes"
5 + "fmt"
6 + "github.com/ipfs/go-ipfs/blocks"
7 + "github.com/ipfs/go-ipfs/blocks/key"
8 + "github.com/ipfs/go-ipfs/util"
9 + "io"
10 + "testing"
11 +)
12 +
13 +func TestRabinChunking(t *testing.T) {
14 + data := make([]byte, 1024*1024*16)
15 + util.NewTimeSeededRand().Read(data)
16 +
17 + r := NewRabin(bytes.NewReader(data), 1024*256)
18 +
19 + var chunks [][]byte
20 +
21 + for {
22 + chunk, err := r.NextBytes()
23 + if err != nil {
24 + if err == io.EOF {
25 + break
26 + }
27 + t.Fatal(err)
28 + }
29 +
30 + chunks = append(chunks, chunk)
31 + }
32 +
33 + fmt.Printf("average block size: %d\n", len(data)/len(chunks))
34 +
35 + unchunked := bytes.Join(chunks, nil)
36 + if !bytes.Equal(unchunked, data) {
37 + fmt.Printf("%d %d\n", len(unchunked), len(data))
38 + t.Fatal("data was chunked incorrectly")
39 + }
40 +}
41 +
42 +func chunkData(t *testing.T, data []byte) map[key.Key]*blocks.Block {
43 + r := NewRabin(bytes.NewReader(data), 1024*256)
44 +
45 + blkmap := make(map[key.Key]*blocks.Block)
46 +
47 + for {
48 + blk, err := r.NextBytes()
49 + if err != nil {
50 + if err == io.EOF {
51 + break
52 + }
53 + t.Fatal(err)
54 + }
55 +
56 + b := blocks.NewBlock(blk)
57 + blkmap[b.Key()] = b
58 + }
59 +
60 + return blkmap
61 +}
62 +
63 +func TestRabinChunkReuse(t *testing.T) {
64 + data := make([]byte, 1024*1024*16)
65 + util.NewTimeSeededRand().Read(data)
66 +
67 + ch1 := chunkData(t, data[1000:])
68 + ch2 := chunkData(t, data)
69 +
70 + var extra int
71 + for k, _ := range ch2 {
72 + _, ok := ch1[k]
73 + if !ok {
74 + extra++
75 + }
76 + }
77 +
78 + if extra > 2 {
79 + t.Fatal("too many spare chunks made")
80 + }
81 + if extra == 2 {
82 + t.Log("why did we get two extra blocks?")
83 + }
84 +}
importer/chunk/splitting.go
+51 -19
@@ -9,39 +9,71 @@ import (
9
10 var log = util.Logger("chunk")
11
12 -var DefaultBlockSize = 1024 * 256
13 -var DefaultSplitter = &SizeSplitter{Size: DefaultBlockSize}
12 +var DefaultBlockSize int64 = 1024 * 256
13
15 -type BlockSplitter interface {
16 - Split(r io.Reader) chan []byte
14 +type Splitter interface {
15 + NextBytes() ([]byte, error)
16 }
17
19 -type SizeSplitter struct {
20 - Size int
18 +type SplitterGen func(r io.Reader) Splitter
19 +
20 +func DefaultSplitter(r io.Reader) Splitter {
21 + return NewSizeSplitter(r, DefaultBlockSize)
22 +}
23 +
24 +func SizeSplitterGen(size int64) SplitterGen {
25 + return func(r io.Reader) Splitter {
26 + return NewSizeSplitter(r, size)
27 + }
28 }
29
23 -func (ss *SizeSplitter) Split(r io.Reader) chan []byte {
30 +func Chan(s Splitter) (<-chan []byte, <-chan error) {
31 out := make(chan []byte)
32 + errs := make(chan error, 1)
33 go func() {
34 defer close(out)
35 + defer close(errs)
36
37 // all-chunks loop (keep creating chunks)
38 for {
30 - // log.Infof("making chunk with size: %d", ss.Size)
31 - chunk := make([]byte, ss.Size)
32 - nread, err := io.ReadFull(r, chunk)
33 - if nread > 0 {
34 - // log.Infof("sending out chunk with size: %d", sofar)
35 - out <- chunk[:nread]
36 - }
37 - if err == io.EOF || err == io.ErrUnexpectedEOF {
38 - return
39 - }
39 + b, err := s.NextBytes()
40 if err != nil {
41 - log.Debugf("Block split error: %s", err)
41 + errs <- err
42 return
43 }
44 +
45 + out <- b
46 }
47 }()
46 - return out
48 + return out, errs
49 +}
50 +
51 +type sizeSplitterv2 struct {
52 + r io.Reader
53 + size int64
54 + err error
55 +}
56 +
57 +func NewSizeSplitter(r io.Reader, size int64) Splitter {
58 + return &sizeSplitterv2{
59 + r: r,
60 + size: size,
61 + }
62 +}
63 +
64 +func (ss *sizeSplitterv2) NextBytes() ([]byte, error) {
65 + if ss.err != nil {
66 + return nil, ss.err
67 + }
68 + buf := make([]byte, ss.size)
69 + n, err := io.ReadFull(ss.r, buf)
70 + if err == io.ErrUnexpectedEOF {
71 + ss.err = io.EOF
72 + err = nil
73 + }
74 + if err != nil {
75 + return nil, err
76 + }
77 +
78 + return buf[:n], nil
79 }
importer/chunk/splitting_test.go
+5 -5
@@ -32,8 +32,8 @@ func TestSizeSplitterIsDeterministic(t *testing.T) {
32 bufA := copyBuf(bufR)
33 bufB := copyBuf(bufR)
34
35 - chunksA := DefaultSplitter.Split(bytes.NewReader(bufA))
36 - chunksB := DefaultSplitter.Split(bytes.NewReader(bufB))
35 + chunksA, _ := Chan(DefaultSplitter(bytes.NewReader(bufA)))
36 + chunksB, _ := Chan(DefaultSplitter(bytes.NewReader(bufB)))
37
38 for n := 0; ; n++ {
39 a, moreA := <-chunksA
@@ -65,8 +65,8 @@ func TestSizeSplitterFillsChunks(t *testing.T) {
65 max := 10000000
66 b := randBuf(t, max)
67 r := &clipReader{r: bytes.NewReader(b), size: 4000}
68 - s := SizeSplitter{Size: 1024 * 256}
69 - c := s.Split(r)
68 + chunksize := int64(1024 * 256)
69 + c, _ := Chan(NewSizeSplitter(r, chunksize))
70
71 sofar := 0
72 whole := make([]byte, max)
@@ -80,7 +80,7 @@ func TestSizeSplitterFillsChunks(t *testing.T) {
80 copy(whole[sofar:], chunk)
81
82 sofar += len(chunk)
83 - if sofar != max && len(chunk) < s.Size {
83 + if sofar != max && len(chunk) < int(chunksize) {
84 t.Fatal("sizesplitter split at a smaller size")
85 }
86 }
importer/helpers/dagbuilder.go
+4 -1
@@ -19,6 +19,8 @@ type DagBuilderHelper struct {
19 dserv dag.DAGService
20 mp pin.ManualPinner
21 in <-chan []byte
22 + errs <-chan error
23 + recvdErr error
24 nextData []byte // the next item to return.
25 maxlinks int
26 ncb NodeCB
@@ -39,7 +41,7 @@ type DagBuilderParams struct {
41
42 // Generate a new DagBuilderHelper from the given params, using 'in' as a
43 // data source
42 -func (dbp *DagBuilderParams) New(in <-chan []byte) *DagBuilderHelper {
44 +func (dbp *DagBuilderParams) New(in <-chan []byte, errs <-chan error) *DagBuilderHelper {
45 ncb := dbp.NodeCB
46 if ncb == nil {
47 ncb = nilFunc
@@ -48,6 +50,7 @@ func (dbp *DagBuilderParams) New(in <-chan []byte) *DagBuilderHelper {
50 return &DagBuilderHelper{
51 dserv: dbp.Dagserv,
52 in: in,
53 + errs: errs,
54 maxlinks: dbp.Maxlinks,
55 ncb: ncb,
56 batch: dbp.Dagserv.Batch(),
importer/importer.go
+7 -8
@@ -4,7 +4,6 @@ package importer
4
5 import (
6 "fmt"
7 - "io"
7 "os"
8
9 bal "github.com/ipfs/go-ipfs/importer/balanced"
@@ -36,12 +35,12 @@ func BuildDagFromFile(fpath string, ds dag.DAGService, mp pin.ManualPinner) (*da
35 }
36 defer f.Close()
37
39 - return BuildDagFromReader(f, ds, chunk.DefaultSplitter, BasicPinnerCB(mp))
38 + return BuildDagFromReader(ds, chunk.NewSizeSplitter(f, chunk.DefaultBlockSize), BasicPinnerCB(mp))
39 }
40
42 -func BuildDagFromReader(r io.Reader, ds dag.DAGService, spl chunk.BlockSplitter, ncb h.NodeCB) (*dag.Node, error) {
41 +func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter, ncb h.NodeCB) (*dag.Node, error) {
42 // Start the splitter
44 - blkch := spl.Split(r)
43 + blkch, errch := chunk.Chan(spl)
44
45 dbp := h.DagBuilderParams{
46 Dagserv: ds,
@@ -49,12 +48,12 @@ func BuildDagFromReader(r io.Reader, ds dag.DAGService, spl chunk.BlockSplitter,
48 NodeCB: ncb,
49 }
50
52 - return bal.BalancedLayout(dbp.New(blkch))
51 + return bal.BalancedLayout(dbp.New(blkch, errch))
52 }
53
55 -func BuildTrickleDagFromReader(r io.Reader, ds dag.DAGService, spl chunk.BlockSplitter, ncb h.NodeCB) (*dag.Node, error) {
54 +func BuildTrickleDagFromReader(ds dag.DAGService, spl chunk.Splitter, ncb h.NodeCB) (*dag.Node, error) {
55 // Start the splitter
57 - blkch := spl.Split(r)
56 + blkch, errch := chunk.Chan(spl)
57
58 dbp := h.DagBuilderParams{
59 Dagserv: ds,
@@ -62,7 +61,7 @@ func BuildTrickleDagFromReader(r io.Reader, ds dag.DAGService, spl chunk.BlockSp
61 NodeCB: ncb,
62 }
63
65 - return trickle.TrickleLayout(dbp.New(blkch))
64 + return trickle.TrickleLayout(dbp.New(blkch, errch))
65 }
66
67 func BasicPinnerCB(p pin.ManualPinner) h.NodeCB {
importer/importer_test.go
+5 -5
@@ -14,20 +14,20 @@ import (
14 u "github.com/ipfs/go-ipfs/util"
15 )
16
17 -func getBalancedDag(t testing.TB, size int64, blksize int) (*dag.Node, dag.DAGService) {
17 +func getBalancedDag(t testing.TB, size int64, blksize int64) (*dag.Node, dag.DAGService) {
18 ds := mdtest.Mock(t)
19 r := io.LimitReader(u.NewTimeSeededRand(), size)
20 - nd, err := BuildDagFromReader(r, ds, &chunk.SizeSplitter{blksize}, nil)
20 + nd, err := BuildDagFromReader(ds, chunk.NewSizeSplitter(r, blksize), nil)
21 if err != nil {
22 t.Fatal(err)
23 }
24 return nd, ds
25 }
26
27 -func getTrickleDag(t testing.TB, size int64, blksize int) (*dag.Node, dag.DAGService) {
27 +func getTrickleDag(t testing.TB, size int64, blksize int64) (*dag.Node, dag.DAGService) {
28 ds := mdtest.Mock(t)
29 r := io.LimitReader(u.NewTimeSeededRand(), size)
30 - nd, err := BuildTrickleDagFromReader(r, ds, &chunk.SizeSplitter{blksize}, nil)
30 + nd, err := BuildTrickleDagFromReader(ds, chunk.NewSizeSplitter(r, blksize), nil)
31 if err != nil {
32 t.Fatal(err)
33 }
@@ -40,7 +40,7 @@ func TestBalancedDag(t *testing.T) {
40 u.NewTimeSeededRand().Read(buf)
41 r := bytes.NewReader(buf)
42
43 - nd, err := BuildDagFromReader(r, ds, chunk.DefaultSplitter, nil)
43 + nd, err := BuildDagFromReader(ds, chunk.DefaultSplitter(r), nil)
44 if err != nil {
45 t.Fatal(err)
46 }
importer/trickle/trickle_test.go
+31 -56
@@ -20,16 +20,16 @@ import (
20 u "github.com/ipfs/go-ipfs/util"
21 )
22
23 -func buildTestDag(r io.Reader, ds merkledag.DAGService, spl chunk.BlockSplitter) (*merkledag.Node, error) {
23 +func buildTestDag(ds merkledag.DAGService, spl chunk.Splitter) (*merkledag.Node, error) {
24 // Start the splitter
25 - blkch := spl.Split(r)
25 + blkch, errs := chunk.Chan(spl)
26
27 dbp := h.DagBuilderParams{
28 Dagserv: ds,
29 Maxlinks: h.DefaultLinksPerBlock,
30 }
31
32 - nd, err := TrickleLayout(dbp.New(blkch))
32 + nd, err := TrickleLayout(dbp.New(blkch, errs))
33 if err != nil {
34 return nil, err
35 }
@@ -42,9 +42,10 @@ func TestSizeBasedSplit(t *testing.T) {
42 if testing.Short() {
43 t.SkipNow()
44 }
45 - bs := &chunk.SizeSplitter{Size: 512}
45 + bs := chunk.SizeSplitterGen(512)
46 testFileConsistency(t, bs, 32*512)
47 - bs = &chunk.SizeSplitter{Size: 4096}
47 +
48 + bs = chunk.SizeSplitterGen(4096)
49 testFileConsistency(t, bs, 32*4096)
50
51 // Uneven offset
@@ -57,13 +58,13 @@ func dup(b []byte) []byte {
58 return o
59 }
60
60 -func testFileConsistency(t *testing.T, bs chunk.BlockSplitter, nbytes int) {
61 +func testFileConsistency(t *testing.T, bs chunk.SplitterGen, nbytes int) {
62 should := make([]byte, nbytes)
63 u.NewTimeSeededRand().Read(should)
64
65 read := bytes.NewReader(should)
66 ds := mdtest.Mock(t)
66 - nd, err := buildTestDag(read, ds, bs)
67 + nd, err := buildTestDag(ds, bs(read))
68 if err != nil {
69 t.Fatal(err)
70 }
@@ -90,7 +91,7 @@ func TestBuilderConsistency(t *testing.T) {
91 io.CopyN(buf, u.NewTimeSeededRand(), int64(nbytes))
92 should := dup(buf.Bytes())
93 dagserv := mdtest.Mock(t)
93 - nd, err := buildTestDag(buf, dagserv, chunk.DefaultSplitter)
94 + nd, err := buildTestDag(dagserv, chunk.DefaultSplitter(buf))
95 if err != nil {
96 t.Fatal(err)
97 }
@@ -122,39 +123,13 @@ func arrComp(a, b []byte) error {
123 return nil
124 }
125
125 -func TestMaybeRabinConsistency(t *testing.T) {
126 - if testing.Short() {
127 - t.SkipNow()
128 - }
129 - testFileConsistency(t, chunk.NewMaybeRabin(4096), 256*4096)
130 -}
131 -
132 -func TestRabinBlockSize(t *testing.T) {
133 - if testing.Short() {
134 - t.SkipNow()
135 - }
136 - buf := new(bytes.Buffer)
137 - nbytes := 1024 * 1024
138 - io.CopyN(buf, u.NewTimeSeededRand(), int64(nbytes))
139 - rab := chunk.NewMaybeRabin(4096)
140 - blkch := rab.Split(buf)
141 -
142 - var blocks [][]byte
143 - for b := range blkch {
144 - blocks = append(blocks, b)
145 - }
146 -
147 - fmt.Printf("Avg block size: %d\n", nbytes/len(blocks))
148 -
149 -}
150 -
126 type dagservAndPinner struct {
127 ds merkledag.DAGService
128 mp pin.ManualPinner
129 }
130
131 func TestIndirectBlocks(t *testing.T) {
157 - splitter := &chunk.SizeSplitter{512}
132 + splitter := chunk.SizeSplitterGen(512)
133 nbytes := 1024 * 1024
134 buf := make([]byte, nbytes)
135 u.NewTimeSeededRand().Read(buf)
@@ -162,7 +137,7 @@ func TestIndirectBlocks(t *testing.T) {
137 read := bytes.NewReader(buf)
138
139 ds := mdtest.Mock(t)
165 - dag, err := buildTestDag(read, ds, splitter)
140 + dag, err := buildTestDag(ds, splitter(read))
141 if err != nil {
142 t.Fatal(err)
143 }
@@ -189,7 +164,7 @@ func TestSeekingBasic(t *testing.T) {
164
165 read := bytes.NewReader(should)
166 ds := mdtest.Mock(t)
192 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
167 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 512))
168 if err != nil {
169 t.Fatal(err)
170 }
@@ -226,7 +201,7 @@ func TestSeekToBegin(t *testing.T) {
201
202 read := bytes.NewReader(should)
203 ds := mdtest.Mock(t)
229 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
204 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
205 if err != nil {
206 t.Fatal(err)
207 }
@@ -270,7 +245,7 @@ func TestSeekToAlmostBegin(t *testing.T) {
245
246 read := bytes.NewReader(should)
247 ds := mdtest.Mock(t)
273 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
248 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
249 if err != nil {
250 t.Fatal(err)
251 }
@@ -314,7 +289,7 @@ func TestSeekEnd(t *testing.T) {
289
290 read := bytes.NewReader(should)
291 ds := mdtest.Mock(t)
317 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
292 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
293 if err != nil {
294 t.Fatal(err)
295 }
@@ -340,7 +315,7 @@ func TestSeekEndSingleBlockFile(t *testing.T) {
315
316 read := bytes.NewReader(should)
317 ds := mdtest.Mock(t)
343 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{5000})
318 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 5000))
319 if err != nil {
320 t.Fatal(err)
321 }
@@ -366,7 +341,7 @@ func TestSeekingStress(t *testing.T) {
341
342 read := bytes.NewReader(should)
343 ds := mdtest.Mock(t)
369 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{1000})
344 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 1000))
345 if err != nil {
346 t.Fatal(err)
347 }
@@ -411,7 +386,7 @@ func TestSeekingConsistency(t *testing.T) {
386
387 read := bytes.NewReader(should)
388 ds := mdtest.Mock(t)
414 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
389 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
390 if err != nil {
391 t.Fatal(err)
392 }
@@ -455,7 +430,7 @@ func TestAppend(t *testing.T) {
430 // Reader for half the bytes
431 read := bytes.NewReader(should[:nbytes/2])
432 ds := mdtest.Mock(t)
458 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
433 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
434 if err != nil {
435 t.Fatal(err)
436 }
@@ -465,10 +440,10 @@ func TestAppend(t *testing.T) {
440 Maxlinks: h.DefaultLinksPerBlock,
441 }
442
468 - spl := &chunk.SizeSplitter{500}
469 - blks := spl.Split(bytes.NewReader(should[nbytes/2:]))
443 + r := bytes.NewReader(should[nbytes/2:])
444 + blks, errs := chunk.Chan(chunk.NewSizeSplitter(r, 500))
445
471 - nnode, err := TrickleAppend(nd, dbp.New(blks))
446 + nnode, err := TrickleAppend(nd, dbp.New(blks, errs))
447 if err != nil {
448 t.Fatal(err)
449 }
@@ -504,7 +479,7 @@ func TestMultipleAppends(t *testing.T) {
479 u.NewTimeSeededRand().Read(should)
480
481 read := bytes.NewReader(nil)
507 - nd, err := buildTestDag(read, ds, &chunk.SizeSplitter{500})
482 + nd, err := buildTestDag(ds, chunk.NewSizeSplitter(read, 500))
483 if err != nil {
484 t.Fatal(err)
485 }
@@ -514,12 +489,12 @@ func TestMultipleAppends(t *testing.T) {
489 Maxlinks: 4,
490 }
491
517 - spl := &chunk.SizeSplitter{500}
492 + spl := chunk.SizeSplitterGen(500)
493
494 for i := 0; i < len(should); i++ {
520 - blks := spl.Split(bytes.NewReader(should[i : i+1]))
495 + blks, errs := chunk.Chan(spl(bytes.NewReader(should[i : i+1])))
496
522 - nnode, err := TrickleAppend(nd, dbp.New(blks))
497 + nnode, err := TrickleAppend(nd, dbp.New(blks, errs))
498 if err != nil {
499 t.Fatal(err)
500 }
@@ -559,18 +534,18 @@ func TestAppendSingleBytesToEmpty(t *testing.T) {
534 Maxlinks: 4,
535 }
536
562 - spl := &chunk.SizeSplitter{500}
537 + spl := chunk.SizeSplitterGen(500)
538
564 - blks := spl.Split(bytes.NewReader(data[:1]))
539 + blks, errs := chunk.Chan(spl(bytes.NewReader(data[:1])))
540
566 - nnode, err := TrickleAppend(nd, dbp.New(blks))
541 + nnode, err := TrickleAppend(nd, dbp.New(blks, errs))
542 if err != nil {
543 t.Fatal(err)
544 }
545
571 - blks = spl.Split(bytes.NewReader(data[1:]))
546 + blks, errs = chunk.Chan(spl(bytes.NewReader(data[1:])))
547
573 - nnode, err = TrickleAppend(nnode, dbp.New(blks))
548 + nnode, err = TrickleAppend(nnode, dbp.New(blks, errs))
549 if err != nil {
550 t.Fatal(err)
551 }
merkledag/merkledag_test.go
+2 -2
@@ -163,9 +163,9 @@ func runBatchFetchTest(t *testing.T, read io.Reader) {
163 dagservs = append(dagservs, NewDAGService(bsi))
164 }
165
166 - spl := &chunk.SizeSplitter{512}
166 + spl := chunk.NewSizeSplitter(read, 512)
167
168 - root, err := imp.BuildDagFromReader(read, dagservs[0], spl, nil)
168 + root, err := imp.BuildDagFromReader(dagservs[0], spl, nil)
169 if err != nil {
170 t.Fatal(err)
171 }
unixfs/mod/dagmodifier.go
+9 -9
@@ -40,7 +40,7 @@ type DagModifier struct {
40 curNode *mdag.Node
41 mp pin.ManualPinner
42
43 - splitter chunk.BlockSplitter
43 + splitter chunk.SplitterGen
44 ctx context.Context
45 readCancel func()
46
@@ -51,7 +51,7 @@ type DagModifier struct {
51 read *uio.DagReader
52 }
53
54 -func NewDagModifier(ctx context.Context, from *mdag.Node, serv mdag.DAGService, mp pin.ManualPinner, spl chunk.BlockSplitter) (*DagModifier, error) {
54 +func NewDagModifier(ctx context.Context, from *mdag.Node, serv mdag.DAGService, mp pin.ManualPinner, spl chunk.SplitterGen) (*DagModifier, error) {
55 return &DagModifier{
56 curNode: from.Copy(),
57 dagserv: serv,
@@ -106,10 +106,10 @@ func (zr zeroReader) Read(b []byte) (int, error) {
106 // expandSparse grows the file with zero blocks of 4096
107 // A small blocksize is chosen to aid in deduplication
108 func (dm *DagModifier) expandSparse(size int64) error {
109 - spl := chunk.SizeSplitter{4096}
109 r := io.LimitReader(zeroReader{}, size)
111 - blks := spl.Split(r)
112 - nnode, err := dm.appendData(dm.curNode, blks)
110 + spl := chunk.NewSizeSplitter(r, 4096)
111 + blks, errs := chunk.Chan(spl)
112 + nnode, err := dm.appendData(dm.curNode, blks, errs)
113 if err != nil {
114 return err
115 }
@@ -196,8 +196,8 @@ func (dm *DagModifier) Sync() error {
196
197 // need to write past end of current dag
198 if !done {
199 - blks := dm.splitter.Split(dm.wrBuf)
200 - nd, err = dm.appendData(dm.curNode, blks)
199 + blks, errs := chunk.Chan(dm.splitter(dm.wrBuf))
200 + nd, err = dm.appendData(dm.curNode, blks, errs)
201 if err != nil {
202 return err
203 }
@@ -306,14 +306,14 @@ func (dm *DagModifier) modifyDag(node *mdag.Node, offset uint64, data io.Reader)
306 }
307
308 // appendData appends the blocks from the given chan to the end of this dag
309 -func (dm *DagModifier) appendData(node *mdag.Node, blks <-chan []byte) (*mdag.Node, error) {
309 +func (dm *DagModifier) appendData(node *mdag.Node, blks <-chan []byte, errs <-chan error) (*mdag.Node, error) {
310 dbp := &help.DagBuilderParams{
311 Dagserv: dm.dagserv,
312 Maxlinks: help.DefaultLinksPerBlock,
313 NodeCB: imp.BasicPinnerCB(dm.mp),
314 }
315
316 - return trickle.TrickleAppend(node, dbp.New(blks))
316 + return trickle.TrickleAppend(node, dbp.New(blks, errs))
317 }
318
319 // Read data from this dag starting at the current offset
unixfs/mod/dagmodifier_test.go
+17 -11
@@ -53,7 +53,7 @@ func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.Blocksto
53
54 func getNode(t testing.TB, dserv mdag.DAGService, size int64, pinner pin.ManualPinner) ([]byte, *mdag.Node) {
55 in := io.LimitReader(u.NewTimeSeededRand(), size)
56 - node, err := imp.BuildTrickleDagFromReader(in, dserv, &chunk.SizeSplitter{500}, imp.BasicPinnerCB(pinner))
56 + node, err := imp.BuildTrickleDagFromReader(dserv, sizeSplitterGen(500)(in), imp.BasicPinnerCB(pinner))
57 if err != nil {
58 t.Fatal(err)
59 }
@@ -117,13 +117,19 @@ func testModWrite(t *testing.T, beg, size uint64, orig []byte, dm *DagModifier)
117 return orig
118 }
119
120 +func sizeSplitterGen(size int64) chunk.SplitterGen {
121 + return func(r io.Reader) chunk.Splitter {
122 + return chunk.NewSizeSplitter(r, size)
123 + }
124 +}
125 +
126 func TestDagModifierBasic(t *testing.T) {
127 dserv, pin := getMockDagServ(t)
128 b, n := getNode(t, dserv, 50000, pin)
129 ctx, cancel := context.WithCancel(context.Background())
130 defer cancel()
131
126 - dagmod, err := NewDagModifier(ctx, n, dserv, pin, &chunk.SizeSplitter{Size: 512})
132 + dagmod, err := NewDagModifier(ctx, n, dserv, pin, sizeSplitterGen(512))
133 if err != nil {
134 t.Fatal(err)
135 }
@@ -178,7 +184,7 @@ func TestMultiWrite(t *testing.T) {
184 ctx, cancel := context.WithCancel(context.Background())
185 defer cancel()
186
181 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
187 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
188 if err != nil {
189 t.Fatal(err)
190 }
@@ -231,7 +237,7 @@ func TestMultiWriteAndFlush(t *testing.T) {
237 ctx, cancel := context.WithCancel(context.Background())
238 defer cancel()
239
234 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
240 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
241 if err != nil {
242 t.Fatal(err)
243 }
@@ -279,7 +285,7 @@ func TestWriteNewFile(t *testing.T) {
285 ctx, cancel := context.WithCancel(context.Background())
286 defer cancel()
287
282 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
288 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
289 if err != nil {
290 t.Fatal(err)
291 }
@@ -322,7 +328,7 @@ func TestMultiWriteCoal(t *testing.T) {
328 ctx, cancel := context.WithCancel(context.Background())
329 defer cancel()
330
325 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
331 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
332 if err != nil {
333 t.Fatal(err)
334 }
@@ -368,7 +374,7 @@ func TestLargeWriteChunks(t *testing.T) {
374 ctx, cancel := context.WithCancel(context.Background())
375 defer cancel()
376
371 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
377 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
378 if err != nil {
379 t.Fatal(err)
380 }
@@ -406,7 +412,7 @@ func TestDagTruncate(t *testing.T) {
412 ctx, cancel := context.WithCancel(context.Background())
413 defer cancel()
414
409 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
415 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
416 if err != nil {
417 t.Fatal(err)
418 }
@@ -437,7 +443,7 @@ func TestSparseWrite(t *testing.T) {
443 ctx, cancel := context.WithCancel(context.Background())
444 defer cancel()
445
440 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
446 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
447 if err != nil {
448 t.Fatal(err)
449 }
@@ -491,7 +497,7 @@ func TestCorrectPinning(t *testing.T) {
497 ctx, cancel := context.WithCancel(context.Background())
498 defer cancel()
499
494 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
500 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
501 if err != nil {
502 t.Fatal(err)
503 }
@@ -598,7 +604,7 @@ func BenchmarkDagmodWrite(b *testing.B) {
604
605 wrsize := 4096
606
601 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, &chunk.SizeSplitter{Size: 512})
607 + dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
608 if err != nil {
609 b.Fatal(err)
610 }