📦 add mapstore

pull/1025/head
Fenny 2020-11-20 11:43:07 +01:00
parent 6ded637712
commit 8fe458011d
9 changed files with 352 additions and 156 deletions

View File

@ -0,0 +1,90 @@
package mapstore
import (
"sync"
"sync/atomic"
"time"
)
type Storage struct {
sync.RWMutex
data map[string]item // data
ts uint64 // timestamp
}
type item struct {
v interface{} // val
e uint64 // exp
}
func New() *Storage {
store := &Storage{
data: make(map[string]item),
ts: uint64(time.Now().Unix()),
}
go store.gc(10 * time.Millisecond)
go store.updater(1 * time.Second)
return store
}
// Get value by key
func (s *Storage) Get(key string) interface{} {
s.RLock()
v, ok := s.data[key]
s.RUnlock()
if !ok || v.e != 0 && v.e <= atomic.LoadUint64(&s.ts) {
return nil
}
return v.v
}
// Set key with value
func (s *Storage) Set(key string, val interface{}, ttl time.Duration) {
var exp uint64
if ttl > 0 {
exp = uint64(ttl.Seconds()) + atomic.LoadUint64(&s.ts)
}
s.Lock()
s.data[key] = item{val, exp}
s.Unlock()
}
// Delete key by key
func (s *Storage) Delete(key string) {
s.Lock()
delete(s.data, key)
s.Unlock()
}
// Reset all keys
func (s *Storage) Reset() {
s.Lock()
s.data = make(map[string]item)
s.Unlock()
}
func (s *Storage) updater(sleep time.Duration) {
for {
time.Sleep(sleep)
atomic.StoreUint64(&s.ts, uint64(time.Now().Unix()))
}
}
func (s *Storage) gc(sleep time.Duration) {
expired := []string{}
for {
time.Sleep(sleep)
expired = expired[:0]
s.RLock()
for key, v := range s.data {
if v.e != 0 && v.e <= atomic.LoadUint64(&s.ts) {
expired = append(expired, key)
}
}
s.RUnlock()
s.Lock()
for i := range expired {
delete(s.data, expired[i])
}
s.Unlock()
}
}

View File

@ -0,0 +1,81 @@
package mapstore
import (
"testing"
"time"
"github.com/gofiber/fiber/v2/utils"
)
// go test -run Test_MapStore -v -race
func Test_MapStore(t *testing.T) {
var store = New()
var (
key = "john"
val interface{} = []byte("doe")
exp = 1 * time.Second
)
store.Set(key, val, 0)
store.Set(key, val, 0)
result := store.Get(key)
utils.AssertEqual(t, val, result)
result = store.Get("empty")
utils.AssertEqual(t, nil, result)
store.Set(key, val, exp)
time.Sleep(1100 * time.Millisecond)
result = store.Get(key)
utils.AssertEqual(t, nil, result)
store.Set(key, val, 0)
result = store.Get(key)
utils.AssertEqual(t, val, result)
store.Delete(key)
result = store.Get(key)
utils.AssertEqual(t, nil, result)
store.Set("john", val, 0)
store.Set("doe", val, 0)
store.Reset()
result = store.Get("john")
utils.AssertEqual(t, nil, result)
result = store.Get("doe")
utils.AssertEqual(t, nil, result)
}
// go test -v -run=^$ -bench=Benchmark_MapStore -benchmem -count=4
func Benchmark_MapStore(b *testing.B) {
keyLength := 1000
keys := make([]string, keyLength)
for i := 0; i < keyLength; i++ {
keys[i] = utils.UUID()
}
value := []string{"some", "random", "value"}
ttl := 2 * time.Second
b.Run("fiber_memory", func(b *testing.B) {
d := New()
b.ReportAllocs()
b.ResetTimer()
for n := 0; n < b.N; n++ {
for _, key := range keys {
d.Set(key, value, ttl)
}
for _, key := range keys {
_ = d.Get(key)
}
for _, key := range keys {
d.Delete(key)
}
}
})
}

View File

@ -1,33 +0,0 @@
package memory
import "time"
// Config defines the config for storage.
type Config struct {
// Time before deleting expired keys
//
// Default is 10 * time.Second
GCInterval time.Duration
}
// ConfigDefault is the default config
var ConfigDefault = Config{
GCInterval: 10 * time.Second,
}
// configDefault is a helper function to set default values
func configDefault(config ...Config) Config {
// Return default config if nothing provided
if len(config) < 1 {
return ConfigDefault
}
// Override default config
cfg := config[0]
// Set default values
if int(cfg.GCInterval.Seconds()) <= 0 {
cfg.GCInterval = ConfigDefault.GCInterval
}
return cfg
}

View File

