package wal import (	"bytes"	"encoding/base64"	"encoding/binary"	"encoding/json"	"errors"	"fmt"	"io/ioutil"	"os"	"path/filepath"	"strconv"	"strings"	"sync"	"unicode/utf8"	"unsafe" 	"github.com/tidwall/gjson"	"github.com/tidwall/tinylru") var (	// ErrCorrupt is returns when the log is corrupt.	ErrCorrupt = errors.New("log corrupt") 	// ErrClosed is returned when an operation cannot be completed because	// the log is closed.	ErrClosed = errors.New("log closed") 	// ErrNotFound is returned when an entry is not found.	ErrNotFound = errors.New("not found") 	// ErrOutOfOrder is returned from Write() when the index is not equal to	// LastIndex()+1. It's required that log monotonically grows by one and has	// no gaps. Thus, the series 10,11,12,13,14 is valid, but 10,11,13,14 is	// not because there's a gap between 11 and 13. Also, 10,12,11,13 is not	// valid because 12 and 11 are out of order.	ErrOutOfOrder = errors.New("out of order") 	// ErrOutOfRange is returned from TruncateFront() and TruncateBack() when	// the index not in the range of the log's first and last index. Or, this	// may be returned when the caller is attempting to remove *all* entries;	// The log requires that at least one entry exists following a truncate.	ErrOutOfRange = errors.New("out of range")) // LogFormat is the format of the log files.type LogFormat byte const (	// Binary format writes entries in binary. This is the default and, unless	// a good reason otherwise, should be used in production.	Binary LogFormat = 0	// JSON format writes entries as JSON lines. This causes larger, human	// readable files.	JSON LogFormat = 1) // Options for Logtype Options struct {	// NoSync disables fsync after writes. This is less durable and puts the	// log at risk of data loss when there's a server crash.	NoSync bool	// SegmentSize of each segment. This is just a target value, actual size	// may differ. Default is 20 MB.	SegmentSize int	// LogFormat is the format of the log files. Default is Binary.	LogFormat LogFormat	// SegmentCacheSize is the maximum number of segments that will be held in	// memory for caching. Increasing this value may enhance performance for	// concurrent read operations. Default is 1	SegmentCacheSize int	// NoCopy allows for the Read() operation to return the raw underlying data	// slice. This is an optimization to help minimize allocations. When this	// option is set, do not modify the returned data because it may affect	// other Read calls. Default false	NoCopy bool	// Perms represents the datafiles modes and permission bits	DirPerms  os.FileMode	FilePerms os.FileMode} // DefaultOptions for Open().var DefaultOptions = &Options{	NoSync:           false,    // Fsync after every write	SegmentSize:      20971520, // 20 MB log segment files.	LogFormat:        Binary,   // Binary format is small and fast.	SegmentCacheSize: 2,        // Number of cached in-memory segments	NoCopy:           false,    // Make a new copy of data for every Read call.	DirPerms:         0750,     // Permissions for the created directories	FilePerms:        0640,     // Permissions for the created data files} // Log represents a write ahead logtype Log struct {	mu         sync.RWMutex	path       string      // absolute path to log directory	opts       Options     // log options	closed     bool        // log is closed	corrupt    bool        // log may be corrupt	segments   []*segment  // all known log segments	firstIndex uint64      // index of the first entry in log	lastIndex  uint64      // index of the last entry in log	sfile      *os.File    // tail segment file handle	wbatch     Batch       // reusable write batch	scache     tinylru.LRU // segment entries cache} // segment represents a single segment file.type segment struct {	path  string // path of segment file	index uint64 // first index of segment	ebuf  []byte // cached entries buffer	epos  []bpos // cached entries positions in buffer} type bpos struct {	pos int // byte position	end int // one byte past pos} // Open a new write ahead logfunc Open(path string, opts *Options) (*Log, error) {	if opts == nil {		opts = DefaultOptions	}	if opts.SegmentCacheSize <= 0 {		opts.SegmentCacheSize = DefaultOptions.SegmentCacheSize	}	if opts.SegmentSize <= 0 {		opts.SegmentSize = DefaultOptions.SegmentSize	}	if opts.DirPerms == 0 {		opts.DirPerms = DefaultOptions.DirPerms	}	if opts.FilePerms == 0 {		opts.FilePerms = DefaultOptions.FilePerms	} 	var err error	path, err = abs(path)	if err != nil {		return nil, err	}	l := &Log{path: path, opts: *opts}	l.scache.Resize(l.opts.SegmentCacheSize)	if err := os.MkdirAll(path, l.opts.DirPerms); err != nil {		return nil, err	}	if err := l.load(); err != nil {		return nil, err	}	return l, nil} func abs(path string) (string, error) {	if path == ":memory:" {		return "", errors.New("in-memory log not supported")	}	return filepath.Abs(path)} func (l *Log) pushCache(segIdx int) {	_, _, _, v, evicted :=		l.scache.SetEvicted(segIdx, l.segments[segIdx])	if evicted {		s := v.(*segment)		s.ebuf = nil		s.epos = nil	}} // load all the segments. This operation also cleans up any START/END segments.func (l *Log) load() error {	fis, err := ioutil.ReadDir(l.path)	if err != nil {		return err	}	startIdx := -1	endIdx := -1	for _, fi := range fis {		name := fi.Name()		if fi.IsDir() || len(name) < 20 {			continue		}		index, err := strconv.ParseUint(name[:20], 10, 64)		if err != nil || index == 0 {			continue		}		isStart := len(name) == 26 && strings.HasSuffix(name, ".START")		isEnd := len(name) == 24 && strings.HasSuffix(name, ".END")		if len(name) == 20 || isStart || isEnd {			if isStart {				startIdx = len(l.segments)			} else if isEnd && endIdx == -1 {				endIdx = len(l.segments)			}			l.segments = append(l.segments, &segment{				index: index,				path:  filepath.Join(l.path, name),			})		}	}	if len(l.segments) == 0 {		// Create a new log		l.segments = append(l.segments, &segment{			index: 1,			path:  filepath.Join(l.path, segmentName(1)),		})		l.firstIndex = 1		l.lastIndex = 0		l.sfile, err = os.OpenFile(l.segments[0].path, os.O_CREATE|os.O_RDWR|os.O_TRUNC, l.opts.FilePerms)		return err	}	// Open existing log. Clean up log if START of END segments exists.	if startIdx != -1 {		if endIdx != -1 {			// There should not be a START and END at the same time			return ErrCorrupt		}		// Delete all files leading up to START		for i := 0; i < startIdx; i++ {			if err := os.Remove(l.segments[i].path); err != nil {				return err			}		}		l.segments = append([]*segment{}, l.segments[startIdx:]...)		// Rename the START segment		orgPath := l.segments[0].path		finalPath := orgPath[:len(orgPath)-len(".START")]		err := os.Rename(orgPath, finalPath)		if err != nil {			return err		}		l.segments[0].path = finalPath	}	if endIdx != -1 {		// Delete all files following END		for i := len(l.segments) - 1; i > endIdx; i-- {			if err := os.Remove(l.segments[i].path); err != nil {				return err			}		}		l.segments = append([]*segment{}, l.segments[:endIdx+1]...)		if len(l.segments) > 1 && l.segments[len(l.segments)-2].index ==			l.segments[len(l.segments)-1].index {			// remove the segment prior to the END segment because it shares			// the same starting index.			l.segments[len(l.segments)-2] = l.segments[len(l.segments)-1]			l.segments = l.segments[:len(l.segments)-1]		}		// Rename the END segment		orgPath := l.segments[len(l.segments)-1].path		finalPath := orgPath[:len(orgPath)-len(".END")]		err := os.Rename(orgPath, finalPath)		if err != nil {			return err		}		l.segments[len(l.segments)-1].path = finalPath	}	l.firstIndex = l.segments[0].index	// Open the last segment for appending	lseg := l.segments[len(l.segments)-1]	l.sfile, err = os.OpenFile(lseg.path, os.O_WRONLY, l.opts.FilePerms)	if err != nil {		return err	}	if _, err := l.sfile.Seek(0, 2); err != nil {		return err	}	// Load the last segment entries	if err := l.loadSegmentEntries(lseg); err != nil {		return err	}	l.lastIndex = lseg.index + uint64(len(lseg.epos)) - 1	return nil} // segmentName returns a 20-byte textual representation of an index// for lexical ordering. This is used for the file names of log segments.func segmentName(index uint64) string {	return fmt.Sprintf("%020d", index)} // Close the log.func (l *Log) Close() error {	l.mu.Lock()	defer l.mu.Unlock()	if l.closed {		if l.corrupt {			return ErrCorrupt		}		return ErrClosed	}	if err := l.sfile.Sync(); err != nil {		return err	}	if err := l.sfile.Close(); err != nil {		return err	}	l.closed = true	if l.corrupt {		return ErrCorrupt	}	return nil} // Write an entry to the log.func (l *Log) Write(index uint64, data []byte) error {	l.mu.Lock()	defer l.mu.Unlock()	if l.corrupt {		return ErrCorrupt	} else if l.closed {		return ErrClosed	}	l.wbatch.Clear()	l.wbatch.Write(index, data)	return l.writeBatch(&l.wbatch)} func (l *Log) appendEntry(dst []byte, index uint64, data []byte) (out []byte,	epos bpos) {	if l.opts.LogFormat == JSON {		return appendJSONEntry(dst, index, data)	}	return appendBinaryEntry(dst, data)} // Cycle the old segment for a new segment.func (l *Log) cycle() error {	if err := l.sfile.Sync(); err != nil {		return err	}	if err := l.sfile.Close(); err != nil {		return err	}	// cache the previous segment	l.pushCache(len(l.segments) - 1)	s := &segment{		index: l.lastIndex + 1,		path:  filepath.Join(l.path, segmentName(l.lastIndex+1)),	}	var err error	l.sfile, err = os.OpenFile(s.path, os.O_CREATE|os.O_RDWR|os.O_TRUNC, l.opts.FilePerms)	if err != nil {		return err	}	l.segments = append(l.segments, s)	return nil} func appendJSONEntry(dst []byte, index uint64, data []byte) (out []byte,	epos bpos) {	// {"index":number,"data":string}	pos := len(dst)	dst = append(dst, `{"index":"`...)	dst = strconv.AppendUint(dst, index, 10)	dst = append(dst, `","data":`...)	dst = appendJSONData(dst, data)	dst = append(dst, '}', '\n')	return dst, bpos{pos, len(dst)}} func appendJSONData(dst []byte, s []byte) []byte {	if utf8.Valid(s) {		b, _ := json.Marshal(*(*string)(unsafe.Pointer(&s)))		dst = append(dst, '"', '+')		return append(dst, b[1:]...)	}	dst = append(dst, '"', '$')	dst = append(dst, base64.URLEncoding.EncodeToString(s)...)	return append(dst, '"')} func appendBinaryEntry(dst []byte, data []byte) (out []byte, epos bpos) {	// data_size + data	pos := len(dst)	dst = appendUvarint(dst, uint64(len(data)))	dst = append(dst, data...)	return dst, bpos{pos, len(dst)}} func appendUvarint(dst []byte, x uint64) []byte {	var buf [10]byte	n := binary.PutUvarint(buf[:], x)	dst = append(dst, buf[:n]...)	return dst} // Batch of entries. Used to write multiple entries at once using WriteBatch().type Batch struct {	entries []batchEntry	datas   []byte} type batchEntry struct {	index uint64	size  int} // Write an entry to the batchfunc (b *Batch) Write(index uint64, data []byte) {	b.entries = append(b.entries, batchEntry{index, len(data)})	b.datas = append(b.datas, data...)} // Clear the batch for reuse.func (b *Batch) Clear() {	b.entries = b.entries[:0]	b.datas = b.datas[:0]} // WriteBatch writes the entries in the batch to the log in the order that they// were added to the batch. The batch is cleared upon a successful return.func (l *Log) WriteBatch(b *Batch) error {	l.mu.Lock()	defer l.mu.Unlock()	if l.corrupt {		return ErrCorrupt	} else if l.closed {		return ErrClosed	}	if len(b.entries) == 0 {		return nil	}	return l.writeBatch(b)} func (l *Log) writeBatch(b *Batch) error {	// check that all indexes in batch are sane	for i := 0; i < len(b.entries); i++ {		if b.entries[i].index != l.lastIndex+uint64(i+1) {			return ErrOutOfOrder		}	}	// load the tail segment	s := l.segments[len(l.segments)-1]	if len(s.ebuf) > l.opts.SegmentSize {		// tail segment has reached capacity. Close it and create a new one.		if err := l.cycle(); err != nil {			return err		}		s = l.segments[len(l.segments)-1]	} 	mark := len(s.ebuf)	datas := b.datas	for i := 0; i < len(b.entries); i++ {		data := datas[:b.entries[i].size]		var epos bpos		s.ebuf, epos = l.appendEntry(s.ebuf, b.entries[i].index, data)		s.epos = append(s.epos, epos)		if len(s.ebuf) >= l.opts.SegmentSize {			// segment has reached capacity, cycle now			if _, err := l.sfile.Write(s.ebuf[mark:]); err != nil {				return err			}			l.lastIndex = b.entries[i].index			if err := l.cycle(); err != nil {				return err			}			s = l.segments[len(l.segments)-1]			mark = 0		}		datas = datas[b.entries[i].size:]	}	if len(s.ebuf)-mark > 0 {		if _, err := l.sfile.Write(s.ebuf[mark:]); err != nil {			return err		}		l.lastIndex = b.entries[len(b.entries)-1].index	}	if !l.opts.NoSync {		if err := l.sfile.Sync(); err != nil {			return err		}	}	b.Clear()	return nil} // FirstIndex returns the index of the first entry in the log. Returns zero// when log has no entries.func (l *Log) FirstIndex() (index uint64, err error) {	l.mu.RLock()	defer l.mu.RUnlock()	if l.corrupt {		return 0, ErrCorrupt	} else if l.closed {		return 0, ErrClosed	}	// We check the lastIndex for zero because the firstIndex is always one or	// more, even when there's no entries	if l.lastIndex == 0 {		return 0, nil	}	return l.firstIndex, nil}

Lines 1–500 of 917. Load more