mirror of
https://github.com/lukaszraczylo/go-simple-graphql.git
synced 2026-10-15 01:07:34 +00:00
General library improvements.
This commit is contained in:
Vendored
+122
-27
@@ -10,22 +10,27 @@ import (
|
||||
)
|
||||
|
||||
type CacheEntry struct {
|
||||
ExpiresAt time.Time
|
||||
Value []byte
|
||||
ExpiresAt time.Time
|
||||
Value []byte
|
||||
IsCompressed bool
|
||||
}
|
||||
|
||||
const shardCount = 256 // Must be power of 2
|
||||
|
||||
type shard struct {
|
||||
entries map[string]CacheEntry
|
||||
entries map[string]CacheEntry
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
type Cache struct {
|
||||
compressPool sync.Pool
|
||||
decompressPool sync.Pool
|
||||
shards [shardCount]*shard
|
||||
globalTTL time.Duration
|
||||
shards [shardCount]*shard
|
||||
globalTTL time.Duration
|
||||
cleanupChan chan struct{}
|
||||
stopChan chan struct{}
|
||||
stopped bool
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// getShard returns the appropriate shard for a given key
|
||||
@@ -37,7 +42,9 @@ func (c *Cache) getShard(key string) *shard {
|
||||
|
||||
func New(globalTTL time.Duration) *Cache {
|
||||
cache := &Cache{
|
||||
globalTTL: globalTTL,
|
||||
globalTTL: globalTTL,
|
||||
cleanupChan: make(chan struct{}, 1),
|
||||
stopChan: make(chan struct{}),
|
||||
compressPool: sync.Pool{
|
||||
New: func() interface{} {
|
||||
w := gzip.NewWriter(nil)
|
||||
@@ -59,31 +66,116 @@ func New(globalTTL time.Duration) *Cache {
|
||||
}
|
||||
}
|
||||
|
||||
go cache.cleanupRoutine(globalTTL)
|
||||
go cache.lazyCleanupWorker()
|
||||
go cache.periodicCleanupRoutine(globalTTL)
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *Cache) cleanupRoutine(globalTTL time.Duration) {
|
||||
ticker := time.NewTicker(globalTTL / 2)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range ticker.C {
|
||||
c.CleanExpiredEntries()
|
||||
func (c *Cache) lazyCleanupWorker() {
|
||||
for {
|
||||
select {
|
||||
case <-c.cleanupChan:
|
||||
c.mu.RLock()
|
||||
if c.stopped {
|
||||
c.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
c.CleanExpiredEntries()
|
||||
case <-c.stopChan:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) periodicCleanupRoutine(globalTTL time.Duration) {
|
||||
// Use a shorter cleanup interval for better responsiveness, but not less than 1 second
|
||||
cleanupInterval := globalTTL / 4
|
||||
if cleanupInterval < time.Second {
|
||||
cleanupInterval = time.Second
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(cleanupInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
c.mu.RLock()
|
||||
if c.stopped {
|
||||
c.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
c.CleanExpiredEntries()
|
||||
case <-c.stopChan:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) triggerLazyCleanup() {
|
||||
c.mu.RLock()
|
||||
if c.stopped {
|
||||
c.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
|
||||
select {
|
||||
case c.cleanupChan <- struct{}{}:
|
||||
// Cleanup triggered
|
||||
default:
|
||||
// Cleanup already pending, skip
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) Stop() {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if c.stopped {
|
||||
return // Already stopped
|
||||
}
|
||||
|
||||
c.stopped = true
|
||||
close(c.stopChan)
|
||||
}
|
||||
|
||||
func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
|
||||
shard := c.getShard(key)
|
||||
shard.Lock()
|
||||
defer shard.Unlock()
|
||||
|
||||
compressedValue, err := c.compress(value)
|
||||
if err != nil {
|
||||
return
|
||||
const compressionThreshold = 1024 // 1KB threshold
|
||||
var finalValue []byte
|
||||
var isCompressed bool
|
||||
|
||||
if len(value) >= compressionThreshold {
|
||||
compressedValue, err := c.compress(value)
|
||||
if err != nil {
|
||||
// If compression fails, store uncompressed
|
||||
finalValue = value
|
||||
isCompressed = false
|
||||
} else {
|
||||
// Only use compression if it actually reduces size
|
||||
if len(compressedValue) < len(value) {
|
||||
finalValue = compressedValue
|
||||
isCompressed = true
|
||||
} else {
|
||||
finalValue = value
|
||||
isCompressed = false
|
||||
}
|
||||
}
|
||||
} else {
|
||||
finalValue = value
|
||||
isCompressed = false
|
||||
}
|
||||
|
||||
shard.entries[key] = CacheEntry{
|
||||
Value: compressedValue,
|
||||
ExpiresAt: time.Now().Add(ttl),
|
||||
Value: finalValue,
|
||||
ExpiresAt: time.Now().Add(ttl),
|
||||
IsCompressed: isCompressed,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -98,19 +190,22 @@ func (c *Cache) Get(key string) ([]byte, bool) {
|
||||
|
||||
if entry.ExpiresAt.Before(time.Now()) {
|
||||
shard.RUnlock()
|
||||
// Clean up expired entry in background
|
||||
go func() {
|
||||
shard.Lock()
|
||||
delete(shard.entries, key)
|
||||
shard.Unlock()
|
||||
}()
|
||||
// Trigger lazy cleanup instead of immediate deletion
|
||||
c.triggerLazyCleanup()
|
||||
return nil, false
|
||||
}
|
||||
shard.RUnlock()
|
||||
|
||||
value, err := c.decompress(entry.Value)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
var value []byte
|
||||
var err error
|
||||
|
||||
if entry.IsCompressed {
|
||||
value, err = c.decompress(entry.Value)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
} else {
|
||||
value = entry.Value
|
||||
}
|
||||
|
||||
return value, true
|
||||
|
||||
Vendored
+11
-7
@@ -8,22 +8,24 @@ import (
|
||||
// Assume that New function initializes the cache and it is defined somewhere in the libpack_cache package.
|
||||
|
||||
func BenchmarkCacheSet(b *testing.B) {
|
||||
cache := New(30 * time.Second) // Initializing the cache with a TTL of 30 seconds
|
||||
cache := New(5 * time.Second) // Use shorter TTL for tests
|
||||
defer cache.Stop()
|
||||
key := "benchmark-key"
|
||||
value := []byte("benchmark-value")
|
||||
|
||||
b.ResetTimer() // Reset the timer to exclude the setup time from the benchmark
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
cache.Set(key, value, 5*time.Second)
|
||||
cache.Set(key, value, 2*time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkCacheGet(b *testing.B) {
|
||||
cache := New(30 * time.Second) // Initializing the cache
|
||||
cache := New(5 * time.Second) // Use shorter TTL for tests
|
||||
defer cache.Stop()
|
||||
key := "benchmark-key"
|
||||
value := []byte("benchmark-value")
|
||||
cache.Set(key, value, 5*time.Second) // Pre-set a value to retrieve
|
||||
cache.Set(key, value, 2*time.Second) // Pre-set a value to retrieve
|
||||
|
||||
b.ResetTimer() // Start timing
|
||||
|
||||
@@ -33,14 +35,16 @@ func BenchmarkCacheGet(b *testing.B) {
|
||||
}
|
||||
|
||||
func BenchmarkCacheExpire(b *testing.B) {
|
||||
key := "benchmark-expire-key"
|
||||
cache := New(5 * time.Second) // Use shorter TTL for tests and reuse cache
|
||||
defer cache.Stop()
|
||||
value := []byte("benchmark-value")
|
||||
ttl := 5 * time.Millisecond // Setting a short TTL for quick expiration
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
cache := New(30 * time.Second)
|
||||
key := "benchmark-expire-key-" + string(rune(i)) // Use unique keys
|
||||
cache.Set(key, value, ttl)
|
||||
time.Sleep(ttl) // Wait for the key to expire
|
||||
time.Sleep(ttl + time.Millisecond) // Wait for the key to expire with buffer
|
||||
_, _ = cache.Get(key)
|
||||
}
|
||||
}
|
||||
|
||||
Vendored
+15
-8
@@ -21,12 +21,15 @@ func TestCachingTestSuite(t *testing.T) {
|
||||
func (suite *CacheTestSuite) Test_New() {
|
||||
suite.T().Run("should return a new cache", func(t *testing.T) {
|
||||
cache := New(2 * time.Second)
|
||||
defer cache.Stop()
|
||||
suite.NotNil(cache)
|
||||
})
|
||||
}
|
||||
|
||||
func (suite *CacheTestSuite) Test_CacheUse() {
|
||||
cache := New(30 * time.Second)
|
||||
cache := New(5 * time.Second)
|
||||
defer cache.Stop()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
cache_value string
|
||||
@@ -42,7 +45,7 @@ func (suite *CacheTestSuite) Test_CacheUse() {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
suite.T().Run(tt.name, func(t *testing.T) {
|
||||
cache.Set(tt.name, []byte(tt.name), 5*time.Second)
|
||||
cache.Set(tt.name, []byte(tt.name), 2*time.Second)
|
||||
c, ok := cache.Get(tt.name)
|
||||
suite.Equal(true, ok)
|
||||
suite.Equal(tt.name, string(c))
|
||||
@@ -51,7 +54,9 @@ func (suite *CacheTestSuite) Test_CacheUse() {
|
||||
}
|
||||
|
||||
func (suite *CacheTestSuite) Test_CacheDelete() {
|
||||
cache := New(30 * time.Second)
|
||||
cache := New(5 * time.Second)
|
||||
defer cache.Stop()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
cache_value string
|
||||
@@ -67,7 +72,7 @@ func (suite *CacheTestSuite) Test_CacheDelete() {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
suite.T().Run(tt.name, func(t *testing.T) {
|
||||
cache.Set(tt.name, []byte(tt.name), 5*time.Second)
|
||||
cache.Set(tt.name, []byte(tt.name), 2*time.Second)
|
||||
c, ok := cache.Get(tt.name)
|
||||
suite.Equal(true, ok)
|
||||
suite.Equal(tt.name, string(c))
|
||||
@@ -80,7 +85,9 @@ func (suite *CacheTestSuite) Test_CacheDelete() {
|
||||
}
|
||||
|
||||
func (suite *CacheTestSuite) Test_CacheExpire() {
|
||||
cache := New(30 * time.Second)
|
||||
cache := New(5 * time.Second)
|
||||
defer cache.Stop()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
cache_value string
|
||||
@@ -89,12 +96,12 @@ func (suite *CacheTestSuite) Test_CacheExpire() {
|
||||
{
|
||||
name: "test1",
|
||||
cache_value: "test1-123",
|
||||
ttl: 2 * time.Second,
|
||||
ttl: 100 * time.Millisecond,
|
||||
},
|
||||
{
|
||||
name: "test2",
|
||||
cache_value: "test2-123",
|
||||
ttl: 5 * time.Second,
|
||||
ttl: 200 * time.Millisecond,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
@@ -103,7 +110,7 @@ func (suite *CacheTestSuite) Test_CacheExpire() {
|
||||
c, ok := cache.Get(tt.name)
|
||||
suite.Equal(true, ok)
|
||||
suite.Equal(tt.name, string(c))
|
||||
time.Sleep(tt.ttl)
|
||||
time.Sleep(tt.ttl + 50*time.Millisecond) // Add small buffer
|
||||
c, ok = cache.Get(tt.name)
|
||||
suite.Equal(false, ok)
|
||||
suite.Equal("", string(c))
|
||||
|
||||
Reference in New Issue
Block a user