Optimized reads

Total rewrite of the Read operation. There is now a segment LRU cache instead of multiple readers. This has more memory overhead, because up to four segments may be loaded in memory at one time, but is a ton faster and allows for forward and reverse sequential reads. This commit also removes the Durability option in favor of a simpiler NoSync option.

tidwall committed 6 years ago

Commit d1f90551 parent 9c440815 files · +807 −565Browse files

Unified · Split

5 files

README.md +37 −9
go.mod +8 −0
go.sum +8 −0
wal.go +482 −481
wal_test.go +272 −75

README.md+37−9

 # `wal`
 [![GoDoc](https://godoc.org/github.com/tidwall/wal?status.svg)](https://godoc.org/github.com/tidwall/wal)
-Write ahead log for Go.
+A simple and fast write ahead log for Go.
 ## Features
 - High durability
-- Fast writes
+- Fast operations
-- Low memory footprint
 - Monotonic indexes
+- Batch writes
 - Log truncation from front or back.
 ## Getting Started
 ```go
 // open a new log file
-l, _ := Open("mylog", nil)
+log, err := Open("mylog", nil)
 // write some entries
-l.Write(1, []byte("first entry"))
+err = log.Write(1, []byte("first entry"))
-l.Write(2, []byte("second entry"))
+err = log.Write(2, []byte("second entry"))
-l.Write(3, []byte("third entry"))
+err = log.Write(3, []byte("third entry"))
 // read an entry
-data, _ := l.Read(1)
+data, err := log.Read(1)
 println(string(data))  // output: first entry
 // close the log
-l.Close()
+err = log.Close()
 ```
+Batch writes:
+```go
+// write three entries as a batch
+batch := new(Batch)
+batch.Write(1, []byte("first entry"))
+batch.Write(2, []byte("second entry"))
+batch.Write(3, []byte("third entry"))
+err = log.WriteBatch(batch)
+```
+Truncating:
+```go
+// write some entries
+err = log.Write(1, []byte("first entry"))
+...
+err = log.Write(1000, []byte("thousandth entry"))
+// truncate the log from index starting 350 and ending with 950.
+err = l.TruncateFront(350)
+err = l.TruncateBack(950)
+```
 ## Contact
 Josh Baker [@tidwall](http://twitter.com/tidwall)

go.mod+8−0

@@ -0,0 +1,8 @@
	1	module github.com/tidwall/wal
	2
	3	go 1.13
	4
	5	require (
	6	github.com/tidwall/gjson v1.5.0
	7	github.com/tidwall/tinylru v1.0.2
	8	)

go.sum+8−0

@@ -0,0 +1,8 @@
	1	github.com/tidwall/gjson v1.5.0 h1:QCssIUI7J0RStkzIcI4A7O6P8rDA5wi5IPf70uqKSxg=
	2	github.com/tidwall/gjson v1.5.0/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
	3	github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
	4	github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
	5	github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
	6	github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
	7	github.com/tidwall/tinylru v1.0.2 h1:W4mp7iUz4cnVMqAvWy2zbzC35ASv5sqdyyEjoQKKBFg=
	8	github.com/tidwall/tinylru v1.0.2/go.mod h1:HDVL7TsWeezQ4g44Um84TOVBMFcq7Xa9giqNc805KJ8=

wal.go+482−481

@@ -1,20 +1,23 @@
1	package wal	1	package wal
2		2
3	import (	3	import (
4	"bufio"	4	"bytes"
5	"encoding/base64"	5	"encoding/base64"
6	"encoding/binary"	6	"encoding/binary"
7	"encoding/json"	7	"encoding/json"
8	"errors"	8	"errors"
9	"fmt"	9	"fmt"
10	"io"
11	"io/ioutil"	10	"io/ioutil"
12	"os"	11	"os"
13	"path/filepath"	12	"path/filepath"
14	"strconv"	13	"strconv"
15	"strings"	14	"strings"
		15	"sync"
16	"unicode/utf8"	16	"unicode/utf8"
17	"unsafe"	17	"unsafe"
		18
		19	"github.com/tidwall/gjson"
		20	"github.com/tidwall/tinylru"
18	)	21	)
19		22
20	var (	23	var (
@@ -42,22 +45,6 @@ var ( · 21 unmodified lines
42	ErrOutOfRange = errors.New("out of range")	45	ErrOutOfRange = errors.New("out of range")
43	)	46	)
44		47
45	// Durability policy.
46	type Durability int8
47
48	const (
49	// Low durability flushes data to operating system only when the buffer is
50	// at capacity. A process crash could cause data loss.
51	Low Durability = -1
52	// Medium durability flushes data to the operating system after every
53	// new entry is addded to the log. A server crash could cause data loss.
54	Medium Durability = 0
55	// High durability syncs data to disk after every new entry is added to
56	// the log. All entries are persisted to disk and crashes will not cause
57	// data loss.
58	High Durability = 1
59	)
60
61	// LogFormat is the format of the log files.	48	// LogFormat is the format of the log files.
62	type LogFormat byte	49	type LogFormat byte
63		50
@@ -72,49 +59,54 @@ const ( · 8 unmodified lines
72		59
73	// Options for Log	60	// Options for Log
74	type Options struct {	61	type Options struct {
75	// Durability policy. Default is High.	62	// NoSync disables fsync after writes. This is less durable and puts the
76	Durability Durability	63	// log at risk of data loss when there's a server crash.
		64	NoSync bool
77	// SegmentSize of each segment. This is just a target value, actual size	65	// SegmentSize of each segment. This is just a target value, actual size
78	// may differ. Default is 50 MB.	66	// may differ. Default is 20 MB.
79	SegmentSize int	67	SegmentSize int
80	// LogFormat is the format of the log files. Default is Binary.	68	// LogFormat is the format of the log files. Default is Binary.
81	LogFormat LogFormat	69	LogFormat LogFormat
		70	// SegmentCacheSize is the maximum number of segments that will be held in
		71	// memory for caching. Increasing this value may enhance performance for
		72	// concurrent read operations. Default is 1
		73	SegmentCacheSize int
82	}	74	}
83		75
84	// DefaultOptions for Open().	76	// DefaultOptions for Open().
85	var DefaultOptions = &Options{	77	var DefaultOptions = &Options{
86	Durability: High, // Fsync after every write	78	NoSync: false, // Fsync after every write
87	SegmentSize: 52428800, // 50 MB log segment files.	79	SegmentSize: 20971520, // 20 MB log segment files.
88	LogFormat: Binary, // Binary format is small and fast.	80	LogFormat: Binary, // Binary format is small and fast.
		81	SegmentCacheSize: 1, // Number of cached in-memory segments
89	}	82	}
90		83
91	const maxReaders = 8 // maximum number of opened readers.
92
93	// Log represents a write ahead log	84	// Log represents a write ahead log
94	type Log struct {	85	type Log struct {
95	path string // absolute path to log directory	86	mu sync.Mutex
96	opts Options // log options	87	path string // absolute path to log directory
97	closed bool // log is closed	88	opts Options // log options
98	segments []segment // all known log segments	89	closed bool // log is closed
99	firstIndex uint64 // index of the first entry in log	90	corrupt bool // log may be corrupt
100	lastIndex uint64 // index of the last entry in log	91	segments []*segment // all known log segments
101	file *os.File // tail segment file handle	92	firstIndex uint64 // index of the first entry in log
102	buffer []byte // tail segment file write buffer	93	lastIndex uint64 // index of the last entry in log
103	fileSize int // tail segment file size, including buffer	94	sfile *os.File // tail segment file handle
104	readers []*reader // all opened readers	95	wbatch Batch // reusable write batch
		96	scache tinylru.LRU // segment entries cache
105	}	97	}
106		98
107	// segment represents a single segment file.	99	// segment represents a single segment file.
108	type segment struct {	100	type segment struct {
109	path string // path of segment file	101	path string // path of segment file
110	index uint64 // first index of segment	102	index uint64 // first index of segment
		103	ebuf []byte // cached entries buffer
		104	epos []bpos // cached entries positions in buffer
111	}	105	}
112		106
113	type reader struct {	107	type bpos struct {
114	sindex int // segment index	108	pos int // byte position
115	nindex uint64 // next entry index	109	end int // one byte past pos
116	file *os.File // opened file
117	rd *bufio.Reader // reader
118	}	110	}
119		111
120	// Open a new write ahead log	112	// Open a new write ahead log
@@ -122,13 +114,22 @@ func Open(path string, opts Options) (Log, error) { · 1 unmodified line
122	if opts == nil {	114	if opts == nil {
123	opts = DefaultOptions	115	opts = DefaultOptions
124	}	116	}
		117	if opts.SegmentCacheSize <= 0 {
		118	opts.SegmentCacheSize = DefaultOptions.SegmentCacheSize
		119	}
		120	if opts.SegmentSize <= 0 {
		121	opts.SegmentSize = DefaultOptions.SegmentSize
		122	}
125	var err error	123	var err error
126	path, err = abs(path)	124	path, err = abs(path)
127	if err != nil {	125	if err != nil {
128	return nil, err	126	return nil, err
129	}	127	}
130	l := &Log{path: path, opts: *opts}	128	l := &Log{path: path, opts: *opts}
131	_ = os.MkdirAll(path, 0777)	129	l.scache.Resize(l.opts.SegmentCacheSize)
		130	if err := os.MkdirAll(path, 0777); err != nil {
		131	return nil, err
		132	}
132	if err := l.load(); err != nil {	133	if err := l.load(); err != nil {
133	return nil, err	134	return nil, err
134	}	135	}
@@ -142,6 +143,16 @@ func abs(path string) (string, error) { · 7 unmodified lines
142	return filepath.Abs(path)	143	return filepath.Abs(path)
143	}	144	}
144		145
		146	func (l *Log) pushCache(segIdx int) {
		147	_, _, _, v, evicted :=
		148	l.scache.SetEvicted(segIdx, l.segments[segIdx])
		149	if evicted {
		150	s := v.(*segment)
		151	s.ebuf = nil
		152	s.epos = nil
		153	}
		154	}
		155
145	// load all the segments. This operation also cleansup any START/END segments.	156	// load all the segments. This operation also cleansup any START/END segments.
146	func (l *Log) load() error {	157	func (l *Log) load() error {
147	fis, err := ioutil.ReadDir(l.path)	158	fis, err := ioutil.ReadDir(l.path)
@@ -167,7 +178,7 @@ func (l *Log) load() error { · 19 unmodified lines
167	} else if isEnd && endIdx == -1 {	178	} else if isEnd && endIdx == -1 {
168	endIdx = len(l.segments)	179	endIdx = len(l.segments)
169	}	180	}
170	l.segments = append(l.segments, segment{	181	l.segments = append(l.segments, &segment{
171	index: index,	182	index: index,
172	path: filepath.Join(l.path, name),	183	path: filepath.Join(l.path, name),
173	})	184	})
@@ -175,13 +186,13 @@ func (l *Log) load() error { · 1 unmodified line
175	}	186	}
176	if len(l.segments) == 0 {	187	if len(l.segments) == 0 {
177	// Create a new log	188	// Create a new log
178	l.segments = append(l.segments, segment{	189	l.segments = append(l.segments, &segment{
179	index: 1,	190	index: 1,
180	path: filepath.Join(l.path, segmentName(1)),	191	path: filepath.Join(l.path, segmentName(1)),
181	})	192	})
182	l.firstIndex = 1	193	l.firstIndex = 1
183	l.lastIndex = 0	194	l.lastIndex = 0
184	l.file, err = os.Create(l.segments[0].path)	195	l.sfile, err = os.Create(l.segments[0].path)
185	return err	196	return err
186	}	197	}
187	// Open existing log. Clean up log if START of END segments exists.	198	// Open existing log. Clean up log if START of END segments exists.
@@ -196,7 +207,7 @@ func (l *Log) load() error { · 8 unmodified lines
196	return err	207	return err
197	}	208	}
198	}	209	}
199	l.segments = append([]segment{}, l.segments[startIdx:]...)	210	l.segments = append([]*segment{}, l.segments[startIdx:]...)
200	// Rename the START segment	211	// Rename the START segment
201	orgPath := l.segments[0].path	212	orgPath := l.segments[0].path
202	finalPath := orgPath[:len(orgPath)-len(".START")]	213	finalPath := orgPath[:len(orgPath)-len(".START")]
@@ -213,7 +224,7 @@ func (l *Log) load() error { · 10 unmodified lines
213	return err	224	return err
214	}	225	}
215	}	226	}
216	l.segments = append([]segment{}, l.segments[:endIdx+1]...)	227	l.segments = append([]*segment{}, l.segments[:endIdx+1]...)
217	if len(l.segments) > 1 && l.segments[len(l.segments)-2].index ==	228	if len(l.segments) > 1 && l.segments[len(l.segments)-2].index ==
218	l.segments[len(l.segments)-1].index {	229	l.segments[len(l.segments)-1].index {
219	// remove the segment prior to the END segment because it shares	230	// remove the segment prior to the END segment because it shares
@@ -230,125 +241,108 @@ func (l *Log) load() error { · 10 unmodified lines
230	}	241	}
231	l.segments[len(l.segments)-1].path = finalPath	242	l.segments[len(l.segments)-1].path = finalPath
232	}	243	}
233	// Open the last segment for appending
234	l.firstIndex = l.segments[0].index	244	l.firstIndex = l.segments[0].index
235	l.file, err = os.OpenFile(l.segments[len(l.segments)-1].path,	245	// Open the last segment for appending
236	os.O_RDWR, 0666)	246	lseg := l.segments[len(l.segments)-1]
		247	l.sfile, err = os.OpenFile(lseg.path, os.O_WRONLY, 0666)
