deployment/docker: switch Go builder from go1.12.8 to go1.12.9

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/150

.travis.yml

@@ -13,7 +13,7 @@ before_install:
   - GO111MODULE=off go get -u github.com/kisielk/errcheck
 
 script:
-  - make check_all
+  - make check-all
   - git diff --exit-code
   - make test-full
   - make test-pure

Makefile

@@ -50,7 +50,7 @@ errcheck: install-errcheck
 install-errcheck:
 	which errcheck || GO111MODULE=off go get -u github.com/kisielk/errcheck
 
-check_all: fmt vet lint errcheck golangci-lint
+check-all: fmt vet lint errcheck golangci-lint
 
 test:
 	GO111MODULE=on go test -tags=integration -mod=vendor ./lib/... ./app/...

README.md

@@ -124,7 +124,7 @@ remote_write:
   - url: http://<victoriametrics-addr>:8428/api/v1/write
     queue_config:
       max_samples_per_send: 10000
-      max_shards: 100
+      max_shards: 30
 ```
 
 Substitute `<victoriametrics-addr>` with the hostname or IP address of VictoriaMetrics.
@@ -221,8 +221,8 @@ foo,tag1=value1,tag2=value2 field1=12,field2=40
 is converted into the following Prometheus data points:
 
 ```
-foo.field1{tag1="value1", tag2="value2"} 12
-foo.field2{tag1="value1", tag2="value2"} 40
+foo_field1{tag1="value1", tag2="value2"} 12
+foo_field2{tag1="value1", tag2="value2"} 40
 ```
 
 Example for writing data with [Influx line protocol](https://docs.influxdata.com/influxdb/v1.7/write_protocols/line_protocol_tutorial/)
@@ -246,6 +246,9 @@ The `/api/v1/export` endpoint should return the following response:
 {"metric":{"__name__":"measurement.field2","tag1":"value1","tag2":"value2"},"values":[1.23],"timestamps":[1560272508147]}
 ```
 
