func (c *Cache) updateCachedBytes(b uint64) {
	atomic.AddInt64(&c.stats.CachedBytes, int64(b))
}
updateCachedBytes increases the cachedBytes counter by b.
package tsm1
import (
	"fmt"
	"math"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/tsdb"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
const ringShards = 16
var (
	ErrSnapshotInProgress = fmt.Errorf("snapshot in progress")
)
func ErrCacheMemorySizeLimitExceeded(n, limit uint64) error {
	return fmt.Errorf("cache-max-memory-size exceeded: (%d/%d)", n, limit)
}
type entry struct {
	mu     sync.RWMutex
	values Values  
	vtype byte
}
func newEntryValues(values []Value) (*entry, error) {
	e := &entry{}
	e.values = make(Values, 0, len(values))
	e.values = append(e.values, values...)
	if len(values) == 0 {
		return e, nil
	}
	et := valueType(values[0])
	for _, v := range values {
		if et != valueType(v) {
			return nil, tsdb.ErrFieldTypeConflict
		}
	}
	e.vtype = et
	return e, nil
}
func (e *entry) add(values []Value) error {
	if len(values) == 0 {
		return nil  
	}
	if e.vtype != 0 {
		for _, v := range values {
			if e.vtype != valueType(v) {
				return tsdb.ErrFieldTypeConflict
			}
		}
	}
	e.mu.Lock()
	if len(e.values) == 0 {
		e.values = values
		e.vtype = valueType(values[0])
		e.mu.Unlock()
		return nil
	}
	e.values = append(e.values, values...)
	e.mu.Unlock()
	return nil
}
func (e *entry) deduplicate() {
	e.mu.Lock()
	defer e.mu.Unlock()
	if len(e.values) <= 1 {
		return
	}
	e.values = e.values.Deduplicate()
}
func (e *entry) count() int {
	e.mu.RLock()
	n := len(e.values)
	e.mu.RUnlock()
	return n
}
func (e *entry) filter(min, max int64) {
	e.mu.Lock()
	if len(e.values) > 1 {
		e.values = e.values.Deduplicate()
	}
	e.values = e.values.Exclude(min, max)
	e.mu.Unlock()
}
func (e *entry) size() int {
	e.mu.RLock()
	sz := e.values.Size()
	e.mu.RUnlock()
	return sz
}
func (e *entry) InfluxQLType() (influxql.DataType, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()
	return e.values.InfluxQLType()
}
const (
	statCacheMemoryBytes = "memBytes"       
	statCacheDiskBytes   = "diskBytes"      
	statSnapshots        = "snapshotCount"  
	statCacheAgeMs       = "cacheAgeMs"     
	statCachedBytes         = "cachedBytes"          
	statWALCompactionTimeMs = "WALCompactionTimeMs"  
	statCacheWriteOK      = "writeOk"
	statCacheWriteErr     = "writeErr"
	statCacheWriteDropped = "writeDropped"
)
type storer interface {
	entry(key []byte) *entry                         
	write(key []byte, values Values) (bool, error)   
	add(key []byte, entry *entry)                    
	remove(key []byte)                               
	keys(sorted bool) [][]byte                       
	apply(f func([]byte, *entry) error) error        
	applySerial(f func([]byte, *entry) error) error  
	reset()                                          
	split(n int) []storer                            
	count() int                                      
}
type Cache struct {
	size         uint64
	snapshotSize uint64
	mu      sync.RWMutex
	store   storer
	maxSize uint64
	snapshot     *Cache
	snapshotting bool
	snapshotAttempts int
	stats        *CacheStatistics
	lastSnapshot time.Time
	initialize       atomic.Value
	initializedCount uint32
}
func NewCache(maxSize uint64, path string) *Cache {
	c := &Cache{
		maxSize:      maxSize,
		store:        emptyStore{},
		stats:        &CacheStatistics{},
		lastSnapshot: time.Now(),
	}
	c.initialize.Store(&sync.Once{})
	c.UpdateAge()
	c.UpdateCompactTime(0)
	c.updateCachedBytes(0)
	c.updateMemSize(0)
	c.updateSnapshots()
	return c
}
type CacheStatistics struct {
	MemSizeBytes        int64
	DiskSizeBytes       int64
	SnapshotCount       int64
	CacheAgeMs          int64
	CachedBytes         int64
	WALCompactionTimeMs int64
	WriteOK             int64
	WriteErr            int64
	WriteDropped        int64
}
func (c *Cache) Statistics(tags map[string]string) []models.Statistic {
	return []models.Statistic{{
		Name: "tsm1_cache",
		Tags: tags,
		Values: map[string]interface{}{
			statCacheMemoryBytes:    atomic.LoadInt64(&c.stats.MemSizeBytes),
			statCacheDiskBytes:      atomic.LoadInt64(&c.stats.DiskSizeBytes),
			statSnapshots:           atomic.LoadInt64(&c.stats.SnapshotCount),
			statCacheAgeMs:          atomic.LoadInt64(&c.stats.CacheAgeMs),
			statCachedBytes:         atomic.LoadInt64(&c.stats.CachedBytes),
			statWALCompactionTimeMs: atomic.LoadInt64(&c.stats.WALCompactionTimeMs),
			statCacheWriteOK:        atomic.LoadInt64(&c.stats.WriteOK),
			statCacheWriteErr:       atomic.LoadInt64(&c.stats.WriteErr),
			statCacheWriteDropped:   atomic.LoadInt64(&c.stats.WriteDropped),
		},
	}}
}
func (c *Cache) init() {
	if !atomic.CompareAndSwapUint32(&c.initializedCount, 0, 1) {
		return
	}
	c.mu.Lock()
	c.store, _ = newring(ringShards)
	c.mu.Unlock()
}
func (c *Cache) Free() {
	if !atomic.CompareAndSwapUint32(&c.initializedCount, 1, 0) {
		return
	}
	c.mu.Lock()
	c.store = emptyStore{}
	c.mu.Unlock()
}
func (c *Cache) Write(key []byte, values []Value) error {
	c.init()
	addedSize := uint64(Values(values).Size())
	limit := c.maxSize
	n := c.Size() + addedSize
	if limit > 0 && n > limit {
		atomic.AddInt64(&c.stats.WriteErr, 1)
		return ErrCacheMemorySizeLimitExceeded(n, limit)
	}
	newKey, err := c.store.write(key, values)
	if err != nil {
		atomic.AddInt64(&c.stats.WriteErr, 1)
		return err
	}
	if newKey {
		addedSize += uint64(len(key))
	}
	c.increaseSize(addedSize)
	c.updateMemSize(int64(addedSize))
	atomic.AddInt64(&c.stats.WriteOK, 1)
	return nil
}
func (c *Cache) WriteMulti(values map[string][]Value) error {
	c.init()
	var addedSize uint64
	for _, v := range values {
		addedSize += uint64(Values(v).Size())
	}
	limit := c.maxSize  
	n := c.Size() + addedSize
	if limit > 0 && n > limit {
		atomic.AddInt64(&c.stats.WriteErr, 1)
		return ErrCacheMemorySizeLimitExceeded(n, limit)
	}
	var werr error
	c.mu.RLock()
	store := c.store
	c.mu.RUnlock()
	c.increaseSize(addedSize)
	for k, v := range values {
		newKey, err := store.write([]byte(k), v)
		if err != nil {
			werr = err
			addedSize -= uint64(Values(v).Size())
			c.decreaseSize(uint64(Values(v).Size()))
		}
		if newKey {
			addedSize += uint64(len(k))
			c.increaseSize(uint64(len(k)))
		}
	}
	if werr != nil {
		atomic.AddInt64(&c.stats.WriteDropped, 1)
		atomic.AddInt64(&c.stats.WriteErr, 1)
	}
	c.updateMemSize(int64(addedSize))
	atomic.AddInt64(&c.stats.WriteOK, 1)
	return werr
}
func (c *Cache) Snapshot() (*Cache, error) {
	c.init()
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.snapshotting {
		return nil, ErrSnapshotInProgress
	}
	c.snapshotting = true
	c.snapshotAttempts++  
	if c.snapshot == nil {
		store, err := newring(ringShards)
		if err != nil {
			return nil, err
		}
		c.snapshot = &Cache{
			store: store,
		}
	}
	if c.snapshot.Size() > 0 {
		return c.snapshot, nil
	}
	c.snapshot.store, c.store = c.store, c.snapshot.store
	snapshotSize := c.Size()
	atomic.StoreUint64(&c.snapshot.size, snapshotSize)
	atomic.StoreUint64(&c.snapshotSize, snapshotSize)
	c.store.reset()
	atomic.StoreUint64(&c.size, 0)
	c.lastSnapshot = time.Now()
	c.updateCachedBytes(snapshotSize)  
	c.updateSnapshots()
	return c.snapshot, nil
}
func (c *Cache) Deduplicate() {
	c.mu.RLock()
	store := c.store
	c.mu.RUnlock()
	_ = store.apply(func(_ []byte, e *entry) error { e.deduplicate(); return nil })
}
func (c *Cache) ClearSnapshot(success bool) {
	c.init()
	c.mu.RLock()
	snapStore := c.snapshot.store
	c.mu.RUnlock()
	if success {
		snapStore.reset()
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.snapshotting = false
	if success {
		c.snapshotAttempts = 0
		c.updateMemSize(-int64(atomic.LoadUint64(&c.snapshotSize)))  
		c.snapshot = &Cache{
			store: c.snapshot.store,
		}
		atomic.StoreUint64(&c.snapshotSize, 0)
		c.updateSnapshots()
	}
}
func (c *Cache) Size() uint64 {
	return atomic.LoadUint64(&c.size) + atomic.LoadUint64(&c.snapshotSize)
}
func (c *Cache) increaseSize(delta uint64) {
	atomic.AddUint64(&c.size, delta)
}
func (c *Cache) decreaseSize(delta uint64) {
	atomic.AddUint64(&c.size, ^(delta - 1))
}
func (c *Cache) MaxSize() uint64 {
	return c.maxSize
}
func (c *Cache) Count() int {
	c.mu.RLock()
	n := c.store.count()
	c.mu.RUnlock()
	return n
}
func (c *Cache) Keys() [][]byte {
	c.mu.RLock()
	store := c.store
	c.mu.RUnlock()
	return store.keys(true)
}
func (c *Cache) Split(n int) []*Cache {
	if n == 1 {
		return []*Cache{c}
	}
	caches := make([]*Cache, n)
	storers := c.store.split(n)
	for i := 0; i < n; i++ {
		caches[i] = &Cache{
			store: storers[i],
		}
	}
	return caches
}
func (c *Cache) Values(key []byte) Values {
	var snapshotEntries *entry
	c.mu.RLock()
	e := c.store.entry(key)
	if c.snapshot != nil {
		snapshotEntries = c.snapshot.store.entry(key)
	}
	c.mu.RUnlock()
	if e == nil {
		if snapshotEntries == nil {
			return nil
		}
	} else {
		e.deduplicate()
	}
	var entries []*entry
	sz := 0
	if snapshotEntries != nil {
		snapshotEntries.deduplicate()  
		entries = append(entries, snapshotEntries)
		sz += snapshotEntries.count()
	}
	if e != nil {
		entries = append(entries, e)
		sz += e.count()
	}
	if sz == 0 {
		return nil
	}
	values := make(Values, sz)
	n := 0
	for _, e := range entries {
		e.mu.RLock()
		n += copy(values[n:], e.values)
		e.mu.RUnlock()
	}
	values = values[:n]
	values = values.Deduplicate()
	return values
}
func (c *Cache) Delete(keys [][]byte) {
	c.DeleteRange(keys, math.MinInt64, math.MaxInt64)
}
func (c *Cache) DeleteRange(keys [][]byte, min, max int64) {
	c.init()
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, k := range keys {
		e := c.store.entry(k)
		if e == nil {
			continue
		}
		origSize := uint64(e.size())
		if min == math.MinInt64 && max == math.MaxInt64 {
			c.decreaseSize(origSize + uint64(len(k)))
			c.store.remove(k)
			continue
		}
		e.filter(min, max)
		if e.count() == 0 {
			c.store.remove(k)
			c.decreaseSize(origSize + uint64(len(k)))
			continue
		}
		c.decreaseSize(origSize - uint64(e.size()))
	}
	atomic.StoreInt64(&c.stats.MemSizeBytes, int64(c.Size()))
}
func (c *Cache) SetMaxSize(size uint64) {
	c.mu.Lock()
	c.maxSize = size
	c.mu.Unlock()
}
func (c *Cache) values(key []byte) Values {
	e := c.store.entry(key)
	if e == nil {
		return nil
	}
	e.mu.RLock()
	v := e.values
	e.mu.RUnlock()
	return v
}
func (c *Cache) ApplyEntryFn(f func(key []byte, entry *entry) error) error {
	c.mu.RLock()
	store := c.store
	c.mu.RUnlock()
	return store.applySerial(f)
}
type CacheLoader struct {
	files []string
	Logger *zap.Logger
}
func NewCacheLoader(files []string) *CacheLoader {
	return &CacheLoader{
		files:  files,
		Logger: zap.NewNop(),
	}
}
func (cl *CacheLoader) Load(cache *Cache) error {
	var r *WALSegmentReader
	for _, fn := range cl.files {
		if err := func() error {
			f, err := os.OpenFile(fn, os.O_CREATE|os.O_RDWR, 0666)
			if err != nil {
				return err
			}
			defer f.Close()
			stat, err := os.Stat(f.Name())
			if err != nil {
				return err
			}
			cl.Logger.Info("Reading file", zap.String("path", f.Name()), zap.Int64("size", stat.Size()))
			if stat.Size() == 0 {
				return nil
			}
			if r == nil {
				r = NewWALSegmentReader(f)
				defer r.Close()
			} else {
				r.Reset(f)
			}
			for r.Next() {
				entry, err := r.Read()
				if err != nil {
					n := r.Count()
					cl.Logger.Info("File corrupt", zap.String("path", f.Name()), zap.Int64("pos", n))
					if err := f.Truncate(n); err != nil {
						return err
					}
					break
				}
				switch t := entry.(type) {
				case *WriteWALEntry:
					if err := cache.WriteMulti(t.Values); err != nil {
						return err
					}
				case *DeleteRangeWALEntry:
					cache.DeleteRange(t.Keys, t.Min, t.Max)
				case *DeleteWALEntry:
					cache.Delete(t.Keys)
				}
			}
			return r.Close()
		}(); err != nil {
			return err
		}
	}
	return nil
}
func (cl *CacheLoader) WithLogger(log *zap.Logger) {
	cl.Logger = log.With(zap.String("service", "cacheloader"))
}
func (c *Cache) UpdateAge() {
	c.mu.RLock()
	defer c.mu.RUnlock()
	ageStat := int64(time.Since(c.lastSnapshot) / time.Millisecond)
	atomic.StoreInt64(&c.stats.CacheAgeMs, ageStat)
}
func (c *Cache) UpdateCompactTime(d time.Duration) {
	atomic.AddInt64(&c.stats.WALCompactionTimeMs, int64(d/time.Millisecond))
}