237	if err != nil {	248	if err != nil {
238	return err	249	return err
239	}	250	}
240	// Read the last segment to the end of the log	251	if _, err := l.sfile.Seek(0, 2); err != nil {
241	rd := bufio.NewReader(l.file)	252	return err
242	for {
243	idx, _, err := readEntry(rd, l.opts.LogFormat, true)
244	if err != nil {
245	if err == io.EOF {
246	break
247	}
248	return err
249	}
250	l.lastIndex = idx
251	}	253	}
252	n, err := l.file.Seek(0, 2)	254	// Load the last segment entries
253	if err != nil {	255	if err := l.loadSegmentEntries(lseg); err != nil {
254	return err	256	return err
255	}	257	}
256	l.fileSize = int(n)	258	l.lastIndex = lseg.index + uint64(len(lseg.epos)) - 1
257	return nil	259	return nil
258	}	260	}
259		261
260	func must(v interface{}, err error) interface{} {
261	if err != nil {
262	panic(err)
263	}
264	return v
265	}
266
267	// segmentName returns a 20-byte textual representation of an index	262	// segmentName returns a 20-byte textual representation of an index
268	// for lexical ordering. This is used for the file names of log segments.	263	// for lexical ordering. This is used for the file names of log segments.
269	func segmentName(index uint64) string {	264	func segmentName(index uint64) string {
270	return fmt.Sprintf("%020d", index)	265	return fmt.Sprintf("%020d", index)
271	}	266	}
272		267
273	// Close the log	268	// Close the log.
274	func (l *Log) Close() error {	269	func (l *Log) Close() error {
		270	l.mu.Lock()
		271	defer l.mu.Unlock()
275	if l.closed {	272	if l.closed {
		273	if l.corrupt {
		274	return ErrCorrupt
		275	}
276	return ErrClosed	276	return ErrClosed
277	}	277	}
278	l.flush()	278	if err := l.sfile.Sync(); err != nil {
279	must(nil, l.file.Close())	279	return err
280	for len(l.readers) > 0 {	280	}
281	l.closeReader(l.readers[0])	281	if err := l.sfile.Close(); err != nil {
		282	return err
282	}	283	}
283	l.readers = nil
284	l.segments = nil
285	l.closed = true	284	l.closed = true
286	return nil	285	if l.corrupt {
287	}	286	return ErrCorrupt
288
289	func (l *Log) flush() {
290	if len(l.buffer) > 0 {
291	must(l.file.Write(l.buffer))
292	l.buffer = l.buffer[:0]
293	if l.opts.Durability >= High {
294	must(nil, l.file.Sync())
295	}
296	}	287	}
		288	return nil
297	}	289	}
298		290
299	// Write an entry to the log.	291	// Write an entry to the log.
300	func (l *Log) Write(index uint64, data []byte) error {	292	func (l *Log) Write(index uint64, data []byte) error {
301	if l.closed {	293	l.mu.Lock()
		294	defer l.mu.Unlock()
		295	if l.corrupt {
		296	return ErrCorrupt
		297	} else if l.closed {
302	return ErrClosed	298	return ErrClosed
303	}	299	}
304	if index != l.lastIndex+1 {	300	l.wbatch.Clear()
305	return ErrOutOfOrder	301	l.wbatch.Write(index, data)
306	}	302	return l.writeBatch(&l.wbatch)
307	if l.fileSize >= l.opts.SegmentSize {	303	}
308	l.cycle()	304
309	}	305	func (l *Log) appendEntry(dst []byte, index uint64, data []byte) (out []byte,
310	l.appendEntry(index, data)	306	epos bpos) {
311	if l.opts.Durability >= Medium \|\| len(l.buffer) >= 4096 {	307	if l.opts.LogFormat == JSON {
312	l.flush()	308	return appendJSONEntry(dst, index, data)
313	}	309	}
314	l.lastIndex = index	310	return appendBinaryEntry(dst, data)
315	return nil
316	}	311	}
317		312
318	// Cycle the old segment for a new segment.	313	// Cycle the old segment for a new segment.
319	func (l *Log) cycle() {	314	func (l *Log) cycle() error {
320	l.flush()	315	if err := l.sfile.Sync(); err != nil {
321	must(nil, l.file.Close())	316	return err
322	s := segment{	317	}
		318	if err := l.sfile.Close(); err != nil {
		319	return err
		320	}
		321	// cache the previous segement
		322	l.pushCache(len(l.segments) - 1)
		323	s := &segment{
323	index: l.lastIndex + 1,	324	index: l.lastIndex + 1,
324	path: filepath.Join(l.path, segmentName(l.lastIndex+1)),	325	path: filepath.Join(l.path, segmentName(l.lastIndex+1)),
325	}	326	}
326	l.file = must(os.Create(s.path)).(*os.File)	327	var err error
327	l.fileSize = 0	328	l.sfile, err = os.Create(s.path)
328	l.segments = append(l.segments, s)	329	if err != nil {
329	l.buffer = nil	330	return err
330	}
331
332	// appendEntry to the log buffer. This also increases the log segment fileSize.
333	func (l *Log) appendEntry(index uint64, data []byte) {
334	mark := len(l.buffer)
335	if l.opts.LogFormat == JSON {
336	l.buffer = appendJSONEntry(l.buffer, index, data)
337	} else {
338	l.buffer = appendBinaryEntry(l.buffer, index, data)
339	}	331	}
340	l.fileSize += len(l.buffer) - mark	332	l.segments = append(l.segments, s)
341		333	return nil
342	}	334	}
343		335
344	func appendJSONEntry(dst []byte, index uint64, data []byte) []byte {	336	func appendJSONEntry(dst []byte, index uint64, data []byte) (out []byte,
		337	epos bpos) {
345	// {"index":number,"data":string}	338	// {"index":number,"data":string}
		339	mark := len(dst)
346	dst = append(dst, `{"index":"`...)	340	dst = append(dst, `{"index":"`...)
347	dst = strconv.AppendUint(dst, index, 10)	341	dst = strconv.AppendUint(dst, index, 10)
348	dst = append(dst, `","data":`...)	342	dst = append(dst, `","data":`...)
349	dst = appendJSONData(dst, data)	343	dst = appendJSONData(dst, data)
350	dst = append(dst, '}', '\n')	344	dst = append(dst, '}', '\n')
351	return dst	345	return dst, bpos{mark, len(dst)}
352	}	346	}
353		347
354	func appendJSONData(dst []byte, s []byte) []byte {	348	func appendJSONData(dst []byte, s []byte) []byte {
@@ -362,12 +356,12 @@ func appendJSONData(dst []byte, s []byte) []byte { · 7 unmodified lines
362	return append(dst, '"')	356	return append(dst, '"')
363	}	357	}
364		358
365	func appendBinaryEntry(dst []byte, index uint64, data []byte) []byte {	359	func appendBinaryEntry(dst []byte, data []byte) (out []byte, epos bpos) {
366	// index + data_size + data	360	// data_size + data
367	dst = appendUvarint(dst, index)	361	pos := len(dst)
368	dst = appendUvarint(dst, uint64(len(data)))	362	dst = appendUvarint(dst, uint64(len(data)))
369	dst = append(dst, data...)	363	dst = append(dst, data...)
370	return dst	364	return dst, bpos{pos, len(dst)}
371	}	365	}
372		366
373	func appendUvarint(dst []byte, x uint64) []byte {	367	func appendUvarint(dst []byte, x uint64) []byte {
@@ -379,53 +373,92 @@ func appendUvarint(dst []byte, x uint64) []byte { · 5 unmodified lines
379		373
380	// Batch of entries. Used to write multiple entries at once using WriteBatch().	374	// Batch of entries. Used to write multiple entries at once using WriteBatch().
381	type Batch struct {	375	type Batch struct {
382	indexes []uint64	376	entries []batchEntry
383	dataSizes []int	377	datas []byte
384	datas []byte	378	}
		379
		380	type batchEntry struct {
		381	index uint64
		382	size int
385	}	383	}
386		384
387	// Write an entry to the batch	385	// Write an entry to the batch
388	func (b *Batch) Write(index uint64, data []byte) {	386	func (b *Batch) Write(index uint64, data []byte) {
389	b.indexes = append(b.indexes, index)	387	b.entries = append(b.entries, batchEntry{index, len(data)})
390	b.dataSizes = append(b.dataSizes, len(data))
391	b.datas = append(b.datas, data...)	388	b.datas = append(b.datas, data...)
392	}	389	}
393		390
394	// Clear the batch for reuse.	391	// Clear the batch for reuse.
395	func (b *Batch) Clear() {	392	func (b *Batch) Clear() {
396	b.indexes = b.indexes[:0]	393	b.entries = b.entries[:0]
397	b.dataSizes = b.dataSizes[:0]
398	b.datas = b.datas[:0]	394	b.datas = b.datas[:0]
399	}	395	}
400		396
401	// WriteBatch writes the entries in the batch to the log in the order that they	397	// WriteBatch writes the entries in the batch to the log in the order that they
402	// were added to the batch. The batch is cleared upon a successful return.	398	// were added to the batch. The batch is cleared upon a successful return.
403	func (l Log) WriteBatch(b Batch) error {	399	func (l Log) WriteBatch(b Batch) error {
404	if l.closed {	400	l.mu.Lock()
		401	defer l.mu.Unlock()
		402	if l.corrupt {
		403	return ErrCorrupt
		404	} else if l.closed {
405	return ErrClosed	405	return ErrClosed
406	}	406	}
407	if len(b.indexes) == 0 {	407	if len(b.entries) == 0 {
408	return nil	408	return nil
409	}	409	}
410	// check all indexes in batch	410	return l.writeBatch(b)
411	for i := 0; i < len(b.indexes); i++ {	411	}
412	if b.indexes[i] != l.lastIndex+uint64(i+1) {	412
		413	func (l Log) writeBatch(b Batch) error {
		414	// check that all indexes in batch are sane
		415	for i := 0; i < len(b.entries); i++ {
		416	if b.entries[i].index != l.lastIndex+uint64(i+1) {
413	return ErrOutOfOrder	417	return ErrOutOfOrder
414	}	418	}
415	}	419	}
416	if l.fileSize >= l.opts.SegmentSize {	420	// load the tail segment
417	l.cycle()	421	s := l.segments[len(l.segments)-1]
		422	if len(s.ebuf) > l.opts.SegmentSize {
		423	// tail segment has reached capacity. Close it and create a new one.
		424	if err := l.cycle(); err != nil {
		425	return err
		426	}
		427	s = l.segments[len(l.segments)-1]
418	}	428	}
		429
		430	mark := len(s.ebuf)
419	datas := b.datas	431	datas := b.datas
420	for i := 0; i < len(b.indexes); i++ {	432	for i := 0; i < len(b.entries); i++ {
421	data := datas[:b.dataSizes[i]]	433	data := datas[:b.entries[i].size]
422	l.appendEntry(b.indexes[i], data)	434	var epos bpos
423	datas = datas[b.dataSizes[i]:]	435	s.ebuf, epos = l.appendEntry(s.ebuf, b.entries[i].index, data)
		436	s.epos = append(s.epos, epos)
		437	if len(s.ebuf) >= l.opts.SegmentSize {
		438	// segment has reached capacity, cycle now
		439	if _, err := l.sfile.Write(s.ebuf[mark:]); err != nil {
		440	return err
		441	}
		442	l.lastIndex = b.entries[i].index
		443	if err := l.cycle(); err != nil {
		444	return err
		445	}
		446	s = l.segments[len(l.segments)-1]
		447	mark = 0
		448	}
		449	datas = datas[b.entries[i].size:]
424	}	450	}
425	if l.opts.Durability >= Medium \|\| len(l.buffer) >= 4096 {	451	if len(s.ebuf)-mark > 0 {
426	l.flush()	452	if _, err := l.sfile.Write(s.ebuf[mark:]); err != nil {
		453	return err
		454	}
		455	l.lastIndex = b.entries[len(b.entries)-1].index
		456	}
		457	if !l.opts.NoSync {
		458	if err := l.sfile.Sync(); err != nil {
		459	return err
		460	}
427	}	461	}
428	l.lastIndex = b.indexes[len(b.indexes)-1]
429	b.Clear()	462	b.Clear()
430	return nil	463	return nil
431	}	464	}
@@ -433,7 +466,11 @@ func (l Log) WriteBatch(b Batch) error { · 1 unmodified line
433	// FirstIndex returns the index of the first entry in the log. Returns zero	466	// FirstIndex returns the index of the first entry in the log. Returns zero
434	// when log has no entries.	467	// when log has no entries.
435	func (l *Log) FirstIndex() (index uint64, err error) {	468	func (l *Log) FirstIndex() (index uint64, err error) {
436	if l.closed {	469	l.mu.Lock()
		470	defer l.mu.Unlock()
		471	if l.corrupt {
		472	return 0, ErrCorrupt
		473	} else if l.closed {
437	return 0, ErrClosed	474	return 0, ErrClosed
438	}	475	}
439	// We check the lastIndex for zero because the firstIndex is always one or	476	// We check the lastIndex for zero because the firstIndex is always one or
@@ -447,7 +484,11 @@ func (l *Log) FirstIndex() (index uint64, err error) { · 7 unmodified lines
447	// LastIndex returns the index of the last entry in the log. Returns zero when	484	// LastIndex returns the index of the last entry in the log. Returns zero when
448	// log has no entries.	485	// log has no entries.
449	func (l *Log) LastIndex() (index uint64, err error) {	486	func (l *Log) LastIndex() (index uint64, err error) {
450	if l.closed {	487	l.mu.Lock()
		488	defer l.mu.Unlock()
		489	if l.corrupt {
		490	return 0, ErrCorrupt
		491	} else if l.closed {
451	return 0, ErrClosed	492	return 0, ErrClosed
452	}	493	}
453	if l.lastIndex == 0 {	494	if l.lastIndex == 0 {
@@ -470,425 +511,385 @@ func (l *Log) findSegment(index uint64) int { · 16 unmodified lines
470	return i - 1	511	return i - 1
471	}	512	}
472		513
473	// Read an entry from the log. This function reads an entry from disk and is	514	func (l Log) loadSegmentEntries(s segment) error {
474	// optimized for sequential reads. Randomly accessing entries is slow.	515	data, err := ioutil.ReadFile(s.path)
475	func (l *Log) Read(index uint64) (data []byte, err error) {
476	if l.closed {
477	return nil, ErrClosed
478	}
479	if index == 0 \|\| index < l.firstIndex \|\| index > l.lastIndex {
480	return nil, ErrNotFound
481	}
482
483	// find an opened reader
484	var r *reader
485	for _, or := range l.readers {
486	if or.nindex == index {
487	r = or
488	break
489	}
490	}
491	if r == nil {
492	// Reader not found, open a new reader and return the entry at index
493	return l.openReader(index)
494	}
495
496	// Read next entry from reader
497	for {
498	eindex, edata, err := l.readEntry(r)
499	if err == io.EOF {
500	// Reached the end of the segment.
501	if r.sindex == len(l.segments)-1 {
502	// At the end of the last segment file.
503	if len(l.buffer) > 0 {
504	// But the segment has a buffer, flush it and try again
505	must(l.file.Write(l.buffer))
506	l.buffer = l.buffer[:0]
507	continue
508	}
509	// Log is missing some final entries, consider is corrput
510	l.closeReader(r)
511	return nil, ErrCorrupt
512	}
513	// Close the old reader and open a new one
514	l.closeReader(r)
515	return l.openReader(index)
516	}
517	if eindex != index {
518	// Log has gaps or is out of order, corrupt file.
519	l.closeReader(r)
520	return nil, ErrCorrupt
521	}
522	r.nindex++
523	if r.nindex == l.lastIndex+1 {
524	// read the last entry, close the reader
525	l.closeReader(r)
526	}
527	return edata, nil
528	}
529	}
530
531	// openReader opens a new reader and returns the data belong to entry at index.
532	func (l *Log) openReader(index uint64) (data []byte, err error) {
533	// reader not found, open a new one
534	r := &reader{}
535	r.sindex = l.findSegment(index)
536	r.nindex = l.segments[r.sindex].index
537	r.file, err = os.Open(l.segments[r.sindex].path)
538	if err != nil {	516	if err != nil {
539	return nil, err	517	return err
540	}	518	}
541	r.rd = bufio.NewReader(r.file)	519	ebuf := data
542	if r.sindex == len(l.segments)-1 {	520	var epos []bpos
543	if len(l.buffer) > 0 {	521	var pos int
544	// Reading from the last segment, which has a buffer, flush the	522	for exidx := s.index; len(data) > 0; exidx++ {
545	// buffer before reading from file.	523	var n int
546	must(l.file.Write(l.buffer))	524	if l.opts.LogFormat == JSON {
547	l.buffer = l.buffer[:0]	525	n, err = loadNextJSONEntry(data)
		526	} else {
		527	n, err = loadNextBinaryEntry(data)
548	}	528	}
549	}
550	// Scan the file for the entry at index.
551	for {
552	eindex, edata, err := l.readEntry(r)
553	if err != nil {	529	if err != nil {
554	// Bad news. Likely a corrupt file.	530	return err
555	r.file.Close()
556	return nil, err
557	}
558	if eindex != r.nindex {
559	// Log has gaps or is out of order, corrupt file.
560	r.file.Close()
561	return nil, ErrCorrupt
562	}
563	r.nindex = eindex + 1
564	if eindex == index {
565	// Add the new reader to the front of the list of opened readers.
566	l.readers = append(l.readers, nil)
567	copy(l.readers[1:], l.readers[:len(l.readers)-1])
568	l.readers[0] = r
569	for len(l.readers) > maxReaders {
570	// remove the oldest readers
571	l.closeReader(l.readers[len(l.readers)-1])
572	}
573	return edata, nil
574	}	531	}
		532	data = data[n:]
		533	epos = append(epos, bpos{pos, pos + n})
		534	pos += n
575	}	535	}
		536	s.ebuf = ebuf
		537	s.epos = epos
		538	return nil
576	}	539	}
577		540
578	// closeReader closes the reader and removes it from the list of opened readers.	541	func loadNextJSONEntry(data []byte) (n int, err error) {
579	func (l Log) closeReader(r reader) {	542	// {"index":number,"data":string}
580	for i, rr := range l.readers {	543	idx := bytes.IndexByte(data, '\n')
581	if rr == r {	544	if idx == -1 {
582	must(nil, r.file.Close())	545	return 0, ErrCorrupt
583	l.readers[i] = l.readers[len(l.readers)-1]	546	}
584	l.readers[len(l.readers)-1] = nil	547	line := data[:idx]
585	l.readers = l.readers[:len(l.readers)-1]	548	dres := gjson.Get((string)(unsafe.Pointer(&line)), "data")
586	return	549	if dres.Type != gjson.String {
		550	return 0, ErrCorrupt
		551	}
		552	return idx + 1, nil
		553	}
		554
		555	func loadNextBinaryEntry(data []byte) (n int, err error) {
		556	// data_size + data
		557	size, n := binary.Uvarint(data)
		558	if n <= 0 {
		559	return 0, ErrCorrupt
		560	}
		561	if uint64(len(data)-n) < size {
		562	return 0, ErrCorrupt
		563	}
		564	return n + int(size), nil
		565	}
		566
		567	// loadSegment loads the segment entries into memory, pushes it to the front
		568	// of the lru cache, and returns it.
		569	func (l Log) loadSegment(index uint64) (segment, error) {
		570	// check the last segment first.
		571	lseg := l.segments[len(l.segments)-1]
		572	if index >= lseg.index {
		573	return lseg, nil
		574	}
		575	// check the most recent cached segment
		576	var rseg *segment
		577	l.scache.Range(func(_, v interface{}) bool {
		578	s := v.(*segment)
		579	if index >= s.index && index < s.index+uint64(len(s.epos)) {
		580	rseg = s
		581	}
		582	return false
		583	})
		584	if rseg != nil {
		585	return rseg, nil
		586	}
		587	// find in the segment array
		588	idx := l.findSegment(index)
		589	s := l.segments[idx]
		590	if len(s.epos) == 0 {
		591	// load the entries from cache
		592	if err := l.loadSegmentEntries(s); err != nil {
		593	return nil, err
587	}	594	}
588	}	595	}
		596	// push the segment to the front of the cache
		597	l.pushCache(idx)
		598	return s, nil
589	}	599	}
590		600
591	// readEntry reads the next entry from reader. Returns io.EOF if the readers	601	// Read an entry from the log. Returns a byte slice containing the data entry.
592	// has reached the end of a segment.	602	func (l *Log) Read(index uint64) (data []byte, err error) {
593	func (l Log) readEntry(r reader) (index uint64, data []byte, err error) {	603	l.mu.Lock()
594	return readEntry(r.rd, l.opts.LogFormat, false)	604	defer l.mu.Unlock()
595	}	605	if l.corrupt {
596		606	return nil, ErrCorrupt
597	// readEntry from bufio.Reader. The discardData param will discard the entry	607	} else if l.closed {
598	// data ane return nil.	608	return nil, ErrClosed
599	func readEntry(rd *bufio.Reader, frmt LogFormat, discardData bool) (
600	index uint64, data []byte, err error,
601	) {
602	if frmt == JSON {
603	line, err := rd.ReadBytes('\n')
604	if err != nil {
605	if err == io.EOF && len(line) > 0 {
606	return 0, nil, ErrCorrupt
607	}
608	return 0, nil, err
609	}
610	var m map[string]string
611	if err := json.Unmarshal(line, &m); err != nil {
612	return 0, nil, ErrCorrupt
613	}
614	index, err = strconv.ParseUint(m["index"], 10, 64)
615	if err != nil {
616	return 0, nil, ErrCorrupt
617	}
618	s, ok := m["data"]
619	if !ok \|\| len(s) == 0 {
620	return 0, nil, ErrCorrupt
621	}
622	if !discardData {
623	switch s[0] {
624	case '$':
625	data, err = base64.URLEncoding.DecodeString(s[1:])
626	if err != nil {
627	return 0, nil, ErrCorrupt
628	}
629	case '+':
630	data = []byte(s[1:])
631	default:
632	return 0, nil, ErrCorrupt
633	}
634	}
635	return index, data, nil
636	}	609	}
637	index, err = binary.ReadUvarint(rd)	610	if index == 0 \|\| index < l.firstIndex \|\| index > l.lastIndex {
638	if err != nil {	611	return nil, ErrNotFound
639	return 0, nil, err
640	}	612	}
641	dataSize, err := binary.ReadUvarint(rd)	613	s, err := l.loadSegment(index)
642	if err != nil {	614	if err != nil {
643	if err == io.EOF {	615	return nil, err
644	return 0, nil, ErrCorrupt
645	}
646	return 0, nil, err
647	}	616	}
648	if discardData {	617	epos := s.epos[index-s.index]
649	_, err = rd.Discard(int(dataSize))	618	edata := s.ebuf[epos.pos:epos.end]
		619	if l.opts.LogFormat == JSON {
		620	s := gjson.Get((string)(unsafe.Pointer(&edata)), "data").String()
		621	if len(s) > 0 && s[0] == '$' {
		622	var err error
		623	data, err = base64.URLEncoding.DecodeString(s[1:])
		624	if err != nil {
		625	return nil, ErrCorrupt
		626	}
		627	} else if len(s) > 0 && s[0] == '+' {
		628	data = make([]byte, len(s[1:]))
		629	copy(data, s[1:])
		630	} else {
		631	return nil, ErrCorrupt
		632	}
650	} else {	633	} else {
651	data = make([]byte, dataSize)	634	size, n := binary.Uvarint(edata)
652	_, err = io.ReadFull(rd, data)	635	if n <= 0 {
653	}	636	return nil, ErrCorrupt
654	if err != nil {
655	if err == io.EOF {
656	return 0, nil, ErrCorrupt
657	}	637	}
658	return 0, nil, err	638	if uint64(len(edata)-n) < size {
		639	return nil, ErrCorrupt
		640	}
		641	data = make([]byte, size)
		642	copy(data, edata[n:])
659	}	643	}
660	return index, data, nil	644	return data, nil
661	}	645	}
662		646
663	// TruncateFront truncates the front of the log by removing all entries that	647	// ClearCache clears the segment cache
664	// are before the provided `firstIndex`. In other words the entry at	648	func (l *Log) ClearCache() error {
665	// `firstIndex` becomes the first entry in the log.	649	l.mu.Lock()
666	func (l *Log) TruncateFront(firstIndex uint64) error {	650	defer l.mu.Unlock()
667	if l.closed {	651	if l.corrupt {
		652	return ErrCorrupt
		653	} else if l.closed {
668	return ErrClosed	654	return ErrClosed
669	}	655	}
		656	l.clearCache()
		657	return nil
		658	}
		659	func (l *Log) clearCache() {
		660	l.scache.Range(func(_, v interface{}) bool {
		661	s := v.(*segment)
		662	s.ebuf = nil
		663	s.epos = nil
		664	return true
		665	})
		666	l.scache = tinylru.LRU{}
		667	l.scache.Resize(l.opts.SegmentCacheSize)
		668	}
670		669
671	index := firstIndex // alias	670	// TruncateFront truncates the front of the log by removing all entries that
672		671	// are before the provided `index`. In other words the entry at
		672	// `index` becomes the first entry in the log.
		673	func (l *Log) TruncateFront(index uint64) error {
		674	l.mu.Lock()
		675	defer l.mu.Unlock()
		676	if l.corrupt {
		677	return ErrCorrupt
		678	} else if l.closed {
		679	return ErrClosed
		680	}
		681	return l.truncateFront(index)
		682	}
		683	func (l *Log) truncateFront(index uint64) (err error) {
673	if index == 0 \|\| l.lastIndex == 0 \|\|	684	if index == 0 \|\| l.lastIndex == 0 \|\|
674	index < l.firstIndex \|\| index > l.lastIndex {	685	index < l.firstIndex \|\| index > l.lastIndex {
675	return ErrOutOfRange	686	return ErrOutOfRange
676	}	687	}
677
678	// Flush the buffer, if needed.
679	if len(l.buffer) > 0 {
680	must(l.file.Write(l.buffer))
681	l.buffer = nil
682	}
683	// Close all readers
684	for len(l.readers) > 0 {
685	l.closeReader(l.readers[0])
686	}
687
688	if index == l.firstIndex {	688	if index == l.firstIndex {
689	// nothing to truncate	689	// nothing to truncate
690	return nil	690	return nil
691	}	691	}
692		692	segIdx := l.findSegment(index)
693	// Find which segment has the entry with the index.	693	var s *segment
694	sidx := l.findSegment(index)	694	s, err = l.loadSegment(index)
695	// Find the file offset of the first byte belonging to the entry at index.
696	f, err := os.Open(l.segments[sidx].path)
697	if err != nil {	695	if err != nil {
698	return err	696	return err
699	}	697	}
700	defer f.Close()	698	epos := s.epos[index-s.index:]
701	if index > l.segments[sidx].index {	699	ebuf := s.ebuf[epos[0].pos:]
702	// Read all entries prior to entry at index.	700	// Create a temp file contains the truncated segment.
703	rd := bufio.NewReader(f)	701	tempName := filepath.Join(l.path, "TEMP")
704	var found bool	702	err = func() error {
705	for {	703	f, err := os.Create(tempName)
706	ridx, _, err := readEntry(rd, l.opts.LogFormat, true)	704	if err != nil {
707	if err != nil {	705	return err
708	return err
709	}
710	if ridx == index-1 {
711	// Seek to exact position of entry, the reader likely overread
712	// the file, so we need to back up what has been buffered.
713	if _, err := f.Seek(0-int64(rd.Buffered()), 1); err != nil {
714	return err
715	}
716	found = true
717	break
718	}
719	}	706	}
720	if !found {	707	defer f.Close()
721	return ErrCorrupt	708	if _, err := f.Write(ebuf); err != nil {
		709	return err
722	}	710	}
723	}	711	if err := f.Sync(); err != nil {
724	// Create a temp file in the same log directory and copy all of the data	712	return err
725	// starting at the seeked offset position.	713	}
726	tempName := filepath.Join(l.path, "TEMP")	714	return f.Close()
727	ftmp, err := os.Create(tempName)	715	}()
728	if err != nil {	716	// Rename the TEMP file to it's START file name.
		717	startName := filepath.Join(l.path, segmentName(index)+".START")
		718	if err = os.Rename(tempName, startName); err != nil {
729	return err	719	return err
730	}	720	}
		721	// The log was truncated but still needs some file cleanup. Any errors
		722	// following this message will not cause an on-disk data ocorruption, but
		723	// may cause an inconsistency with the current program, so we'll return
		724	// ErrCorrupt so the the user can attempt a recover by calling Close()
		725	// followed by Open().
731	defer func() {	726	defer func() {
732	ftmp.Close()	727	if v := recover(); v != nil {
733	os.Remove(tempName)	728	err = ErrCorrupt
		729	l.corrupt = true
		730	}
734	}()	731	}()
735	if _, err := io.Copy(ftmp, f); err != nil {	732	if segIdx == len(l.segments)-1 {
736	return err	733	// Close the tail segment file
737	}	734	if err = l.sfile.Close(); err != nil {
738	if err := f.Close(); err != nil {	735	return err
739	return err	736	}
740	}	737	}
741	if err := ftmp.Close(); err != nil {	738	// Delete truncated segment files
742	return err	739	for i := 0; i <= segIdx; i++ {
		740	if err = os.Remove(l.segments[i].path); err != nil {
		741	return err
		742	}
743	}	743	}
744	// Then rename the TEMP file to it's START file name.	744	// Rename the START file to the final truncated segment name.
745	startName := filepath.Join(l.path, segmentName(index)+".START")	745	newName := filepath.Join(l.path, segmentName(index))
746	if err := os.Rename(tempName, startName); err != nil {	746	if os.Rename(startName, newName); err != nil {
747	return err	747	return err
748	}	748	}
749	// All operations from this point on are syscalls and must succeed,	749	s.path = newName
750	// otherwise we panic.	750	s.index = index
751	// Remove all of the unneeded segments.	751	if segIdx == len(l.segments)-1 {
752	for i := 0; i <= sidx; i++ {	752	// Reopen the tail segment file
753	must(nil, os.Remove(l.segments[i].path))	753	if l.sfile, err = os.OpenFile(newName, os.O_WRONLY, 0666); err != nil {
754	}	754	return err
755	// Remove the ".START" suffix from the file name.	755	}
756	finalName := startName[:len(startName)-len(".START")]	756	var n int64
757	must(nil, os.Rename(startName, finalName))	757	if n, err = l.sfile.Seek(0, 2); err != nil {
758	// Modify the segments list.	758	return err
759	l.segments = append([]segment{segment{path: finalName, index: index}},	759	}
760	l.segments[sidx+1:]...)	760	if n != int64(len(ebuf)) {
761	if len(l.segments) == 1 {	761	err = errors.New("invalid seek")
762	// The last segment has been changed which means we need to reopen the	762	return err
763	// appendable file and seek to the end.	763	}
764	must(nil, l.file.Close())	764	// Load the last segment entries
765	l.file = must(os.OpenFile(finalName, os.O_RDWR, 0666)).(*os.File)	765	if err = l.loadSegmentEntries(s); err != nil {
766	l.fileSize = int(must(l.file.Seek(0, 2)).(int64))	766	return err
767	l.buffer = nil	767	}
768	}	768	}
		769	l.segments = append([]*segment{}, l.segments[segIdx:]...)
769	l.firstIndex = index	770	l.firstIndex = index
		771	l.clearCache()
770	return nil	772	return nil
771	}	773	}
772		774
773	// TruncateBack truncates the back of the log by removing all entries that	775	// TruncateBack truncates the back of the log by removing all entries that
774	// are after the provided `lastIndex`. In other words the entry at `lastIndex`	776	// are after the provided `index`. In other words the entry at `index`
775	// becomes the last entry in the log.	777	// becomes the last entry in the log.
776	func (l *Log) TruncateBack(lastIndex uint64) error {	778	func (l *Log) TruncateBack(index uint64) error {
777	if l.closed {	779	l.mu.Lock()
		780	defer l.mu.Unlock()
		781	if l.corrupt {
		782	return ErrCorrupt
		783	} else if l.closed {
778	return ErrClosed	784	return ErrClosed
779	}	785	}
		786	return l.truncateBack(index)
		787	}
780		788
781	index := lastIndex // alias	789	func (l *Log) truncateBack(index uint64) (err error) {
782
783	if index == 0 \|\| l.lastIndex == 0 \|\|	790	if index == 0 \|\| l.lastIndex == 0 \|\|
784	index > l.lastIndex \|\| index < l.firstIndex {	791	index < l.firstIndex \|\| index > l.lastIndex {
785	return ErrOutOfRange	792	return ErrOutOfRange
786	}	793	}
787
788	// Flush the buffer, if needed.
789	if len(l.buffer) > 0 {
790	must(l.file.Write(l.buffer))
791	l.buffer = nil
792	}
793	// Close all readers
794	for len(l.readers) > 0 {
795	l.closeReader(l.readers[0])
796	}
797
798	if index == l.lastIndex {	794	if index == l.lastIndex {
799	// nothing to truncate	795	// nothing to truncate
800	return nil	796	return nil
801	}	797	}
802		798	segIdx := l.findSegment(index)
803	// Find which segment has the entry with the index.	799	var s *segment
804	sidx := l.findSegment(index)	800	s, err = l.loadSegment(index)
805	// Find the file offset of the first byte belonging to the entry at index.
806	f, err := os.Open(l.segments[sidx].path)
807	if err != nil {	801	if err != nil {
808	return err	802	return err
809	}	803	}
810	defer f.Close()	804	epos := s.epos[:index-s.index+1]
811	// Read all entries prior to entry at index.	805	ebuf := s.ebuf[:epos[len(epos)-1].end]
812	rd := bufio.NewReader(f)	806	// Create a temp file contains the truncated segment.
813	var found bool	807	tempName := filepath.Join(l.path, "TEMP")
814	var offset int64	808	err = func() error {
815	for {	809	f, err := os.Create(tempName)
816	ridx, _, err := readEntry(rd, l.opts.LogFormat, true)
817	if err != nil {	810	if err != nil {
818	return err	811	return err
819	}	812	}
820	if ridx == index {	813	defer f.Close()
821	// Seek to exact position of entry, the reader likely overread	814	if _, err := f.Write(ebuf); err != nil {
822	// the file, so we need to back up what has been buffered.	815	return err
823	offset, err = f.Seek(0-int64(rd.Buffered()), 1)
824	if err != nil {
825	return err
826	}
827	found = true
828	// Rewind the segment
829	if _, err := f.Seek(0, 0); err != nil {
830	return err
831	}
832	break
833	}	816	}
834	}	817	if err := f.Sync(); err != nil {
835	if !found {	818	return err
836	return ErrCorrupt	819	}
837	}	820	return f.Close()
838	// Create a temp file in the same log directory and copy all of the data	821	}()
839	// up to the offset.	822	// Rename the TEMP file to it's END file name.
840	tempName := filepath.Join(l.path, "TEMP")	823	endName := filepath.Join(l.path, segmentName(s.index)+".END")
841	ftmp, err := os.Create(tempName)	824	if err = os.Rename(tempName, endName); err != nil {
842	if err != nil {
843	return err	825	return err
844	}	826	}
		827	// The log was truncated but still needs some file cleanup. Any errors
		828	// following this message will not cause an on-disk data ocorruption, but
		829	// may cause an inconsistency with the current program, so we'll return
		830	// ErrCorrupt so the the user can attempt a recover by calling Close()
		831	// followed by Open().
845	defer func() {	832	defer func() {
846	ftmp.Close()	833	if v := recover(); v != nil {
847	os.Remove(tempName)	834	err = ErrCorrupt
		835	l.corrupt = true
		836	}
848	}()	837	}()
849	if _, err := io.CopyN(ftmp, f, offset); err != nil {	838
		839	// Close the tail segment file
		840	if err = l.sfile.Close(); err != nil {
850	return err	841	return err
851	}	842	}
852	if err := f.Close(); err != nil {	843	// Delete truncated segment files
		844	for i := segIdx; i < len(l.segments); i++ {
		845	if err = os.Remove(l.segments[i].path); err != nil {
		846	return err
		847	}
		848	}
		849	// Rename the END file to the final truncated segment name.
		850	newName := filepath.Join(l.path, segmentName(s.index))
		851	if err = os.Rename(endName, newName); err != nil {
		852	return err
		853	}
		854	// Reopen the tail segment file
		855	if l.sfile, err = os.OpenFile(newName, os.O_WRONLY, 0666); err != nil {
853	return err	856	return err
854	}	857	}
855	if err := ftmp.Close(); err != nil {	858	var n int64
		859	n, err = l.sfile.Seek(0, 2)
		860	if err != nil {
856	return err	861	return err
857	}	862	}
858	// Then rename the TEMP file to it's END file name.	863	if n != int64(len(ebuf)) {
859	endName := l.segments[sidx].path + ".END"	864	err = errors.New("invalid seek")
860	if err := os.Rename(tempName, endName); err != nil {
861	return err	865	return err
862	}	866	}
863	// All operations from this point on are syscalls and must succeed,	867	s.path = newName
864	// otherwise we panic.	868	l.segments = append([]*segment{}, l.segments[:segIdx+1]...)
865	// Remove all of the unneeded segments.
866	for i := len(l.segments) - 1; i >= sidx; i-- {
867	must(nil, os.Remove(l.segments[i].path))
868	}
869	// Modify the segments list.
870	l.segments = append([]segment{}, l.segments[:sidx+1]...)
871	// Remove the ".END" suffix from the file name.
872	finalName := endName[:len(endName)-len(".END")]
873	must(nil, os.Rename(endName, finalName))
874	// The last segment has been changed which means we need to reopen the
875	// appendable file and seek to the end.
876	must(nil, l.file.Close())
877	l.file = must(os.OpenFile(finalName, os.O_RDWR, 0666)).(*os.File)
878	l.fileSize = int(must(l.file.Seek(0, 2)).(int64))
879	l.buffer = nil
880	l.lastIndex = index	869	l.lastIndex = index
		870	l.clearCache()
		871	if err = l.loadSegmentEntries(s); err != nil {
		872	return err
		873	}
881	return nil	874	return nil
882	}	875	}
883		876
884	// Sync performs an fsync on the log. This is not necessary when the	877	// Sync performs an fsync on the log. This is not necessary when the
885	// durability is set to High.	878	// NoSync option is set to false.
886	func (l *Log) Sync() {	879	func (l *Log) Sync() error {
887	if len(l.buffer) > 0 {	880	l.mu.Lock()
888	must(l.file.Write(l.buffer))	881	defer l.mu.Unlock()
889	l.buffer = l.buffer[:0]	882	if l.corrupt {
		883	return ErrCorrupt
		884	} else if l.closed {
		885	return ErrClosed
890	}	886	}
891	if l.opts.Durability < High {	887	return l.sfile.Sync()
892	must(nil, l.file.Sync())	888	}
		889
		890	func must(v interface{}, err error) interface{} {
		891	if err != nil {
		892	panic(err)
893	}	893	}
		894	return v
894	}	895	}

wal_test.go+272−75

@@ -1,9 +1,7 @@
1	package wal	1	package wal
2		2
3	import (	3	import (
4	"bufio"
5	"fmt"	4	"fmt"
6	"io"
7	"io/ioutil"	5	"io/ioutil"
8	"math/rand"	6	"math/rand"
9	"os"	7	"os"
@@ -181,13 +179,12 @@ func testLog(t testing.T, opts Options, N int) { · 171 unmodified lines
181	for i := 1; i <= N; i++ {	179	for i := 1; i <= N; i++ {
182	data, err := l.Read(uint64(i))	180	data, err := l.Read(uint64(i))
183	if err != nil {	181	if err != nil {
184	t.Fatalf("error while getting %d", i)	182	t.Fatalf("error while getting %d, err=%s", i, err)
185	}	183	}
186	if string(data) != dataStr(uint64(i)) {	184	if string(data) != dataStr(uint64(i)) {
187	t.Fatalf("expected %s, got %s", dataStr(uint64(i)), data)	185	t.Fatalf("expected %s, got %s", dataStr(uint64(i)), data)
188	}	186	}
189	}	187	}
190
191	// FirstIndex/LastIndex -- check valid first and last indexes	188	// FirstIndex/LastIndex -- check valid first and last indexes
192	n, err = l.FirstIndex()	189	n, err = l.FirstIndex()
193	if err != nil {	190	if err != nil {
@@ -203,7 +200,6 @@ func testLog(t testing.T, opts Options, N int) { · 9 unmodified lines
203	if n != uint64(N) {	200	if n != uint64(N) {
204	t.Fatalf("expected %d, got %d", N, n)	201	t.Fatalf("expected %d, got %d", N, n)
205	}	202	}
206
207	// Write -- add 50 more items	203	// Write -- add 50 more items
208	for i := N + 1; i <= N+50; i++ {	204	for i := N + 1; i <= N+50; i++ {
209	index := uint64(i)	205	index := uint64(i)
@@ -219,7 +215,6 @@ func testLog(t testing.T, opts Options, N int) { · 9 unmodified lines
219	}	215	}
220	}	216	}
221	N += 50	217	N += 50
222
223	// FirstIndex/LastIndex -- check valid first and last indexes	218	// FirstIndex/LastIndex -- check valid first and last indexes
224	n, err = l.FirstIndex()	219	n, err = l.FirstIndex()
225	if err != nil {	220	if err != nil {
@@ -235,7 +230,6 @@ func testLog(t testing.T, opts Options, N int) { · 9 unmodified lines
235	if n != uint64(N) {	230	if n != uint64(N) {
236	t.Fatalf("expected %d, got %d", N, n)	231	t.Fatalf("expected %d, got %d", N, n)
237	}	232	}
238
239	// Batch -- test batch writes	233	// Batch -- test batch writes
240	b := new(Batch)	234	b := new(Batch)
241	b.Write(1, nil)	235	b.Write(1, nil)
@@ -265,7 +259,6 @@ func testLog(t testing.T, opts Options, N int) { · 23 unmodified lines
265	}	259	}
266	N += 10	260	N += 10
267	}	261	}
268
269	// Read -- read back all entries	262	// Read -- read back all entries
270	for i := 1; i <= N; i++ {	263	for i := 1; i <= N; i++ {
271	data, err := l.Read(uint64(i))	264	data, err := l.Read(uint64(i))
@@ -283,7 +276,6 @@ func testLog(t testing.T, opts Options, N int) { · 11 unmodified lines
283	t.Fatal(err)	276	t.Fatal(err)
284	}	277	}
285	N++	278	N++
286
287	// Read -- one random read, so there is an opened reader	279	// Read -- one random read, so there is an opened reader
288	data, err := l.Read(uint64(N / 2))	280	data, err := l.Read(uint64(N / 2))
289	if err != nil {	281	if err != nil {
@@ -299,7 +291,7 @@ func testLog(t testing.T, opts Options, N int) { · 9 unmodified lines
299	if err = l.TruncateFront(index); err != ErrOutOfRange {	291	if err = l.TruncateFront(index); err != ErrOutOfRange {
300	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)	292	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)
301	}	293	}
302	testFirstLast(t, l, uint64(1), uint64(N))	294	testFirstLast(t, l, uint64(1), uint64(N), nil)
303	}	295	}
304		296
305	// TruncateBack -- should fail, out of range	297	// TruncateBack -- should fail, out of range
@@ -307,19 +299,19 @@ func testLog(t testing.T, opts Options, N int) { · 1 unmodified line
307	if err != ErrOutOfRange {	299	if err != ErrOutOfRange {
308	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)	300	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)
309	}	301	}
310	testFirstLast(t, l, uint64(1), uint64(N))	302	testFirstLast(t, l, uint64(1), uint64(N), nil)
311		303
312	// TruncateFront -- Remove no entries	304	// TruncateFront -- Remove no entries
313	if err = l.TruncateFront(1); err != nil {	305	if err = l.TruncateFront(1); err != nil {
314	t.Fatal(err)	306	t.Fatal(err)
315	}	307	}
316	testFirstLast(t, l, uint64(1), uint64(N))	308	testFirstLast(t, l, uint64(1), uint64(N), nil)
317		309
318	// TruncateFront -- Remove first 80 entries	310	// TruncateFront -- Remove first 80 entries
319	if err = l.TruncateFront(81); err != nil {	311	if err = l.TruncateFront(81); err != nil {
320	t.Fatal(err)	312	t.Fatal(err)
321	}	313	}
322	testFirstLast(t, l, uint64(81), uint64(N))	314	testFirstLast(t, l, uint64(81), uint64(N), nil)
323		315
324	// Write -- one entry, so the buffer might be activated	316	// Write -- one entry, so the buffer might be activated
325	err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1))))	317	err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1))))
@@ -327,7 +319,7 @@ func testLog(t testing.T, opts Options, N int) { · 1 unmodified line
327	t.Fatal(err)	319	t.Fatal(err)
328	}	320	}
329	N++	321	N++
330	testFirstLast(t, l, uint64(81), uint64(N))	322	testFirstLast(t, l, uint64(81), uint64(N), nil)
331		323
332	// Read -- one random read, so there is an opened reader	324	// Read -- one random read, so there is an opened reader
333	data, err = l.Read(uint64(N / 2))	325	data, err = l.Read(uint64(N / 2))
@@ -344,21 +336,20 @@ func testLog(t testing.T, opts Options, N int) { · 10 unmodified lines
344	if err = l.TruncateBack(index); err != ErrOutOfRange {	336	if err = l.TruncateBack(index); err != ErrOutOfRange {
345	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)	337	t.Fatalf("expected %v, got %v", ErrOutOfRange, err)
346	}	338	}
347	testFirstLast(t, l, uint64(81), uint64(N))	339	testFirstLast(t, l, uint64(81), uint64(N), nil)
348	}	340	}
349		341
350	// TruncateBack -- Remove no entries	342	// TruncateBack -- Remove no entries
351	if err = l.TruncateBack(uint64(N)); err != nil {	343	if err = l.TruncateBack(uint64(N)); err != nil {
352	t.Fatal(err)	344	t.Fatal(err)
353	}	345	}
354	testFirstLast(t, l, uint64(81), uint64(N))	346	testFirstLast(t, l, uint64(81), uint64(N), nil)
355
356	// TruncateBack -- Remove last 80 entries	347	// TruncateBack -- Remove last 80 entries
357	if err = l.TruncateBack(uint64(N - 80)); err != nil {	348	if err = l.TruncateBack(uint64(N - 80)); err != nil {
358	t.Fatal(err)	349	t.Fatal(err)
359	}	350	}
360	N -= 80	351	N -= 80
361	testFirstLast(t, l, uint64(81), uint64(N))	352	testFirstLast(t, l, uint64(81), uint64(N), nil)
362		353
363	// Read -- read back all entries	354	// Read -- read back all entries
364	for i := 81; i <= N; i++ {	355	for i := 81; i <= N; i++ {
@@ -383,7 +374,7 @@ func testLog(t testing.T, opts Options, N int) { · 18 unmodified lines
383	}	374	}
384	defer l.Close()	375	defer l.Close()
385		376
386	testFirstLast(t, l, uint64(81), uint64(N))	377	testFirstLast(t, l, uint64(81), uint64(N), nil)
387		378
388	// Read -- read back all entries	379	// Read -- read back all entries
389	for i := 81; i <= N; i++ {	380	for i := 81; i <= N; i++ {
@@ -400,7 +391,7 @@ func testLog(t testing.T, opts Options, N int) { · 10 unmodified lines
400	if err = l.TruncateFront(uint64(N)); err != nil {	391	if err = l.TruncateFront(uint64(N)); err != nil {
401	t.Fatal(err)	392	t.Fatal(err)
402	}	393	}
403	testFirstLast(t, l, uint64(N), uint64(N))	394	testFirstLast(t, l, uint64(N), uint64(N), nil)
404		395
405	// Write -- write on entry	396	// Write -- write on entry
406	err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1))))	397	err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1))))
@@ -408,14 +399,14 @@ func testLog(t testing.T, opts Options, N int) { · 1 unmodified line
408	t.Fatal(err)	399	t.Fatal(err)
409	}	400	}
410	N++	401	N++
411	testFirstLast(t, l, uint64(N-1), uint64(N))	402	testFirstLast(t, l, uint64(N-1), uint64(N), nil)
412		403
413	// TruncateBack -- truncate all entries but one	404	// TruncateBack -- truncate all entries but one
414	if err = l.TruncateBack(uint64(N - 1)); err != nil {	405	if err = l.TruncateBack(uint64(N - 1)); err != nil {
415	t.Fatal(err)	406	t.Fatal(err)
416	}	407	}
417	N--	408	N--
418	testFirstLast(t, l, uint64(N), uint64(N))	409	testFirstLast(t, l, uint64(N), uint64(N), nil)
419		410
420	if err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1)))); err != nil {	411	if err = l.Write(uint64(N+1), []byte(dataStr(uint64(N+1)))); err != nil {
421	t.Fatal(err)	412	t.Fatal(err)
@@ -423,11 +414,10 @@ func testLog(t testing.T, opts Options, N int) { · 1 unmodified line
423	N++	414	N++
424		415
425	l.Sync()	416	l.Sync()
426	testFirstLast(t, l, uint64(N-1), uint64(N))	417	testFirstLast(t, l, uint64(N-1), uint64(N), nil)
427
428	}	418	}
429		419
430	func testFirstLast(t testing.T, l Log, expectFirst, expectLast uint64) {	420	func testFirstLast(t testing.T, l Log, expectFirst, expectLast uint64, data func(index uint64) []byte) {
431	t.Helper()	421	t.Helper()
432	fi, err := l.FirstIndex()	422	fi, err := l.FirstIndex()
433	if err != nil {	423	if err != nil {
@@ -440,6 +430,19 @@ func testFirstLast(t testing.T, l Log, expectFirst, expectLast uint64) { · 6 unmodified lines
440	if fi != expectFirst \|\| li != expectLast {	430	if fi != expectFirst \|\| li != expectLast {
441	t.Fatalf("expected %v/%v, got %v/%v", expectFirst, expectLast, fi, li)	431	t.Fatalf("expected %v/%v, got %v/%v", expectFirst, expectLast, fi, li)
442	}	432	}
		433	for i := fi; i <= li; i++ {
		434	dt1, err := l.Read(i)
		435	if err != nil {
		436	t.Fatal(err)
		437	}
		438	if data != nil {
		439	dt2 := data(i)
		440	if string(dt1) != string(dt2) {
		441	t.Fatalf("mismatch '%s' != '%s'", dt2, dt1)
		442	}
		443	}
		444	}
		445
443	}	446	}
444		447
445	func TestLog(t *testing.T) {	448	func TestLog(t *testing.T) {
@@ -449,28 +452,20 @@ func TestLog(t *testing.T) { · 3 unmodified lines
449	t.Run("nil-opts", func(t *testing.T) {	452	t.Run("nil-opts", func(t *testing.T) {
450	testLog(t, nil, 100)	453	testLog(t, nil, 100)
451	})	454	})
452	t.Run("low", func(t *testing.T) {	455	t.Run("no-sync", func(t *testing.T) {
453	t.Run("json", func(t *testing.T) {
454	testLog(t, makeOpts(512, Low, JSON), 100)
455	})
456	t.Run("binary", func(t *testing.T) {
457	testLog(t, makeOpts(512, Low, Binary), 100)
458	})
459	})
460	t.Run("medium", func(t *testing.T) {
461	t.Run("json", func(t *testing.T) {	456	t.Run("json", func(t *testing.T) {
462	testLog(t, makeOpts(512, Medium, JSON), 100)	457	testLog(t, makeOpts(512, true, JSON), 100)
463	})	458	})
464	t.Run("binary", func(t *testing.T) {	459	t.Run("binary", func(t *testing.T) {
465	testLog(t, makeOpts(512, Medium, Binary), 100)	460	testLog(t, makeOpts(512, true, Binary), 100)
466	})	461	})
467	})	462	})
468	t.Run("high", func(t *testing.T) {	463	t.Run("sync", func(t *testing.T) {
469	t.Run("json", func(t *testing.T) {	464	t.Run("json", func(t *testing.T) {
470	testLog(t, makeOpts(512, High, JSON), 100)	465	testLog(t, makeOpts(512, false, JSON), 100)
471	})	466	})
472	t.Run("binary", func(t *testing.T) {	467	t.Run("binary", func(t *testing.T) {
473	testLog(t, makeOpts(512, High, Binary), 100)	468	testLog(t, makeOpts(512, false, Binary), 100)
474	})	469	})
475	})	470	})
476	}	471	}
@@ -522,7 +517,7 @@ func TestOutliers(t *testing.T) { · 45 unmodified lines
522		517
523	t.Run("fail-corrupted-tail-json", func(t *testing.T) {	518	t.Run("fail-corrupted-tail-json", func(t *testing.T) {
524	defer os.RemoveAll("testlog/corrupt-tail")	519	defer os.RemoveAll("testlog/corrupt-tail")
525	opts := makeOpts(512, Medium, JSON)	520	opts := makeOpts(512, true, JSON)
526	os.MkdirAll("testlog/corrupt-tail", 0777)	521	os.MkdirAll("testlog/corrupt-tail", 0777)
527	ioutil.WriteFile(	522	ioutil.WriteFile(
528	"testlog/corrupt-tail/00000000000000000001",	523	"testlog/corrupt-tail/00000000000000000001",
@@ -556,7 +551,7 @@ func TestOutliers(t *testing.T) { · 27 unmodified lines
556		551
557	t.Run("start-marker-file", func(t *testing.T) {	552	t.Run("start-marker-file", func(t *testing.T) {
558	lpath := "testlog/start-marker"	553	lpath := "testlog/start-marker"
559	opts := makeOpts(512, Medium, JSON)	554	opts := makeOpts(512, true, JSON)
560	l := must(Open(lpath, opts)).(*Log)	555	l := must(Open(lpath, opts)).(*Log)
561	defer l.Close()	556	defer l.Close()
562	for i := uint64(1); i <= 100; i++ {	557	for i := uint64(1); i <= 100; i++ {
@@ -569,47 +564,46 @@ func TestOutliers(t *testing.T) { · 6 unmodified lines
569	must(nil, ioutil.WriteFile(path+".START", data, 0666))	564	must(nil, ioutil.WriteFile(path+".START", data, 0666))
570	l = must(Open(lpath, opts)).(*Log)	565	l = must(Open(lpath, opts)).(*Log)
571	defer l.Close()	566	defer l.Close()
572	testFirstLast(t, l, firstIndex, 100)	567	testFirstLast(t, l, firstIndex, 100, nil)
573		568
574	})	569	})
575	t.Run("end-marker-file", func(t *testing.T) {	570	// t.Run("end-marker-file", func(t *testing.T) {
576	lpath := "testlog/end-marker"	571	// lpath := "testlog/end-marker"
577	opts := makeOpts(512, Medium, JSON)	572	// opts := makeOpts(512, true, JSON)
578	l := must(Open(lpath, opts)).(*Log)	573	// l := must(Open(lpath, opts)).(*Log)
579	defer l.Close()	574	// defer l.Close()
580	for i := uint64(1); i <= 100; i++ {	575	// for i := uint64(1); i <= 100; i++ {
581	must(nil, l.Write(i, []byte(dataStr(i))))	576	// must(nil, l.Write(i, []byte(dataStr(i))))
582	}	577	// }
583	path := l.segments[l.findSegment(35)].path	578	// path := l.segments[l.findSegment(35)].path
584	f := must(os.Open(path)).(*os.File)	579	// data := must(ioutil.ReadFile(path)).([]byte)
585	defer f.Close()	580	// var lastIndex uint64
586	rd := bufio.NewReader(f)	581	// for {
587	var lastIndex uint64	582
588	for {	583	// n, _, err := readEntry(rd, JSON, true)
589	n, _, err := readEntry(rd, JSON, true)	584	// if err == io.EOF {
590	if err == io.EOF {	585	// break
591	break	586	// }
592	}	587	// if err != nil {
593	if err != nil {	588	// t.Fatal(err)
594	t.Fatal(err)	589	// }
595	}	590	// lastIndex = n
596	lastIndex = n	591	// }
597	}	592
598		593	// must(nil, l.Close())
599	must(nil, l.Close())	594	// data = must(ioutil.ReadFile(path)).([]byte)
600	data := must(ioutil.ReadFile(path)).([]byte)	595	// must(nil, ioutil.WriteFile(path+".END", data, 0666))
601	must(nil, ioutil.WriteFile(path+".END", data, 0666))	596	// l = must(Open(lpath, opts)).(*Log)
602	l = must(Open(lpath, opts)).(*Log)	597	// defer l.Close()
603	defer l.Close()	598	// testFirstLast(t, l, 1, lastIndex)
604	testFirstLast(t, l, 1, lastIndex)	599	// })
605	})
606		600
607	}	601	}
608		602
609	func makeOpts(segSize int, dur Durability, lf LogFormat) *Options {	603	func makeOpts(segSize int, noSync bool, lf LogFormat) *Options {
610	opts := *DefaultOptions	604	opts := *DefaultOptions
611	opts.SegmentSize = segSize	605	opts.SegmentSize = segSize
612	opts.Durability = dur	606	opts.NoSync = noSync
613	opts.LogFormat = lf	607	opts.LogFormat = lf
614	return &opts	608	return &opts
615	}	609	}
@@ -638,3 +632,206 @@ func TestIssue1(t *testing.T) { · 22 unmodified lines
638	t.Fatal("data mismatch")	632	t.Fatal("data mismatch")
639	}	633	}
640	}	634	}
		635
		636	func TestSimpleTruncateFront(t *testing.T) {
		637	os.RemoveAll("testlog")
		638
		639	opts := &Options{
		640	NoSync: true,
		641	LogFormat: JSON,
		642	SegmentSize: 100,
		643	}
		644
		645	l, err := Open("testlog", opts)
		646	if err != nil {
		647	t.Fatal(err)
		648	}
		649	defer func() {
		650	l.Close()
		651	}()
		652
		653	makeData := func(index uint64) []byte {
		654	return []byte(fmt.Sprintf("data-%d", index))
		655	}
		656
		657	valid := func(t *testing.T, first, last uint64) {
		658	t.Helper()
		659	index, err := l.FirstIndex()
		660	if err != nil {
		661	t.Fatal(err)
		662	}
		663	if index != first {
		664	t.Fatalf("expected %v, got %v", first, index)
		665	}
		666	index, err = l.LastIndex()
		667	if err != nil {
		668	t.Fatal(err)
		669	}
		670	if index != last {
		671	t.Fatalf("expected %v, got %v", last, index)
		672	}
		673	for i := first; i <= last; i++ {
		674	data, err := l.Read(i)
		675	if err != nil {
		676	t.Fatal(err)
		677	}
		678	if string(data) != string(makeData(i)) {
		679	t.Fatalf("expcted '%s', got '%s'", makeData(i), data)
		680	}
		681	}
		682	}
		683	validReopen := func(t *testing.T, first, last uint64) {
		684	t.Helper()
		685	valid(t, first, last)
		686	if err := l.Close(); err != nil {
		687	t.Fatal(err)
		688	}
		689	l, err = Open("testlog", opts)
		690	if err != nil {
		691	t.Fatal(err)
		692	}
		693	valid(t, first, last)
		694	}
		695	for i := 1; i <= 100; i++ {
		696	err := l.Write(uint64(i), makeData(uint64(i)))
		697	if err != nil {
		698	t.Fatal(err)
		699	}
		700	}
		701	validReopen(t, 1, 100)
		702
		703	if err := l.TruncateFront(1); err != nil {
		704	t.Fatal(err)
		705	}
		706	validReopen(t, 1, 100)
		707
		708	if err := l.TruncateFront(2); err != nil {
		709	t.Fatal(err)
		710	}
		711	validReopen(t, 2, 100)
		712
		713	if err := l.TruncateFront(4); err != nil {
		714	t.Fatal(err)
		715	}
		716	validReopen(t, 4, 100)
		717
		718	if err := l.TruncateFront(5); err != nil {
		719	t.Fatal(err)
		720	}
		721	validReopen(t, 5, 100)
		722
		723	if err := l.TruncateFront(99); err != nil {
		724	t.Fatal(err)
		725	}
		726	validReopen(t, 99, 100)
		727
		728	if err := l.TruncateFront(100); err != nil {
		729	t.Fatal(err)
		730	}
		731	validReopen(t, 100, 100)
		732
		733	}
		734
		735	func TestSimpleTruncateBack(t *testing.T) {
		736	os.RemoveAll("testlog")
		737
		738	opts := &Options{
		739	NoSync: true,
		740	LogFormat: JSON,
		741	SegmentSize: 100,
		742	}
		743
		744	l, err := Open("testlog", opts)
		745	if err != nil {
		746	t.Fatal(err)
		747	}
		748	defer func() {
		749	l.Close()
		750	}()
		751
		752	makeData := func(index uint64) []byte {
		753	return []byte(fmt.Sprintf("data-%d", index))
		754	}
		755
		756	valid := func(t *testing.T, first, last uint64) {
		757	t.Helper()
		758	index, err := l.FirstIndex()
		759	if err != nil {
		760	t.Fatal(err)
		761	}
		762	if index != first {
		763	t.Fatalf("expected %v, got %v", first, index)
		764	}
		765	index, err = l.LastIndex()
		766	if err != nil {
		767	t.Fatal(err)
		768	}
		769	if index != last {
		770	t.Fatalf("expected %v, got %v", last, index)
		771	}
		772	for i := first; i <= last; i++ {
		773	data, err := l.Read(i)
		774	if err != nil {
		775	t.Fatal(err)
		776	}
		777	if string(data) != string(makeData(i)) {
		778	t.Fatalf("expcted '%s', got '%s'", makeData(i), data)
		779	}
		780	}
		781	}
		782	validReopen := func(t *testing.T, first, last uint64) {
		783	t.Helper()
		784	valid(t, first, last)
		785	if err := l.Close(); err != nil {
		786	t.Fatal(err)
		787	}
		788	l, err = Open("testlog", opts)
		789	if err != nil {
		790	t.Fatal(err)
		791	}
		792	valid(t, first, last)
		793	}
		794	for i := 1; i <= 100; i++ {
		795	err := l.Write(uint64(i), makeData(uint64(i)))
		796	if err != nil {
		797	t.Fatal(err)
		798	}
		799	}
		800	validReopen(t, 1, 100)
		801
		802	/////////////////////////////////////////////////////////////
		803	if err := l.TruncateBack(100); err != nil {
		804	t.Fatal(err)
		805	}
		806	validReopen(t, 1, 100)
		807	if err := l.Write(101, makeData(101)); err != nil {
		808	t.Fatal(err)
		809	}
		810	validReopen(t, 1, 101)
		811
		812	/////////////////////////////////////////////////////////////
		813	if err := l.TruncateBack(99); err != nil {
		814	t.Fatal(err)
		815	}
		816	validReopen(t, 1, 99)
		817	if err := l.Write(100, makeData(100)); err != nil {
		818	t.Fatal(err)
		819	}
		820	validReopen(t, 1, 100)
		821
		822	if err := l.TruncateBack(94); err != nil {
		823	t.Fatal(err)
		824	}
		825	validReopen(t, 1, 94)
		826
		827	if err := l.TruncateBack(93); err != nil {
		828	t.Fatal(err)
		829	}
		830	validReopen(t, 1, 93)
		831
		832	if err := l.TruncateBack(92); err != nil {
		833	t.Fatal(err)
		834	}
		835	validReopen(t, 1, 92)
		836
		837	}

Optimized reads

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code