+Note that Influx line protocol expects [timestamps in *nanoseconds* by default](https://docs.influxdata.com/influxdb/v1.7/write_protocols/line_protocol_tutorial/#timestamp),
+while VictoriaMetrics stores them with *milliseconds* precision.
+
 
 ### How to send data from Graphite-compatible agents such as [StatsD](https://github.com/etsy/statsd)?
 
@@ -524,7 +527,7 @@ kill -HUP `pidof prometheus`
 
 
 If you have Prometheus HA pairs with replicas `r1` and `r2` in each pair, then configure each `r1`
-to write data to `<victoriametrics-addr-1`, while each `r2` should write data to `victoriametrics-addr-2`.
+to write data to `victoriametrics-addr-1`, while each `r2` should write data to `victoriametrics-addr-2`.
 
 
 ### Multiple retentions

app/vminsert/concurrencylimiter/concurrencylimiter.go

@@ -32,6 +32,17 @@ func Init() {
 func Do(f func() error) error {
 	// Limit the number of conurrent f calls in order to prevent from excess
 	// memory usage and CPU trashing.
+	select {
+	case ch <- struct{}{}:
+		err := f()
+		<-ch
+		return err
+	default:
+	}
+
+	// All the workers are busy.
+	// Sleep for up to waitDuration.
+	concurrencyLimitReached.Inc()
 	t := timerpool.Get(waitDuration)
 	select {
 	case ch <- struct{}{}:
@@ -41,9 +52,19 @@ func Do(f func() error) error {
 		return err
 	case <-t.C:
 		timerpool.Put(t)
-		concurrencyLimitErrors.Inc()
+		concurrencyLimitTimeout.Inc()
 		return fmt.Errorf("the server is overloaded with %d concurrent inserts; either increase -maxConcurrentInserts or reduce the load", cap(ch))
 	}
 }
 
-var concurrencyLimitErrors = metrics.NewCounter(`vm_concurrency_limit_errors_total`)
+var (
+	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_insert_limit_reached_total`)
+	concurrencyLimitTimeout = metrics.NewCounter(`vm_concurrent_insert_limit_timeout_total`)
+
+	_ = metrics.NewGauge(`vm_concurrent_insert_capacity`, func() float64 {
+		return float64(cap(ch))
+	})
+	_ = metrics.NewGauge(`vm_concurrent_insert_current`, func() float64 {
+		return float64(len(ch))
+	})
+)

app/vmselect/main.go

@@ -30,29 +30,49 @@ func Init() {
 	fs.RemoveDirContents(tmpDirPath)
 	netstorage.InitTmpBlocksDir(tmpDirPath)
 	promql.InitRollupResultCache(*vmstorage.DataPath + "/cache/rollupResult")
+
 	concurrencyCh = make(chan struct{}, *maxConcurrentRequests)
 }
 
-var concurrencyCh chan struct{}
-
 // Stop stops vmselect
 func Stop() {
 	promql.StopRollupResultCache()
 }
 
+var concurrencyCh chan struct{}
+
+var (
+	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_select_limit_reached_total`)
+	concurrencyLimitTimeout = metrics.NewCounter(`vm_concurrent_select_limit_timeout_total`)
+
+	_ = metrics.NewGauge(`vm_concurrent_select_capacity`, func() float64 {
+		return float64(cap(concurrencyCh))
+	})
+	_ = metrics.NewGauge(`vm_concurrent_select_current`, func() float64 {
+		return float64(len(concurrencyCh))
+	})
+)
+
 // RequestHandler handles remote read API requests for Prometheus
 func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	// Limit the number of concurrent queries.
-	// Sleep for a while until giving up. This should resolve short bursts in requests.
-	t := timerpool.Get(*maxQueueDuration)
 	select {
 	case concurrencyCh <- struct{}{}:
-		timerpool.Put(t)
 		defer func() { <-concurrencyCh }()
-	case <-t.C:
-		timerpool.Put(t)
-		httpserver.Errorf(w, "cannot handle more than %d concurrent requests", cap(concurrencyCh))
-		return true
+	default:
+		// Sleep for a while until giving up. This should resolve short bursts in requests.
+		concurrencyLimitReached.Inc()
+		t := timerpool.Get(*maxQueueDuration)
+		select {
+		case concurrencyCh <- struct{}{}:
+			timerpool.Put(t)
+			defer func() { <-concurrencyCh }()
+		case <-t.C:
+			timerpool.Put(t)
+			concurrencyLimitTimeout.Inc()
+			httpserver.Errorf(w, "cannot handle more than %d concurrent requests", cap(concurrencyCh))
+			return true
+		}
 	}
 
 	path := strings.Replace(r.URL.Path, "//", "/", -1)

app/vmselect/promql/binary_op.go

@@ -416,10 +416,25 @@ func binaryOpIfnot(left, right float64) float64 {
 }
 
 func binaryOpEq(left, right float64) bool {
+	// Special handling for nan == nan.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/150 .
+	if math.IsNaN(left) {
+		return math.IsNaN(right)
+	}
+
 	return left == right
 }
 
 func binaryOpNeq(left, right float64) bool {
+	// Special handling for comparison with nan.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/150 .
+	if math.IsNaN(left) {
+		return !math.IsNaN(right)
+	}
+	if math.IsNaN(right) {
+		return true
+	}
+
 	return left != right
 }
 

app/vmselect/promql/lexer.go

@@ -149,12 +149,6 @@ func scanString(s string) (string, error) {
 }
 
 func scanPositiveNumber(s string) (string, error) {
-	if strings.HasPrefix(s, "Inf") {
-		return "Inf", nil
-	}
-	if strings.HasPrefix(s, "NaN") {
-		return "NaN", nil
-	}
 	// Scan integer part. It may be empty if fractional part exists.
 	i := 0
 	for i < len(s) && isDecimalChar(s[i]) {
@@ -333,6 +327,14 @@ func scanTagFilterOpPrefix(s string) int {
 	return -1
 }
 
+func isInfOrNaN(s string) bool {
+	if len(s) != 3 {
+		return false
+	}
+	s = strings.ToLower(s)
+	return s == "inf" || s == "nan"
+}
+
 func isOffset(s string) bool {
 	s = strings.ToLower(s)
 	return s == "offset"
@@ -361,7 +363,7 @@ func isPositiveNumberPrefix(s string) bool {
 
 	// Check for .234 numbers
 	if s[0] != '.' || len(s) < 2 {
-		return strings.HasPrefix(s, "Inf") || strings.HasPrefix(s, "NaN")
+		return false
 	}
 	return isDecimalChar(s[1])
 }

app/vmselect/promql/parser.go

@@ -373,7 +373,7 @@ func (p *parser) parseSingleExpr() (expr, error) {
 }
 
 func (p *parser) parseSingleExprWithoutRollupSuffix() (expr, error) {
-	if isPositiveNumberPrefix(p.lex.Token) {
+	if isPositiveNumberPrefix(p.lex.Token) || isInfOrNaN(p.lex.Token) {
 		return p.parsePositiveNumberExpr()
 	}
 	if isStringPrefix(p.lex.Token) {
@@ -417,7 +417,7 @@ func (p *parser) parseSingleExprWithoutRollupSuffix() (expr, error) {
 }
 
 func (p *parser) parsePositiveNumberExpr() (*numberExpr, error) {
-	if !isPositiveNumberPrefix(p.lex.Token) {
+	if !isPositiveNumberPrefix(p.lex.Token) && !isInfOrNaN(p.lex.Token) {
 		return nil, fmt.Errorf(`positiveNumberExpr: unexpected token %q; want "number"`, p.lex.Token)
 	}
 

app/vmselect/promql/parser_test.go

@@ -170,14 +170,34 @@ func TestParsePromQLSuccess(t *testing.T) {
 	another(`-.2`, `-0.2`)
 	another(`-.2E-2`, `-0.002`)
 	same(`NaN`)
+	another(`nan`, `NaN`)
+	another(`NAN`, `NaN`)
+	another(`nAN`, `NaN`)
 	another(`Inf`, `+Inf`)
+	another(`INF`, `+Inf`)
+	another(`inf`, `+Inf`)
 	another(`+Inf`, `+Inf`)
 	another(`-Inf`, `-Inf`)
+	another(`-inF`, `-Inf`)
 
 	// binaryOpExpr
-	another(`NaN + 2 *3 * Inf`, `NaN`)
-	another(`Inf - Inf`, `NaN`)
-	another(`Inf + Inf`, `+Inf`)
+	another(`nan == nan`, `NaN`)
+	another(`nan ==bool nan`, `1`)
+	another(`nan !=bool nan`, `0`)
+	another(`nan !=bool 2`, `1`)
+	another(`2 !=bool nan`, `1`)
+	another(`nan >bool nan`, `0`)
+	another(`nan <bool nan`, `0`)
+	another(`1 ==bool nan`, `0`)
+	another(`NaN !=bool 1`, `1`)
+	another(`inf >=bool 2`, `1`)
+	another(`-1 >bool -inf`, `1`)
+	another(`-1 <bool -inf`, `0`)
+	another(`nan + 2 *3 * inf`, `NaN`)
+	another(`INF - Inf`, `NaN`)
+	another(`Inf + inf`, `+Inf`)
+	another(`1/0`, `+Inf`)
+	another(`0/0`, `NaN`)
 	another(`-m`, `0 - m`)
 	same(`m + ignoring () n[5m]`)
 	another(`M + IGNORING () N[5m]`, `M + ignoring () N[5m]`)

app/vmselect/promql/rollup.go

@@ -45,6 +45,8 @@ var rollupFuncs = map[string]newRollupFunc{
 	"distinct_over_time": newRollupFuncOneArg(rollupDistinct),
 	"integrate":          newRollupFuncOneArg(rollupIntegrate),
 	"ideriv":             newRollupFuncOneArg(rollupIderiv),
+	"lifetime":           newRollupFuncOneArg(rollupLifetime),
+	"scrape_interval":    newRollupFuncOneArg(rollupScrapeInterval),
 	"rollup":             newRollupFuncOneArg(rollupFake),
 	"rollup_rate":        newRollupFuncOneArg(rollupFake), // + rollupFuncsRemoveCounterResets
 	"rollup_deriv":       newRollupFuncOneArg(rollupFake),
@@ -61,6 +63,8 @@ var rollupFuncsMayAdjustWindow = map[string]bool{
 	"deriv_fast":      true,
 	"irate":           true,
 	"rate":            true,
+	"lifetime":        true,
+	"scrape_interval": true,
 }
 
 var rollupFuncsRemoveCounterResets = map[string]bool{
@@ -615,10 +619,15 @@ func rollupDelta(rfa *rollupFuncArg) float64 {
 		if len(values) == 0 {
 			return nan
 		}
+		if len(values) == 1 {
+			// Assume that the previous non-existing value was 0.
+			return values[0]
+		}
 		prevValue = values[0]
 		values = values[1:]
 	}
 	if len(values) == 0 {
+		// Assume that the value didn't change on the given interval.
 		return 0
 	}
 	return values[len(values)-1] - prevValue
@@ -632,6 +641,7 @@ func rollupIdelta(rfa *rollupFuncArg) float64 {
 		if math.IsNaN(rfa.prevValue) {
 			return nan
 		}
+		// Assume that the value didn't change on the given interval.
 		return 0
 	}
 	lastValue := values[len(values)-1]
@@ -639,7 +649,8 @@ func rollupIdelta(rfa *rollupFuncArg) float64 {
 	if len(values) == 0 {
 		prevValue := rfa.prevValue
 		if math.IsNaN(prevValue) {
-			return 0
+			// Assume that the previous non-existing value was 0.
+			return lastValue
 		}
 		return lastValue - prevValue
 	}
@@ -661,7 +672,8 @@ func rollupDerivFast(rfa *rollupFuncArg) float64 {
 	prevValue := rfa.prevValue
 	prevTimestamp := rfa.prevTimestamp
 	if math.IsNaN(prevValue) {
-		if len(values) == 0 {
+		if len(values) < 2 {
+			// It is impossible to calculate derivative on 0 or 1 values.
 			return nan
 		}
 		prevValue = values[0]
@@ -670,6 +682,7 @@ func rollupDerivFast(rfa *rollupFuncArg) float64 {
 		timestamps = timestamps[1:]
 	}
 	if len(values) == 0 {
+		// Assume that the value didn't change on the given interval.
 		return 0
 	}
 	vEnd := values[len(values)-1]
@@ -684,11 +697,12 @@ func rollupIderiv(rfa *rollupFuncArg) float64 {
 	// before calling rollup funcs.
 	values := rfa.values
 	timestamps := rfa.timestamps
-	if len(values) == 0 {
-		if math.IsNaN(rfa.prevValue) {
+	if len(values) < 2 {
+		if len(values) == 0 || math.IsNaN(rfa.prevValue) {
+			// It is impossible to calculate derivative on 0 or 1 values.
 			return nan
 		}
-		return 0
+		return (values[0] - rfa.prevValue) / (float64(timestamps[0]-rfa.prevTimestamp) * 1e-3)
 	}
 	vEnd := values[len(values)-1]
 	tEnd := timestamps[len(timestamps)-1]
@@ -712,7 +726,37 @@ func rollupIderiv(rfa *rollupFuncArg) float64 {
 	}
 	dv := vEnd - vStart
 	dt := tEnd - tStart
-	return dv / (float64(dt) / 1000)
+	return dv / (float64(dt) * 1e-3)
+}
+
+func rollupLifetime(rfa *rollupFuncArg) float64 {
+	// Calculate the duration between the first and the last data points.
+	timestamps := rfa.timestamps
+	if math.IsNaN(rfa.prevValue) {
+		if len(timestamps) < 2 {
+			return nan
+		}
+		return float64(timestamps[len(timestamps)-1]-timestamps[0]) * 1e-3
+	}
+	if len(timestamps) == 0 {
+		return nan
+	}
+	return float64(timestamps[len(timestamps)-1]-rfa.prevTimestamp) * 1e-3
+}
+
+func rollupScrapeInterval(rfa *rollupFuncArg) float64 {
+	// Calculate the average interval between data points.
+	timestamps := rfa.timestamps
+	if math.IsNaN(rfa.prevValue) {
+		if len(timestamps) < 2 {
+			return nan
+		}
+		return float64(timestamps[len(timestamps)-1]-timestamps[0]) * 1e-3 / float64(len(timestamps)-1)
+	}
+	if len(timestamps) == 0 {
+		return nan
+	}
+	return (float64(timestamps[len(timestamps)-1]-rfa.prevTimestamp) * 1e-3) / float64(len(timestamps))
 }
 
 func rollupChanges(rfa *rollupFuncArg) float64 {

app/vmselect/promql/rollup_result_cache.go

@@ -4,14 +4,15 @@ import (
 	"crypto/rand"
 	"flag"
 	"fmt"
-	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 	"github.com/VictoriaMetrics/fastcache"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -19,7 +20,7 @@ import (
 var disableCache = flag.Bool("search.disableCache", false, "Whether to disable response caching. This may be useful during data backfilling")
 
 var rollupResultCacheV = &rollupResultCache{
-	fastcache.New(1024 * 1024), // This is a cache for testing.
+	c: workingsetcache.New(1024*1024, time.Hour), // This is a cache for testing.
 }
 var rollupResultCachePath string
 
@@ -43,15 +44,17 @@ var (
 func InitRollupResultCache(cachePath string) {
 	rollupResultCachePath = cachePath
 	startTime := time.Now()
-	var c *fastcache.Cache
+	cacheSize := getRollupResultCacheSize()
+	var c *workingsetcache.Cache
 	if len(rollupResultCachePath) > 0 {
 		logger.Infof("loading rollupResult cache from %q...", rollupResultCachePath)
-		c = fastcache.LoadFromFileOrNew(rollupResultCachePath, getRollupResultCacheSize())
+		c = workingsetcache.Load(rollupResultCachePath, cacheSize, time.Hour)
 	} else {
-		c = fastcache.New(getRollupResultCacheSize())
+		c = workingsetcache.New(cacheSize, time.Hour)
 	}
 	if *disableCache {
-		c.Reset()
+		c.Stop()
+		c = nil
 	}
 
 	stats := &fastcache.Stats{}
@@ -96,25 +99,28 @@ func InitRollupResultCache(cachePath string) {
 // StopRollupResultCache closes the rollupResult cache.
 func StopRollupResultCache() {
 	if len(rollupResultCachePath) == 0 {
-		rollupResultCacheV.c.Reset()
+		if !*disableCache {
+			rollupResultCacheV.c.Stop()
+			rollupResultCacheV.c = nil
+		}
 		return
 	}
-	gomaxprocs := runtime.GOMAXPROCS(-1)
 	logger.Infof("saving rollupResult cache to %q...", rollupResultCachePath)
 	startTime := time.Now()
-	if err := rollupResultCacheV.c.SaveToFileConcurrent(rollupResultCachePath, gomaxprocs); err != nil {
+	if err := rollupResultCacheV.c.Save(rollupResultCachePath); err != nil {
 		logger.Errorf("cannot close rollupResult cache at %q: %s", rollupResultCachePath, err)
-	} else {
-		var fcs fastcache.Stats
-		rollupResultCacheV.c.UpdateStats(&fcs)
-		rollupResultCacheV.c.Reset()
-		logger.Infof("saved rollupResult cache to %q in %s; entriesCount: %d, sizeBytes: %d",
-			rollupResultCachePath, time.Since(startTime), fcs.EntriesCount, fcs.BytesSize)
+		return
 	}
+	var fcs fastcache.Stats
+	rollupResultCacheV.c.UpdateStats(&fcs)
+	rollupResultCacheV.c.Stop()
+	rollupResultCacheV.c = nil
+	logger.Infof("saved rollupResult cache to %q in %s; entriesCount: %d, sizeBytes: %d",
+		rollupResultCachePath, time.Since(startTime), fcs.EntriesCount, fcs.BytesSize)
 }
 
 type rollupResultCache struct {
-	c *fastcache.Cache
+	c *workingsetcache.Cache
 }
 
 var rollupResultCacheResets = metrics.NewCounter(`vm_cache_resets_total{type="promql/rollupResult"}`)
@@ -148,15 +154,23 @@ func (rrc *rollupResultCache) Get(funcName string, ec *EvalConfig, me *metricExp
 		return nil, ec.Start
 	}
 	bb.B = key.Marshal(bb.B[:0])
-	resultBuf := rrc.c.GetBig(nil, bb.B)
-	if len(resultBuf) == 0 {
+	compressedResultBuf := resultBufPool.Get()
+	defer resultBufPool.Put(compressedResultBuf)
+	compressedResultBuf.B = rrc.c.GetBig(compressedResultBuf.B[:0], bb.B)
+	if len(compressedResultBuf.B) == 0 {
 		mi.RemoveKey(key)
 		metainfoBuf = mi.Marshal(metainfoBuf[:0])
 		bb.B = marshalRollupResultCacheKey(bb.B[:0], funcName, me, iafc, window, ec.Step)
 		rrc.c.Set(bb.B, metainfoBuf)
 		return nil, ec.Start
 	}
-	tss, err := unmarshalTimeseriesFast(resultBuf)
+	// Decompress into newly allocated byte slice, since tss returned from unmarshalTimeseriesFast
+	// refers to the byte slice, so it cannot be returned to the resultBufPool.
+	resultBuf, err := encoding.DecompressZSTD(nil, compressedResultBuf.B)
+	if err != nil {
+		logger.Panicf("BUG: cannot decompress resultBuf from rollupResultCache: %s; it looks like it was improperly saved", err)
+	}
+	tss, err = unmarshalTimeseriesFast(resultBuf)
 	if err != nil {
 		logger.Panicf("BUG: cannot unmarshal timeseries from rollupResultCache: %s; it looks like it was improperly saved", err)
 	}
@@ -196,6 +210,8 @@ func (rrc *rollupResultCache) Get(funcName string, ec *EvalConfig, me *metricExp
 	return tss, newStart
 }
 
+var resultBufPool bytesutil.ByteBufferPool
+
 func (rrc *rollupResultCache) Put(funcName string, ec *EvalConfig, me *metricExpr, iafc *incrementalAggrFuncContext, window int64, tss []*timeseries) {
 	if *disableCache || len(tss) == 0 || !ec.mayCache() {
 		return
@@ -227,11 +243,16 @@ func (rrc *rollupResultCache) Put(funcName string, ec *EvalConfig, me *metricExp
 
 	// Store tss in the cache.
 	maxMarshaledSize := getRollupResultCacheSize() / 4
-	tssMarshaled := marshalTimeseriesFast(tss, maxMarshaledSize, ec.Step)
-	if tssMarshaled == nil {
+	resultBuf := resultBufPool.Get()
+	defer resultBufPool.Put(resultBuf)
+	resultBuf.B = marshalTimeseriesFast(resultBuf.B[:0], tss, maxMarshaledSize, ec.Step)
+	if len(resultBuf.B) == 0 {
 		tooBigRollupResults.Inc()
 		return
 	}
+	compressedResultBuf := resultBufPool.Get()
+	defer resultBufPool.Put(compressedResultBuf)
+	compressedResultBuf.B = encoding.CompressZSTDLevel(compressedResultBuf.B[:0], resultBuf.B, 1)
 
 	bb := bbPool.Get()
 	defer bbPool.Put(bb)
@@ -240,7 +261,7 @@ func (rrc *rollupResultCache) Put(funcName string, ec *EvalConfig, me *metricExp
 	key.prefix = rollupResultCacheKeyPrefix
 	key.suffix = atomic.AddUint64(&rollupResultCacheKeySuffix, 1)
 	bb.B = key.Marshal(bb.B[:0])
-	rrc.c.SetBig(bb.B, tssMarshaled)
+	rrc.c.SetBig(bb.B, compressedResultBuf.B)
 
 	bb.B = marshalRollupResultCacheKey(bb.B[:0], funcName, me, iafc, window, ec.Step)
 	metainfoBuf := rrc.c.Get(nil, bb.B)
@@ -270,7 +291,7 @@ var (
 var tooBigRollupResults = metrics.NewCounter("vm_too_big_rollup_results_total")
 
 // Increment this value every time the format of the cache changes.
-const rollupResultCacheVersion = 5
+const rollupResultCacheVersion = 6
 
 func marshalRollupResultCacheKey(dst []byte, funcName string, me *metricExpr, iafc *incrementalAggrFuncContext, window, step int64) []byte {
 	dst = append(dst, rollupResultCacheVersion)

app/vmselect/promql/rollup_test.go

@@ -45,8 +45,19 @@ func TestRollupIderivDuplicateTimestamps(t *testing.T) {
 		timestamps: []int64{100},
 	}
 	n = rollupIderiv(rfa)
-	if n != 0 {
-		t.Fatalf("unexpected value; got %v; want %v", n, 0)
+	if !math.IsNaN(n) {
+		t.Fatalf("unexpected value; got %v; want %v", n, nan)
+	}
+
+	rfa = &rollupFuncArg{
+		prevTimestamp: 90,
+		prevValue:     10,
+		values:        []float64{15},
+		timestamps:    []int64{100},
+	}
+	n = rollupIderiv(rfa)
+	if n != 500 {
+		t.Fatalf("unexpected value; got %v; want %v", n, 0.5)
 	}
 
 	rfa = &rollupFuncArg{
@@ -569,10 +580,66 @@ func TestRollupFuncsNoWindow(t *testing.T) {
 		}
 		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
 		values := rc.Do(nil, testValues, testTimestamps)
-		valuesExpected := []float64{0, 33, -87, 0}
+		valuesExpected := []float64{123, 33, -87, 0}
 		timestampsExpected := []int64{10, 50, 90, 130}
 		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
 	})
+	t.Run("lifetime", func(t *testing.T) {
+		rc := rollupConfig{
+			Func:   rollupLifetime,
+			Start:  0,
+			End:    160,
+			Step:   40,
+			Window: 0,
+		}
+		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
+		values := rc.Do(nil, testValues, testTimestamps)
+		valuesExpected := []float64{nan, 0.031, 0.044, 0.04, 0.01}
+		timestampsExpected := []int64{0, 40, 80, 120, 160}
+		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
+	})
+	t.Run("lifetime", func(t *testing.T) {
+		rc := rollupConfig{
+			Func:   rollupLifetime,
+			Start:  0,
+			End:    160,
+			Step:   40,
+			Window: 200,
+		}
+		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
+		values := rc.Do(nil, testValues, testTimestamps)
+		valuesExpected := []float64{nan, 0.031, 0.075, 0.115, 0.125}
+		timestampsExpected := []int64{0, 40, 80, 120, 160}
+		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
+	})
+	t.Run("scrape_interval", func(t *testing.T) {
+		rc := rollupConfig{
+			Func:   rollupScrapeInterval,
+			Start:  0,
+			End:    160,
+			Step:   40,
+			Window: 0,
+		}
+		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
+		values := rc.Do(nil, testValues, testTimestamps)
+		valuesExpected := []float64{nan, 0.010333333333333333, 0.011, 0.013333333333333334, 0.01}
+		timestampsExpected := []int64{0, 40, 80, 120, 160}
+		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
+	})
+	t.Run("scrape_interval", func(t *testing.T) {
+		rc := rollupConfig{
+			Func:   rollupScrapeInterval,
+			Start:  0,
+			End:    160,
+			Step:   40,
+			Window: 80,
+		}
+		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
+		values := rc.Do(nil, testValues, testTimestamps)
+		valuesExpected := []float64{nan, 0.010333333333333333, 0.010714285714285714, 0.012, 0.0125}
+		timestampsExpected := []int64{0, 40, 80, 120, 160}
+		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
+	})
 	t.Run("changes", func(t *testing.T) {
 		rc := rollupConfig{
 			Func:   rollupChanges,

app/vmselect/promql/timeseries.go

@@ -76,7 +76,7 @@ func putTimeseries(ts *timeseries) {
 
 var timeseriesPool sync.Pool
 
-func marshalTimeseriesFast(tss []*timeseries, maxSize int, step int64) []byte {
+func marshalTimeseriesFast(dst []byte, tss []*timeseries, maxSize int, step int64) []byte {
 	if len(tss) == 0 {
 		logger.Panicf("BUG: tss cannot be empty")
 	}
@@ -92,13 +92,13 @@ func marshalTimeseriesFast(tss []*timeseries, maxSize int, step int64) []byte {
 
 	if size > maxSize {
 		// Do not marshal tss, since it would occupy too much space
-		return nil
+		return dst
 	}
 
 	// Allocate the buffer for the marshaled tss before its' marshaling.
 	// This should reduce memory fragmentation and memory usage.
-	dst := make([]byte, 0, size)
-	dst = marshalFastTimestamps(dst, tss[0].Timestamps)
+	dst = bytesutil.Resize(dst, size)
+	dst = marshalFastTimestamps(dst[:0], tss[0].Timestamps)
 	for _, ts := range tss {
 		dst = ts.marshalFastNoTimestamps(dst)
 	}

app/vmselect/promql/timeseries_test.go

@@ -74,7 +74,7 @@ func TestTimeseriesMarshalUnmarshalFast(t *testing.T) {
 
 			tssOrig = append(tssOrig, &ts)
 		}
-		buf := marshalTimeseriesFast(tssOrig, 1e6, 123)
+		buf := marshalTimeseriesFast(nil, tssOrig, 1e6, 123)
 		tssGot, err := unmarshalTimeseriesFast(buf)
 		if err != nil {
 			t.Fatalf("error in unmarshalTimeseriesFast: %s", err)

app/vmstorage/main.go

@@ -365,6 +365,22 @@ func registerStorageMetrics() {
 		return float64(m().TooSmallTimestampRows)
 	})
 
+	metrics.NewGauge(`vm_concurrent_addrows_limit_reached_total`, func() float64 {
+		return float64(m().AddRowsConcurrencyLimitReached)
+	})
+	metrics.NewGauge(`vm_concurrent_addrows_limit_timeout_total`, func() float64 {
+		return float64(m().AddRowsConcurrencyLimitTimeout)
+	})
+	metrics.NewGauge(`vm_concurrent_addrows_dropped_rows_total`, func() float64 {
+		return float64(m().AddRowsConcurrencyDroppedRows)
+	})
+	metrics.NewGauge(`vm_concurrent_addrows_capacity`, func() float64 {
+		return float64(m().AddRowsConcurrencyCapacity)
+	})
+	metrics.NewGauge(`vm_concurrent_addrows_current`, func() float64 {
+		return float64(m().AddRowsConcurrencyCurrent)
+	})
+
 	metrics.NewGauge(`vm_rows{type="storage/big"}`, func() float64 {
 		return float64(tm().BigRowsCount)
 	})

deployment/docker/Makefile

@@ -1,5 +1,5 @@
 DOCKER_NAMESPACE := victoriametrics
-BUILDER_IMAGE := local/builder:go1.12.7
+BUILDER_IMAGE := local/builder:go1.12.9
 CERTS_IMAGE := local/certs:1.0.2
 
 package-certs:

deployment/docker/builder/Dockerfile

@@ -1,2 +1,2 @@
-FROM golang:1.12.7
+FROM golang:1.12.9
 STOPSIGNAL SIGINT

go.mod

@@ -2,17 +2,17 @@ module github.com/VictoriaMetrics/VictoriaMetrics
 
 require (
 	github.com/VictoriaMetrics/fastcache v1.5.1
-	github.com/VictoriaMetrics/metrics v1.7.0
+	github.com/VictoriaMetrics/metrics v1.7.1
 	github.com/cespare/xxhash/v2 v2.0.1-0.20190104013014-3767db7a7e18
 	github.com/golang/snappy v0.0.1
 	github.com/google/go-cmp v0.3.0 // indirect
-	github.com/klauspost/compress v1.7.4
+	github.com/klauspost/compress v1.7.5
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/valyala/fastjson v1.4.1
-	github.com/valyala/gozstd v1.5.1
+	github.com/valyala/gozstd v1.6.0
 	github.com/valyala/histogram v1.0.1
 	github.com/valyala/quicktemplate v1.1.1
-	golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7
+	golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa
 )
 
 go 1.12

go.sum

@@ -3,8 +3,8 @@ github.com/OneOfOne/xxhash v1.2.5 h1:zl/OfRA6nftbBK9qTohYBJ5xvw6C/oNKizR7cZGl3cI
 github.com/OneOfOne/xxhash v1.2.5/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/VictoriaMetrics/fastcache v1.5.1 h1:qHgHjyoNFV7jgucU8QZUuU4gcdhfs8QW1kw68OD2Lag=
 github.com/VictoriaMetrics/fastcache v1.5.1/go.mod h1:+jv9Ckb+za/P1ZRg/sulP5Ni1v49daAVERr0H3CuscE=
-github.com/VictoriaMetrics/metrics v1.7.0 h1:+bdBpPEMOSgOwoQFf4KHqgeAy6xiXn/uzlrKx2YSCT8=
-github.com/VictoriaMetrics/metrics v1.7.0/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
+github.com/VictoriaMetrics/metrics v1.7.1 h1:g2qrY6Upn8rvlvR40cGHFY0crwi4hpqF0n9vJMNsCSg=
+github.com/VictoriaMetrics/metrics v1.7.1/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
@@ -20,8 +20,8 @@ github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.4.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.7.4 h1:4UqAIzZ1Ns2epCTyJ1d2xMWvxtX+FNSCYWeOFogK9nc=
-github.com/klauspost/compress v1.7.4/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
+github.com/klauspost/compress v1.7.5 h1:NMapGoDIKPKpk2hpcgAU6XHfsREHG2p8PIg7C3f/jpI=
+github.com/klauspost/compress v1.7.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid v1.2.0 h1:NMpwD2G9JSFOE1/TJjGSo5zG7Yb2bTe7eq1jH+irmeE=
 github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
@@ -41,13 +41,13 @@ github.com/valyala/fastjson v1.4.1 h1:hrltpHpIpkaxll8QltMU8c3QZ5+qIiCL8yKqPFJI/y
 github.com/valyala/fastjson v1.4.1/go.mod h1:nV6MsjxL2IMJQUoHDIrjEI7oLyeqK6aBD7EFWPsvP8o=
 github.com/valyala/fastrand v1.0.0 h1:LUKT9aKer2dVQNUi3waewTbKV+7H17kvWFNKs2ObdkI=
 github.com/valyala/fastrand v1.0.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
-github.com/valyala/gozstd v1.5.1 h1:ZLepItgu2g+B2CfVQy6KCV/as8lnJ7ef1KU6DPxQSS0=
-github.com/valyala/gozstd v1.5.1/go.mod h1:oYOS+oJovjw9ewtrwEYb9+ybolEXd6pHyLMuAWN5zts=
+github.com/valyala/gozstd v1.6.0 h1:34qKK75C6Dx9zof2JqUiunfJQ87Up6vTHXABWDyCH+g=
+github.com/valyala/gozstd v1.6.0/go.mod h1:y5Ew47GLlP37EkTB+B4s7r6A5rdaeB7ftbl9zoYiIPQ=
 github.com/valyala/histogram v1.0.1 h1:FzA7n2Tz/wKRMejgu3PV1vw3htAklTjjuoI6z3d4KDg=
 github.com/valyala/histogram v1.0.1/go.mod h1:lQy0xA4wUz2+IUnf97SivorsJIp8FxsnRd6x25q7Mto=
 github.com/valyala/quicktemplate v1.1.1 h1:C58y/wN0FMTi2PR0n3onltemfFabany53j7M6SDDB8k=
 github.com/valyala/quicktemplate v1.1.1/go.mod h1:EH+4AkTd43SvgIbQHYu59/cJyxDoOVRUAfrukLPuGJ4=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
 golang.org/x/net v0.0.0-20180911220305-26e67e76b6c3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7 h1:LepdCS8Gf/MVejFIt8lsiexZATdoGVyp5bcyS+rYoUI=
-golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa h1:KIDDMLT1O0Nr7TSxp8xM5tJcdn8tgyAONntO829og1M=
+golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

lib/fs/fs.go

@@ -5,12 +5,15 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/filestream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/metrics"
+	"golang.org/x/sys/unix"
 )
 
 // ReadAtCloser is rand-access read interface.
@@ -87,26 +90,42 @@ func MustSyncPath(path string) {
 	}
 }
 
-// WriteFile writes data to the given file path.
+var tmpFileNum uint64
+
+// WriteFileAtomically atomically writes data to the given file path.
 //
-// WriteFile returns only after the file is fully written
+// WriteFile returns only after the file is fully written and synced
 // to the underlying storage.
-func WriteFile(path string, data []byte) error {
+func WriteFileAtomically(path string, data []byte) error {
+	// Check for the existing file. It is expected that
+	// the WriteFileAtomically function cannot be called concurrently
+	// with the same `path`.
 	if IsPathExist(path) {
 		return fmt.Errorf("cannot create file %q, since it already exists", path)
 	}
-	f, err := filestream.Create(path, false)
+
+	n := atomic.AddUint64(&tmpFileNum, 1)
+	tmpPath := fmt.Sprintf("%s.tmp.%d", path, n)
+	f, err := filestream.Create(tmpPath, false)
 	if err != nil {
-		return fmt.Errorf("cannot create file %q: %s", path, err)
+		return fmt.Errorf("cannot create file %q: %s", tmpPath, err)
 	}
 	if _, err := f.Write(data); err != nil {
 		f.MustClose()
-		return fmt.Errorf("cannot write %d bytes to file %q: %s", len(data), path, err)
+		MustRemoveAll(tmpPath)
+		return fmt.Errorf("cannot write %d bytes to file %q: %s", len(data), tmpPath, err)
 	}
 
 	// Sync and close the file.
 	f.MustClose()
 
+	// Atomically move the file from tmpPath to path.
+	if err := os.Rename(tmpPath, path); err != nil {
+		// do not call MustRemoveAll(tmpPath) here, so the user could inspect
+		// the file contents during investigating the issue.
+		return fmt.Errorf("cannot move %q to %q: %s", tmpPath, path, err)
+	}
+
 	// Sync the containing directory, so the file is guaranteed to appear in the directory.
 	// See https://www.quora.com/When-should-you-fsync-the-containing-directory-in-addition-to-the-file-itself
 	absPath, err := filepath.Abs(path)
@@ -119,6 +138,15 @@ func WriteFile(path string, data []byte) error {
 	return nil
 }
 
+// IsTemporaryFileName returns true if fn matches temporary file name pattern
+// from WriteFileAtomically.
+func IsTemporaryFileName(fn string) bool {
+	return tmpFileNameRe.MatchString(fn)
+}
+
+// tmpFileNameRe is regexp for temporary file name - see WriteFileAtomically for details.
+var tmpFileNameRe = regexp.MustCompile(`\.tmp\.\d+$`)
+
 // MkdirAllIfNotExist creates the given path dir if it isn't exist.
 func MkdirAllIfNotExist(path string) error {
 	if IsPathExist(path) {
@@ -358,3 +386,17 @@ func MustWriteData(w io.Writer, data []byte) {
 		logger.Panicf("BUG: writer wrote %d bytes instead of %d bytes", n, len(data))
 	}
 }
+
+// CreateFlockFile creates flock.lock file in the directory dir
+// and returns the handler to the file.
+func CreateFlockFile(dir string) (*os.File, error) {
+	flockFile := dir + "/flock.lock"
+	flockF, err := os.Create(flockFile)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create lock file %q: %s", flockFile, err)
+	}
+	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
+		return nil, fmt.Errorf("cannot acquire lock on file %q: %s", flockFile, err)
+	}
+	return flockF, nil
+}

lib/fs/fs_test.go

@@ -0,0 +1,24 @@
+package fs
+
+import (
+	"testing"
+)
+
+func TestIsTemporaryFileName(t *testing.T) {
+	f := func(s string, resultExpected bool) {
+		t.Helper()
+		result := IsTemporaryFileName(s)
+		if result != resultExpected {
+			t.Fatalf("unexpected IsTemporaryFileName(%q); got %v; want %v", s, result, resultExpected)
+		}
+	}
+	f("", false)
+	f(".", false)
+	f(".tmp", false)
+	f("tmp.123", false)
+	f(".tmp.123.xx", false)
+	f(".tmp.1", true)
+	f("asdf.dff.tmp.123", true)
+	f("asdf.sdfds.tmp.dfd", false)
+	f("dfd.sdfds.dfds.1232", false)
+}

lib/mergeset/part_header.go

@@ -164,7 +164,7 @@ func (ph *partHeader) WriteMetadata(partPath string) error {
 		return fmt.Errorf("cannot marshal metadata: %s", err)
 	}
 	metadataPath := partPath + "/metadata.json"
-	if err := fs.WriteFile(metadataPath, metadata); err != nil {
+	if err := fs.WriteFileAtomically(metadataPath, metadata); err != nil {
 		return fmt.Errorf("cannot create %q: %s", metadataPath, err)
 	}
 	return nil

lib/mergeset/table.go

@@ -134,13 +134,9 @@ func OpenTable(path string) (*Table, error) {
 	}
 
 	// Protect from concurrent opens.
-	flockFile := path + "/flock.lock"
-	flockF, err := os.Create(flockFile)
+	flockF, err := fs.CreateFlockFile(path)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create lock file %q: %s", flockFile, err)
-	}
-	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
-		return nil, fmt.Errorf("cannot acquire lock on file %q: %s", flockFile, err)
+		return nil, err
 	}
 
 	// Open table parts.
@@ -715,7 +711,7 @@ func (tb *Table) mergeParts(pws []*partWrapper, stopCh <-chan struct{}, isOuterP
 	dstPartPath := ph.Path(tb.path, mergeIdx)
 	fmt.Fprintf(&bb, "%s -> %s\n", tmpPartPath, dstPartPath)
 	txnPath := fmt.Sprintf("%s/txn/%016X", tb.path, mergeIdx)
-	if err := fs.WriteFile(txnPath, bb.B); err != nil {
+	if err := fs.WriteFileAtomically(txnPath, bb.B); err != nil {
 		return fmt.Errorf("cannot create transaction file %q: %s", txnPath, err)
 	}
 
@@ -994,7 +990,12 @@ func runTransactions(txnLock *sync.RWMutex, path string) error {
 	})
 
 	for _, fi := range fis {
-		txnPath := txnDir + "/" + fi.Name()
+		fn := fi.Name()
+		if fs.IsTemporaryFileName(fn) {
+			// Skip temporary files, which could be left after unclean shutdown.
+			continue
+		}
+		txnPath := txnDir + "/" + fn
 		if err := runTransaction(txnLock, path, txnPath); err != nil {
 			return fmt.Errorf("cannot run transaction from %q: %s", txnPath, err)
 		}

lib/storage/block.go

@@ -205,19 +205,6 @@ func (b *Block) MarshalData(timestampsBlockOffset, valuesBlockOffset uint64) ([]
 	b.bh.ValuesBlockSize = uint32(len(b.valuesData))
 	b.values = b.values[:0]
 
-	if len(timestamps) > 1 && (b.bh.ValuesMarshalType == encoding.MarshalTypeConst || b.bh.ValuesMarshalType == encoding.MarshalTypeDeltaConst) {
-		// Special case - values are constant or are changed with constant rate.
-		// In this case we may 'cheat' by assuming timestamps are changed
-		// at ideal constant rate. This improves timestamps' compression rate.
-		minTimestamp := timestamps[0]
-		maxTimestamp := timestamps[len(timestamps)-1]
-		delta := (maxTimestamp - minTimestamp) / int64(len(timestamps)-1)
-		ts := minTimestamp
-		for i := 1; i < len(timestamps); i++ {
-			ts += delta
-			timestamps[i] = ts
-		}
-	}
 	b.timestampsData, b.bh.TimestampsMarshalType, b.bh.MinTimestamp = encoding.MarshalTimestamps(b.timestampsData[:0], timestamps, b.bh.PrecisionBits)
 	b.bh.TimestampsBlockOffset = timestampsBlockOffset
 	b.bh.TimestampsBlockSize = uint32(len(b.timestampsData))

lib/storage/index_db.go

@@ -18,6 +18,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 	"github.com/VictoriaMetrics/fastcache"
 	xxhash "github.com/cespare/xxhash/v2"
 )
@@ -52,17 +53,17 @@ type indexDB struct {
 	extDBLock sync.Mutex
 
 	// Cache for fast TagFilters -> TSIDs lookup.
-	tagCache *fastcache.Cache
+	tagCache *workingsetcache.Cache
 
 	// Cache for fast MetricID -> TSID lookup.
-	metricIDCache *fastcache.Cache
+	metricIDCache *workingsetcache.Cache
 
 	// Cache for fast MetricID -> MetricName lookup.
-	metricNameCache *fastcache.Cache
+	metricNameCache *workingsetcache.Cache
 
 	// Cache holding useless TagFilters entries, which have no tag filters
 	// matching low number of metrics.
-	uselessTagFiltersCache *fastcache.Cache
+	uselessTagFiltersCache *workingsetcache.Cache
 
 	indexSearchPool sync.Pool
 
@@ -101,7 +102,7 @@ type indexDB struct {
 }
 
 // openIndexDB opens index db from the given path with the given caches.
-func openIndexDB(path string, metricIDCache, metricNameCache *fastcache.Cache, currHourMetricIDs, prevHourMetricIDs *atomic.Value) (*indexDB, error) {
+func openIndexDB(path string, metricIDCache, metricNameCache *workingsetcache.Cache, currHourMetricIDs, prevHourMetricIDs *atomic.Value) (*indexDB, error) {
 	if metricIDCache == nil {
 		logger.Panicf("BUG: metricIDCache must be non-nil")
 	}
@@ -130,10 +131,10 @@ func openIndexDB(path string, metricIDCache, metricNameCache *fastcache.Cache, c
 		tb:       tb,
 		name:     name,
 
-		tagCache:               fastcache.New(mem / 32),
+		tagCache:               workingsetcache.New(mem/32, time.Hour),
 		metricIDCache:          metricIDCache,
 		metricNameCache:        metricNameCache,
-		uselessTagFiltersCache: fastcache.New(mem / 128),
+		uselessTagFiltersCache: workingsetcache.New(mem/128, time.Hour),
 
 		currHourMetricIDs: currHourMetricIDs,
 		prevHourMetricIDs: prevHourMetricIDs,
@@ -273,8 +274,8 @@ func (db *indexDB) decRef() {
 	db.SetExtDB(nil)
 
 	// Free space occupied by caches owned by db.
-	db.tagCache.Reset()
-	db.uselessTagFiltersCache.Reset()
+	db.tagCache.Stop()
+	db.uselessTagFiltersCache.Stop()
 
 	db.tagCache = nil
 	db.metricIDCache = nil
@@ -291,20 +292,36 @@ func (db *indexDB) decRef() {
 }
 
 func (db *indexDB) getFromTagCache(key []byte) ([]TSID, bool) {
-	value := db.tagCache.GetBig(nil, key)
-	if len(value) == 0 {
+	compressedBuf := tagBufPool.Get()
+	defer tagBufPool.Put(compressedBuf)
+	compressedBuf.B = db.tagCache.GetBig(compressedBuf.B[:0], key)
+	if len(compressedBuf.B) == 0 {
 		return nil, false
 	}
-	tsids, err := unmarshalTSIDs(nil, value)
+	buf := tagBufPool.Get()
+	defer tagBufPool.Put(buf)
+	var err error
+	buf.B, err = encoding.DecompressZSTD(buf.B[:0], compressedBuf.B)
+	if err != nil {
+		logger.Panicf("FATAL: cannot decompress tsids from tagCache: %s", err)
+	}
+	tsids, err := unmarshalTSIDs(nil, buf.B)
 	if err != nil {
 		logger.Panicf("FATAL: cannot unmarshal tsids from tagCache: %s", err)
 	}
 	return tsids, true
 }
 
+var tagBufPool bytesutil.ByteBufferPool
+
 func (db *indexDB) putToTagCache(tsids []TSID, key []byte) {
-	value := marshalTSIDs(nil, tsids)
-	db.tagCache.SetBig(key, value)
+	buf := tagBufPool.Get()
+	buf.B = marshalTSIDs(buf.B[:0], tsids)
+	compressedBuf := tagBufPool.Get()
+	compressedBuf.B = encoding.CompressZSTDLevel(compressedBuf.B[:0], buf.B, 1)
+	tagBufPool.Put(buf)
+	db.tagCache.SetBig(key, compressedBuf.B)
+	tagBufPool.Put(compressedBuf)
 }
 
 func (db *indexDB) getFromMetricIDCache(dst *TSID, metricID uint64) error {
@@ -974,7 +991,8 @@ func (db *indexDB) searchTSIDs(tfss []*TagFilters, tr TimeRange, maxMetrics int)
 		extTSIDs, err = is.searchTSIDs(tfss, tr, maxMetrics)
 		extDB.putIndexSearch(is)
 
-		db.putToTagCache(tsids, tfKeyExtBuf.B)
+		sort.Slice(extTSIDs, func(i, j int) bool { return extTSIDs[i].Less(&extTSIDs[j]) })
+		extDB.putToTagCache(extTSIDs, tfKeyExtBuf.B)
 	}) {
 		if err != nil {
 			return nil, err

lib/storage/index_db_test.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/fastcache"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 )
 
 func TestMarshalUnmarshalTSIDs(t *testing.T) {
@@ -57,10 +57,10 @@ func TestMarshalUnmarshalTSIDs(t *testing.T) {
 }
 
 func TestIndexDBOpenClose(t *testing.T) {
-	metricIDCache := fastcache.New(1234)
-	metricNameCache := fastcache.New(1234)
-	defer metricIDCache.Reset()
-	defer metricNameCache.Reset()
+	metricIDCache := workingsetcache.New(1234, time.Hour)
+	metricNameCache := workingsetcache.New(1234, time.Hour)
+	defer metricIDCache.Stop()
+	defer metricNameCache.Stop()
 
 	var hmCurr atomic.Value
 	hmCurr.Store(&hourMetricIDs{})
@@ -85,10 +85,10 @@ func TestIndexDB(t *testing.T) {
 	const metricGroups = 10
 
 	t.Run("serial", func(t *testing.T) {
-		metricIDCache := fastcache.New(1234)
-		metricNameCache := fastcache.New(1234)
-		defer metricIDCache.Reset()
-		defer metricNameCache.Reset()
+		metricIDCache := workingsetcache.New(1234, time.Hour)
+		metricNameCache := workingsetcache.New(1234, time.Hour)
+		defer metricIDCache.Stop()
+		defer metricNameCache.Stop()
 
 		var hmCurr atomic.Value
 		hmCurr.Store(&hourMetricIDs{})
@@ -142,10 +142,10 @@ func TestIndexDB(t *testing.T) {
 	})
 
 	t.Run("concurrent", func(t *testing.T) {
-		metricIDCache := fastcache.New(1234)
-		metricNameCache := fastcache.New(1234)
-		defer metricIDCache.Reset()
-		defer metricNameCache.Reset()
+		metricIDCache := workingsetcache.New(1234, time.Hour)
+		metricNameCache := workingsetcache.New(1234, time.Hour)
+		defer metricIDCache.Stop()
+		defer metricNameCache.Stop()
 
 		var hmCurr atomic.Value
 		hmCurr.Store(&hourMetricIDs{})

lib/storage/index_db_timing_test.go

@@ -6,17 +6,18 @@ import (
 	"strconv"
 	"sync/atomic"
 	"testing"
+	"time"
 
-	"github.com/VictoriaMetrics/fastcache"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 )
 
 func BenchmarkIndexDBAddTSIDs(b *testing.B) {
 	const recordsPerLoop = 1e3
 
-	metricIDCache := fastcache.New(1234)
-	metricNameCache := fastcache.New(1234)
-	defer metricIDCache.Reset()
-	defer metricNameCache.Reset()
+	metricIDCache := workingsetcache.New(1234, time.Hour)
+	metricNameCache := workingsetcache.New(1234, time.Hour)
+	defer metricIDCache.Stop()
+	defer metricNameCache.Stop()
 
 	var hmCurr atomic.Value
 	hmCurr.Store(&hourMetricIDs{})
@@ -78,86 +79,11 @@ func benchmarkIndexDBAddTSIDs(db *indexDB, tsid *TSID, mn *MetricName, startOffs
 	}
 }
 
-func BenchmarkIndexDBSearchTSIDs(b *testing.B) {
-	metricIDCache := fastcache.New(1234)
-	metricNameCache := fastcache.New(1234)
-	defer metricIDCache.Reset()
-	defer metricNameCache.Reset()
-
-	var hmCurr atomic.Value
-	hmCurr.Store(&hourMetricIDs{})
-	var hmPrev atomic.Value
-	hmPrev.Store(&hourMetricIDs{})
-
-	const dbName = "bench-index-db-search-tsids"
-	db, err := openIndexDB(dbName, metricIDCache, metricNameCache, &hmCurr, &hmPrev)
-	if err != nil {
-		b.Fatalf("cannot open indexDB: %s", err)
-	}
-	defer func() {
-		db.MustClose()
-		if err := os.RemoveAll(dbName); err != nil {
-			b.Fatalf("cannot remove indexDB: %s", err)
-		}
-	}()
-
-	const recordsCount = 1e5
-
-	// Fill the db with recordsCount records.
-	var mn MetricName
-	mn.MetricGroup = []byte("rps")
-	for i := 0; i < 2; i++ {
-		key := fmt.Sprintf("key_%d", i)
-		value := fmt.Sprintf("value_%d", i)
-		mn.AddTag(key, value)
-	}
-	var tsid TSID
-	var metricName []byte
-	is := db.getIndexSearch()
-	defer db.putIndexSearch(is)
-	for i := 0; i < recordsCount; i++ {
-		mn.sortTags()
-		metricName = mn.Marshal(metricName[:0])
-		if err := is.GetOrCreateTSIDByName(&tsid, metricName); err != nil {
-			b.Fatalf("cannot insert record: %s", err)
-		}
-	}
-
-	b.SetBytes(1)
-	b.ReportAllocs()
-	b.ResetTimer()
-	b.RunParallel(func(pb *testing.PB) {
-		tags := []Tag{
-			{[]byte("key_0"), []byte("value_0")},
-			{[]byte("key_1"), []byte("value_1")},
-		}
-		var tfs TagFilters
-		tfss := []*TagFilters{&tfs}
-		i := 0
-		for pb.Next() {
-			tfs.Reset()
-			for j := range tags {
-				if err := tfs.Add(tags[j].Key, tags[j].Value, false, false); err != nil {
-					panic(fmt.Errorf("BUG: unexpected error: %s", err))
-				}
-			}
-			tsids, err := db.searchTSIDs(tfss, TimeRange{}, 1e5)
-			if err != nil {
-				panic(fmt.Errorf("unexpected error in search for tfs=%s: %s", &tfs, err))
-			}
-			if len(tsids) == 0 && i < recordsCount {
-				panic(fmt.Errorf("zero tsids found for tfs=%s", &tfs))
-			}
-			i++
-		}
-	})
-}
-
 func BenchmarkIndexDBGetTSIDs(b *testing.B) {
-	metricIDCache := fastcache.New(1234)
-	metricNameCache := fastcache.New(1234)
-	defer metricIDCache.Reset()
-	defer metricNameCache.Reset()
+	metricIDCache := workingsetcache.New(1234, time.Hour)
+	metricNameCache := workingsetcache.New(1234, time.Hour)
+	defer metricIDCache.Stop()
+	defer metricNameCache.Stop()
 
 	var hmCurr atomic.Value
 	hmCurr.Store(&hourMetricIDs{})

lib/storage/part_search.go

@@ -51,6 +51,7 @@ func (ps *partSearch) reset() {
 	ps.p = nil
 	ps.tsids = ps.tsids[:0]
 	ps.tsidIdx = 0
+	ps.fetchData = true
 	ps.metaindex = nil
 	ps.ibCache = nil
 	ps.bhs = nil

lib/storage/partition.go

@@ -1008,7 +1008,7 @@ func (pt *partition) mergeParts(pws []*partWrapper, stopCh <-chan struct{}) erro
 	}
 	fmt.Fprintf(&bb, "%s -> %s\n", tmpPartPath, dstPartPath)
 	txnPath := fmt.Sprintf("%s/txn/%016X", ptPath, mergeIdx)
-	if err := fs.WriteFile(txnPath, bb.B); err != nil {
+	if err := fs.WriteFileAtomically(txnPath, bb.B); err != nil {
 		return fmt.Errorf("cannot create transaction file %q: %s", txnPath, err)
 	}
 
@@ -1367,7 +1367,12 @@ func runTransactions(txnLock *sync.RWMutex, pathPrefix1, pathPrefix2, path strin
 	})
 
 	for _, fi := range fis {
-		txnPath := txnDir + "/" + fi.Name()
+		fn := fi.Name()
+		if fs.IsTemporaryFileName(fn) {
+			// Skip temporary files, which could be left after unclean shutdown.
+			continue
+		}
+		txnPath := txnDir + "/" + fn
 		if err := runTransaction(txnLock, pathPrefix1, pathPrefix2, txnPath); err != nil {
 			return fmt.Errorf("cannot run transaction from %q: %s", txnPath, err)
 		}

lib/storage/storage.go

@@ -20,8 +20,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 	"github.com/VictoriaMetrics/fastcache"
-	"golang.org/x/sys/unix"
 )
 
 const maxRetentionMonths = 12 * 100
@@ -40,16 +40,16 @@ type Storage struct {
 	tb *table
 
 	// tsidCache is MetricName -> TSID cache.
-	tsidCache *fastcache.Cache
+	tsidCache *workingsetcache.Cache
 
 	// metricIDCache is MetricID -> TSID cache.
-	metricIDCache *fastcache.Cache
+	metricIDCache *workingsetcache.Cache
 
 	// metricNameCache is MetricID -> MetricName cache.
-	metricNameCache *fastcache.Cache
+	metricNameCache *workingsetcache.Cache
 
 	// dateMetricIDCache is (Date, MetricID) cache.
-	dateMetricIDCache *fastcache.Cache
+	dateMetricIDCache *workingsetcache.Cache
 
 	// Fast cache for MetricID values occured during the current hour.
 	currHourMetricIDs atomic.Value
@@ -68,6 +68,10 @@ type Storage struct {
 
 	tooSmallTimestampRows uint64
 	tooBigTimestampRows   uint64
+
+	addRowsConcurrencyLimitReached uint64
+	addRowsConcurrencyLimitTimeout uint64
+	addRowsConcurrencyDroppedRows  uint64
 }
 
 // OpenStorage opens storage on the given path with the given number of retention months.
@@ -99,13 +103,10 @@ func OpenStorage(path string, retentionMonths int) (*Storage, error) {
 		return nil, fmt.Errorf("cannot create %q: %s", snapshotsPath, err)
 	}
 
-	flockFile := path + "/flock.lock"
-	flockF, err := os.Create(flockFile)
+	// Protect from concurrent opens.
+	flockF, err := fs.CreateFlockFile(path)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create lock file %q: %s", flockFile, err)
-	}
-	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
-		return nil, fmt.Errorf("cannot acquire lock on file %q: %s", flockFile, err)
+		return nil, err
 	}
 	s.flockF = flockF
 
@@ -277,6 +278,12 @@ type Metrics struct {
 	TooSmallTimestampRows uint64
 	TooBigTimestampRows   uint64
 
+	AddRowsConcurrencyLimitReached uint64
+	AddRowsConcurrencyLimitTimeout uint64
+	AddRowsConcurrencyDroppedRows  uint64
+	AddRowsConcurrencyCapacity     uint64
+	AddRowsConcurrencyCurrent      uint64
+
 	TSIDCacheSize       uint64
 	TSIDCacheSizeBytes  uint64
 	TSIDCacheRequests   uint64
@@ -317,6 +324,12 @@ func (s *Storage) UpdateMetrics(m *Metrics) {
 	m.TooSmallTimestampRows += atomic.LoadUint64(&s.tooSmallTimestampRows)
 	m.TooBigTimestampRows += atomic.LoadUint64(&s.tooBigTimestampRows)
 
+	m.AddRowsConcurrencyLimitReached += atomic.LoadUint64(&s.addRowsConcurrencyLimitReached)
+	m.AddRowsConcurrencyLimitTimeout += atomic.LoadUint64(&s.addRowsConcurrencyLimitTimeout)
+	m.AddRowsConcurrencyDroppedRows += atomic.LoadUint64(&s.addRowsConcurrencyDroppedRows)
+	m.AddRowsConcurrencyCapacity = uint64(cap(addRowsConcurrencyCh))
+	m.AddRowsConcurrencyCurrent = uint64(len(addRowsConcurrencyCh))
+
 	var cs fastcache.Stats
 	s.tsidCache.UpdateStats(&cs)
 	m.TSIDCacheSize += cs.EntriesCount
@@ -448,10 +461,10 @@ func (s *Storage) MustClose() {
 	s.idb().MustClose()
 
 	// Save caches.
-	s.mustSaveCache(s.tsidCache, "MetricName->TSID", "metricName_tsid")
-	s.mustSaveCache(s.metricIDCache, "MetricID->TSID", "metricID_tsid")
-	s.mustSaveCache(s.metricNameCache, "MetricID->MetricName", "metricID_metricName")
-	s.mustSaveCache(s.dateMetricIDCache, "Date->MetricID", "date_metricID")
+	s.mustSaveAndStopCache(s.tsidCache, "MetricName->TSID", "metricName_tsid")
+	s.mustSaveAndStopCache(s.metricIDCache, "MetricID->TSID", "metricID_tsid")
+	s.mustSaveAndStopCache(s.metricNameCache, "MetricID->MetricName", "metricID_metricName")
+	s.mustSaveAndStopCache(s.dateMetricIDCache, "Date->MetricID", "date_metricID")
 
 	hmCurr := s.currHourMetricIDs.Load().(*hourMetricIDs)
 	s.mustSaveHourMetricIDs(hmCurr, "curr_hour_metric_ids")
@@ -530,11 +543,11 @@ func (s *Storage) mustSaveHourMetricIDs(hm *hourMetricIDs, name string) {
 	logger.Infof("saved %s to %q in %s; entriesCount: %d; sizeBytes: %d", name, path, time.Since(startTime), len(hm.m), len(dst))
 }
 
-func (s *Storage) mustLoadCache(info, name string, sizeBytes int) *fastcache.Cache {
+func (s *Storage) mustLoadCache(info, name string, sizeBytes int) *workingsetcache.Cache {
 	path := s.cachePath + "/" + name
 	logger.Infof("loading %s cache from %q...", info, path)
 	startTime := time.Now()
-	c := fastcache.LoadFromFileOrNew(path, sizeBytes)
+	c := workingsetcache.Load(path, sizeBytes, time.Hour)
 	var cs fastcache.Stats
 	c.UpdateStats(&cs)
 	logger.Infof("loaded %s cache from %q in %s; entriesCount: %d; sizeBytes: %d",
@@ -542,17 +555,16 @@ func (s *Storage) mustLoadCache(info, name string, sizeBytes int) *fastcache.Cac
 	return c
 }
 
-func (s *Storage) mustSaveCache(c *fastcache.Cache, info, name string) {
-	gomaxprocs := runtime.GOMAXPROCS(-1)
+func (s *Storage) mustSaveAndStopCache(c *workingsetcache.Cache, info, name string) {
 	path := s.cachePath + "/" + name
 	logger.Infof("saving %s cache to %q...", info, path)
 	startTime := time.Now()
-	if err := c.SaveToFileConcurrent(path, gomaxprocs); err != nil {
+	if err := c.Save(path); err != nil {
 		logger.Panicf("FATAL: cannot save %s cache to %q: %s", info, path, err)
 	}
 	var cs fastcache.Stats
 	c.UpdateStats(&cs)
-	c.Reset()
+	c.Stop()
 	logger.Infof("saved %s cache to %q in %s; entriesCount: %d; sizeBytes: %d",
 		info, path, time.Since(startTime), cs.EntriesCount, cs.BytesSize)
 }
@@ -722,15 +734,24 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 	// Limit the number of concurrent goroutines that may add rows to the storage.
 	// This should prevent from out of memory errors and CPU trashing when too many
 	// goroutines call AddRows.
-	t := timerpool.Get(addRowsTimeout)
 	select {
 	case addRowsConcurrencyCh <- struct{}{}:
-		timerpool.Put(t)
 		defer func() { <-addRowsConcurrencyCh }()
-	case <-t.C:
-		timerpool.Put(t)
-		return fmt.Errorf("Cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers. Add more CPUs or reduce load",
-			len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh))
+	default:
+		// Sleep for a while until giving up
+		atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1)
+		t := timerpool.Get(addRowsTimeout)
+		select {
+		case addRowsConcurrencyCh <- struct{}{}:
+			timerpool.Put(t)
+			defer func() { <-addRowsConcurrencyCh }()
+		case <-t.C:
+			timerpool.Put(t)
+			atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1)
+			atomic.AddUint64(&s.addRowsConcurrencyDroppedRows, uint64(len(mrs)))
+			return fmt.Errorf("Cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers. Add more CPUs or reduce load",
+				len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh))
+		}
 	}
 
 	// Add rows to the storage.
@@ -950,7 +971,7 @@ func (s *Storage) putTSIDToCache(tsid *TSID, metricName []byte) {
 	s.tsidCache.Set(metricName, buf)
 }
 
-func openIndexDBTables(path string, metricIDCache, metricNameCache *fastcache.Cache, currHourMetricIDs, prevHourMetricIDs *atomic.Value) (curr, prev *indexDB, err error) {
+func openIndexDBTables(path string, metricIDCache, metricNameCache *workingsetcache.Cache, currHourMetricIDs, prevHourMetricIDs *atomic.Value) (curr, prev *indexDB, err error) {
 	if err := fs.MkdirAllIfNotExist(path); err != nil {
 		return nil, nil, fmt.Errorf("cannot create directory %q: %s", path, err)
 	}

lib/storage/table.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"golang.org/x/sys/unix"
 )
 
 // table represents a single table with time series data.
@@ -84,13 +83,10 @@ func openTable(path string, retentionMonths int, getDeletedMetricIDs func() map[
 		return nil, fmt.Errorf("cannot create directory for table %q: %s", path, err)
 	}
 
-	flockFile := path + "/flock.lock"
-	flockF, err := os.Create(flockFile)
+	// Protect from concurrent opens.
+	flockF, err := fs.CreateFlockFile(path)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create lock file %q: %s", flockFile, err)
-	}
-	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
-		return nil, fmt.Errorf("cannot acquire lock on file %q: %s", flockFile, err)
+		return nil, err
 	}
 
 	// Create directories for small and big partitions if they don't exist yet.

lib/workingsetcache/cache.go

@@ -0,0 +1,255 @@
+package workingsetcache
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/VictoriaMetrics/fastcache"
+)
+
+// Cache is a cache for working set entries.
+//
+// The cache evicts inactive entries after the given expireDuration.
+// Recently accessed entries survive expireDuration.
+//
+// Comparing to fastcache, this cache minimizes the required RAM size
+// to values smaller than maxBytes.
+type Cache struct {
+	curr atomic.Value
+	prev atomic.Value
+
+	// skipPrev indicates whether to use only curr and skip prev.
+	//
+	// This flag is set if curr is filled for more than 50% space.
+	// In this case using prev would result in RAM waste,
+	// it is better to use only curr cache with doubled size.
+	skipPrev uint64
+
+	// mu serializes access to curr, prev and skipPrev
+	// in expirationWorker and cacheSizeWatcher.
+	mu sync.Mutex
+
+	wg     sync.WaitGroup
+	stopCh chan struct{}
+
+	misses uint64
+}
+
+// Load loads the cache from filePath and limits its size to maxBytes
+// and evicts inactive entires after expireDuration.
+//
+// Stop must be called on the returned cache when it is no longer needed.
+func Load(filePath string, maxBytes int, expireDuration time.Duration) *Cache {
+	// Split maxBytes between curr and prev caches.
+	maxBytes /= 2
+	curr := fastcache.LoadFromFileOrNew(filePath, maxBytes)
+	return newWorkingSetCache(curr, maxBytes, expireDuration)
+}
+
+// New creates new cache with the given maxBytes size and the given expireDuration
+// for inactive entries.
+//
+// Stop must be called on the returned cache when it is no longer needed.
+func New(maxBytes int, expireDuration time.Duration) *Cache {
+	// Split maxBytes between curr and prev caches.
+	maxBytes /= 2
+	curr := fastcache.New(maxBytes)
+	return newWorkingSetCache(curr, maxBytes, expireDuration)
+}
+
+func newWorkingSetCache(curr *fastcache.Cache, maxBytes int, expireDuration time.Duration) *Cache {
+	prev := fastcache.New(1024)
+	var c Cache
+	c.curr.Store(curr)
+	c.prev.Store(prev)
+	c.stopCh = make(chan struct{})
+
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		c.expirationWorker(maxBytes, expireDuration)
+	}()
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		c.cacheSizeWatcher(maxBytes)
+	}()
+	return &c
+}
+
+func (c *Cache) expirationWorker(maxBytes int, expireDuration time.Duration) {
+	t := time.NewTicker(expireDuration / 2)
+	for {
+		select {
+		case <-c.stopCh:
+			t.Stop()
+			return
+		case <-t.C:
+		}
+
+		c.mu.Lock()
+		if atomic.LoadUint64(&c.skipPrev) != 0 {
+			// Expire prev cache and create fresh curr cache.
+			// Do not reuse prev cache, since it can have too big capacity.
+			prev := c.prev.Load().(*fastcache.Cache)
+			prev.Reset()
+			curr := c.curr.Load().(*fastcache.Cache)
+			c.prev.Store(curr)
+			curr = fastcache.New(maxBytes)
+			c.curr.Store(curr)
+		}
+		c.mu.Unlock()
+	}
+}
+
+func (c *Cache) cacheSizeWatcher(maxBytes int) {
+	t := time.NewTicker(time.Minute)
+	for {
+		select {
+		case <-c.stopCh:
+			t.Stop()
+			return
+		case <-t.C:
+		}
+		var cs fastcache.Stats
+		curr := c.curr.Load().(*fastcache.Cache)
+		curr.UpdateStats(&cs)
+		if cs.BytesSize < uint64(maxBytes)/2 {
+			continue
+		}
+
+		// curr cache size exceeds 50% of its capacity. It is better
+		// to double the size of curr cache and stop using prev cache,
+		// since this will result in higher summary cache capacity.
+		c.mu.Lock()
+		curr.Reset()
+		prev := c.prev.Load().(*fastcache.Cache)
+		prev.Reset()
+		curr = fastcache.New(maxBytes * 2)
+		c.curr.Store(curr)
+		atomic.StoreUint64(&c.skipPrev, 1)
+		c.mu.Unlock()
+		return
+	}
+}
+
+// Save safes the cache to filePath.
+func (c *Cache) Save(filePath string) error {
+	curr := c.curr.Load().(*fastcache.Cache)
+	concurrency := runtime.GOMAXPROCS(-1)
+	return curr.SaveToFileConcurrent(filePath, concurrency)
+}
+
+// Stop stops the cache.
+//
+// The cache cannot be used after the Stop call.
+func (c *Cache) Stop() {
+	close(c.stopCh)
+	c.wg.Wait()
+
+	c.Reset()
+}
+
+// Reset resets the cache.
+func (c *Cache) Reset() {
+	prev := c.prev.Load().(*fastcache.Cache)
+	prev.Reset()
+	curr := c.curr.Load().(*fastcache.Cache)
+	curr.Reset()
+
+	c.misses = 0
+}
+
+// UpdateStats updates fcs with cache stats.
+func (c *Cache) UpdateStats(fcs *fastcache.Stats) {
+	curr := c.curr.Load().(*fastcache.Cache)
+	fcsOrig := *fcs
+	curr.UpdateStats(fcs)
+	if atomic.LoadUint64(&c.skipPrev) != 0 {
+		return
+	}
+
+	fcs.Misses = fcsOrig.Misses + atomic.LoadUint64(&c.misses)
+	fcsOrig.Reset()
+	prev := c.prev.Load().(*fastcache.Cache)
+	prev.UpdateStats(&fcsOrig)
+	fcs.EntriesCount += fcsOrig.EntriesCount
+	fcs.BytesSize += fcsOrig.BytesSize
+}
+
+// Get appends the found value for the given key to dst and returns the result.
+func (c *Cache) Get(dst, key []byte) []byte {
+	curr := c.curr.Load().(*fastcache.Cache)
+	result := curr.Get(dst, key)
+	if len(result) > len(dst) {
+		// Fast path - the entry is found in the current cache.
+		return result
+	}
+	if atomic.LoadUint64(&c.skipPrev) != 0 {
+		return result
+	}
+
+	// Search for the entry in the previous cache.
+	prev := c.prev.Load().(*fastcache.Cache)
+	result = prev.Get(dst, key)
+	if len(result) <= len(dst) {
+		// Nothing found.
+		atomic.AddUint64(&c.misses, 1)
+		return result
+	}
+	// Cache the found entry in the current cache.
+	curr.Set(key, result[len(dst):])
+	return result
+}
+
+// Has verifies whether the cahce contains the given key.
+func (c *Cache) Has(key []byte) bool {
+	curr := c.curr.Load().(*fastcache.Cache)
+	if curr.Has(key) {
+		return true
+	}
+	if atomic.LoadUint64(&c.skipPrev) != 0 {
+		return false
+	}
+	prev := c.prev.Load().(*fastcache.Cache)
+	return prev.Has(key)
+}
+
+// Set sets the given value for the given key.
+func (c *Cache) Set(key, value []byte) {
+	curr := c.curr.Load().(*fastcache.Cache)
+	curr.Set(key, value)
+}
+
+// GetBig appends the found value for the given key to dst and returns the result.
+func (c *Cache) GetBig(dst, key []byte) []byte {
+	curr := c.curr.Load().(*fastcache.Cache)
+	result := curr.GetBig(dst, key)
+	if len(result) > len(dst) {
+		// Fast path - the entry is found in the current cache.
+		return result
+	}
+	if atomic.LoadUint64(&c.skipPrev) != 0 {
+		return result
+	}
+
+	// Search for the entry in the previous cache.
+	prev := c.prev.Load().(*fastcache.Cache)
+	result = prev.GetBig(dst, key)
+	if len(result) <= len(dst) {
+		// Nothing found.
+		atomic.AddUint64(&c.misses, 1)
+		return result
+	}
+	// Cache the found entry in the current cache.
+	curr.SetBig(key, result[len(dst):])
+	return result
+}
+
+// SetBig sets the given value for the given key.
+func (c *Cache) SetBig(key, value []byte) {
+	curr := c.curr.Load().(*fastcache.Cache)
+	curr.SetBig(key, value)
+}

vendor/github.com/VictoriaMetrics/metrics/process_metrics_linux.go

@@ -67,7 +67,11 @@ func writeProcessMetrics(w io.Writer) {
 	// It is expensive obtaining `process_open_fds` when big number of file descriptors is opened,
 	// don't do it here.
 
-	fmt.Fprintf(w, "process_cpu_seconds_total %g\n", float64(p.Utime+p.Stime)/userHZ)
+	utime := float64(p.Utime) / userHZ
+	stime := float64(p.Stime) / userHZ
+	fmt.Fprintf(w, "process_cpu_seconds_system_total %g\n", stime)
+	fmt.Fprintf(w, "process_cpu_seconds_total %g\n", utime+stime)
+	fmt.Fprintf(w, "process_cpu_seconds_user_total %g\n", utime)
 	fmt.Fprintf(w, "process_major_pagefaults_total %d\n", p.Majflt)
 	fmt.Fprintf(w, "process_minor_pagefaults_total %d\n", p.Minflt)
 	fmt.Fprintf(w, "process_num_threads %d\n", p.NumThreads)

vendor/github.com/klauspost/compress/LICENSE

@@ -1,4 +1,5 @@
 Copyright (c) 2012 The Go Authors. All rights reserved.
+Copyright (c) 2019 Klaus Post. All rights reserved.
 
 Redistribution and use in source and binary forms, with or without
 modification, are permitted provided that the following conditions are

vendor/github.com/klauspost/compress/huff0/decompress.go

@@ -247,9 +247,13 @@ func (s *Scratch) Decompress4X(in []byte, dstSize int) (out []byte, err error) {
 	dstOut := s.Out
 	dstEvery := (dstSize + 3) / 4
 
+	const tlSize = 1 << tableLogMax
+	const tlMask = tlSize - 1
+	single := s.dt.single[:tlSize]
+
 	decode := func(br *bitReader) byte {
 		val := br.peekBitsFast(s.actualTableLog) /* note : actualTableLog >= 1 */
-		v := s.dt.single[val]
+		v := single[val&tlMask]
 		br.bitsRead += v.nBits
 		return v.byte
 	}

vendor/github.com/klauspost/compress/zstd/README.md

@@ -34,7 +34,8 @@ For now, a high speed (fastest) and medium-fast (default) compressor has been im
 The "Fastest" compression ratio is roughly equivalent to zstd level 1. 
 The "Default" compression ration is roughly equivalent to zstd level 3 (default).
 
-In terms of speed, it is typically 2x as fast as the stdlib deflate/gzip in its fastest mode. The compression ratio compared to stdlib is around level 3, but usually 3x as fast.
+In terms of speed, it is typically 2x as fast as the stdlib deflate/gzip in its fastest mode. 
+The compression ratio compared to stdlib is around level 3, but usually 3x as fast.
 
 Compared to cgo zstd, the speed is around level 3 (default), but compression slightly worse, between level 1&2.
 
@@ -217,7 +218,8 @@ silesia.tar zstd    3   211947520   66793301    1377    146.79
 
 As part of the development process a *Snappy* -> *Zstandard* converter was also built.
 
-This can convert a *framed* [Snappy Stream](https://godoc.org/github.com/golang/snappy#Writer) to a zstd stream. Note that a single block is not framed.
+This can convert a *framed* [Snappy Stream](https://godoc.org/github.com/golang/snappy#Writer) to a zstd stream. 
+Note that a single block is not framed.
 
 Conversion is done by converting the stream directly from Snappy without intermediate full decoding.
 Therefore the compression ratio is much less than what can be done by a full decompression

vendor/github.com/klauspost/compress/zstd/blockenc.go

@@ -155,14 +155,17 @@ func (h *literalsHeader) setSize(regenLen int) {
 }
 
 // setSizes will set the size of a compressed literals section and the input length.
-func (h *literalsHeader) setSizes(compLen, inLen int) {
+func (h *literalsHeader) setSizes(compLen, inLen int, single bool) {
 	compBits, inBits := bits.Len32(uint32(compLen)), bits.Len32(uint32(inLen))
 	// Only retain 2 bits
 	const mask = 3
 	lh := uint64(*h & mask)
 	switch {
 	case compBits <= 10 && inBits <= 10:
-		lh |= (1 << 2) | (uint64(inLen) << 4) | (uint64(compLen) << (10 + 4)) | (3 << 60)
+		if !single {
+			lh |= 1 << 2
+		}
+		lh |= (uint64(inLen) << 4) | (uint64(compLen) << (10 + 4)) | (3 << 60)
 		if debug {
 			const mmask = (1 << 24) - 1
 			n := (lh >> 4) & mmask
@@ -175,8 +178,14 @@ func (h *literalsHeader) setSizes(compLen, inLen int) {
 		}
 	case compBits <= 14 && inBits <= 14:
 		lh |= (2 << 2) | (uint64(inLen) << 4) | (uint64(compLen) << (14 + 4)) | (4 << 60)
+		if single {
+			panic("single stream used with more than 10 bits length.")
+		}
 	case compBits <= 18 && inBits <= 18:
 		lh |= (3 << 2) | (uint64(inLen) << 4) | (uint64(compLen) << (18 + 4)) | (5 << 60)
+		if single {
+			panic("single stream used with more than 10 bits length.")
+		}
 	default:
 		panic("internal error: block too big")
 	}
@@ -307,12 +316,30 @@ func (b *blockEnc) encodeLits() error {
 		return nil
 	}
 
-	// TODO: Switch to 1X when less than x bytes.
-	out, reUsed, err := huff0.Compress4X(b.literals, b.litEnc)
-	// Bail out of compression is too little.
-	if len(out) > (len(b.literals) - len(b.literals)>>4) {
+	var (
+		out            []byte
+		reUsed, single bool
+		err            error
+	)
+	if len(b.literals) >= 1024 {
+		// Use 4 Streams.
+		out, reUsed, err = huff0.Compress4X(b.literals, b.litEnc)
+		if len(out) > len(b.literals)-len(b.literals)>>4 {
+			// Bail out of compression is too little.
+			err = huff0.ErrIncompressible
+		}
+	} else if len(b.literals) > 32 {
+		// Use 1 stream
+		single = true
+		out, reUsed, err = huff0.Compress1X(b.literals, b.litEnc)
+		if len(out) > len(b.literals)-len(b.literals)>>4 {
+			// Bail out of compression is too little.
+			err = huff0.ErrIncompressible
+		}
+	} else {
 		err = huff0.ErrIncompressible
 	}
+
 	switch err {
 	case huff0.ErrIncompressible:
 		if debug {
@@ -351,7 +378,7 @@ func (b *blockEnc) encodeLits() error {
 		lh.setType(literalsBlockCompressed)
 	}
 	// Set sizes
-	lh.setSizes(len(out), len(b.literals))
+	lh.setSizes(len(out), len(b.literals), single)
 	bh.setSize(uint32(len(out) + lh.size() + 1))
 
 	// Write block headers.
@@ -381,16 +408,23 @@ func (b *blockEnc) encode() error {
 	b.output = bh.appendTo(b.output)
 
 	var (
-		out    []byte
-		reUsed bool
-		err    error
+		out            []byte
+		reUsed, single bool
+		err            error
 	)
-	if len(b.literals) > 32 {
-		// TODO: Switch to 1X on small blocks.
+	if len(b.literals) >= 1024 {
+		// Use 4 Streams.
 		out, reUsed, err = huff0.Compress4X(b.literals, b.litEnc)
 		if len(out) > len(b.literals)-len(b.literals)>>4 {
 			err = huff0.ErrIncompressible
 		}
+	} else if len(b.literals) > 32 {
+		// Use 1 stream
+		single = true
+		out, reUsed, err = huff0.Compress1X(b.literals, b.litEnc)
+		if len(out) > len(b.literals)-len(b.literals)>>4 {
+			err = huff0.ErrIncompressible
+		}
 	} else {
 		err = huff0.ErrIncompressible
 	}
@@ -435,7 +469,7 @@ func (b *blockEnc) encode() error {
 				}
 			}
 		}
-		lh.setSizes(len(out), len(b.literals))
+		lh.setSizes(len(out), len(b.literals), single)
 		if debug {
 			printf("Compressed %d literals to %d bytes", len(b.literals), len(out))
 			println("Adding literal header:", lh)

vendor/github.com/klauspost/compress/zstd/decoder.go

@@ -281,17 +281,17 @@ func (d *Decoder) DecodeAll(input, dst []byte) ([]byte, error) {
 		}
 		d.decoders <- block
 		frame.rawInput = nil
+		frame.bBuf = nil
 		d.frames <- frame
 	}()
+	frame.bBuf = input
 	if cap(dst) == 0 {
 		// Allocate 1MB by default if nothing is provided.
 		dst = make([]byte, 0, 1<<20)
 	}
 
-	// Allocation here:
-	br := byteBuf(input)
 	for {
-		err := frame.reset(&br)
+		err := frame.reset(&frame.bBuf)
 		if err == io.EOF {
 			return dst, nil
 		}
@@ -313,7 +313,7 @@ func (d *Decoder) DecodeAll(input, dst []byte) ([]byte, error) {
 		if err != nil {
 			return dst, err
 		}
-		if len(br) == 0 {
+		if len(frame.bBuf) == 0 {
 			break
 		}
 	}

vendor/github.com/klauspost/compress/zstd/enc_dfast.go

@@ -82,16 +82,11 @@ func (e *doubleFastEncoder) Encode(blk *blockEnc, src []byte) {
 		stepSize++
 	}
 
-	// TEMPLATE
-
 	const kSearchStrength = 8
 
 	// nextEmit is where in src the next emitLiteral should start from.
 	nextEmit := s
 	cv := load6432(src, s)
-	// nextHash is the hash at s
-	nextHashS := hash5(cv, dFastShortTableBits)
-	nextHashL := hash8(cv, dFastLongTableBits)
 
 	// Relative offsets
 	offset1 := int32(blk.recentOffsets[0])
@@ -119,8 +114,8 @@ encodeLoop:
 				panic("offset0 was 0")
 			}
 
-			nextHashS = nextHashS & dFastShortTableMask
-			nextHashL = nextHashL & dFastLongTableMask
+			nextHashS := hash5(cv, dFastShortTableBits)
+			nextHashL := hash8(cv, dFastLongTableBits)
 			candidateL := e.longTable[nextHashL]
 			candidateS := e.table[nextHashS]
 
@@ -172,8 +167,6 @@ encodeLoop:
 						break encodeLoop
 					}
 					cv = load6432(src, s)
-					nextHashS = hash5(cv, dFastShortTableBits)
-					nextHashL = hash8(cv, dFastLongTableBits)
 					continue
 				}
 				const repOff2 = 1
@@ -221,8 +214,6 @@ encodeLoop:
 						break encodeLoop
 					}
 					cv = load6432(src, s)
-					nextHashS = hash5(cv, dFastShortTableBits)
-					nextHashL = hash8(cv, dFastLongTableBits)
 					// Swap offsets
 					offset1, offset2 = offset2, offset1
 					continue
@@ -296,8 +287,6 @@ encodeLoop:
 				break encodeLoop
 			}
 			cv = load6432(src, s)
-			nextHashS = hash5(cv, dFastShortTableBits)
-			nextHashL = hash8(cv, dFastLongTableBits)
 		}
 
 		// A 4-byte match has been found. Update recent offsets.
@@ -354,20 +343,18 @@ encodeLoop:
 		cv1 := load6432(src, index1)
 		te0 := tableEntry{offset: index0 + e.cur, val: uint32(cv0)}
 		te1 := tableEntry{offset: index1 + e.cur, val: uint32(cv1)}
-		e.longTable[hash8(cv0, dFastLongTableBits)&dFastLongTableMask] = te0
-		e.longTable[hash8(cv1, dFastLongTableBits)&dFastLongTableMask] = te1
+		e.longTable[hash8(cv0, dFastLongTableBits)] = te0
+		e.longTable[hash8(cv1, dFastLongTableBits)] = te1
 		cv0 >>= 8
 		cv1 >>= 8
 		te0.offset++
 		te1.offset++
 		te0.val = uint32(cv0)
 		te1.val = uint32(cv1)
-		e.table[hash5(cv0, dFastShortTableBits)&dFastShortTableMask] = te0
-		e.table[hash5(cv1, dFastShortTableBits)&dFastShortTableMask] = te1
+		e.table[hash5(cv0, dFastShortTableBits)] = te0
+		e.table[hash5(cv1, dFastShortTableBits)] = te1
 
 		cv = load6432(src, s)
-		nextHashS = hash5(cv1>>8, dFastShortTableBits)
-		nextHashL = hash8(cv, dFastLongTableBits)
 
 		if !canRepeat {
 			continue
@@ -381,14 +368,17 @@ encodeLoop:
 				break
 			}
 
+			// Store this, since we have it.
+			nextHashS := hash5(cv1>>8, dFastShortTableBits)
+			nextHashL := hash8(cv, dFastLongTableBits)
+
 			// We have at least 4 byte match.
 			// No need to check backwards. We come straight from a match
 			l := 4 + e.matchlen(s+4, o2+4, src)
 
-			// Store this, since we have it.
 			entry := tableEntry{offset: s + e.cur, val: uint32(cv)}
-			e.longTable[nextHashL&dFastLongTableMask] = entry
-			e.table[nextHashS&dFastShortTableMask] = entry
+			e.longTable[nextHashL] = entry
+			e.table[nextHashS] = entry
 			seq.matchLen = uint32(l) - zstdMinMatch
 			seq.litLen = 0
 
@@ -408,8 +398,6 @@ encodeLoop:
 				break encodeLoop
 			}
 			cv = load6432(src, s)
-			nextHashS = hash5(cv, dFastShortTableBits)
-			nextHashL = hash8(cv, dFastLongTableBits)
 		}
 	}
 

vendor/github.com/klauspost/compress/zstd/enc_fast.go

@@ -124,8 +124,6 @@ func (e *fastEncoder) Encode(blk *blockEnc, src []byte) {
 	// nextEmit is where in src the next emitLiteral should start from.
 	nextEmit := s
 	cv := load6432(src, s)
-	// nextHash is the hash at s
-	nextHash := hash6(cv, hashLog)
 
 	// Relative offsets
 	offset1 := int32(blk.recentOffsets[0])
@@ -157,8 +155,8 @@ encodeLoop:
 				panic("offset0 was 0")
 			}
 
-			nextHash2 := hash6(cv>>8, hashLog) & tableMask
-			nextHash = nextHash & tableMask
+			nextHash := hash6(cv, hashLog)
+			nextHash2 := hash6(cv>>8, hashLog)
 			candidate := e.table[nextHash]
 			candidate2 := e.table[nextHash2]
 			repIndex := s - offset1 + 2
@@ -207,8 +205,6 @@ encodeLoop:
 					break encodeLoop
 				}
 				cv = load6432(src, s)
-				//nextHash = hashLen(cv, hashLog, mls)
-				nextHash = hash6(cv, hashLog)
 				continue
 			}
 			coffset0 := s - (candidate.offset - e.cur)
@@ -245,7 +241,6 @@ encodeLoop:
 				break encodeLoop
 			}
 			cv = load6432(src, s)
-			nextHash = hash6(cv, hashLog)
 		}
 		// A 4-byte match has been found. We'll later see if more than 4 bytes.
 		offset2 = offset1
@@ -292,15 +287,16 @@ encodeLoop:
 			break encodeLoop
 		}
 		cv = load6432(src, s)
-		nextHash = hash6(cv, hashLog)
 
 		// Check offset 2
-		if o2 := s - offset2; canRepeat && o2 > 0 && load3232(src, o2) == uint32(cv) {
+		if o2 := s - offset2; canRepeat && load3232(src, o2) == uint32(cv) {
 			// We have at least 4 byte match.
 			// No need to check backwards. We come straight from a match
 			l := 4 + e.matchlen(s+4, o2+4, src)
+
 			// Store this, since we have it.
-			e.table[nextHash&tableMask] = tableEntry{offset: s + e.cur, val: uint32(cv)}
+			nextHash := hash6(cv, hashLog)
+			e.table[nextHash] = tableEntry{offset: s + e.cur, val: uint32(cv)}
 			seq.matchLen = uint32(l) - zstdMinMatch
 			seq.litLen = 0
 			// Since litlen is always 0, this is offset 1.
@@ -319,7 +315,6 @@ encodeLoop:
 			}
 			// Prepare next loop.
 			cv = load6432(src, s)
-			nextHash = hash6(cv, hashLog)
 		}
 	}
 

vendor/github.com/klauspost/compress/zstd/framedec.go

@@ -39,6 +39,9 @@ type frameDec struct {
 
 	rawInput byteBuffer
 
+	// Byte buffer that can be reused for small input blocks.
+	bBuf byteBuf
+
 	// asyncRunning indicates whether the async routine processes input on 'decoding'.
 	asyncRunning   bool
 	asyncRunningMu sync.Mutex

vendor/github.com/klauspost/compress/zstd/fse_decoder.go

@@ -184,29 +184,75 @@ func (s *fseDecoder) readNCount(b *byteReader, maxSymbol uint16) error {
 // decSymbol contains information about a state entry,
 // Including the state offset base, the output symbol and
 // the number of bits to read for the low part of the destination state.
-type decSymbol struct {
-	newState uint16
-	addBits  uint8 // Used for symbols until transformed.
-	nbBits   uint8
-	baseline uint32
+// Using a composite uint64 is faster than a struct with separate members.
+type decSymbol uint64
+
+func newDecSymbol(nbits, addBits uint8, newState uint16, baseline uint32) decSymbol {
+	return decSymbol(nbits) | (decSymbol(addBits) << 8) | (decSymbol(newState) << 16) | (decSymbol(baseline) << 32)
+}
+
+func (d decSymbol) nbBits() uint8 {
+	return uint8(d)
+}
+
+func (d decSymbol) addBits() uint8 {
+	return uint8(d >> 8)
+}
+
+func (d decSymbol) newState() uint16 {
+	return uint16(d >> 16)
+}
+
+func (d decSymbol) baseline() uint32 {
+	return uint32(d >> 32)
+}
+
+func (d decSymbol) baselineInt() int {
+	return int(d >> 32)
+}
+
+func (d *decSymbol) set(nbits, addBits uint8, newState uint16, baseline uint32) {
+	*d = decSymbol(nbits) | (decSymbol(addBits) << 8) | (decSymbol(newState) << 16) | (decSymbol(baseline) << 32)
+}
+
+func (d *decSymbol) setNBits(nBits uint8) {
+	const mask = 0xffffffffffffff00
+	*d = (*d & mask) | decSymbol(nBits)
+}
+
+func (d *decSymbol) setAddBits(addBits uint8) {
+	const mask = 0xffffffffffff00ff
+	*d = (*d & mask) | (decSymbol(addBits) << 8)
+}
+
+func (d *decSymbol) setNewState(state uint16) {
+	const mask = 0xffffffff0000ffff
+	*d = (*d & mask) | decSymbol(state)<<16
+}
+
+func (d *decSymbol) setBaseline(baseline uint32) {
+	const mask = 0xffffffff
+	*d = (*d & mask) | decSymbol(baseline)<<32
+}
+
+func (d *decSymbol) setExt(addBits uint8, baseline uint32) {
+	const mask = 0xffff00ff
+	*d = (*d & mask) | (decSymbol(addBits) << 8) | (decSymbol(baseline) << 32)
 }
 
 // decSymbolValue returns the transformed decSymbol for the given symbol.
 func decSymbolValue(symb uint8, t []baseOffset) (decSymbol, error) {
 	if int(symb) >= len(t) {
-		return decSymbol{}, fmt.Errorf("rle symbol %d >= max %d", symb, len(t))
+		return 0, fmt.Errorf("rle symbol %d >= max %d", symb, len(t))
 	}
 	lu := t[symb]
-	return decSymbol{
-		addBits:  lu.addBits,
-		baseline: lu.baseLine,
-	}, nil
+	return newDecSymbol(0, lu.addBits, 0, lu.baseLine), nil
 }
 
 // setRLE will set the decoder til RLE mode.
 func (s *fseDecoder) setRLE(symbol decSymbol) {
 	s.actualTableLog = 0
-	s.maxBits = symbol.addBits
+	s.maxBits = symbol.addBits()
 	s.dt[0] = symbol
 }
 
@@ -220,7 +266,7 @@ func (s *fseDecoder) buildDtable() error {
 	{
 		for i, v := range s.norm[:s.symbolLen] {
 			if v == -1 {
-				s.dt[highThreshold].addBits = uint8(i)
+				s.dt[highThreshold].setAddBits(uint8(i))
 				highThreshold--
 				symbolNext[i] = 1
 			} else {
@@ -235,7 +281,7 @@ func (s *fseDecoder) buildDtable() error {
 		position := uint32(0)
 		for ss, v := range s.norm[:s.symbolLen] {
 			for i := 0; i < int(v); i++ {
-				s.dt[position].addBits = uint8(ss)
+				s.dt[position].setAddBits(uint8(ss))
 				position = (position + step) & tableMask
 				for position > highThreshold {
 					// lowprob area
@@ -253,11 +299,11 @@ func (s *fseDecoder) buildDtable() error {
 	{
 		tableSize := uint16(1 << s.actualTableLog)
 		for u, v := range s.dt[:tableSize] {
-			symbol := v.addBits
+			symbol := v.addBits()
 			nextState := symbolNext[symbol]
 			symbolNext[symbol] = nextState + 1
 			nBits := s.actualTableLog - byte(highBits(uint32(nextState)))
-			s.dt[u&maxTableMask].nbBits = nBits
+			s.dt[u&maxTableMask].setNBits(nBits)
 			newState := (nextState << nBits) - tableSize
 			if newState > tableSize {
 				return fmt.Errorf("newState (%d) outside table size (%d)", newState, tableSize)
@@ -266,7 +312,7 @@ func (s *fseDecoder) buildDtable() error {
 				// Seems weird that this is possible with nbits > 0.
 				return fmt.Errorf("newState (%d) == oldState (%d) and no bits", newState, u)
 			}
-			s.dt[u&maxTableMask].newState = newState
+			s.dt[u&maxTableMask].setNewState(newState)
 		}
 	}
 	return nil
@@ -279,25 +325,21 @@ func (s *fseDecoder) transform(t []baseOffset) error {
 	tableSize := uint16(1 << s.actualTableLog)
 	s.maxBits = 0
 	for i, v := range s.dt[:tableSize] {
-		if int(v.addBits) >= len(t) {
-			return fmt.Errorf("invalid decoding table entry %d, symbol %d >= max (%d)", i, v.addBits, len(t))
+		add := v.addBits()
+		if int(add) >= len(t) {
+			return fmt.Errorf("invalid decoding table entry %d, symbol %d >= max (%d)", i, v.addBits(), len(t))
 		}
-		lu := t[v.addBits]
+		lu := t[add]
 		if lu.addBits > s.maxBits {
 			s.maxBits = lu.addBits
 		}
-		s.dt[i&maxTableMask] = decSymbol{
-			newState: v.newState,
-			nbBits:   v.nbBits,
-			addBits:  lu.addBits,
-			baseline: lu.baseLine,
-		}
+		v.setExt(lu.addBits, lu.baseLine)
+		s.dt[i] = v
 	}
 	return nil
 }
 
 type fseState struct {
-	// TODO: Check if *[1 << maxTablelog]decSymbol is faster.
 	dt    []decSymbol
 	state decSymbol
 }
@@ -312,26 +354,31 @@ func (s *fseState) init(br *bitReader, tableLog uint8, dt []decSymbol) {
 // next returns the current symbol and sets the next state.
 // At least tablelog bits must be available in the bit reader.
 func (s *fseState) next(br *bitReader) {
-	lowBits := uint16(br.getBits(s.state.nbBits))
-	s.state = s.dt[s.state.newState+lowBits]
+	lowBits := uint16(br.getBits(s.state.nbBits()))
+	s.state = s.dt[s.state.newState()+lowBits]
 }
 
 // finished returns true if all bits have been read from the bitstream
 // and the next state would require reading bits from the input.
 func (s *fseState) finished(br *bitReader) bool {
-	return br.finished() && s.state.nbBits > 0
+	return br.finished() && s.state.nbBits() > 0
 }
 
 // final returns the current state symbol without decoding the next.
 func (s *fseState) final() (int, uint8) {
-	return int(s.state.baseline), s.state.addBits
+	return s.state.baselineInt(), s.state.addBits()
+}
+
+// final returns the current state symbol without decoding the next.
+func (s decSymbol) final() (int, uint8) {
+	return s.baselineInt(), s.addBits()
 }
 
 // nextFast returns the next symbol and sets the next state.
 // This can only be used if no symbols are 0 bits.
 // At least tablelog bits must be available in the bit reader.
 func (s *fseState) nextFast(br *bitReader) (uint32, uint8) {
-	lowBits := uint16(br.getBitsFast(s.state.nbBits))
-	s.state = s.dt[s.state.newState+lowBits]
-	return s.state.baseline, s.state.addBits
+	lowBits := uint16(br.getBitsFast(s.state.nbBits()))
+	s.state = s.dt[s.state.newState()+lowBits]
+	return s.state.baseline(), s.state.addBits()
 }

vendor/github.com/klauspost/compress/zstd/seqdec.go

@@ -89,6 +89,10 @@ func (s *sequenceDecs) initialize(br *bitReader, hist *history, literals, out []
 // decode sequences from the stream with the provided history.
 func (s *sequenceDecs) decode(seqs int, br *bitReader, hist []byte) error {
 	startSize := len(s.out)
+	// Grab full sizes tables, to avoid bounds checks.
+	llTable, mlTable, ofTable := s.litLengths.fse.dt[:maxTablesize], s.matchLengths.fse.dt[:maxTablesize], s.offsets.fse.dt[:maxTablesize]
+	llState, mlState, ofState := s.litLengths.state.state, s.matchLengths.state.state, s.offsets.state.state
+
 	for i := seqs - 1; i >= 0; i-- {
 		if br.overread() {
 			printf("reading sequence %d, exceeded available data\n", seqs-i)
@@ -96,10 +100,10 @@ func (s *sequenceDecs) decode(seqs int, br *bitReader, hist []byte) error {
 		}
 		var litLen, matchOff, matchLen int
 		if br.off > 4+((maxOffsetBits+16+16)>>3) {
-			litLen, matchOff, matchLen = s.nextFast(br)
+			litLen, matchOff, matchLen = s.nextFast(br, llState, mlState, ofState)
 			br.fillFast()
 		} else {
-			litLen, matchOff, matchLen = s.next(br)
+			litLen, matchOff, matchLen = s.next(br, llState, mlState, ofState)
 			br.fill()
 		}
 
@@ -175,30 +179,25 @@ func (s *sequenceDecs) decode(seqs int, br *bitReader, hist []byte) error {
 			// This is the last sequence, so we shouldn't update state.
 			break
 		}
-		if true {
-			// Manually inlined, ~ 5-20% faster
-			// Update all 3 states at once. Approx 20% faster.
-			a, b, c := s.litLengths.state.state, s.matchLengths.state.state, s.offsets.state.state
 
-			nBits := a.nbBits + b.nbBits + c.nbBits
-			if nBits == 0 {
-				s.litLengths.state.state = s.litLengths.state.dt[a.newState]
-				s.matchLengths.state.state = s.matchLengths.state.dt[b.newState]
-				s.offsets.state.state = s.offsets.state.dt[c.newState]
-			} else {
-				bits := br.getBitsFast(nBits)
-				lowBits := uint16(bits >> ((c.nbBits + b.nbBits) & 31))
-				s.litLengths.state.state = s.litLengths.state.dt[a.newState+lowBits]
-
-				lowBits = uint16(bits >> (c.nbBits & 31))
-				lowBits &= bitMask[b.nbBits&15]
-				s.matchLengths.state.state = s.matchLengths.state.dt[b.newState+lowBits]
-
-				lowBits = uint16(bits) & bitMask[c.nbBits&15]
-				s.offsets.state.state = s.offsets.state.dt[c.newState+lowBits]
-			}
+		// Manually inlined, ~ 5-20% faster
+		// Update all 3 states at once. Approx 20% faster.
+		nBits := llState.nbBits() + mlState.nbBits() + ofState.nbBits()
+		if nBits == 0 {
+			llState = llTable[llState.newState()&maxTableMask]
+			mlState = mlTable[mlState.newState()&maxTableMask]
+			ofState = ofTable[ofState.newState()&maxTableMask]
 		} else {
-			s.updateAlt(br)
+			bits := br.getBitsFast(nBits)
+			lowBits := uint16(bits >> ((ofState.nbBits() + mlState.nbBits()) & 31))
+			llState = llTable[(llState.newState()+lowBits)&maxTableMask]
+
+			lowBits = uint16(bits >> (ofState.nbBits() & 31))
+			lowBits &= bitMask[mlState.nbBits()&15]
+			mlState = mlTable[(mlState.newState()+lowBits)&maxTableMask]
+
+			lowBits = uint16(bits) & bitMask[ofState.nbBits()&15]
+			ofState = ofTable[(ofState.newState()+lowBits)&maxTableMask]
 		}
 	}
 
@@ -230,55 +229,49 @@ func (s *sequenceDecs) updateAlt(br *bitReader) {
 	// Update all 3 states at once. Approx 20% faster.
 	a, b, c := s.litLengths.state.state, s.matchLengths.state.state, s.offsets.state.state
 
-	nBits := a.nbBits + b.nbBits + c.nbBits
+	nBits := a.nbBits() + b.nbBits() + c.nbBits()
 	if nBits == 0 {
-		s.litLengths.state.state = s.litLengths.state.dt[a.newState]
-		s.matchLengths.state.state = s.matchLengths.state.dt[b.newState]
-		s.offsets.state.state = s.offsets.state.dt[c.newState]
+		s.litLengths.state.state = s.litLengths.state.dt[a.newState()]
+		s.matchLengths.state.state = s.matchLengths.state.dt[b.newState()]
+		s.offsets.state.state = s.offsets.state.dt[c.newState()]
 		return
 	}
 	bits := br.getBitsFast(nBits)
-	lowBits := uint16(bits >> ((c.nbBits + b.nbBits) & 31))
-	s.litLengths.state.state = s.litLengths.state.dt[a.newState+lowBits]
+	lowBits := uint16(bits >> ((c.nbBits() + b.nbBits()) & 31))
+	s.litLengths.state.state = s.litLengths.state.dt[a.newState()+lowBits]
 
-	lowBits = uint16(bits >> (c.nbBits & 31))
-	lowBits &= bitMask[b.nbBits&15]
-	s.matchLengths.state.state = s.matchLengths.state.dt[b.newState+lowBits]
+	lowBits = uint16(bits >> (c.nbBits() & 31))
+	lowBits &= bitMask[b.nbBits()&15]
+	s.matchLengths.state.state = s.matchLengths.state.dt[b.newState()+lowBits]
 
-	lowBits = uint16(bits) & bitMask[c.nbBits&15]
-	s.offsets.state.state = s.offsets.state.dt[c.newState+lowBits]
+	lowBits = uint16(bits) & bitMask[c.nbBits()&15]
+	s.offsets.state.state = s.offsets.state.dt[c.newState()+lowBits]
 }
 
 // nextFast will return new states when there are at least 4 unused bytes left on the stream when done.
-func (s *sequenceDecs) nextFast(br *bitReader) (ll, mo, ml int) {
+func (s *sequenceDecs) nextFast(br *bitReader, llState, mlState, ofState decSymbol) (ll, mo, ml int) {
 	// Final will not read from stream.
-	ll, llB := s.litLengths.state.final()
-	ml, mlB := s.matchLengths.state.final()
-	mo, moB := s.offsets.state.final()
+	ll, llB := llState.final()
+	ml, mlB := mlState.final()
+	mo, moB := ofState.final()
 
 	// extra bits are stored in reverse order.
 	br.fillFast()
-	if s.maxBits <= 32 {
-		mo += br.getBits(moB)
-		ml += br.getBits(mlB)
-		ll += br.getBits(llB)
-	} else {
-		mo += br.getBits(moB)
+	mo += br.getBits(moB)
+	if s.maxBits > 32 {
 		br.fillFast()
-		// matchlength+literal length, max 32 bits
-		ml += br.getBits(mlB)
-		ll += br.getBits(llB)
 	}
+	ml += br.getBits(mlB)
+	ll += br.getBits(llB)
 
-	// mo = s.adjustOffset(mo, ll, moB)
-	// Inlined for rather big speedup
 	if moB > 1 {
 		s.prevOffset[2] = s.prevOffset[1]
 		s.prevOffset[1] = s.prevOffset[0]
 		s.prevOffset[0] = mo
 		return
 	}
-
+	// mo = s.adjustOffset(mo, ll, moB)
+	// Inlined for rather big speedup
 	if ll == 0 {
 		// There is an exception though, when current sequence's literals_length = 0.
 		// In this case, repeated offsets are shifted by one, so an offset_value of 1 means Repeated_Offset2,
@@ -312,11 +305,11 @@ func (s *sequenceDecs) nextFast(br *bitReader) (ll, mo, ml int) {
 	return
 }
 
-func (s *sequenceDecs) next(br *bitReader) (ll, mo, ml int) {
+func (s *sequenceDecs) next(br *bitReader, llState, mlState, ofState decSymbol) (ll, mo, ml int) {
 	// Final will not read from stream.
-	ll, llB := s.litLengths.state.final()
-	ml, mlB := s.matchLengths.state.final()
-	mo, moB := s.offsets.state.final()
+	ll, llB := llState.final()
+	ml, mlB := mlState.final()
+	mo, moB := ofState.final()
 
 	// extra bits are stored in reverse order.
 	br.fill()

vendor/github.com/valyala/gozstd/Makefile

@@ -3,6 +3,7 @@ GOARCH ?= $(shell go env GOARCH)
 GOOS_GOARCH := $(GOOS)_$(GOARCH)
 GOOS_GOARCH_NATIVE := $(shell go env GOHOSTOS)_$(shell go env GOHOSTARCH)
 LIBZSTD_NAME := libzstd_$(GOOS_GOARCH).a
+ZSTD_VERSION ?= master
 
 .PHONY: libzstd.a
 
@@ -10,15 +11,15 @@ libzstd.a: $(LIBZSTD_NAME)
 
 $(LIBZSTD_NAME):
 ifeq ($(GOOS_GOARCH),$(GOOS_GOARCH_NATIVE))
-	cd zstd/lib && ZSTD_LEGACY_SUPPORT=0 $(MAKE) clean libzstd.a
+	cd zstd/lib && ZSTD_LEGACY_SUPPORT=0 MOREFLAGS=$(MOREFLAGS) $(MAKE) clean libzstd.a
 	mv zstd/lib/libzstd.a $(LIBZSTD_NAME)
 else
 ifeq ($(GOOS_GOARCH),linux_arm)
-	cd zstd/lib && CC=arm-linux-gnueabi-gcc ZSTD_LEGACY_SUPPORT=0 $(MAKE) clean libzstd.a
+	cd zstd/lib && CC=arm-linux-gnueabi-gcc ZSTD_LEGACY_SUPPORT=0 MOREFLAGS=$(MOREFLAGS) $(MAKE) clean libzstd.a
 	mv zstd/lib/libzstd.a libzstd_linux_arm.a
 endif
 ifeq ($(GOOS_GOARCH),linux_arm64)
-	cd zstd/lib && CC=aarch64-linux-gnu-gcc ZSTD_LEGACY_SUPPORT=0 $(MAKE) clean libzstd.a
+	cd zstd/lib && CC=aarch64-linux-gnu-gcc ZSTD_LEGACY_SUPPORT=0 MOREFLAGS=$(MOREFLAGS) $(MAKE) clean libzstd.a
 	mv zstd/lib/libzstd.a libzstd_linux_arm64.a
 endif
 endif
@@ -29,7 +30,7 @@ clean:
 
 update-zstd:
 	rm -rf zstd-tmp
-	git clone --branch master --depth 1 https://github.com/Facebook/zstd zstd-tmp
+	git clone --branch $(ZSTD_VERSION) --depth 1 https://github.com/Facebook/zstd zstd-tmp
 	rm -rf zstd-tmp/.git
 	rm -rf zstd
 	mv zstd-tmp zstd

vendor/github.com/valyala/gozstd/README.md

@@ -73,5 +73,8 @@ and [Reader](https://godoc.org/github.com/valyala/gozstd#Reader) for stream deco
   * Q: _I don't trust `libzstd*.a` binary files from the repo or these files dont't work on my OS/ARCH. How to rebuild them?_
     A: Just run `make clean libzstd.a` if your OS/ARCH is supported.
     
+  * Q: _How do I specify custom build flags when recompiling `libzstd*.a`?_
+    A: You can specify MOREFLAGS=... variable when running `make` like this: `MOREFLAGS=-fPIC make clean libzstd.a`.
+
   * Q: _Why the repo contains `libzstd*.a` binary files?_  
     A: This simplifies package installation to usual `go get` without additional steps for building the `libzstd*.a`

vendor/github.com/valyala/gozstd/dict.go

@@ -1,6 +1,8 @@
 package gozstd
 
 /*
+#cgo CFLAGS: -O3
+
 #define ZSTD_STATIC_LINKING_ONLY
 #include "zstd.h"
 
@@ -86,7 +88,10 @@ func BuildDict(samples [][]byte, desiredDictLen int) []byte {
 		&samplesSizes[0],
 		C.unsigned(len(samplesSizes)))
 	buildDictLock.Unlock()
-	ensureNoError("ZDICT_trainFromBuffer", result)
+	if C.ZDICT_isError(result) != 0 {
+		// Return empty dictionary, since the original samples are too small.
+		return nil
+	}
 
 	dictLen := int(result)
 	return dict[:dictLen]

vendor/github.com/valyala/gozstd/go.mod

@@ -1 +1,3 @@
 module github.com/valyala/gozstd
+
+go 1.12

vendor/github.com/valyala/gozstd/gozstd.go

@@ -66,72 +66,55 @@ func CompressDict(dst, src []byte, cd *CDict) []byte {
 }
 
 func compressDictLevel(dst, src []byte, cd *CDict, compressionLevel int) []byte {
-	compressInitOnce.Do(compressInit)
+	concurrencyLimitCh <- struct{}{}
+
+	var cctx, cctxDict *cctxWrapper
+	if cd == nil {
+		cctx = cctxPool.Get().(*cctxWrapper)
+	} else {
+		cctxDict = cctxDictPool.Get().(*cctxWrapper)
+	}
+
+	dst = compress(cctx, cctxDict, dst, src, cd, compressionLevel)
+
+	if cd == nil {
+		cctxPool.Put(cctx)
+	} else {
+		cctxDictPool.Put(cctxDict)
+	}
+
+	<-concurrencyLimitCh
 
-	cw := getCompressWork()
-	cw.dst = dst
-	cw.src = src
-	cw.cd = cd
-	cw.compressionLevel = compressionLevel
-	compressWorkCh <- cw
-	<-cw.done
-	dst = cw.dst
-	putCompressWork(cw)
 	return dst
 }
 
-func getCompressWork() *compressWork {
-	v := compressWorkPool.Get()
-	if v == nil {
-		v = &compressWork{
-			done: make(chan struct{}),
-		}
-	}
-	return v.(*compressWork)
+var cctxPool = &sync.Pool{
+	New: newCCtx,
 }
 
-func putCompressWork(cw *compressWork) {
-	cw.src = nil
-	cw.dst = nil
-	cw.cd = nil
-	cw.compressionLevel = 0
-	compressWorkPool.Put(cw)
+var cctxDictPool = &sync.Pool{
+	New: newCCtx,
 }
 
-type compressWork struct {
-	dst              []byte
-	src              []byte
-	cd               *CDict
-	compressionLevel int
-	done             chan struct{}
-}
-
-var (
-	compressWorkCh   chan *compressWork
-	compressWorkPool sync.Pool
-	compressInitOnce sync.Once
-)
-
-func compressInit() {
-	gomaxprocs := runtime.GOMAXPROCS(-1)
-
-	compressWorkCh = make(chan *compressWork, gomaxprocs)
-	for i := 0; i < gomaxprocs; i++ {
-		go compressWorker()
-	}
-}
-
-func compressWorker() {
+func newCCtx() interface{} {
 	cctx := C.ZSTD_createCCtx()
-	cctxDict := C.ZSTD_createCCtx()
-
-	for cw := range compressWorkCh {
-		cw.dst = compress(cctx, cctxDict, cw.dst, cw.src, cw.cd, cw.compressionLevel)
-		cw.done <- struct{}{}
+	cw := &cctxWrapper{
+		cctx: cctx,
 	}
+	runtime.SetFinalizer(cw, freeCCtx)
+	return cw
 }
 
-func compress(cctx, cctxDict *C.ZSTD_CCtx, dst, src []byte, cd *CDict, compressionLevel int) []byte {
+func freeCCtx(cw *cctxWrapper) {
+	C.ZSTD_freeCCtx(cw.cctx)
+	cw.cctx = nil
+}
+
+type cctxWrapper struct {
+	cctx *C.ZSTD_CCtx
+}
+
+func compress(cctx, cctxDict *cctxWrapper, dst, src []byte, cd *CDict, compressionLevel int) []byte {
 	if len(src) == 0 {
 		return dst
 	}
@@ -167,9 +150,9 @@ func compress(cctx, cctxDict *C.ZSTD_CCtx, dst, src []byte, cd *CDict, compressi
 	return dst[:dstLen+compressedSize]
 }
 
-func compressInternal(cctx, cctxDict *C.ZSTD_CCtx, dst, src []byte, cd *CDict, compressionLevel int, mustSucceed bool) C.size_t {
+func compressInternal(cctx, cctxDict *cctxWrapper, dst, src []byte, cd *CDict, compressionLevel int, mustSucceed bool) C.size_t {
 	if cd != nil {
-		result := C.ZSTD_compress_usingCDict_wrapper(cctxDict,
+		result := C.ZSTD_compress_usingCDict_wrapper(cctxDict.cctx,
 			C.uintptr_t(uintptr(unsafe.Pointer(&dst[0]))),
 			C.size_t(cap(dst)),
 			C.uintptr_t(uintptr(unsafe.Pointer(&src[0]))),
@@ -183,7 +166,7 @@ func compressInternal(cctx, cctxDict *C.ZSTD_CCtx, dst, src []byte, cd *CDict, c
 		}
 		return result
 	}
-	result := C.ZSTD_compressCCtx_wrapper(cctx,
+	result := C.ZSTD_compressCCtx_wrapper(cctx.cctx,
 		C.uintptr_t(uintptr(unsafe.Pointer(&dst[0]))),
 		C.size_t(cap(dst)),
 		C.uintptr_t(uintptr(unsafe.Pointer(&src[0]))),
@@ -207,72 +190,56 @@ func Decompress(dst, src []byte) ([]byte, error) {
 //
 // The given dictionary dd is used for the decompression.
 func DecompressDict(dst, src []byte, dd *DDict) ([]byte, error) {
-	decompressInitOnce.Do(decompressInit)
+	concurrencyLimitCh <- struct{}{}
+
+	var dctx, dctxDict *dctxWrapper
+	if dd == nil {
+		dctx = dctxPool.Get().(*dctxWrapper)
+	} else {
+		dctxDict = dctxDictPool.Get().(*dctxWrapper)
+	}
+
+	var err error
+	dst, err = decompress(dctx, dctxDict, dst, src, dd)
+
+	if dd == nil {
+		dctxPool.Put(dctx)
+	} else {
+		dctxDictPool.Put(dctxDict)
+	}
+
+	<-concurrencyLimitCh
 
-	dw := getDecompressWork()
-	dw.dst = dst
-	dw.src = src
-	dw.dd = dd
-	decompressWorkCh <- dw
-	<-dw.done
-	dst = dw.dst
-	err := dw.err
-	putDecompressWork(dw)
 	return dst, err
 }
 
-func getDecompressWork() *decompressWork {
-	v := decompressWorkPool.Get()
-	if v == nil {
-		v = &decompressWork{
-			done: make(chan struct{}),
-		}
-	}
-	return v.(*decompressWork)
+var dctxPool = &sync.Pool{
+	New: newDCtx,
 }
 
-func putDecompressWork(dw *decompressWork) {
-	dw.dst = nil
-	dw.src = nil
-	dw.dd = nil
-	dw.err = nil
-	decompressWorkPool.Put(dw)
+var dctxDictPool = &sync.Pool{
+	New: newDCtx,
 }
 
-type decompressWork struct {
-	dst  []byte
-	src  []byte
-	dd   *DDict
-	err  error
-	done chan struct{}
-}
-
-var (
-	decompressWorkCh   chan *decompressWork
-	decompressWorkPool sync.Pool
-	decompressInitOnce sync.Once
-)
-
-func decompressInit() {
-	gomaxprocs := runtime.GOMAXPROCS(-1)
-
-	decompressWorkCh = make(chan *decompressWork, gomaxprocs)
-	for i := 0; i < gomaxprocs; i++ {
-		go decompressWorker()
-	}
-}
-
-func decompressWorker() {
+func newDCtx() interface{} {
 	dctx := C.ZSTD_createDCtx()
-	dctxDict := C.ZSTD_createDCtx()
-
-	for dw := range decompressWorkCh {
-		dw.dst, dw.err = decompress(dctx, dctxDict, dw.dst, dw.src, dw.dd)
-		dw.done <- struct{}{}
+	dw := &dctxWrapper{
+		dctx: dctx,
 	}
+	runtime.SetFinalizer(dw, freeDCtx)
+	return dw
 }
 
-func decompress(dctx, dctxDict *C.ZSTD_DCtx, dst, src []byte, dd *DDict) ([]byte, error) {
+func freeDCtx(dw *dctxWrapper) {
+	C.ZSTD_freeDCtx(dw.dctx)
+	dw.dctx = nil
+}
+
+type dctxWrapper struct {
+	dctx *C.ZSTD_DCtx
+}
+
+func decompress(dctx, dctxDict *dctxWrapper, dst, src []byte, dd *DDict) ([]byte, error) {
 	if len(src) == 0 {
 		return dst, nil
 	}
@@ -325,17 +292,17 @@ func decompress(dctx, dctxDict *C.ZSTD_DCtx, dst, src []byte, dd *DDict) ([]byte
 	return dst[:dstLen], fmt.Errorf("decompression error: %s", errStr(result))
 }
 
-func decompressInternal(dctx, dctxDict *C.ZSTD_DCtx, dst, src []byte, dd *DDict) C.size_t {
+func decompressInternal(dctx, dctxDict *dctxWrapper, dst, src []byte, dd *DDict) C.size_t {
 	var n C.size_t
 	if dd != nil {
-		n = C.ZSTD_decompress_usingDDict_wrapper(dctxDict,
+		n = C.ZSTD_decompress_usingDDict_wrapper(dctxDict.dctx,
 			C.uintptr_t(uintptr(unsafe.Pointer(&dst[0]))),
 			C.size_t(cap(dst)),
 			C.uintptr_t(uintptr(unsafe.Pointer(&src[0]))),
 			C.size_t(len(src)),
 			dd.p)
 	} else {
-		n = C.ZSTD_decompressDCtx_wrapper(dctx,
+		n = C.ZSTD_decompressDCtx_wrapper(dctx.dctx,
 			C.uintptr_t(uintptr(unsafe.Pointer(&dst[0]))),
 			C.size_t(cap(dst)),
 			C.uintptr_t(uintptr(unsafe.Pointer(&src[0]))),
@@ -347,6 +314,11 @@ func decompressInternal(dctx, dctxDict *C.ZSTD_DCtx, dst, src []byte, dd *DDict)
 	return n
 }
 
+var concurrencyLimitCh = func() chan struct{} {
+	gomaxprocs := runtime.GOMAXPROCS(-1)
+	return make(chan struct{}, gomaxprocs)
+}()
+
 func errStr(result C.size_t) string {
 	errCode := C.ZSTD_getErrorCode(result)
 	errCStr := C.ZSTD_getErrorString(errCode)

vendor/github.com/valyala/gozstd/writer.go

@@ -164,14 +164,18 @@ func (zw *Writer) ReadFrom(r io.Reader) (int64, error) {
 		// Fill the inBuf.
 		for zw.inBuf.size < cstreamInBufSize {
 			n, err := r.Read(zw.inBufGo[zw.inBuf.size:cstreamInBufSize])
+
+			// Sometimes n > 0 even when Read() returns an error.
+			// This is true especially if the error is io.EOF.
+			zw.inBuf.size += C.size_t(n)
+			nn += int64(n)
+
 			if err != nil {
 				if err == io.EOF {
 					return nn, nil
 				}
 				return nn, err
 			}
-			zw.inBuf.size += C.size_t(n)
-			nn += int64(n)
 		}
 
 		// Flush the inBuf.

vendor/github.com/valyala/gozstd/zdict.h

@@ -94,6 +94,8 @@ typedef struct {
     unsigned steps;              /* Number of steps : Only used for optimization : 0 means default (40) : Higher means more parameters checked */
     unsigned nbThreads;          /* Number of threads : constraint: 0 < nbThreads : 1 means single-threaded : Only used for optimization : Ignored if ZSTD_MULTITHREAD is not defined */
     double splitPoint;           /* Percentage of samples used for training: Only used for optimization : the first nbSamples * splitPoint samples will be used to training, the last nbSamples * (1 - splitPoint) samples will be used for testing, 0 means default (1.0), 1.0 when all samples are used for both training and testing */
+    unsigned shrinkDict;         /* Train dictionaries to shrink in size starting from the minimum size and selects the smallest dictionary that is shrinkDictMaxRegression% worse than the largest dictionary. 0 means no shrinking and 1 means shrinking  */
+    unsigned shrinkDictMaxRegression; /* Sets shrinkDictMaxRegression so that a smaller dictionary can be at worse shrinkDictMaxRegression% worse than the max dict size dictionary. */
     ZDICT_params_t zParams;
 } ZDICT_cover_params_t;
 
@@ -105,6 +107,9 @@ typedef struct {
     unsigned nbThreads;          /* Number of threads : constraint: 0 < nbThreads : 1 means single-threaded : Only used for optimization : Ignored if ZSTD_MULTITHREAD is not defined */
     double splitPoint;           /* Percentage of samples used for training: Only used for optimization : the first nbSamples * splitPoint samples will be used to training, the last nbSamples * (1 - splitPoint) samples will be used for testing, 0 means default (0.75), 1.0 when all samples are used for both training and testing */
     unsigned accel;              /* Acceleration level: constraint: 0 < accel <= 10, higher means faster and less accurate, 0 means default(1) */
+    unsigned shrinkDict;         /* Train dictionaries to shrink in size starting from the minimum size and selects the smallest dictionary that is shrinkDictMaxRegression% worse than the largest dictionary. 0 means no shrinking and 1 means shrinking  */
+    unsigned shrinkDictMaxRegression; /* Sets shrinkDictMaxRegression so that a smaller dictionary can be at worse shrinkDictMaxRegression% worse than the max dict size dictionary. */
+
     ZDICT_params_t zParams;
 } ZDICT_fastCover_params_t;
 

vendor/github.com/valyala/gozstd/zstd.h

@@ -71,7 +71,7 @@ extern "C" {
 /*------   Version   ------*/
 #define ZSTD_VERSION_MAJOR    1
 #define ZSTD_VERSION_MINOR    4
-#define ZSTD_VERSION_RELEASE  0
+#define ZSTD_VERSION_RELEASE  2
 
 #define ZSTD_VERSION_NUMBER  (ZSTD_VERSION_MAJOR *100*100 + ZSTD_VERSION_MINOR *100 + ZSTD_VERSION_RELEASE)
 ZSTDLIB_API unsigned ZSTD_versionNumber(void);   /**< to check runtime library version */
@@ -82,16 +82,16 @@ ZSTDLIB_API unsigned ZSTD_versionNumber(void);   /**< to check runtime library v
 #define ZSTD_VERSION_STRING ZSTD_EXPAND_AND_QUOTE(ZSTD_LIB_VERSION)
 ZSTDLIB_API const char* ZSTD_versionString(void);   /* requires v1.3.0+ */
 
-/***************************************
-*  Default constant
-***************************************/
+/* *************************************
+ *  Default constant
+ ***************************************/
 #ifndef ZSTD_CLEVEL_DEFAULT
 #  define ZSTD_CLEVEL_DEFAULT 3
 #endif
 
-/***************************************
-*  Constants
-***************************************/
+/* *************************************
+ *  Constants
+ ***************************************/
 
 /* All magic numbers are supposed read/written to/from files/memory using little-endian convention */
 #define ZSTD_MAGICNUMBER            0xFD2FB528    /* valid since v0.8.0 */
@@ -183,9 +183,14 @@ ZSTDLIB_API int         ZSTD_maxCLevel(void);               /*!< maximum compres
 ***************************************/
 /*= Compression context
  *  When compressing many times,
- *  it is recommended to allocate a context just once, and re-use it for each successive compression operation.
+ *  it is recommended to allocate a context just once,
+ *  and re-use it for each successive compression operation.
  *  This will make workload friendlier for system's memory.
- *  Use one context per thread for parallel execution in multi-threaded environments. */
+ *  Note : re-using context is just a speed / resource optimization.
+ *         It doesn't change the compression ratio, which remains identical.
+ *  Note 2 : In multi-threaded environments,
+ *         use one different context per thread for parallel execution.
+ */
 typedef struct ZSTD_CCtx_s ZSTD_CCtx;
 ZSTDLIB_API ZSTD_CCtx* ZSTD_createCCtx(void);
 ZSTDLIB_API size_t     ZSTD_freeCCtx(ZSTD_CCtx* cctx);
@@ -380,6 +385,7 @@ typedef enum {
      * ZSTD_c_forceMaxWindow
      * ZSTD_c_forceAttachDict
      * ZSTD_c_literalCompressionMode
+     * ZSTD_c_targetCBlockSize
      * Because they are not stable, it's necessary to define ZSTD_STATIC_LINKING_ONLY to access them.
      * note : never ever use experimentalParam? names directly;
      *        also, the enums values themselves are unstable and can still change.
@@ -389,6 +395,7 @@ typedef enum {
      ZSTD_c_experimentalParam3=1000,
      ZSTD_c_experimentalParam4=1001,
      ZSTD_c_experimentalParam5=1002,
+     ZSTD_c_experimentalParam6=1003,
 } ZSTD_cParameter;
 
 typedef struct {
@@ -657,17 +664,33 @@ ZSTDLIB_API size_t ZSTD_compressStream2( ZSTD_CCtx* cctx,
                                          ZSTD_inBuffer* input,
                                          ZSTD_EndDirective endOp);
 
-ZSTDLIB_API size_t ZSTD_CStreamInSize(void);    /**< recommended size for input buffer */
-ZSTDLIB_API size_t ZSTD_CStreamOutSize(void);   /**< recommended size for output buffer. Guarantee to successfully flush at least one complete compressed block in all circumstances. */
 
-/*******************************************************************************
- * This is a legacy streaming API, and can be replaced by ZSTD_CCtx_reset() and
- * ZSTD_compressStream2(). It is redundant, but is still fully supported.
+/* These buffer sizes are softly recommended.
+ * They are not required : ZSTD_compressStream*() happily accepts any buffer size, for both input and output.
+ * Respecting the recommended size just makes it a bit easier for ZSTD_compressStream*(),
+ * reducing the amount of memory shuffling and buffering, resulting in minor performance savings.
+ *
+ * However, note that these recommendations are from the perspective of a C caller program.
+ * If the streaming interface is invoked from some other language,
+ * especially managed ones such as Java or Go, through a foreign function interface such as jni or cgo,
+ * a major performance rule is to reduce crossing such interface to an absolute minimum.
+ * It's not rare that performance ends being spent more into the interface, rather than compression itself.
+ * In which cases, prefer using large buffers, as large as practical,
+ * for both input and output, to reduce the nb of roundtrips.
+ */
+ZSTDLIB_API size_t ZSTD_CStreamInSize(void);    /**< recommended size for input buffer */
+ZSTDLIB_API size_t ZSTD_CStreamOutSize(void);   /**< recommended size for output buffer. Guarantee to successfully flush at least one complete compressed block. */
+
+
+/* *****************************************************************************
+ * This following is a legacy streaming API.
+ * It can be replaced by ZSTD_CCtx_reset() and ZSTD_compressStream2().
+ * It is redundant, but remains fully supported.
  * Advanced parameters and dictionary compression can only be used through the
  * new API.
  ******************************************************************************/
 
-/**
+/*!
  * Equivalent to:
  *
  *     ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only);
@@ -675,16 +698,16 @@ ZSTDLIB_API size_t ZSTD_CStreamOutSize(void);   /**< recommended size for output
  *     ZSTD_CCtx_setParameter(zcs, ZSTD_c_compressionLevel, compressionLevel);
  */
 ZSTDLIB_API size_t ZSTD_initCStream(ZSTD_CStream* zcs, int compressionLevel);
-/**
+/*!
  * Alternative for ZSTD_compressStream2(zcs, output, input, ZSTD_e_continue).
  * NOTE: The return value is different. ZSTD_compressStream() returns a hint for
  * the next read size (if non-zero and not an error). ZSTD_compressStream2()
- * returns the number of bytes left to flush (if non-zero and not an error).
+ * returns the minimum nb of bytes left to flush (if non-zero and not an error).
  */
 ZSTDLIB_API size_t ZSTD_compressStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output, ZSTD_inBuffer* input);
-/** Equivalent to ZSTD_compressStream2(zcs, output, &emptyInput, ZSTD_e_flush). */
+/*! Equivalent to ZSTD_compressStream2(zcs, output, &emptyInput, ZSTD_e_flush). */
 ZSTDLIB_API size_t ZSTD_flushStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output);
-/** Equivalent to ZSTD_compressStream2(zcs, output, &emptyInput, ZSTD_e_end). */
+/*! Equivalent to ZSTD_compressStream2(zcs, output, &emptyInput, ZSTD_e_end). */
 ZSTDLIB_API size_t ZSTD_endStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output);
 
 
@@ -969,7 +992,7 @@ ZSTDLIB_API size_t ZSTD_sizeof_DDict(const ZSTD_DDict* ddict);
 #endif  /* ZSTD_H_235446 */
 
 
-/****************************************************************************************
+/* **************************************************************************************
  *   ADVANCED AND EXPERIMENTAL FUNCTIONS
  ****************************************************************************************
  * The definitions in the following section are considered experimental.
@@ -1037,6 +1060,10 @@ ZSTDLIB_API size_t ZSTD_sizeof_DDict(const ZSTD_DDict* ddict);
 #define ZSTD_LDM_HASHRATELOG_MIN     0
 #define ZSTD_LDM_HASHRATELOG_MAX (ZSTD_WINDOWLOG_MAX - ZSTD_HASHLOG_MIN)
 
+/* Advanced parameter bounds */
+#define ZSTD_TARGETCBLOCKSIZE_MIN   64
+#define ZSTD_TARGETCBLOCKSIZE_MAX   ZSTD_BLOCKSIZE_MAX
+
 /* internal */
 #define ZSTD_HASHLOG3_MAX           17
 
@@ -1162,7 +1189,7 @@ typedef enum {
  *            however it does mean that all frame data must be present and valid. */
 ZSTDLIB_API unsigned long long ZSTD_findDecompressedSize(const void* src, size_t srcSize);
 
-/** ZSTD_decompressBound() :
+/*! ZSTD_decompressBound() :
  *  `src` should point to the start of a series of ZSTD encoded and/or skippable frames
  *  `srcSize` must be the _exact_ size of this series
  *       (i.e. there should be a frame boundary at `src + srcSize`)
@@ -1409,6 +1436,11 @@ ZSTDLIB_API size_t ZSTD_CCtx_refPrefix_advanced(ZSTD_CCtx* cctx, const void* pre
  */
 #define ZSTD_c_literalCompressionMode ZSTD_c_experimentalParam5
 
+/* Tries to fit compressed block size to be around targetCBlockSize.
+ * No target when targetCBlockSize == 0.
+ * There is no guarantee on compressed block size (default:0) */
+#define ZSTD_c_targetCBlockSize ZSTD_c_experimentalParam6
+
 /*! ZSTD_CCtx_getParameter() :
  *  Get the requested compression parameter value, selected by enum ZSTD_cParameter,
  *  and store it into int* value.
@@ -1843,7 +1875,7 @@ typedef struct {
     unsigned checksumFlag;
 } ZSTD_frameHeader;
 
-/** ZSTD_getFrameHeader() :
+/*! ZSTD_getFrameHeader() :
  *  decode Frame Header, or requires larger `srcSize`.
  * @return : 0, `zfhPtr` is correctly filled,
  *          >0, `srcSize` is too small, value is wanted `srcSize` amount,

vendor/golang.org/x/sys/unix/affinity_linux.go

@@ -91,9 +91,13 @@ func onesCount64(x uint64) int {
 	const m0 = 0x5555555555555555 // 01010101 ...
 	const m1 = 0x3333333333333333 // 00110011 ...
 	const m2 = 0x0f0f0f0f0f0f0f0f // 00001111 ...
-	const m3 = 0x00ff00ff00ff00ff // etc.
-	const m4 = 0x0000ffff0000ffff
 
+	// Unused in this function, but definitions preserved for
+	// documentation purposes:
+	//
+	//   const m3 = 0x00ff00ff00ff00ff // etc.
+	//   const m4 = 0x0000ffff0000ffff
+	//
 	// Implementation: Parallel summing of adjacent bits.
 	// See "Hacker's Delight", Chap. 5: Counting Bits.
 	// The following pattern shows the general approach:

vendor/golang.org/x/sys/unix/dirent.go

@@ -2,7 +2,7 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-// +build aix darwin dragonfly freebsd linux nacl netbsd openbsd solaris
+// +build aix darwin dragonfly freebsd linux netbsd openbsd solaris
 
 package unix
 

vendor/golang.org/x/sys/unix/endian_little.go

@@ -2,7 +2,7 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 //
-// +build 386 amd64 amd64p32 arm arm64 ppc64le mipsle mips64le
+// +build 386 amd64 amd64p32 arm arm64 ppc64le mipsle mips64le riscv64
 
 package unix
 

vendor/modules.txt

@@ -1,12 +1,12 @@
 # github.com/VictoriaMetrics/fastcache v1.5.1
 github.com/VictoriaMetrics/fastcache
-# github.com/VictoriaMetrics/metrics v1.7.0
+# github.com/VictoriaMetrics/metrics v1.7.1
 github.com/VictoriaMetrics/metrics
 # github.com/cespare/xxhash/v2 v2.0.1-0.20190104013014-3767db7a7e18
 github.com/cespare/xxhash/v2
 # github.com/golang/snappy v0.0.1
 github.com/golang/snappy
-# github.com/klauspost/compress v1.7.4
+# github.com/klauspost/compress v1.7.5
 github.com/klauspost/compress/fse
 github.com/klauspost/compress/huff0
 github.com/klauspost/compress/snappy
@@ -18,11 +18,11 @@ github.com/valyala/bytebufferpool
 github.com/valyala/fastjson/fastfloat
 # github.com/valyala/fastrand v1.0.0
 github.com/valyala/fastrand
-# github.com/valyala/gozstd v1.5.1
+# github.com/valyala/gozstd v1.6.0
 github.com/valyala/gozstd
 # github.com/valyala/histogram v1.0.1
 github.com/valyala/histogram
 # github.com/valyala/quicktemplate v1.1.1
 github.com/valyala/quicktemplate
-# golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7
+# golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa
 golang.org/x/sys/unix