VDB-327 Constantinople prep (#135)

* Bump geth to 1.8.20 for Constantinople

* Fix conflicting import/toml source for logrus
This commit is contained in:
Edvard Hübinette
2019-01-14 11:31:28 +01:00
committed by GitHub
parent 5efd683c54
commit d41209d293
235 changed files with 11114 additions and 7199 deletions
+5
View File
@@ -0,0 +1,5 @@
.idea
.DS_Store
/server/server.exe
/server/server
CHANGELOG.md
+31
View File
@@ -0,0 +1,31 @@
language: go
go:
- 1.x
- tip
matrix:
allow_failures:
- go: tip
fast_finish: true
before_install:
- go get github.com/modocache/gover
- go get github.com/mattn/goveralls
- go get golang.org/x/tools/cmd/cover
- go get golang.org/x/tools/cmd/goimports
- go get github.com/golang/lint/golint
- go get github.com/stretchr/testify/assert
- go get github.com/gordonklaus/ineffassign
script:
- gofiles=$(find ./ -name '*.go') && [ -z "$gofiles" ] || unformatted=$(goimports -l $gofiles) && [ -z "$unformatted" ] || (echo >&2 "Go files must be formatted with gofmt. Following files has problem:\n $unformatted" && false)
- diff <(echo -n) <(gofmt -s -d .)
- golint ./... # This won't break the build, just show warnings
- ineffassign .
- go vet ./...
- go test -race -count=1 -coverprofile=queue.coverprofile ./queue
- go test -race -count=1 -coverprofile=server.coverprofile ./server
- go test -race -count=1 -coverprofile=main.coverprofile
- $HOME/gopath/bin/gover
- $HOME/gopath/bin/goveralls -coverprofile=gover.coverprofile -service travis-ci
+201
View File
@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright {yyyy} {name of copyright owner}
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
+145
View File
@@ -0,0 +1,145 @@
# BigCache [![Build Status](https://travis-ci.org/allegro/bigcache.svg?branch=master)](https://travis-ci.org/allegro/bigcache)&nbsp;[![Coverage Status](https://coveralls.io/repos/github/allegro/bigcache/badge.svg?branch=master)](https://coveralls.io/github/allegro/bigcache?branch=master)&nbsp;[![GoDoc](https://godoc.org/github.com/allegro/bigcache?status.svg)](https://godoc.org/github.com/allegro/bigcache)&nbsp;[![Go Report Card](https://goreportcard.com/badge/github.com/allegro/bigcache)](https://goreportcard.com/report/github.com/allegro/bigcache)
Fast, concurrent, evicting in-memory cache written to keep big number of entries without impact on performance.
BigCache keeps entries on heap but omits GC for them. To achieve that operations on bytes arrays take place,
therefore entries (de)serialization in front of the cache will be needed in most use cases.
## Usage
### Simple initialization
```go
import "github.com/allegro/bigcache"
cache, _ := bigcache.NewBigCache(bigcache.DefaultConfig(10 * time.Minute))
cache.Set("my-unique-key", []byte("value"))
entry, _ := cache.Get("my-unique-key")
fmt.Println(string(entry))
```
### Custom initialization
When cache load can be predicted in advance then it is better to use custom initialization because additional memory
allocation can be avoided in that way.
```go
import (
"log"
"github.com/allegro/bigcache"
)
config := bigcache.Config {
// number of shards (must be a power of 2)
Shards: 1024,
// time after which entry can be evicted
LifeWindow: 10 * time.Minute,
// rps * lifeWindow, used only in initial memory allocation
MaxEntriesInWindow: 1000 * 10 * 60,
// max entry size in bytes, used only in initial memory allocation
MaxEntrySize: 500,
// prints information about additional memory allocation
Verbose: true,
// cache will not allocate more memory than this limit, value in MB
// if value is reached then the oldest entries can be overridden for the new ones
// 0 value means no size limit
HardMaxCacheSize: 8192,
// callback fired when the oldest entry is removed because of its
// expiration time or no space left for the new entry. Default value is nil which
// means no callback and it prevents from unwrapping the oldest entry.
OnRemove: nil,
}
cache, initErr := bigcache.NewBigCache(config)
if initErr != nil {
log.Fatal(initErr)
}
cache.Set("my-unique-key", []byte("value"))
if entry, err := cache.Get("my-unique-key"); err == nil {
fmt.Println(string(entry))
}
```
## Benchmarks
Three caches were compared: bigcache, [freecache](https://github.com/coocood/freecache) and map.
Benchmark tests were made using an i7-6700K with 32GB of RAM on Windows 10.
### Writes and reads
```bash
cd caches_bench; go test -bench=. -benchtime=10s ./... -timeout 30m
BenchmarkMapSet-8 2000000 716 ns/op 336 B/op 3 allocs/op
BenchmarkConcurrentMapSet-8 1000000 1292 ns/op 347 B/op 8 allocs/op
BenchmarkFreeCacheSet-8 3000000 501 ns/op 371 B/op 3 allocs/op
BenchmarkBigCacheSet-8 3000000 482 ns/op 303 B/op 2 allocs/op
BenchmarkMapGet-8 5000000 309 ns/op 24 B/op 1 allocs/op
BenchmarkConcurrentMapGet-8 2000000 659 ns/op 24 B/op 2 allocs/op
BenchmarkFreeCacheGet-8 3000000 541 ns/op 152 B/op 3 allocs/op
BenchmarkBigCacheGet-8 3000000 420 ns/op 152 B/op 3 allocs/op
BenchmarkBigCacheSetParallel-8 10000000 184 ns/op 313 B/op 3 allocs/op
BenchmarkFreeCacheSetParallel-8 10000000 195 ns/op 357 B/op 4 allocs/op
BenchmarkConcurrentMapSetParallel-8 5000000 242 ns/op 200 B/op 6 allocs/op
BenchmarkBigCacheGetParallel-8 20000000 100 ns/op 152 B/op 4 allocs/op
BenchmarkFreeCacheGetParallel-8 10000000 133 ns/op 152 B/op 4 allocs/op
BenchmarkConcurrentMapGetParallel-8 10000000 202 ns/op 24 B/op 2 allocs/op
```
Writes and reads in bigcache are faster than in freecache.
Writes to map are the slowest.
### GC pause time
```bash
cd caches_bench; go run caches_gc_overhead_comparison.go
Number of entries: 20000000
GC pause for bigcache: 5.8658ms
GC pause for freecache: 32.4341ms
GC pause for map: 52.9661ms
```
Test shows how long are the GC pauses for caches filled with 20mln of entries.
Bigcache and freecache have very similar GC pause time.
It is clear that both reduce GC overhead in contrast to map
which GC pause time took more than 10 seconds.
## How it works
BigCache relies on optimization presented in 1.5 version of Go ([issue-9477](https://github.com/golang/go/issues/9477)).
This optimization states that if map without pointers in keys and values is used then GC will omit its content.
Therefore BigCache uses `map[uint64]uint32` where keys are hashed and values are offsets of entries.
Entries are kept in bytes array, to omit GC again.
Bytes array size can grow to gigabytes without impact on performance
because GC will only see single pointer to it.
## Bigcache vs Freecache
Both caches provide the same core features but they reduce GC overhead in different ways.
Bigcache relies on `map[uint64]uint32`, freecache implements its own mapping built on
slices to reduce number of pointers.
Results from benchmark tests are presented above.
One of the advantage of bigcache over freecache is that you dont need to know
the size of the cache in advance, because when bigcache is full,
it can allocate additional memory for new entries instead of
overwriting existing ones as freecache does currently.
However hard max size in bigcache also can be set, check [HardMaxCacheSize](https://godoc.org/github.com/allegro/bigcache#Config).
## HTTP Server
This package also includes an easily deployable HTTP implementation of BigCache, which can be found in the [server](/server) package.
## More
Bigcache genesis is described in allegro.tech blog post: [writing a very fast cache service in Go](http://allegro.tech/2016/03/writing-fast-cache-service-in-go.html)
## License
BigCache is released under the Apache 2.0 license (see [LICENSE](LICENSE))
+155
View File
@@ -0,0 +1,155 @@
package bigcache
import (
"fmt"
"time"
)
const (
minimumEntriesInShard = 10 // Minimum number of entries in single shard
)
// BigCache is fast, concurrent, evicting cache created to keep big number of entries without impact on performance.
// It keeps entries on heap but omits GC for them. To achieve that operations on bytes arrays take place,
// therefore entries (de)serialization in front of the cache will be needed in most use cases.
type BigCache struct {
shards []*cacheShard
lifeWindow uint64
clock clock
hash Hasher
config Config
shardMask uint64
maxShardSize uint32
}
// NewBigCache initialize new instance of BigCache
func NewBigCache(config Config) (*BigCache, error) {
return newBigCache(config, &systemClock{})
}
func newBigCache(config Config, clock clock) (*BigCache, error) {
if !isPowerOfTwo(config.Shards) {
return nil, fmt.Errorf("Shards number must be power of two")
}
if config.Hasher == nil {
config.Hasher = newDefaultHasher()
}
cache := &BigCache{
shards: make([]*cacheShard, config.Shards),
lifeWindow: uint64(config.LifeWindow.Seconds()),
clock: clock,
hash: config.Hasher,
config: config,
shardMask: uint64(config.Shards - 1),
maxShardSize: uint32(config.maximumShardSize()),
}
var onRemove func(wrappedEntry []byte)
if config.OnRemove == nil {
onRemove = cache.notProvidedOnRemove
} else {
onRemove = cache.providedOnRemove
}
for i := 0; i < config.Shards; i++ {
cache.shards[i] = initNewShard(config, onRemove, clock)
}
if config.CleanWindow > 0 {
go func() {
for t := range time.Tick(config.CleanWindow) {
cache.cleanUp(uint64(t.Unix()))
}
}()
}
return cache, nil
}
// Get reads entry for the key.
// It returns an EntryNotFoundError when
// no entry exists for the given key.
func (c *BigCache) Get(key string) ([]byte, error) {
hashedKey := c.hash.Sum64(key)
shard := c.getShard(hashedKey)
return shard.get(key, hashedKey)
}
// Set saves entry under the key
func (c *BigCache) Set(key string, entry []byte) error {
hashedKey := c.hash.Sum64(key)
shard := c.getShard(hashedKey)
return shard.set(key, hashedKey, entry)
}
// Delete removes the key
func (c *BigCache) Delete(key string) error {
hashedKey := c.hash.Sum64(key)
shard := c.getShard(hashedKey)
return shard.del(key, hashedKey)
}
// Reset empties all cache shards
func (c *BigCache) Reset() error {
for _, shard := range c.shards {
shard.reset(c.config)
}
return nil
}
// Len computes number of entries in cache
func (c *BigCache) Len() int {
var len int
for _, shard := range c.shards {
len += shard.len()
}
return len
}
// Stats returns cache's statistics
func (c *BigCache) Stats() Stats {
var s Stats
for _, shard := range c.shards {
tmp := shard.getStats()
s.Hits += tmp.Hits
s.Misses += tmp.Misses
s.DelHits += tmp.DelHits
s.DelMisses += tmp.DelMisses
s.Collisions += tmp.Collisions
}
return s
}
// Iterator returns iterator function to iterate over EntryInfo's from whole cache.
func (c *BigCache) Iterator() *EntryInfoIterator {
return newIterator(c)
}
func (c *BigCache) onEvict(oldestEntry []byte, currentTimestamp uint64, evict func() error) bool {
oldestTimestamp := readTimestampFromEntry(oldestEntry)
if currentTimestamp-oldestTimestamp > c.lifeWindow {
evict()
return true
}
return false
}
func (c *BigCache) cleanUp(currentTimestamp uint64) {
for _, shard := range c.shards {
shard.cleanUp(currentTimestamp)
}
}
func (c *BigCache) getShard(hashedKey uint64) (shard *cacheShard) {
return c.shards[hashedKey&c.shardMask]
}
func (c *BigCache) providedOnRemove(wrappedEntry []byte) {
c.config.OnRemove(readKeyFromEntry(wrappedEntry), readEntry(wrappedEntry))
}
func (c *BigCache) notProvidedOnRemove(wrappedEntry []byte) {
}
+141
View File
@@ -0,0 +1,141 @@
package bigcache
import (
"fmt"
"math/rand"
"strconv"
"testing"
"time"
)
var message = blob('a', 256)
func BenchmarkWriteToCacheWith1Shard(b *testing.B) {
writeToCache(b, 1, 100*time.Second, b.N)
}
func BenchmarkWriteToLimitedCacheWithSmallInitSizeAnd1Shard(b *testing.B) {
m := blob('a', 1024)
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 100 * time.Second,
MaxEntriesInWindow: 100,
MaxEntrySize: 256,
HardMaxCacheSize: 1,
})
b.ReportAllocs()
for i := 0; i < b.N; i++ {
cache.Set(fmt.Sprintf("key-%d", i), m)
}
}
func BenchmarkWriteToUnlimitedCacheWithSmallInitSizeAnd1Shard(b *testing.B) {
m := blob('a', 1024)
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 100 * time.Second,
MaxEntriesInWindow: 100,
MaxEntrySize: 256,
})
b.ReportAllocs()
for i := 0; i < b.N; i++ {
cache.Set(fmt.Sprintf("key-%d", i), m)
}
}
func BenchmarkWriteToCache(b *testing.B) {
for _, shards := range []int{1, 512, 1024, 8192} {
b.Run(fmt.Sprintf("%d-shards", shards), func(b *testing.B) {
writeToCache(b, shards, 100*time.Second, b.N)
})
}
}
func BenchmarkReadFromCache(b *testing.B) {
for _, shards := range []int{1, 512, 1024, 8192} {
b.Run(fmt.Sprintf("%d-shards", shards), func(b *testing.B) {
readFromCache(b, 1024)
})
}
}
func BenchmarkIterateOverCache(b *testing.B) {
m := blob('a', 1)
for _, shards := range []int{512, 1024, 8192} {
b.Run(fmt.Sprintf("%d-shards", shards), func(b *testing.B) {
cache, _ := NewBigCache(Config{
Shards: shards,
LifeWindow: 1000 * time.Second,
MaxEntriesInWindow: max(b.N, 100),
MaxEntrySize: 500,
})
for i := 0; i < b.N; i++ {
cache.Set(fmt.Sprintf("key-%d", i), m)
}
b.ResetTimer()
it := cache.Iterator()
b.RunParallel(func(pb *testing.PB) {
b.ReportAllocs()
for pb.Next() {
if it.SetNext() {
it.Value()
}
}
})
})
}
}
func BenchmarkWriteToCacheWith1024ShardsAndSmallShardInitSize(b *testing.B) {
writeToCache(b, 1024, 100*time.Second, 100)
}
func writeToCache(b *testing.B, shards int, lifeWindow time.Duration, requestsInLifeWindow int) {
cache, _ := NewBigCache(Config{
Shards: shards,
LifeWindow: lifeWindow,
MaxEntriesInWindow: max(requestsInLifeWindow, 100),
MaxEntrySize: 500,
})
rand.Seed(time.Now().Unix())
b.RunParallel(func(pb *testing.PB) {
id := rand.Int()
counter := 0
b.ReportAllocs()
for pb.Next() {
cache.Set(fmt.Sprintf("key-%d-%d", id, counter), message)
counter = counter + 1
}
})
}
func readFromCache(b *testing.B, shards int) {
cache, _ := NewBigCache(Config{
Shards: shards,
LifeWindow: 1000 * time.Second,
MaxEntriesInWindow: max(b.N, 100),
MaxEntrySize: 500,
})
for i := 0; i < b.N; i++ {
cache.Set(strconv.Itoa(i), message)
}
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
b.ReportAllocs()
for pb.Next() {
cache.Get(strconv.Itoa(rand.Intn(b.N)))
}
})
}
+579
View File
@@ -0,0 +1,579 @@
package bigcache
import (
"fmt"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
var sink []byte
func TestParallel(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(DefaultConfig(5 * time.Second))
value := []byte("value")
var wg sync.WaitGroup
wg.Add(3)
keys := 1337
// when
go func() {
defer wg.Done()
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), value)
}
}()
go func() {
defer wg.Done()
for i := 0; i < keys; i++ {
sink, _ = cache.Get(fmt.Sprintf("key%d", i))
}
}()
go func() {
defer wg.Done()
for i := 0; i < keys; i++ {
cache.Delete(fmt.Sprintf("key%d", i))
}
}()
// then
wg.Wait()
}
func TestWriteAndGetOnCache(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(DefaultConfig(5 * time.Second))
value := []byte("value")
// when
cache.Set("key", value)
cachedValue, err := cache.Get("key")
// then
assert.NoError(t, err)
assert.Equal(t, value, cachedValue)
}
func TestConstructCacheWithDefaultHasher(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 16,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 10,
MaxEntrySize: 256,
})
assert.IsType(t, fnv64a{}, cache.hash)
}
func TestWillReturnErrorOnInvalidNumberOfPartitions(t *testing.T) {
t.Parallel()
// given
cache, error := NewBigCache(Config{
Shards: 18,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 10,
MaxEntrySize: 256,
})
assert.Nil(t, cache)
assert.Error(t, error, "Shards number must be power of two")
}
func TestEntryNotFound(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 16,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 10,
MaxEntrySize: 256,
})
// when
_, err := cache.Get("nonExistingKey")
// then
assert.EqualError(t, err, "Entry \"nonExistingKey\" not found")
}
func TestTimingEviction(t *testing.T) {
t.Parallel()
// given
clock := mockedClock{value: 0}
cache, _ := newBigCache(Config{
Shards: 1,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
}, &clock)
// when
cache.Set("key", []byte("value"))
clock.set(5)
cache.Set("key2", []byte("value2"))
_, err := cache.Get("key")
// then
assert.EqualError(t, err, "Entry \"key\" not found")
}
func TestTimingEvictionShouldEvictOnlyFromUpdatedShard(t *testing.T) {
t.Parallel()
// given
clock := mockedClock{value: 0}
cache, _ := newBigCache(Config{
Shards: 4,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
}, &clock)
// when
cache.Set("key", []byte("value"))
clock.set(5)
cache.Set("key2", []byte("value 2"))
value, err := cache.Get("key")
// then
assert.NoError(t, err, "Entry \"key\" not found")
assert.Equal(t, []byte("value"), value)
}
func TestCleanShouldEvictAll(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 4,
LifeWindow: time.Second,
CleanWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
// when
cache.Set("key", []byte("value"))
<-time.After(3 * time.Second)
value, err := cache.Get("key")
// then
assert.EqualError(t, err, "Entry \"key\" not found")
assert.Equal(t, value, []byte(nil))
}
func TestOnRemoveCallback(t *testing.T) {
t.Parallel()
// given
clock := mockedClock{value: 0}
onRemoveInvoked := false
onRemove := func(key string, entry []byte) {
onRemoveInvoked = true
assert.Equal(t, "key", key)
assert.Equal(t, []byte("value"), entry)
}
cache, _ := newBigCache(Config{
Shards: 1,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
OnRemove: onRemove,
}, &clock)
// when
cache.Set("key", []byte("value"))
clock.set(5)
cache.Set("key2", []byte("value2"))
// then
assert.True(t, onRemoveInvoked)
}
func TestCacheLen(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
keys := 1337
// when
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
// then
assert.Equal(t, keys, cache.Len())
}
func TestCacheStats(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
// when
for i := 0; i < 100; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
for i := 0; i < 10; i++ {
value, err := cache.Get(fmt.Sprintf("key%d", i))
assert.Nil(t, err)
assert.Equal(t, string(value), "value")
}
for i := 100; i < 110; i++ {
_, err := cache.Get(fmt.Sprintf("key%d", i))
assert.Error(t, err)
}
for i := 10; i < 20; i++ {
err := cache.Delete(fmt.Sprintf("key%d", i))
assert.Nil(t, err)
}
for i := 110; i < 120; i++ {
err := cache.Delete(fmt.Sprintf("key%d", i))
assert.Error(t, err)
}
// then
stats := cache.Stats()
assert.Equal(t, stats.Hits, int64(10))
assert.Equal(t, stats.Misses, int64(10))
assert.Equal(t, stats.DelHits, int64(10))
assert.Equal(t, stats.DelMisses, int64(10))
}
func TestCacheDel(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(DefaultConfig(time.Second))
// when
err := cache.Delete("nonExistingKey")
// then
assert.Equal(t, err.Error(), "Entry \"nonExistingKey\" not found")
// and when
cache.Set("existingKey", nil)
err = cache.Delete("existingKey")
cachedValue, _ := cache.Get("existingKey")
// then
assert.Nil(t, err)
assert.Len(t, cachedValue, 0)
}
func TestCacheReset(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
keys := 1337
// when
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
// then
assert.Equal(t, keys, cache.Len())
// and when
cache.Reset()
// then
assert.Equal(t, 0, cache.Len())
// and when
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
// then
assert.Equal(t, keys, cache.Len())
}
func TestIterateOnResetCache(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
keys := 1337
// when
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
cache.Reset()
// then
iterator := cache.Iterator()
assert.Equal(t, false, iterator.SetNext())
}
func TestGetOnResetCache(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
keys := 1337
// when
for i := 0; i < keys; i++ {
cache.Set(fmt.Sprintf("key%d", i), []byte("value"))
}
cache.Reset()
// then
value, err := cache.Get("key1")
assert.Equal(t, err.Error(), "Entry \"key1\" not found")
assert.Equal(t, value, []byte(nil))
}
func TestEntryUpdate(t *testing.T) {
t.Parallel()
// given
clock := mockedClock{value: 0}
cache, _ := newBigCache(Config{
Shards: 1,
LifeWindow: 6 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
}, &clock)
// when
cache.Set("key", []byte("value"))
clock.set(5)
cache.Set("key", []byte("value2"))
clock.set(7)
cache.Set("key2", []byte("value3"))
cachedValue, _ := cache.Get("key")
// then
assert.Equal(t, []byte("value2"), cachedValue)
}
func TestOldestEntryDeletionWhenMaxCacheSizeIsReached(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 1,
HardMaxCacheSize: 1,
})
// when
cache.Set("key1", blob('a', 1024*400))
cache.Set("key2", blob('b', 1024*400))
cache.Set("key3", blob('c', 1024*800))
_, key1Err := cache.Get("key1")
_, key2Err := cache.Get("key2")
entry3, _ := cache.Get("key3")
// then
assert.EqualError(t, key1Err, "Entry \"key1\" not found")
assert.EqualError(t, key2Err, "Entry \"key2\" not found")
assert.Equal(t, blob('c', 1024*800), entry3)
}
func TestRetrievingEntryShouldCopy(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 1,
HardMaxCacheSize: 1,
})
cache.Set("key1", blob('a', 1024*400))
value, key1Err := cache.Get("key1")
// when
// override queue
cache.Set("key2", blob('b', 1024*400))
cache.Set("key3", blob('c', 1024*400))
cache.Set("key4", blob('d', 1024*400))
cache.Set("key5", blob('d', 1024*400))
// then
assert.Nil(t, key1Err)
assert.Equal(t, blob('a', 1024*400), value)
}
func TestEntryBiggerThanMaxShardSizeError(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 1,
HardMaxCacheSize: 1,
})
// when
err := cache.Set("key1", blob('a', 1024*1025))
// then
assert.EqualError(t, err, "entry is bigger than max shard size")
}
func TestHashCollision(t *testing.T) {
t.Parallel()
ml := &mockedLogger{}
// given
cache, _ := NewBigCache(Config{
Shards: 16,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 10,
MaxEntrySize: 256,
Verbose: true,
Hasher: hashStub(5),
Logger: ml,
})
// when
cache.Set("liquid", []byte("value"))
cachedValue, err := cache.Get("liquid")
// then
assert.NoError(t, err)
assert.Equal(t, []byte("value"), cachedValue)
// when
cache.Set("costarring", []byte("value 2"))
cachedValue, err = cache.Get("costarring")
// then
assert.NoError(t, err)
assert.Equal(t, []byte("value 2"), cachedValue)
// when
cachedValue, err = cache.Get("liquid")
// then
assert.Error(t, err)
assert.Nil(t, cachedValue)
assert.NotEqual(t, "", ml.lastFormat)
assert.Equal(t, cache.Stats().Collisions, int64(1))
}
func TestNilValueCaching(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: 5 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 1,
HardMaxCacheSize: 1,
})
// when
cache.Set("Kierkegaard", []byte{})
cachedValue, err := cache.Get("Kierkegaard")
// then
assert.NoError(t, err)
assert.Equal(t, []byte{}, cachedValue)
// when
cache.Set("Sartre", nil)
cachedValue, err = cache.Get("Sartre")
// then
assert.NoError(t, err)
assert.Equal(t, []byte{}, cachedValue)
// when
cache.Set("Nietzsche", []byte(nil))
cachedValue, err = cache.Get("Nietzsche")
// then
assert.NoError(t, err)
assert.Equal(t, []byte{}, cachedValue)
}
type mockedLogger struct {
lastFormat string
lastArgs []interface{}
}
func (ml *mockedLogger) Printf(format string, v ...interface{}) {
ml.lastFormat = format
ml.lastArgs = v
}
type mockedClock struct {
value int64
}
func (mc *mockedClock) epoch() int64 {
return mc.value
}
func (mc *mockedClock) set(value int64) {
mc.value = value
}
func blob(char byte, len int) []byte {
b := make([]byte, len)
for index := range b {
b[index] = char
}
return b
}
+219
View File
@@ -0,0 +1,219 @@
package main
import (
"fmt"
"math/rand"
"sync"
"testing"
"time"
"github.com/allegro/bigcache"
"github.com/coocood/freecache"
)
const maxEntrySize = 256
func BenchmarkMapSet(b *testing.B) {
m := make(map[string][]byte)
for i := 0; i < b.N; i++ {
m[key(i)] = value()
}
}
func BenchmarkConcurrentMapSet(b *testing.B) {
var m sync.Map
for i := 0; i < b.N; i++ {
m.Store(key(i), value())
}
}
func BenchmarkFreeCacheSet(b *testing.B) {
cache := freecache.NewCache(b.N * maxEntrySize)
for i := 0; i < b.N; i++ {
cache.Set([]byte(key(i)), value(), 0)
}
}
func BenchmarkBigCacheSet(b *testing.B) {
cache := initBigCache(b.N)
for i := 0; i < b.N; i++ {
cache.Set(key(i), value())
}
}
func BenchmarkMapGet(b *testing.B) {
b.StopTimer()
m := make(map[string][]byte)
for i := 0; i < b.N; i++ {
m[key(i)] = value()
}
b.StartTimer()
hitCount := 0
for i := 0; i < b.N; i++ {
if m[key(i)] != nil {
hitCount++
}
}
}
func BenchmarkConcurrentMapGet(b *testing.B) {
b.StopTimer()
var m sync.Map
for i := 0; i < b.N; i++ {
m.Store(key(i), value())
}
b.StartTimer()
hitCounter := 0
for i := 0; i < b.N; i++ {
_, ok := m.Load(key(i))
if ok {
hitCounter++
}
}
}
func BenchmarkFreeCacheGet(b *testing.B) {
b.StopTimer()
cache := freecache.NewCache(b.N * maxEntrySize)
for i := 0; i < b.N; i++ {
cache.Set([]byte(key(i)), value(), 0)
}
b.StartTimer()
for i := 0; i < b.N; i++ {
cache.Get([]byte(key(i)))
}
}
func BenchmarkBigCacheGet(b *testing.B) {
b.StopTimer()
cache := initBigCache(b.N)
for i := 0; i < b.N; i++ {
cache.Set(key(i), value())
}
b.StartTimer()
for i := 0; i < b.N; i++ {
cache.Get(key(i))
}
}
func BenchmarkBigCacheSetParallel(b *testing.B) {
cache := initBigCache(b.N)
rand.Seed(time.Now().Unix())
b.RunParallel(func(pb *testing.PB) {
id := rand.Intn(1000)
counter := 0
for pb.Next() {
cache.Set(parallelKey(id, counter), value())
counter = counter + 1
}
})
}
func BenchmarkFreeCacheSetParallel(b *testing.B) {
cache := freecache.NewCache(b.N * maxEntrySize)
rand.Seed(time.Now().Unix())
b.RunParallel(func(pb *testing.PB) {
id := rand.Intn(1000)
counter := 0
for pb.Next() {
cache.Set([]byte(parallelKey(id, counter)), value(), 0)
counter = counter + 1
}
})
}
func BenchmarkConcurrentMapSetParallel(b *testing.B) {
var m sync.Map
b.RunParallel(func(pb *testing.PB) {
id := rand.Intn(1000)
for pb.Next() {
m.Store(key(id), value())
}
})
}
func BenchmarkBigCacheGetParallel(b *testing.B) {
b.StopTimer()
cache := initBigCache(b.N)
for i := 0; i < b.N; i++ {
cache.Set(key(i), value())
}
b.StartTimer()
b.RunParallel(func(pb *testing.PB) {
counter := 0
for pb.Next() {
cache.Get(key(counter))
counter = counter + 1
}
})
}
func BenchmarkFreeCacheGetParallel(b *testing.B) {
b.StopTimer()
cache := freecache.NewCache(b.N * maxEntrySize)
for i := 0; i < b.N; i++ {
cache.Set([]byte(key(i)), value(), 0)
}
b.StartTimer()
b.RunParallel(func(pb *testing.PB) {
counter := 0
for pb.Next() {
cache.Get([]byte(key(counter)))
counter = counter + 1
}
})
}
func BenchmarkConcurrentMapGetParallel(b *testing.B) {
b.StopTimer()
var m sync.Map
for i := 0; i < b.N; i++ {
m.Store(key(i), value())
}
b.StartTimer()
hitCount := 0
b.RunParallel(func(pb *testing.PB) {
id := rand.Intn(1000)
for pb.Next() {
_, ok := m.Load(key(id))
if ok {
hitCount++
}
}
})
}
func key(i int) string {
return fmt.Sprintf("key-%010d", i)
}
func value() []byte {
return make([]byte, 100)
}
func parallelKey(threadID int, counter int) string {
return fmt.Sprintf("key-%04d-%06d", threadID, counter)
}
func initBigCache(entriesInWindow int) *bigcache.BigCache {
cache, _ := bigcache.NewBigCache(bigcache.Config{
Shards: 256,
LifeWindow: 10 * time.Minute,
MaxEntriesInWindow: entriesInWindow,
MaxEntrySize: maxEntrySize,
Verbose: true,
})
return cache
}
@@ -0,0 +1,96 @@
package main
import (
"fmt"
"runtime"
"runtime/debug"
"time"
"github.com/allegro/bigcache"
"github.com/coocood/freecache"
)
func gcPause() time.Duration {
runtime.GC()
var stats debug.GCStats
debug.ReadGCStats(&stats)
return stats.PauseTotal
}
const (
entries = 20000000
valueSize = 100
)
func main() {
debug.SetGCPercent(10)
fmt.Println("Number of entries: ", entries)
config := bigcache.Config{
Shards: 256,
LifeWindow: 100 * time.Minute,
MaxEntriesInWindow: entries,
MaxEntrySize: 200,
Verbose: true,
}
bigcache, _ := bigcache.NewBigCache(config)
for i := 0; i < entries; i++ {
key, val := generateKeyValue(i, valueSize)
bigcache.Set(key, val)
}
firstKey, _ := generateKeyValue(1, valueSize)
checkFirstElement(bigcache.Get(firstKey))
fmt.Println("GC pause for bigcache: ", gcPause())
bigcache = nil
gcPause()
//------------------------------------------
freeCache := freecache.NewCache(entries * 200) //allocate entries * 200 bytes
for i := 0; i < entries; i++ {
key, val := generateKeyValue(i, valueSize)
if err := freeCache.Set([]byte(key), val, 0); err != nil {
fmt.Println("Error in set: ", err.Error())
}
}
firstKey, _ = generateKeyValue(1, valueSize)
checkFirstElement(freeCache.Get([]byte(firstKey)))
if freeCache.OverwriteCount() != 0 {
fmt.Println("Overwritten: ", freeCache.OverwriteCount())
}
fmt.Println("GC pause for freecache: ", gcPause())
freeCache = nil
gcPause()
//------------------------------------------
mapCache := make(map[string][]byte)
for i := 0; i < entries; i++ {
key, val := generateKeyValue(i, valueSize)
mapCache[key] = val
}
fmt.Println("GC pause for map: ", gcPause())
}
func checkFirstElement(val []byte, err error) {
_, expectedVal := generateKeyValue(1, valueSize)
if err != nil {
fmt.Println("Error in get: ", err.Error())
} else if string(val) != string(expectedVal) {
fmt.Println("Wrong first element: ", string(val))
}
}
func generateKeyValue(index int, valSize int) (string, []byte) {
key := fmt.Sprintf("key-%010d", index)
fixedNumber := []byte(fmt.Sprintf("%010d", index))
val := append(make([]byte, valSize-10), fixedNumber...)
return key, val
}
+14
View File
@@ -0,0 +1,14 @@
package bigcache
import "time"
type clock interface {
epoch() int64
}
type systemClock struct {
}
func (c systemClock) epoch() int64 {
return time.Now().Unix()
}
+67
View File
@@ -0,0 +1,67 @@
package bigcache
import "time"
// Config for BigCache
type Config struct {
// Number of cache shards, value must be a power of two
Shards int
// Time after which entry can be evicted
LifeWindow time.Duration
// Interval between removing expired entries (clean up).
// If set to <= 0 then no action is performed. Setting to < 1 second is counterproductive — bigcache has a one second resolution.
CleanWindow time.Duration
// Max number of entries in life window. Used only to calculate initial size for cache shards.
// When proper value is set then additional memory allocation does not occur.
MaxEntriesInWindow int
// Max size of entry in bytes. Used only to calculate initial size for cache shards.
MaxEntrySize int
// Verbose mode prints information about new memory allocation
Verbose bool
// Hasher used to map between string keys and unsigned 64bit integers, by default fnv64 hashing is used.
Hasher Hasher
// HardMaxCacheSize is a limit for cache size in MB. Cache will not allocate more memory than this limit.
// It can protect application from consuming all available memory on machine, therefore from running OOM Killer.
// Default value is 0 which means unlimited size. When the limit is higher than 0 and reached then
// the oldest entries are overridden for the new ones.
HardMaxCacheSize int
// OnRemove is a callback fired when the oldest entry is removed because of its expiration time or no space left
// for the new entry. Default value is nil which means no callback and it prevents from unwrapping the oldest entry.
OnRemove func(key string, entry []byte)
// Logger is a logging interface and used in combination with `Verbose`
// Defaults to `DefaultLogger()`
Logger Logger
}
// DefaultConfig initializes config with default values.
// When load for BigCache can be predicted in advance then it is better to use custom config.
func DefaultConfig(eviction time.Duration) Config {
return Config{
Shards: 1024,
LifeWindow: eviction,
CleanWindow: 0,
MaxEntriesInWindow: 1000 * 10 * 60,
MaxEntrySize: 500,
Verbose: true,
Hasher: newDefaultHasher(),
HardMaxCacheSize: 0,
Logger: DefaultLogger(),
}
}
// initialShardSize computes initial shard size
func (c Config) initialShardSize() int {
return max(c.MaxEntriesInWindow/c.Shards, minimumEntriesInShard)
}
// maximumShardSize computes maximum shard size
func (c Config) maximumShardSize() int {
maxShardSize := 0
if c.HardMaxCacheSize > 0 {
maxShardSize = convertMBToBytes(c.HardMaxCacheSize) / c.Shards
}
return maxShardSize
}
+70
View File
@@ -0,0 +1,70 @@
package bigcache
import (
"encoding/binary"
"reflect"
"unsafe"
)
const (
timestampSizeInBytes = 8 // Number of bytes used for timestamp
hashSizeInBytes = 8 // Number of bytes used for hash
keySizeInBytes = 2 // Number of bytes used for size of entry key
headersSizeInBytes = timestampSizeInBytes + hashSizeInBytes + keySizeInBytes // Number of bytes used for all headers
)
func wrapEntry(timestamp uint64, hash uint64, key string, entry []byte, buffer *[]byte) []byte {
keyLength := len(key)
blobLength := len(entry) + headersSizeInBytes + keyLength
if blobLength > len(*buffer) {
*buffer = make([]byte, blobLength)
}
blob := *buffer
binary.LittleEndian.PutUint64(blob, timestamp)
binary.LittleEndian.PutUint64(blob[timestampSizeInBytes:], hash)
binary.LittleEndian.PutUint16(blob[timestampSizeInBytes+hashSizeInBytes:], uint16(keyLength))
copy(blob[headersSizeInBytes:], key)
copy(blob[headersSizeInBytes+keyLength:], entry)
return blob[:blobLength]
}
func readEntry(data []byte) []byte {
length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
// copy on read
dst := make([]byte, len(data)-int(headersSizeInBytes+length))
copy(dst, data[headersSizeInBytes+length:])
return dst
}
func readTimestampFromEntry(data []byte) uint64 {
return binary.LittleEndian.Uint64(data)
}
func readKeyFromEntry(data []byte) string {
length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
// copy on read
dst := make([]byte, length)
copy(dst, data[headersSizeInBytes:headersSizeInBytes+length])
return bytesToString(dst)
}
func bytesToString(b []byte) string {
bytesHeader := (*reflect.SliceHeader)(unsafe.Pointer(&b))
strHeader := reflect.StringHeader{Data: bytesHeader.Data, Len: bytesHeader.Len}
return *(*string)(unsafe.Pointer(&strHeader))
}
func readHashFromEntry(data []byte) uint64 {
return binary.LittleEndian.Uint64(data[timestampSizeInBytes:])
}
func resetKeyFromEntry(data []byte) {
binary.LittleEndian.PutUint64(data[timestampSizeInBytes:], 0)
}
+46
View File
@@ -0,0 +1,46 @@
package bigcache
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestEncodeDecode(t *testing.T) {
// given
now := uint64(time.Now().Unix())
hash := uint64(42)
key := "key"
data := []byte("data")
buffer := make([]byte, 100)
// when
wrapped := wrapEntry(now, hash, key, data, &buffer)
// then
assert.Equal(t, key, readKeyFromEntry(wrapped))
assert.Equal(t, hash, readHashFromEntry(wrapped))
assert.Equal(t, now, readTimestampFromEntry(wrapped))
assert.Equal(t, data, readEntry(wrapped))
assert.Equal(t, 100, len(buffer))
}
func TestAllocateBiggerBuffer(t *testing.T) {
//given
now := uint64(time.Now().Unix())
hash := uint64(42)
key := "1"
data := []byte("2")
buffer := make([]byte, 1)
// when
wrapped := wrapEntry(now, hash, key, data, &buffer)
// then
assert.Equal(t, key, readKeyFromEntry(wrapped))
assert.Equal(t, hash, readHashFromEntry(wrapped))
assert.Equal(t, now, readTimestampFromEntry(wrapped))
assert.Equal(t, data, readEntry(wrapped))
assert.Equal(t, 2+headersSizeInBytes, len(buffer))
}
+17
View File
@@ -0,0 +1,17 @@
package bigcache
import "fmt"
// EntryNotFoundError is an error type struct which is returned when entry was not found for provided key
type EntryNotFoundError struct {
message string
}
func notFound(key string) error {
return &EntryNotFoundError{fmt.Sprintf("Entry %q not found", key)}
}
// Error returned when entry does not exist.
func (e EntryNotFoundError) Error() string {
return e.message
}
+28
View File
@@ -0,0 +1,28 @@
package bigcache
// newDefaultHasher returns a new 64-bit FNV-1a Hasher which makes no memory allocations.
// Its Sum64 method will lay the value out in big-endian byte order.
// See https://en.wikipedia.org/wiki/FowlerNollVo_hash_function
func newDefaultHasher() Hasher {
return fnv64a{}
}
type fnv64a struct{}
const (
// offset64 FNVa offset basis. See https://en.wikipedia.org/wiki/FowlerNollVo_hash_function#FNV-1a_hash
offset64 = 14695981039346656037
// prime64 FNVa prime value. See https://en.wikipedia.org/wiki/FowlerNollVo_hash_function#FNV-1a_hash
prime64 = 1099511628211
)
// Sum64 gets the string and returns its uint64 hash value.
func (f fnv64a) Sum64(key string) uint64 {
var hash uint64 = offset64
for i := 0; i < len(key); i++ {
hash ^= uint64(key[i])
hash *= prime64
}
return hash
}
+18
View File
@@ -0,0 +1,18 @@
package bigcache
import "testing"
var text = "abcdefg"
func BenchmarkFnvHashSum64(b *testing.B) {
h := newDefaultHasher()
for i := 0; i < b.N; i++ {
h.Sum64(text)
}
}
func BenchmarkFnvHashStdLibSum64(b *testing.B) {
for i := 0; i < b.N; i++ {
stdLibFnvSum64(text)
}
}
+35
View File
@@ -0,0 +1,35 @@
package bigcache
import (
"hash/fnv"
"testing"
)
type testCase struct {
text string
expectedHash uint64
}
var testCases = []testCase{
{"", stdLibFnvSum64("")},
{"a", stdLibFnvSum64("a")},
{"ab", stdLibFnvSum64("ab")},
{"abc", stdLibFnvSum64("abc")},
{"some longer and more complicated text", stdLibFnvSum64("some longer and more complicated text")},
}
func TestFnvHashSum64(t *testing.T) {
h := newDefaultHasher()
for _, testCase := range testCases {
hashed := h.Sum64(testCase.text)
if hashed != testCase.expectedHash {
t.Errorf("hash(%q) = %d want %d", testCase.text, hashed, testCase.expectedHash)
}
}
}
func stdLibFnvSum64(key string) uint64 {
h := fnv.New64a()
h.Write([]byte(key))
return h.Sum64()
}
+8
View File
@@ -0,0 +1,8 @@
package bigcache
// Hasher is responsible for generating unsigned, 64 bit hash of provided string. Hasher should minimize collisions
// (generating same hash for different strings) and while performance is also important fast functions are preferable (i.e.
// you can use FarmHash family).
type Hasher interface {
Sum64(string) uint64
}
+7
View File
@@ -0,0 +1,7 @@
package bigcache
type hashStub uint64
func (stub hashStub) Sum64(_ string) uint64 {
return uint64(stub)
}
+122
View File
@@ -0,0 +1,122 @@
package bigcache
import "sync"
type iteratorError string
func (e iteratorError) Error() string {
return string(e)
}
// ErrInvalidIteratorState is reported when iterator is in invalid state
const ErrInvalidIteratorState = iteratorError("Iterator is in invalid state. Use SetNext() to move to next position")
// ErrCannotRetrieveEntry is reported when entry cannot be retrieved from underlying
const ErrCannotRetrieveEntry = iteratorError("Could not retrieve entry from cache")
var emptyEntryInfo = EntryInfo{}
// EntryInfo holds informations about entry in the cache
type EntryInfo struct {
timestamp uint64
hash uint64
key string
value []byte
}
// Key returns entry's underlying key
func (e EntryInfo) Key() string {
return e.key
}
// Hash returns entry's hash value
func (e EntryInfo) Hash() uint64 {
return e.hash
}
// Timestamp returns entry's timestamp (time of insertion)
func (e EntryInfo) Timestamp() uint64 {
return e.timestamp
}
// Value returns entry's underlying value
func (e EntryInfo) Value() []byte {
return e.value
}
// EntryInfoIterator allows to iterate over entries in the cache
type EntryInfoIterator struct {
mutex sync.Mutex
cache *BigCache
currentShard int
currentIndex int
elements []uint32
elementsCount int
valid bool
}
// SetNext moves to next element and returns true if it exists.
func (it *EntryInfoIterator) SetNext() bool {
it.mutex.Lock()
it.valid = false
it.currentIndex++
if it.elementsCount > it.currentIndex {
it.valid = true
it.mutex.Unlock()
return true
}
for i := it.currentShard + 1; i < it.cache.config.Shards; i++ {
it.elements, it.elementsCount = it.cache.shards[i].copyKeys()
// Non empty shard - stick with it
if it.elementsCount > 0 {
it.currentIndex = 0
it.currentShard = i
it.valid = true
it.mutex.Unlock()
return true
}
}
it.mutex.Unlock()
return false
}
func newIterator(cache *BigCache) *EntryInfoIterator {
elements, count := cache.shards[0].copyKeys()
return &EntryInfoIterator{
cache: cache,
currentShard: 0,
currentIndex: -1,
elements: elements,
elementsCount: count,
}
}
// Value returns current value from the iterator
func (it *EntryInfoIterator) Value() (EntryInfo, error) {
it.mutex.Lock()
if !it.valid {
it.mutex.Unlock()
return emptyEntryInfo, ErrInvalidIteratorState
}
entry, err := it.cache.shards[it.currentShard].getEntry(int(it.elements[it.currentIndex]))
if err != nil {
it.mutex.Unlock()
return emptyEntryInfo, ErrCannotRetrieveEntry
}
it.mutex.Unlock()
return EntryInfo{
timestamp: readTimestampFromEntry(entry),
hash: readHashFromEntry(entry),
key: readKeyFromEntry(entry),
value: readEntry(entry),
}, nil
}
+150
View File
@@ -0,0 +1,150 @@
package bigcache
import (
"fmt"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestEntriesIterator(t *testing.T) {
t.Parallel()
// given
keysCount := 1000
cache, _ := NewBigCache(Config{
Shards: 8,
LifeWindow: 6 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
value := []byte("value")
for i := 0; i < keysCount; i++ {
cache.Set(fmt.Sprintf("key%d", i), value)
}
// when
keys := make(map[string]struct{})
iterator := cache.Iterator()
for iterator.SetNext() {
current, err := iterator.Value()
if err == nil {
keys[current.Key()] = struct{}{}
}
}
// then
assert.Equal(t, keysCount, len(keys))
}
func TestEntriesIteratorWithMostShardsEmpty(t *testing.T) {
t.Parallel()
// given
clock := mockedClock{value: 0}
cache, _ := newBigCache(Config{
Shards: 8,
LifeWindow: 6 * time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
}, &clock)
cache.Set("key", []byte("value"))
// when
iterator := cache.Iterator()
// then
if !iterator.SetNext() {
t.Errorf("Iterator should contain at least single element")
}
current, err := iterator.Value()
// then
assert.Nil(t, err)
assert.Equal(t, "key", current.Key())
assert.Equal(t, uint64(0x3dc94a19365b10ec), current.Hash())
assert.Equal(t, []byte("value"), current.Value())
assert.Equal(t, uint64(0), current.Timestamp())
}
func TestEntriesIteratorWithConcurrentUpdate(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
cache.Set("key", []byte("value"))
// when
iterator := cache.Iterator()
// then
if !iterator.SetNext() {
t.Errorf("Iterator should contain at least single element")
}
// Quite ugly but works
for i := 0; i < cache.config.Shards; i++ {
if oldestEntry, err := cache.shards[i].getOldestEntry(); err == nil {
cache.onEvict(oldestEntry, 10, cache.shards[i].removeOldestEntry)
}
}
current, err := iterator.Value()
// then
assert.Equal(t, ErrCannotRetrieveEntry, err)
assert.Equal(t, "Could not retrieve entry from cache", err.Error())
assert.Equal(t, EntryInfo{}, current)
}
func TestEntriesIteratorWithAllShardsEmpty(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
// when
iterator := cache.Iterator()
// then
if iterator.SetNext() {
t.Errorf("Iterator should not contain any elements")
}
}
func TestEntriesIteratorInInvalidState(t *testing.T) {
t.Parallel()
// given
cache, _ := NewBigCache(Config{
Shards: 1,
LifeWindow: time.Second,
MaxEntriesInWindow: 1,
MaxEntrySize: 256,
})
// when
iterator := cache.Iterator()
// then
_, err := iterator.Value()
assert.Equal(t, ErrInvalidIteratorState, err)
assert.Equal(t, "Iterator is in invalid state. Use SetNext() to move to next position", err.Error())
}
+30
View File
@@ -0,0 +1,30 @@
package bigcache
import (
"log"
"os"
)
// Logger is invoked when `Config.Verbose=true`
type Logger interface {
Printf(format string, v ...interface{})
}
// this is a safeguard, breaking on compile time in case
// `log.Logger` does not adhere to our `Logger` interface.
// see https://golang.org/doc/faq#guarantee_satisfies_interface
var _ Logger = &log.Logger{}
// DefaultLogger returns a `Logger` implementation
// backed by stdlib's log
func DefaultLogger() *log.Logger {
return log.New(os.Stdout, "", log.LstdFlags)
}
func newLogger(custom Logger) Logger {
if custom != nil {
return custom
}
return DefaultLogger()
}
+210
View File
@@ -0,0 +1,210 @@
package queue
import (
"encoding/binary"
"log"
"time"
)
const (
// Number of bytes used to keep information about entry size
headerEntrySize = 4
// Bytes before left margin are not used. Zero index means element does not exist in queue, useful while reading slice from index
leftMarginIndex = 1
// Minimum empty blob size in bytes. Empty blob fills space between tail and head in additional memory allocation.
// It keeps entries indexes unchanged
minimumEmptyBlobSize = 32 + headerEntrySize
)
// BytesQueue is a non-thread safe queue type of fifo based on bytes array.
// For every push operation index of entry is returned. It can be used to read the entry later
type BytesQueue struct {
array []byte
capacity int
maxCapacity int
head int
tail int
count int
rightMargin int
headerBuffer []byte
verbose bool
initialCapacity int
}
type queueError struct {
message string
}
// NewBytesQueue initialize new bytes queue.
// Initial capacity is used in bytes array allocation
// When verbose flag is set then information about memory allocation are printed
func NewBytesQueue(initialCapacity int, maxCapacity int, verbose bool) *BytesQueue {
return &BytesQueue{
array: make([]byte, initialCapacity),
capacity: initialCapacity,
maxCapacity: maxCapacity,
headerBuffer: make([]byte, headerEntrySize),
tail: leftMarginIndex,
head: leftMarginIndex,
rightMargin: leftMarginIndex,
verbose: verbose,
initialCapacity: initialCapacity,
}
}
// Reset removes all entries from queue
func (q *BytesQueue) Reset() {
// Just reset indexes
q.tail = leftMarginIndex
q.head = leftMarginIndex
q.rightMargin = leftMarginIndex
q.count = 0
}
// Push copies entry at the end of queue and moves tail pointer. Allocates more space if needed.
// Returns index for pushed data or error if maximum size queue limit is reached.
func (q *BytesQueue) Push(data []byte) (int, error) {
dataLen := len(data)
if q.availableSpaceAfterTail() < dataLen+headerEntrySize {
if q.availableSpaceBeforeHead() >= dataLen+headerEntrySize {
q.tail = leftMarginIndex
} else if q.capacity+headerEntrySize+dataLen >= q.maxCapacity && q.maxCapacity > 0 {
return -1, &queueError{"Full queue. Maximum size limit reached."}
} else {
q.allocateAdditionalMemory(dataLen + headerEntrySize)
}
}
index := q.tail
q.push(data, dataLen)
return index, nil
}
func (q *BytesQueue) allocateAdditionalMemory(minimum int) {
start := time.Now()
if q.capacity < minimum {
q.capacity += minimum
}
q.capacity = q.capacity * 2
if q.capacity > q.maxCapacity && q.maxCapacity > 0 {
q.capacity = q.maxCapacity
}
oldArray := q.array
q.array = make([]byte, q.capacity)
if leftMarginIndex != q.rightMargin {
copy(q.array, oldArray[:q.rightMargin])
if q.tail < q.head {
emptyBlobLen := q.head - q.tail - headerEntrySize
q.push(make([]byte, emptyBlobLen), emptyBlobLen)
q.head = leftMarginIndex
q.tail = q.rightMargin
}
}
if q.verbose {
log.Printf("Allocated new queue in %s; Capacity: %d \n", time.Since(start), q.capacity)
}
}
func (q *BytesQueue) push(data []byte, len int) {
binary.LittleEndian.PutUint32(q.headerBuffer, uint32(len))
q.copy(q.headerBuffer, headerEntrySize)
q.copy(data, len)
if q.tail > q.head {
q.rightMargin = q.tail
}
q.count++
}
func (q *BytesQueue) copy(data []byte, len int) {
q.tail += copy(q.array[q.tail:], data[:len])
}
// Pop reads the oldest entry from queue and moves head pointer to the next one
func (q *BytesQueue) Pop() ([]byte, error) {
data, size, err := q.peek(q.head)
if err != nil {
return nil, err
}
q.head += headerEntrySize + size
q.count--
if q.head == q.rightMargin {
q.head = leftMarginIndex
if q.tail == q.rightMargin {
q.tail = leftMarginIndex
}
q.rightMargin = q.tail
}
return data, nil
}
// Peek reads the oldest entry from list without moving head pointer
func (q *BytesQueue) Peek() ([]byte, error) {
data, _, err := q.peek(q.head)
return data, err
}
// Get reads entry from index
func (q *BytesQueue) Get(index int) ([]byte, error) {
data, _, err := q.peek(index)
return data, err
}
// Capacity returns number of allocated bytes for queue
func (q *BytesQueue) Capacity() int {
return q.capacity
}
// Len returns number of entries kept in queue
func (q *BytesQueue) Len() int {
return q.count
}
// Error returns error message
func (e *queueError) Error() string {
return e.message
}
func (q *BytesQueue) peek(index int) ([]byte, int, error) {
if q.count == 0 {
return nil, 0, &queueError{"Empty queue"}
}
if index <= 0 {
return nil, 0, &queueError{"Index must be grater than zero. Invalid index."}
}
if index+headerEntrySize >= len(q.array) {
return nil, 0, &queueError{"Index out of range"}
}
blockSize := int(binary.LittleEndian.Uint32(q.array[index : index+headerEntrySize]))
return q.array[index+headerEntrySize : index+headerEntrySize+blockSize], blockSize, nil
}
func (q *BytesQueue) availableSpaceAfterTail() int {
if q.tail >= q.head {
return q.capacity - q.tail
}
return q.head - q.tail - minimumEmptyBlobSize
}
func (q *BytesQueue) availableSpaceBeforeHead() int {
if q.tail >= q.head {
return q.head - leftMarginIndex - minimumEmptyBlobSize
}
return q.head - q.tail - minimumEmptyBlobSize
}
+365
View File
@@ -0,0 +1,365 @@
package queue
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestPushAndPop(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(10, 0, true)
entry := []byte("hello")
// when
_, err := queue.Pop()
// then
assert.EqualError(t, err, "Empty queue")
// when
queue.Push(entry)
// then
assert.Equal(t, entry, pop(queue))
}
func TestLen(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
entry := []byte("hello")
assert.Zero(t, queue.Len())
// when
queue.Push(entry)
// then
assert.Equal(t, queue.Len(), 1)
}
func TestPeek(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
entry := []byte("hello")
// when
read, err := queue.Peek()
// then
assert.EqualError(t, err, "Empty queue")
assert.Nil(t, read)
// when
queue.Push(entry)
read, err = queue.Peek()
// then
assert.NoError(t, err)
assert.Equal(t, pop(queue), read)
assert.Equal(t, entry, read)
}
func TestReset(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
entry := []byte("hello")
// when
queue.Push(entry)
queue.Push(entry)
queue.Push(entry)
queue.Reset()
read, err := queue.Peek()
// then
assert.EqualError(t, err, "Empty queue")
assert.Nil(t, read)
// when
queue.Push(entry)
read, err = queue.Peek()
// then
assert.NoError(t, err)
assert.Equal(t, pop(queue), read)
assert.Equal(t, entry, read)
// when
read, err = queue.Peek()
// then
assert.EqualError(t, err, "Empty queue")
assert.Nil(t, read)
}
func TestReuseAvailableSpace(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
// when
queue.Push(blob('a', 70))
queue.Push(blob('b', 20))
queue.Pop()
queue.Push(blob('c', 20))
// then
assert.Equal(t, 100, queue.Capacity())
assert.Equal(t, blob('b', 20), pop(queue))
}
func TestAllocateAdditionalSpace(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(11, 0, false)
// when
queue.Push([]byte("hello1"))
queue.Push([]byte("hello2"))
// then
assert.Equal(t, 22, queue.Capacity())
}
func TestAllocateAdditionalSpaceForInsufficientFreeFragmentedSpaceWhereHeadIsBeforeTail(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(25, 0, false)
// when
queue.Push(blob('a', 3)) // header + entry + left margin = 8 bytes
queue.Push(blob('b', 6)) // additional 10 bytes
queue.Pop() // space freed, 7 bytes available at the beginning
queue.Push(blob('c', 6)) // 10 bytes needed, 14 available but not in one segment, allocate additional memory
// then
assert.Equal(t, 50, queue.Capacity())
assert.Equal(t, blob('b', 6), pop(queue))
assert.Equal(t, blob('c', 6), pop(queue))
}
func TestUnchangedEntriesIndexesAfterAdditionalMemoryAllocationWhereHeadIsBeforeTail(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(25, 0, false)
// when
queue.Push(blob('a', 3)) // header + entry + left margin = 8 bytes
index, _ := queue.Push(blob('b', 6)) // additional 10 bytes
queue.Pop() // space freed, 7 bytes available at the beginning
newestIndex, _ := queue.Push(blob('c', 6)) // 10 bytes needed, 14 available but not in one segment, allocate additional memory
// then
assert.Equal(t, 50, queue.Capacity())
assert.Equal(t, blob('b', 6), get(queue, index))
assert.Equal(t, blob('c', 6), get(queue, newestIndex))
}
func TestAllocateAdditionalSpaceForInsufficientFreeFragmentedSpaceWhereTailIsBeforeHead(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
// when
queue.Push(blob('a', 70)) // header + entry + left margin = 75 bytes
queue.Push(blob('b', 10)) // 75 + 10 + 4 = 89 bytes
queue.Pop() // space freed at the beginning
queue.Push(blob('c', 30)) // 34 bytes used at the beginning, tail pointer is before head pointer
queue.Push(blob('d', 40)) // 44 bytes needed but no available in one segment, allocate new memory
// then
assert.Equal(t, 200, queue.Capacity())
assert.Equal(t, blob('c', 30), pop(queue))
// empty blob fills space between tail and head,
// created when additional memory was allocated,
// it keeps current entries indexes unchanged
assert.Equal(t, blob(0, 36), pop(queue))
assert.Equal(t, blob('b', 10), pop(queue))
assert.Equal(t, blob('d', 40), pop(queue))
}
func TestUnchangedEntriesIndexesAfterAdditionalMemoryAllocationWhereTailIsBeforeHead(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(100, 0, false)
// when
queue.Push(blob('a', 70)) // header + entry + left margin = 75 bytes
index, _ := queue.Push(blob('b', 10)) // 75 + 10 + 4 = 89 bytes
queue.Pop() // space freed at the beginning
queue.Push(blob('c', 30)) // 34 bytes used at the beginning, tail pointer is before head pointer
newestIndex, _ := queue.Push(blob('d', 40)) // 44 bytes needed but no available in one segment, allocate new memory
// then
assert.Equal(t, 200, queue.Capacity())
assert.Equal(t, blob('b', 10), get(queue, index))
assert.Equal(t, blob('d', 40), get(queue, newestIndex))
}
func TestAllocateAdditionalSpaceForValueBiggerThanInitQueue(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(11, 0, false)
// when
queue.Push(blob('a', 100))
// then
assert.Equal(t, blob('a', 100), pop(queue))
assert.Equal(t, 230, queue.Capacity())
}
func TestAllocateAdditionalSpaceForValueBiggerThanQueue(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(21, 0, false)
// when
queue.Push(make([]byte, 2))
queue.Push(make([]byte, 2))
queue.Push(make([]byte, 100))
// then
queue.Pop()
queue.Pop()
assert.Equal(t, make([]byte, 100), pop(queue))
assert.Equal(t, 250, queue.Capacity())
}
func TestPopWholeQueue(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(13, 0, false)
// when
queue.Push([]byte("a"))
queue.Push([]byte("b"))
queue.Pop()
queue.Pop()
queue.Push([]byte("c"))
// then
assert.Equal(t, 13, queue.Capacity())
assert.Equal(t, []byte("c"), pop(queue))
}
func TestGetEntryFromIndex(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(20, 0, false)
// when
queue.Push([]byte("a"))
index, _ := queue.Push([]byte("b"))
queue.Push([]byte("c"))
result, _ := queue.Get(index)
// then
assert.Equal(t, []byte("b"), result)
}
func TestGetEntryFromInvalidIndex(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(1, 0, false)
queue.Push([]byte("a"))
// when
result, err := queue.Get(0)
// then
assert.Nil(t, result)
assert.EqualError(t, err, "Index must be grater than zero. Invalid index.")
}
func TestGetEntryFromIndexOutOfRange(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(1, 0, false)
queue.Push([]byte("a"))
// when
result, err := queue.Get(42)
// then
assert.Nil(t, result)
assert.EqualError(t, err, "Index out of range")
}
func TestGetEntryFromEmptyQueue(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(13, 0, false)
// when
result, err := queue.Get(1)
// then
assert.Nil(t, result)
assert.EqualError(t, err, "Empty queue")
}
func TestMaxSizeLimit(t *testing.T) {
t.Parallel()
// given
queue := NewBytesQueue(30, 50, false)
// when
queue.Push(blob('a', 25))
queue.Push(blob('b', 5))
capacity := queue.Capacity()
_, err := queue.Push(blob('c', 15))
// then
assert.Equal(t, 50, capacity)
assert.EqualError(t, err, "Full queue. Maximum size limit reached.")
assert.Equal(t, blob('a', 25), pop(queue))
assert.Equal(t, blob('b', 5), pop(queue))
}
func pop(queue *BytesQueue) []byte {
entry, err := queue.Pop()
if err != nil {
panic(err)
}
return entry
}
func get(queue *BytesQueue, index int) []byte {
entry, err := queue.Get(index)
if err != nil {
panic(err)
}
return entry
}
func blob(char byte, len int) []byte {
b := make([]byte, len)
for index := range b {
b[index] = char
}
return b
}
+105
View File
@@ -0,0 +1,105 @@
# BigCache HTTP Server
This is a basic HTTP server implementation for BigCache. It has a basic RESTful API and is designed for easy operational deployments. This server is intended to be consumed as a standalone executable, for things like Cloud Foundry, Heroku, etc. A design goal is versatility, so if you want to cache pictures, software artifacts, text, or any type of bit, the BigCache HTTP Server should fit your needs.
```bash
# cache API.
GET /api/v1/cache/{key}
PUT /api/v1/cache/{key}
DELETE /api/v1/cache/{key}
# stats API.
GET /api/v1/stats
```
The cache API is designed for ease-of-use caching and accepts any content type. The stats API will return hit and miss statistics about the cache since the last time the server was started - they will reset whenever the server is restarted.
### Notes for Operators
1. No SSL support, currently.
1. No authentication, currently.
1. Stats from the stats API are not persistent.
1. The easiest way to clean the cache is to restart the process; it takes less than a second to initialise.
1. There is no replication or clustering.
### Command-line Interface
```powershell
PS C:\go\src\github.com\mxplusb\bigcache\server> .\server.exe -h
Usage of C:\go\src\github.com\mxplusb\bigcache\server\server.exe:
-lifetime duration
Lifetime of each cache object. (default 10m0s)
-logfile string
Location of the logfile.
-max int
Maximum amount of data in the cache in MB. (default 8192)
-maxInWindow int
Used only in initial memory allocation. (default 600000)
-maxShardEntrySize int
The maximum size of each object stored in a shard. Used only in initial memory allocation. (default 500)
-port int
The port to listen on. (default 9090)
-shards int
Number of shards for the cache. (default 1024)
-v Verbose logging.
-version
Print server version.
```
Example:
```bash
$ curl -v -XPUT localhost:9090/api/v1/cache/example -d "yay!"
* Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 9090 (#0)
> PUT /api/v1/cache/example HTTP/1.1
> Host: localhost:9090
> User-Agent: curl/7.47.0
> Accept: */*
> Content-Length: 4
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 4 out of 4 bytes
< HTTP/1.1 201 Created
< Date: Fri, 17 Nov 2017 03:50:07 GMT
< Content-Length: 0
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
$
$ curl -v -XGET localhost:9090/api/v1/cache/example
Note: Unnecessary use of -X or --request, GET is already inferred.
* Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 9090 (#0)
> GET /api/v1/cache/example HTTP/1.1
> Host: localhost:9090
> User-Agent: curl/7.47.0
> Accept: */*
>
< HTTP/1.1 200 OK
< Date: Fri, 17 Nov 2017 03:50:23 GMT
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
yay!
```
The server does log basic metrics:
```bash
$ ./server
2017/11/16 22:49:22 cache initialised.
2017/11/16 22:49:22 starting server on :9090
2017/11/16 22:50:07 stored "example" in cache.
2017/11/16 22:50:07 request took 277000ns.
2017/11/16 22:50:23 request took 9000ns.
```
### Acquiring Natively
This is native Go with no external dependencies, so it will compile for all supported Golang platforms. To build:
```bash
go build server.go
```
+87
View File
@@ -0,0 +1,87 @@
package main
import (
"io/ioutil"
"log"
"net/http"
"strings"
)
func cacheIndexHandler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
getCacheHandler(w, r)
case http.MethodPut:
putCacheHandler(w, r)
case http.MethodDelete:
deleteCacheHandler(w, r)
}
})
}
// handles get requests.
func getCacheHandler(w http.ResponseWriter, r *http.Request) {
target := r.URL.Path[len(cachePath):]
if target == "" {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("can't get a key if there is no key."))
log.Print("empty request.")
return
}
entry, err := cache.Get(target)
if err != nil {
errMsg := (err).Error()
if strings.Contains(errMsg, "not found") {
log.Print(err)
w.WriteHeader(http.StatusNotFound)
return
}
log.Print(err)
w.WriteHeader(http.StatusInternalServerError)
return
}
w.Write(entry)
}
func putCacheHandler(w http.ResponseWriter, r *http.Request) {
target := r.URL.Path[len(cachePath):]
if target == "" {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("can't put a key if there is no key."))
log.Print("empty request.")
return
}
entry, err := ioutil.ReadAll(r.Body)
if err != nil {
log.Print(err)
w.WriteHeader(http.StatusInternalServerError)
return
}
if err := cache.Set(target, []byte(entry)); err != nil {
log.Print(err)
w.WriteHeader(http.StatusInternalServerError)
return
}
log.Printf("stored \"%s\" in cache.", target)
w.WriteHeader(http.StatusCreated)
}
// delete cache objects.
func deleteCacheHandler(w http.ResponseWriter, r *http.Request) {
target := r.URL.Path[len(cachePath):]
if err := cache.Delete(target); err != nil {
if strings.Contains((err).Error(), "not found") {
w.WriteHeader(http.StatusNotFound)
log.Printf("%s not found.", target)
return
}
w.WriteHeader(http.StatusInternalServerError)
log.Printf("internal cache error: %s", err)
}
// this is what the RFC says to use when calling DELETE.
w.WriteHeader(http.StatusOK)
return
}
+29
View File
@@ -0,0 +1,29 @@
package main
import (
"log"
"net/http"
"time"
)
// our base middleware implementation.
type service func(http.Handler) http.Handler
// chain load middleware services.
func serviceLoader(h http.Handler, svcs ...service) http.Handler {
for _, svc := range svcs {
h = svc(h)
}
return h
}
// middleware for request length metrics.
func requestMetrics(l *log.Logger) service {
return func(h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
h.ServeHTTP(w, r)
l.Printf("%s request to %s took %vns.", r.Method, r.URL.Path, time.Now().Sub(start).Nanoseconds())
})
}
}
+85
View File
@@ -0,0 +1,85 @@
package main
import (
"flag"
"fmt"
"log"
"net/http"
"os"
"strconv"
"github.com/allegro/bigcache"
)
const (
// base HTTP paths.
apiVersion = "v1"
apiBasePath = "/api/" + apiVersion + "/"
// path to cache.
cachePath = apiBasePath + "cache/"
statsPath = apiBasePath + "stats"
// server version.
version = "1.0.0"
)
var (
port int
logfile string
ver bool
// cache-specific settings.
cache *bigcache.BigCache
config = bigcache.Config{}
)
func init() {
flag.BoolVar(&config.Verbose, "v", false, "Verbose logging.")
flag.IntVar(&config.Shards, "shards", 1024, "Number of shards for the cache.")
flag.IntVar(&config.MaxEntriesInWindow, "maxInWindow", 1000*10*60, "Used only in initial memory allocation.")
flag.DurationVar(&config.LifeWindow, "lifetime", 100000*100000*60, "Lifetime of each cache object.")
flag.IntVar(&config.HardMaxCacheSize, "max", 8192, "Maximum amount of data in the cache in MB.")
flag.IntVar(&config.MaxEntrySize, "maxShardEntrySize", 500, "The maximum size of each object stored in a shard. Used only in initial memory allocation.")
flag.IntVar(&port, "port", 9090, "The port to listen on.")
flag.StringVar(&logfile, "logfile", "", "Location of the logfile.")
flag.BoolVar(&ver, "version", false, "Print server version.")
}
func main() {
flag.Parse()
if ver {
fmt.Printf("BigCache HTTP Server v%s", version)
os.Exit(0)
}
var logger *log.Logger
if logfile == "" {
logger = log.New(os.Stdout, "", log.LstdFlags)
} else {
f, err := os.OpenFile(logfile, os.O_APPEND|os.O_WRONLY, 0600)
if err != nil {
panic(err)
}
logger = log.New(f, "", log.LstdFlags)
}
var err error
cache, err = bigcache.NewBigCache(config)
if err != nil {
logger.Fatal(err)
}
logger.Print("cache initialised.")
// let the middleware log.
http.Handle(cachePath, serviceLoader(cacheIndexHandler(), requestMetrics(logger)))
http.Handle(statsPath, serviceLoader(statsIndexHandler(), requestMetrics(logger)))
logger.Printf("starting server on :%d", port)
strPort := ":" + strconv.Itoa(port)
log.Fatal("ListenAndServe: ", http.ListenAndServe(strPort, nil))
}
+185
View File
@@ -0,0 +1,185 @@
package main
import (
"bytes"
"encoding/json"
"io/ioutil"
"net/http/httptest"
"testing"
"time"
"github.com/allegro/bigcache"
)
const (
testBaseString = "http://bigcache.org"
)
func testCacheSetup() {
cache, _ = bigcache.NewBigCache(bigcache.Config{
Shards: 1024,
LifeWindow: 10 * time.Minute,
MaxEntriesInWindow: 1000 * 10 * 60,
MaxEntrySize: 500,
Verbose: true,
HardMaxCacheSize: 8192,
OnRemove: nil,
})
}
func TestMain(m *testing.M) {
testCacheSetup()
m.Run()
}
func TestGetWithNoKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("GET", testBaseString+"/api/v1/cache/", nil)
rr := httptest.NewRecorder()
getCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 400 {
t.Errorf("want: 400; got: %d", resp.StatusCode)
}
}
func TestGetWithMissingKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("GET", testBaseString+"/api/v1/cache/doesNotExist", nil)
rr := httptest.NewRecorder()
getCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 404 {
t.Errorf("want: 404; got: %d", resp.StatusCode)
}
}
func TestGetKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("GET", testBaseString+"/api/v1/cache/getKey", nil)
rr := httptest.NewRecorder()
// set something.
cache.Set("getKey", []byte("123"))
getCacheHandler(rr, req)
resp := rr.Result()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
t.Errorf("cannot deserialise test response: %s", err)
}
if string(body) != "123" {
t.Errorf("want: 123; got: %s.\n\tcan't get existing key getKey.", string(body))
}
}
func TestPutKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("PUT", testBaseString+"/api/v1/cache/putKey", bytes.NewBuffer([]byte("123")))
rr := httptest.NewRecorder()
putCacheHandler(rr, req)
testPutKeyResult, err := cache.Get("putKey")
if err != nil {
t.Errorf("error returning cache entry: %s", err)
}
if string(testPutKeyResult) != "123" {
t.Errorf("want: 123; got: %s.\n\tcan't get PUT key putKey.", string(testPutKeyResult))
}
}
func TestPutEmptyKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("PUT", testBaseString+"/api/v1/cache/", bytes.NewBuffer([]byte("123")))
rr := httptest.NewRecorder()
putCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 400 {
t.Errorf("want: 400; got: %d.\n\tempty key insertion should return with 400", resp.StatusCode)
}
}
func TestDeleteEmptyKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("DELETE", testBaseString+"/api/v1/cache/", bytes.NewBuffer([]byte("123")))
rr := httptest.NewRecorder()
deleteCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 404 {
t.Errorf("want: 404; got: %d.\n\tapparently we're trying to delete empty keys.", resp.StatusCode)
}
}
func TestDeleteInvalidKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("DELETE", testBaseString+"/api/v1/cache/invalidDeleteKey", bytes.NewBuffer([]byte("123")))
rr := httptest.NewRecorder()
deleteCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 404 {
t.Errorf("want: 404; got: %d.\n\tapparently we're trying to delete invalid keys.", resp.StatusCode)
}
}
func TestDeleteKey(t *testing.T) {
t.Parallel()
req := httptest.NewRequest("DELETE", testBaseString+"/api/v1/cache/testDeleteKey", bytes.NewBuffer([]byte("123")))
rr := httptest.NewRecorder()
if err := cache.Set("testDeleteKey", []byte("123")); err != nil {
t.Errorf("can't set key for testing. %s", err)
}
deleteCacheHandler(rr, req)
resp := rr.Result()
if resp.StatusCode != 200 {
t.Errorf("want: 200; got: %d.\n\tcan't delete keys.", resp.StatusCode)
}
}
func TestGetStats(t *testing.T) {
t.Parallel()
var testStats bigcache.Stats
req := httptest.NewRequest("GET", testBaseString+"/api/v1/stats", nil)
rr := httptest.NewRecorder()
// manually enter a key so there are some stats. get it so there's at least 1 hit.
if err := cache.Set("incrementStats", []byte("123")); err != nil {
t.Errorf("error setting cache value. error %s", err)
}
// it's okay if this fails, since we'll catch it downstream.
if _, err := cache.Get("incrementStats"); err != nil {
t.Errorf("can't find incrementStats. error: %s", err)
}
getCacheStatsHandler(rr, req)
resp := rr.Result()
if err := json.NewDecoder(resp.Body).Decode(&testStats); err != nil {
t.Errorf("error decoding cache stats. error: %s", err)
}
if testStats.Hits == 0 {
t.Errorf("want: > 0; got: 0.\n\thandler not properly returning stats info.")
}
}
+33
View File
@@ -0,0 +1,33 @@
package main
import (
"encoding/json"
"log"
"net/http"
)
// index for stats handle
func statsIndexHandler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
getCacheStatsHandler(w, r)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
})
}
// returns the cache's statistics.
func getCacheStatsHandler(w http.ResponseWriter, r *http.Request) {
target, err := json.Marshal(cache.Stats())
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
log.Printf("cannot marshal cache stats. error: %s", err)
return
}
// since we're sending a struct, make it easy for consumers to interface.
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.Write(target)
return
}
+229
View File
@@ -0,0 +1,229 @@
package bigcache
import (
"fmt"
"sync"
"sync/atomic"
"github.com/allegro/bigcache/queue"
)
type cacheShard struct {
hashmap map[uint64]uint32
entries queue.BytesQueue
lock sync.RWMutex
entryBuffer []byte
onRemove func(wrappedEntry []byte)
isVerbose bool
logger Logger
clock clock
lifeWindow uint64
stats Stats
}
type onRemoveCallback func(wrappedEntry []byte)
func (s *cacheShard) get(key string, hashedKey uint64) ([]byte, error) {
s.lock.RLock()
itemIndex := s.hashmap[hashedKey]
if itemIndex == 0 {
s.lock.RUnlock()
s.miss()
return nil, notFound(key)
}
wrappedEntry, err := s.entries.Get(int(itemIndex))
if err != nil {
s.lock.RUnlock()
s.miss()
return nil, err
}
if entryKey := readKeyFromEntry(wrappedEntry); key != entryKey {
if s.isVerbose {
s.logger.Printf("Collision detected. Both %q and %q have the same hash %x", key, entryKey, hashedKey)
}
s.lock.RUnlock()
s.collision()
return nil, notFound(key)
}
s.lock.RUnlock()
s.hit()
return readEntry(wrappedEntry), nil
}
func (s *cacheShard) set(key string, hashedKey uint64, entry []byte) error {
currentTimestamp := uint64(s.clock.epoch())
s.lock.Lock()
if previousIndex := s.hashmap[hashedKey]; previousIndex != 0 {
if previousEntry, err := s.entries.Get(int(previousIndex)); err == nil {
resetKeyFromEntry(previousEntry)
}
}
if oldestEntry, err := s.entries.Peek(); err == nil {
s.onEvict(oldestEntry, currentTimestamp, s.removeOldestEntry)
}
w := wrapEntry(currentTimestamp, hashedKey, key, entry, &s.entryBuffer)
for {
if index, err := s.entries.Push(w); err == nil {
s.hashmap[hashedKey] = uint32(index)
s.lock.Unlock()
return nil
}
if s.removeOldestEntry() != nil {
s.lock.Unlock()
return fmt.Errorf("entry is bigger than max shard size")
}
}
}
func (s *cacheShard) del(key string, hashedKey uint64) error {
s.lock.RLock()
itemIndex := s.hashmap[hashedKey]
if itemIndex == 0 {
s.lock.RUnlock()
s.delmiss()
return notFound(key)
}
wrappedEntry, err := s.entries.Get(int(itemIndex))
if err != nil {
s.lock.RUnlock()
s.delmiss()
return err
}
s.lock.RUnlock()
s.lock.Lock()
{
delete(s.hashmap, hashedKey)
s.onRemove(wrappedEntry)
resetKeyFromEntry(wrappedEntry)
}
s.lock.Unlock()
s.delhit()
return nil
}
func (s *cacheShard) onEvict(oldestEntry []byte, currentTimestamp uint64, evict func() error) bool {
oldestTimestamp := readTimestampFromEntry(oldestEntry)
if currentTimestamp-oldestTimestamp > s.lifeWindow {
evict()
return true
}
return false
}
func (s *cacheShard) cleanUp(currentTimestamp uint64) {
s.lock.Lock()
for {
if oldestEntry, err := s.entries.Peek(); err != nil {
break
} else if evicted := s.onEvict(oldestEntry, currentTimestamp, s.removeOldestEntry); !evicted {
break
}
}
s.lock.Unlock()
}
func (s *cacheShard) getOldestEntry() ([]byte, error) {
return s.entries.Peek()
}
func (s *cacheShard) getEntry(index int) ([]byte, error) {
return s.entries.Get(index)
}
func (s *cacheShard) copyKeys() (keys []uint32, next int) {
keys = make([]uint32, len(s.hashmap))
s.lock.RLock()
for _, index := range s.hashmap {
keys[next] = index
next++
}
s.lock.RUnlock()
return keys, next
}
func (s *cacheShard) removeOldestEntry() error {
oldest, err := s.entries.Pop()
if err == nil {
hash := readHashFromEntry(oldest)
delete(s.hashmap, hash)
s.onRemove(oldest)
return nil
}
return err
}
func (s *cacheShard) reset(config Config) {
s.lock.Lock()
s.hashmap = make(map[uint64]uint32, config.initialShardSize())
s.entryBuffer = make([]byte, config.MaxEntrySize+headersSizeInBytes)
s.entries.Reset()
s.lock.Unlock()
}
func (s *cacheShard) len() int {
s.lock.RLock()
res := len(s.hashmap)
s.lock.RUnlock()
return res
}
func (s *cacheShard) getStats() Stats {
var stats = Stats{
Hits: atomic.LoadInt64(&s.stats.Hits),
Misses: atomic.LoadInt64(&s.stats.Misses),
DelHits: atomic.LoadInt64(&s.stats.DelHits),
DelMisses: atomic.LoadInt64(&s.stats.DelMisses),
Collisions: atomic.LoadInt64(&s.stats.Collisions),
}
return stats
}
func (s *cacheShard) hit() {
atomic.AddInt64(&s.stats.Hits, 1)
}
func (s *cacheShard) miss() {
atomic.AddInt64(&s.stats.Misses, 1)
}
func (s *cacheShard) delhit() {
atomic.AddInt64(&s.stats.DelHits, 1)
}
func (s *cacheShard) delmiss() {
atomic.AddInt64(&s.stats.DelMisses, 1)
}
func (s *cacheShard) collision() {
atomic.AddInt64(&s.stats.Collisions, 1)
}
func initNewShard(config Config, callback onRemoveCallback, clock clock) *cacheShard {
return &cacheShard{
hashmap: make(map[uint64]uint32, config.initialShardSize()),
entries: *queue.NewBytesQueue(config.initialShardSize()*config.MaxEntrySize, config.maximumShardSize(), config.Verbose),
entryBuffer: make([]byte, config.MaxEntrySize+headersSizeInBytes),
onRemove: callback,
isVerbose: config.Verbose,
logger: newLogger(config.Logger),
clock: clock,
lifeWindow: uint64(config.LifeWindow.Seconds()),
}
}
+15
View File
@@ -0,0 +1,15 @@
package bigcache
// Stats stores cache statistics
type Stats struct {
// Hits is a number of successfully found keys
Hits int64 `json:"hits"`
// Misses is a number of not found keys
Misses int64 `json:"misses"`
// DelHits is a number of successfully deleted keys
DelHits int64 `json:"delete_hits"`
// DelMisses is a number of not deleted keys
DelMisses int64 `json:"delete_misses"`
// Collisions is a number of happened key-collisions
Collisions int64 `json:"collisions"`
}
+16
View File
@@ -0,0 +1,16 @@
package bigcache
func max(a, b int) int {
if a > b {
return a
}
return b
}
func convertMBToBytes(value int) int {
return value * 1024 * 1024
}
func isPowerOfTwo(number int) bool {
return (number & (number - 1)) == 0
}