@ -14,10 +14,8 @@ type Storage struct {
done chan struct{} done chan struct{}
} }
// Common storage errors
// ErrNotFound means that a get call did not find the requested key. var ErrNotExist = errors.New("key does not exist")
var ErrNotFound = errors.New("key not found")
var ErrKeyNotExist = ErrNotFound
type entry struct { type entry struct {
data []byte data []byte
@ -25,14 +23,11 @@ type entry struct {
} }
// New creates a new memory storage // New creates a new memory storage
func New(config ...Config) *Storage { func New() *Storage {
// Set default config
cfg := configDefault(config...)
// Create storage // Create storage
store := &Storage{ store := &Storage{
db: make(map[string]entry), db: make(map[string]entry),
gcInterval: cfg.GCInterval, gcInterval: 10 * time.Second,
done: make(chan struct{}), done: make(chan struct{}),
} }

View File

@ -4,7 +4,6 @@ package cache
import ( import (
"strconv" "strconv"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -31,11 +30,7 @@ func New(config ...Config) fiber.Handler {
) )
// create storage handler // create storage handler
store := &storage{ store := newStorage(&cfg)
cfg: &cfg,
mux: &sync.RWMutex{},
entries: make(map[string]*entry),
}
// Update timestamp every second // Update timestamp every second
go func() { go func() {
@ -61,15 +56,16 @@ func New(config ...Config) fiber.Handler {
key := cfg.KeyGenerator(c) key := cfg.KeyGenerator(c)
// Get/Create new entry // Get/Create new entry
var e = store.get(key) e := store.get(key)
if e == nil {
e = &entry{}
}
// Get timestamp // Get timestamp
ts := atomic.LoadUint64(&timestamp) ts := atomic.LoadUint64(&timestamp)
// Set expiration if entry does not exist // Set expiration if entry does not exist
if e.exp == 0 { if e.exp == 0 {
e.exp = ts + expiration e.exp = ts + expiration
} else if ts >= e.exp { } else if ts >= e.exp {
// Check if entry is expired // Check if entry is expired
store.delete(key) store.delete(key)

View File

@ -6,14 +6,11 @@ import (
"bytes" "bytes"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"net/http"
"net/http/httptest" "net/http/httptest"
"sync"
"testing" "testing"
"time" "time"
"github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/internal/storage/memory"
"github.com/gofiber/fiber/v2/utils" "github.com/gofiber/fiber/v2/utils"
"github.com/valyala/fasthttp" "github.com/valyala/fasthttp"
) )
@ -94,54 +91,54 @@ func Test_Cache(t *testing.T) {
utils.AssertEqual(t, cachedBody, body) utils.AssertEqual(t, cachedBody, body)
} }
// go test -run Test_Cache_Concurrency_Storage -race -v // // go test -run Test_Cache_Concurrency_Storage -race -v
func Test_Cache_Concurrency_Storage(t *testing.T) { // func Test_Cache_Concurrency_Storage(t *testing.T) {
// Test concurrency using a custom store // // Test concurrency using a custom store
app := fiber.New() // app := fiber.New()
app.Use(New(Config{ // app.Use(New(Config{
Storage: memory.New(), // Storage: memory.New(),
})) // }))
app.Get("/", func(c *fiber.Ctx) error { // app.Get("/", func(c *fiber.Ctx) error {
return c.SendString("Hello tester!") // return c.SendString("Hello tester!")
}) // })
var wg sync.WaitGroup // var wg sync.WaitGroup
singleRequest := func(wg *sync.WaitGroup) { // singleRequest := func(wg *sync.WaitGroup) {
defer wg.Done() // defer wg.Done()
resp, err := app.Test(httptest.NewRequest(http.MethodGet, "/", nil)) // resp, err := app.Test(httptest.NewRequest(http.MethodGet, "/", nil))
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
utils.AssertEqual(t, fiber.StatusOK, resp.StatusCode) // utils.AssertEqual(t, fiber.StatusOK, resp.StatusCode)
body, err := ioutil.ReadAll(resp.Body) // body, err := ioutil.ReadAll(resp.Body)
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
utils.AssertEqual(t, "Hello tester!", string(body)) // utils.AssertEqual(t, "Hello tester!", string(body))
} // }
for i := 0; i <= 49; i++ { // for i := 0; i <= 49; i++ {
wg.Add(1) // wg.Add(1)
go singleRequest(&wg) // go singleRequest(&wg)
} // }
wg.Wait() // wg.Wait()
req := httptest.NewRequest("GET", "/", nil) // req := httptest.NewRequest("GET", "/", nil)
resp, err := app.Test(req) // resp, err := app.Test(req)
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
cachedReq := httptest.NewRequest("GET", "/", nil) // cachedReq := httptest.NewRequest("GET", "/", nil)
cachedResp, err := app.Test(cachedReq) // cachedResp, err := app.Test(cachedReq)
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
body, err := ioutil.ReadAll(resp.Body) // body, err := ioutil.ReadAll(resp.Body)
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
cachedBody, err := ioutil.ReadAll(cachedResp.Body) // cachedBody, err := ioutil.ReadAll(cachedResp.Body)
utils.AssertEqual(t, nil, err) // utils.AssertEqual(t, nil, err)
utils.AssertEqual(t, cachedBody, body) // utils.AssertEqual(t, cachedBody, body)
} // }
func Test_Cache_Invalid_Expiration(t *testing.T) { func Test_Cache_Invalid_Expiration(t *testing.T) {
app := fiber.New() app := fiber.New()
@ -285,7 +282,7 @@ func Benchmark_Cache_Storage(b *testing.B) {
app := fiber.New() app := fiber.New()
app.Use(New(Config{ app.Use(New(Config{
Store: memory.New(), //// Store: memory.New(),
})) }))
app.Get("/demo", func(c *fiber.Ctx) error { app.Get("/demo", func(c *fiber.Ctx) error {

View File

@ -1,6 +1,8 @@
package cache package cache
import "sync" import (
"github.com/gofiber/fiber/v2/internal/mapstore"
)
// go:generate msgp // go:generate msgp
// msgp -file="store.go" -o="store_msgp.go" -tests=false -unexported // msgp -file="store.go" -o="store_msgp.go" -tests=false -unexported
@ -15,31 +17,42 @@ type entry struct {
//msgp:ignore storage //msgp:ignore storage
type storage struct { type storage struct {
cfg *Config cfg *Config
mux *sync.RWMutex store *mapstore.MapStore
entries map[string]*entry
} }
func (s *storage) get(key string) (e *entry) { func newStorage(cfg *Config) *storage {
store := &storage{
cfg: cfg,
}
if cfg.Storage == nil {
store.store = mapstore.New()
}
return store
}
func (s *storage) get(key string) *entry {
if s.cfg.Storage != nil { if s.cfg.Storage != nil {
raw, err := s.cfg.Storage.Get(key) raw, err := s.cfg.Storage.Get(key)
if err != nil || raw == nil { if err != nil || raw == nil {
return e return nil
} }
e := &entry{}
if _, err := e.UnmarshalMsg(raw); err != nil { if _, err := e.UnmarshalMsg(raw); err != nil {
return e return nil
} }
body, err := s.cfg.Storage.Get(key + "_body") body, err := s.cfg.Storage.Get(key + "_body")
if err != nil || body == nil { if err != nil || body == nil {
return e return nil
} }
e.body = body e.body = body
return e
} else { } else {
s.mux.Lock() val := s.store.Get(key)
e = s.entries[key] if val != nil {
s.mux.Lock() return val.(*entry)
}
} }
return e return nil
} }
func (s *storage) set(key string, e *entry) { func (s *storage) set(key string, e *entry) {
@ -53,9 +66,7 @@ func (s *storage) set(key string, e *entry) {
_ = s.cfg.Storage.Set(key+"_body", body, s.cfg.Expiration) _ = s.cfg.Storage.Set(key+"_body", body, s.cfg.Expiration)
} }
} else { } else {
s.mux.Lock() s.store.Set(key, e, s.cfg.Expiration)
s.entries[key] = e
s.mux.Unlock()
} }
} }
@ -64,8 +75,6 @@ func (s *storage) delete(key string) {
_ = s.cfg.Storage.Delete(key) _ = s.cfg.Storage.Delete(key)
_ = s.cfg.Storage.Delete(key + "_body") _ = s.cfg.Storage.Delete(key + "_body")
} else { } else {
s.mux.Lock() s.store.Delete(key)
delete(s.entries, key)
s.mux.Unlock()
} }
} }

View File

@ -1,9 +1,7 @@
package limiter package limiter
import ( import (
"fmt"
"strconv" "strconv"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -30,12 +28,14 @@ func New(config ...Config) fiber.Handler {
max = strconv.Itoa(cfg.Max) max = strconv.Itoa(cfg.Max)
timestamp = uint64(time.Now().Unix()) timestamp = uint64(time.Now().Unix())
expiration = uint64(cfg.Expiration.Seconds()) expiration = uint64(cfg.Expiration.Seconds())
mux = &sync.RWMutex{} // mux = &sync.RWMutex{}
// Default store logic (if no Store is provided) // // Default store logic (if no Store is provided)
entries = make(map[string]entry) // entries = make(map[string]entry)
) )
store := newStorage(&cfg)
// Update timestamp every second // Update timestamp every second
go func() { go func() {
for { for {
@ -54,65 +54,67 @@ func New(config ...Config) fiber.Handler {
// Get key from request // Get key from request
key := cfg.KeyGenerator(c) key := cfg.KeyGenerator(c)
// Create new entry e := store.get(key)
entry := entry{} // // Create new entry
// entry := entry{}
// Lock entry // // Lock entry
mux.Lock() // mux.Lock()
defer mux.Unlock() // defer mux.Unlock()
// Use Storage if provided // // Use Storage if provided
if cfg.Storage != nil { // if cfg.Storage != nil {
val, err := cfg.Storage.Get(key) // val, err := cfg.Storage.Get(key)
if val != nil && len(val) > 0 { // if val != nil && len(val) > 0 {
if _, err := entry.UnmarshalMsg(val); err != nil { // if _, err := entry.UnmarshalMsg(val); err != nil {
return err // return err
} // }
} // }
if err != nil && err.Error() != errNotExist { // if err != nil && err.Error() != errNotExist {
fmt.Println("[LIMITER]", err.Error()) // fmt.Println("[LIMITER]", err.Error())
} // }
} else { // } else {
entry = entries[key] // entry = entries[key]
} // }
// Get timestamp // Get timestamp
ts := atomic.LoadUint64(&timestamp) ts := atomic.LoadUint64(&timestamp)
// Set expiration if entry does not exist // Set expiration if entry does not exist
if entry.exp == 0 { if e.exp == 0 {
entry.exp = ts + expiration e.exp = ts + expiration
} else if ts >= entry.exp { } else if ts >= e.exp {
// Check if entry is expired // Check if entry is expired
entry.hits = 0 e.hits = 0
entry.exp = ts + expiration e.exp = ts + expiration
} }
// Increment hits // Increment hits
entry.hits++ e.hits++
// Use Storage if provided store.set(key, e)
if cfg.Storage != nil { // // Use Storage if provided
// Marshal entry to bytes // if cfg.Storage != nil {
val, err := entry.MarshalMsg(nil) // // Marshal entry to bytes
if err != nil { // val, err := entry.MarshalMsg(nil)
return err // if err != nil {
} // return err
// }
// Pass value to Storage // // Pass value to Storage
if err = cfg.Storage.Set(key, val, cfg.Expiration); err != nil { // if err = cfg.Storage.Set(key, val, cfg.Expiration); err != nil {
return err // return err
} // }
} else { // } else {
entries[key] = entry // entries[key] = entry
} // }
// Calculate when it resets in seconds // Calculate when it resets in seconds
expire := entry.exp - ts expire := e.exp - ts
// Set how many hits we have left // Set how many hits we have left
remaining := cfg.Max - entry.hits remaining := cfg.Max - e.hits
// Check if hits exceed the cfg.Max // Check if hits exceed the cfg.Max
if remaining < 0 { if remaining < 0 {

View File

@ -1,5 +1,9 @@
package limiter package limiter
import (
"github.com/gofiber/fiber/v2/internal/mapstore"
)
// go:generate msgp // go:generate msgp
// msgp -file="store.go" -o="store_msgp.go" -tests=false -unexported // msgp -file="store.go" -o="store_msgp.go" -tests=false -unexported
// don't forget to replace the msgp import path to: // don't forget to replace the msgp import path to:
@ -8,3 +12,58 @@ type entry struct {
hits int `msg:"hits"` hits int `msg:"hits"`
exp uint64 `msg:"exp"` exp uint64 `msg:"exp"`
} }
//msgp:ignore storage
type storage struct {
cfg *Config
store *mapstore.Storage
}
func newStorage(cfg *Config) *storage {
store := &storage{
cfg: cfg,
}
if cfg.Storage == nil {
store.store = mapstore.New()
}
return store
}
func (s *storage) get(key string) (e entry) {
if s.cfg.Storage != nil {
raw, err := s.cfg.Storage.Get(key)
if err != nil || raw == nil {
return
}
if _, err := e.UnmarshalMsg(raw); err != nil {
return
}
return
} else {
// val := s.mem.Get(key).(*entry)
var ok bool
e, ok = s.store.Get(key).(entry)
if !ok {
return
}
}
return
}
func (s *storage) set(key string, e entry) {
if s.cfg.Storage != nil {
if data, err := e.MarshalMsg(nil); err == nil {
_ = s.cfg.Storage.Set(key, data, s.cfg.Expiration)
}
} else {
s.store.Set(key, e, s.cfg.Expiration)
}
}
func (s *storage) delete(key string) {
if s.cfg.Storage != nil {
_ = s.cfg.Storage.Delete(key)
} else {
s.store.Delete(key)
}
}