-
Notifications
You must be signed in to change notification settings - Fork 102
/
Copy pathissues_test.go
191 lines (171 loc) · 4.74 KB
/
issues_test.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
package diskv
import (
"bytes"
"io/ioutil"
"math/rand"
"sync"
"testing"
"time"
)
// ReadStream from cache shouldn't panic on a nil dereference from a nonexistent
// Compression :)
func TestIssue2A(t *testing.T) {
d := New(Options{
BasePath: "test-issue-2a",
CacheSizeMax: 1024,
})
defer d.EraseAll()
input := "abcdefghijklmnopqrstuvwxy"
key, writeBuf, sync := "a", bytes.NewBufferString(input), false
if err := d.WriteStream(key, writeBuf, sync); err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
began := time.Now()
rc, err := d.ReadStream(key, false)
if err != nil {
t.Fatal(err)
}
buf, err := ioutil.ReadAll(rc)
if err != nil {
t.Fatal(err)
}
if !cmpBytes(buf, []byte(input)) {
t.Fatalf("read #%d: '%s' != '%s'", i+1, string(buf), input)
}
rc.Close()
t.Logf("read #%d in %s", i+1, time.Since(began))
}
}
// ReadStream on a key that resolves to a directory should return an error.
func TestIssue2B(t *testing.T) {
blockTransform := func(s string) []string {
transformBlockSize := 3
sliceSize := len(s) / transformBlockSize
pathSlice := make([]string, sliceSize)
for i := 0; i < sliceSize; i++ {
from, to := i*transformBlockSize, (i*transformBlockSize)+transformBlockSize
pathSlice[i] = s[from:to]
}
return pathSlice
}
d := New(Options{
BasePath: "test-issue-2b",
Transform: blockTransform,
CacheSizeMax: 0,
})
defer d.EraseAll()
v := []byte{'1', '2', '3'}
if err := d.Write("abcabc", v); err != nil {
t.Fatal(err)
}
_, err := d.ReadStream("abc", false)
if err == nil {
t.Fatal("ReadStream('abc') should return error")
}
t.Logf("ReadStream('abc') returned error: %v", err)
}
// Ensure ReadStream with direct=true isn't racy.
func TestIssue17(t *testing.T) {
var (
basePath = "test-data"
)
dWrite := New(Options{
BasePath: basePath,
CacheSizeMax: 0,
})
defer dWrite.EraseAll()
dRead := New(Options{
BasePath: basePath,
CacheSizeMax: 50,
})
cases := map[string]string{
"a": `1234567890`,
"b": `2345678901`,
"c": `3456789012`,
"d": `4567890123`,
"e": `5678901234`,
}
for k, v := range cases {
if err := dWrite.Write(k, []byte(v)); err != nil {
t.Fatalf("during write: %s", err)
}
dRead.Read(k) // ensure it's added to cache
}
var wg sync.WaitGroup
start := make(chan struct{})
for k, v := range cases {
wg.Add(1)
go func(k, v string) {
<-start
dRead.ReadStream(k, true)
wg.Done()
}(k, v)
}
close(start)
wg.Wait()
}
// Test for issue #40, where acquiring two stream readers on the same k/v pair
// caused the value to be written into the cache twice, messing up the
// size calculations.
func TestIssue40(t *testing.T) {
var (
basePath = "test-data"
)
// Simplest transform function: put all the data files into the base dir.
flatTransform := func(s string) []string { return []string{} }
// Initialize a new diskv store, rooted at "my-data-dir",
// with a 100 byte cache.
d := New(Options{
BasePath: basePath,
Transform: flatTransform,
CacheSizeMax: 100,
})
defer d.EraseAll()
// Write a 50 byte value, filling the cache half-way
k1 := "key1"
d1 := make([]byte, 50)
rand.Read(d1)
d.Write(k1, d1)
// Get *two* read streams on it. Because the key is not yet in the cache,
// and will not be in the cache until a stream is fully read, both
// readers use the 'siphon' object, which always writes to the cache
// after reading.
s1, err := d.ReadStream(k1, false)
if err != nil {
t.Fatal(err)
}
s2, err := d.ReadStream(k1, false)
if err != nil {
t.Fatal(err)
}
// When each stream is drained, the underlying siphon will write
// the value into the cache's map and increment the cache size.
// This means we will have 1 entry in the cache map
// ("key1" mapping to a 50 byte slice) but the cache size will be 100,
// because the buggy code does not check if an entry already exists
// in the map.
// s1 drains:
// cache[k] = v
// cacheSize += len(v)
// s2 drains:
// cache[k] = v /* overwrites existing */
// cacheSize += len(v) /* blindly adds to the cache size */
ioutil.ReadAll(s1)
ioutil.ReadAll(s2)
// Now write a different k/v pair, with a 60 byte array.
k2 := "key2"
d2 := make([]byte, 60)
rand.Read(d2)
d.Write(k2, d2)
// The act of reading the k/v pair back out causes it to be cached.
// Because the cache is only 100 bytes, it needs to delete existing
// entries to make room.
// If the cache is buggy, it will delete the single 50-byte entry
// from the cache map & decrement cacheSize by 50... but because
// cacheSize was improperly incremented twice earlier, this will
// leave us with no entries in the cacheMap but with cacheSize==50.
// Since CacheSizeMax-cacheSize (100-50) is less than 60, there
// is no room in the cache for this entry and it panics.
d.Read(k2)
}