lukaszraczylo/traefikoidc
1
0
Fork 0
mirror of https://github.com/lukaszraczylo/traefikoidc.git synced 2026-10-18 04:47:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pooling cleanup.

Browse Source
This commit is contained in:
lukaszraczylo
2025-09-30 10:47:26 +01:00
parent 913264c0b7
commit ac0f95dd14
14 changed files with 229 additions and 1437 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cache/typed_cache.go
Vendored
+17 -3
View File
@@ -1,9 +1,12 @@
package cache package cache
import ( import (
"bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"time" "time"
"github.com/lukaszraczylo/traefikoidc/internal/pool"
) )
// TypedCache provides a type-safe wrapper around Cache for specific types // TypedCache provides a type-safe wrapper around Cache for specific types
@@ -42,13 +45,24 @@ func (tc *TypedCache[T]) Get(key string) (T, bool) {
} }
// If that fails, try JSON marshaling/unmarshaling for complex types // If that fails, try JSON marshaling/unmarshaling for complex types
data, err := json.Marshal(value) // Use pooled buffer for encoding
if err != nil { pm := pool.Get()
buf := pm.GetBuffer(256)
defer pm.PutBuffer(buf)
encoder := pm.GetJSONEncoder(buf)
defer pm.PutJSONEncoder(encoder)
if err := encoder.Encode(value); err != nil {
return zero, false return zero, false
} }
// Decode using pooled decoder
var result T var result T
if err := json.Unmarshal(data, &result); err != nil { decoder := pm.GetJSONDecoder(bytes.NewReader(buf.Bytes()))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&result); err != nil {
return zero, false return zero, false
} }
internal/optimization/memory_pool.go
-100
View File
@@ -1,100 +0,0 @@
// Package optimization provides memory and performance optimizations
package optimization
import (
"sync"
)
// StringBuilderPool provides a pool of reusable string builders
type StringBuilderPool struct {
pool sync.Pool
}
// NewStringBuilderPool creates a new string builder pool
func NewStringBuilderPool() *StringBuilderPool {
return &StringBuilderPool{
pool: sync.Pool{
New: func() any {
return make([]byte, 0, 256) // Pre-allocate 256 bytes
},
},
}
}
// Get retrieves a string builder from the pool
func (p *StringBuilderPool) Get() []byte {
buf := p.pool.Get().([]byte)
// Return the slice with reset length but preserved capacity
return buf[:0]
}
// Put returns a string builder to the pool
func (p *StringBuilderPool) Put(buf []byte) {
if buf != nil && cap(buf) < 4096 { // Don't pool overly large buffers or nil buffers
// Intentional design: slice API is more ergonomic than *[]byte.
// The 24-byte boxing allocation is negligible compared to 256+ byte buffer allocation we avoid.
//lint:ignore SA6002 slice interface is preferred over pointer interface for ergonomics
p.pool.Put(buf[:0]) // Reset length and store the slice directly
}
}
// ByteSlicePool provides a pool of reusable byte slices
type ByteSlicePool struct {
pool sync.Pool
size int
}
// NewByteSlicePool creates a new byte slice pool with specified size
func NewByteSlicePool(size int) *ByteSlicePool {
return &ByteSlicePool{
size: size,
pool: sync.Pool{
New: func() any {
return make([]byte, size)
},
},
}
}
// Get retrieves a byte slice from the pool
func (p *ByteSlicePool) Get() []byte {
buf := p.pool.Get().([]byte)
// Ensure we're returning the full-sized slice
return buf[:p.size]
}
// Put returns a byte slice to the pool
func (p *ByteSlicePool) Put(buf []byte) {
if buf != nil && len(buf) == p.size {
// Intentional design: slice API is more ergonomic than *[]byte.
// The 24-byte boxing allocation is negligible compared to the sized buffer allocation we avoid.
//lint:ignore SA6002 slice interface is preferred over pointer interface for ergonomics
p.pool.Put(buf)
}
}
// Global pools for common use cases
var (
globalStringBuilderPool = NewStringBuilderPool()
globalByteSlicePool = NewByteSlicePool(2048)
)
// GetStringBuilder gets a string builder from the global pool
func GetStringBuilder() []byte {
return globalStringBuilderPool.Get()
}
// PutStringBuilder returns a string builder to the global pool
func PutStringBuilder(buf []byte) {
globalStringBuilderPool.Put(buf)
}
// GetByteSlice gets a byte slice from the global pool
func GetByteSlice() []byte {
return globalByteSlicePool.Get()
}
// PutByteSlice returns a byte slice to the global pool
func PutByteSlice(buf []byte) {
globalByteSlicePool.Put(buf)
}
internal/optimization/memory_pool_test.go
-532
View File
@@ -1,532 +0,0 @@
package optimization
import (
"runtime"
"sync"
"testing"
)
// TestNewStringBuilderPool tests StringBuilderPool creation
func TestNewStringBuilderPool(t *testing.T) {
pool := NewStringBuilderPool()
if pool == nil {
t.Error("Expected NewStringBuilderPool to return non-nil pool")
}
}
// TestStringBuilderPool tests StringBuilderPool functionality
func TestStringBuilderPool(t *testing.T) {
pool := NewStringBuilderPool()
// Test Get
buf1 := pool.Get()
if buf1 == nil {
t.Error("Expected Get to return non-nil buffer")
}
// Check initial capacity
if cap(buf1) != 256 {
t.Errorf("Expected initial capacity of 256, got %d", cap(buf1))
}
// Check initial length
if len(buf1) != 0 {
t.Errorf("Expected initial length of 0, got %d", len(buf1))
}
// Test Put and reuse
// Simulate some usage
buf1 = append(buf1, []byte("hello world")...)
pool.Put(buf1)
// Get another buffer - should reuse the one we just put
buf2 := pool.Get()
if buf2 == nil {
t.Error("Expected Get to return non-nil buffer after Put")
}
// Buffer should be reset to length 0 but keep capacity
if len(buf2) != 0 {
t.Errorf("Expected buffer length to be reset to 0, got %d", len(buf2))
}
// Should reuse the same underlying array
if cap(buf2) < 256 {
t.Errorf("Expected buffer capacity to be at least 256, got %d", cap(buf2))
}
}
// TestStringBuilderPoolLargeBuffer tests that large buffers are not pooled
func TestStringBuilderPoolLargeBuffer(t *testing.T) {
pool := NewStringBuilderPool()
// Get a buffer and make it very large
buf := pool.Get()
largeBuf := make([]byte, 5000) // Larger than 4096 threshold
buf = append(buf, largeBuf...)
// Put it back - should not be pooled due to size
pool.Put(buf)
// Get a new buffer - should be a fresh one with default capacity
newBuf := pool.Get()
if cap(newBuf) != 256 {
t.Errorf("Expected fresh buffer with capacity 256, got %d", cap(newBuf))
}
}
// TestNewByteSlicePool tests ByteSlicePool creation
func TestNewByteSlicePool(t *testing.T) {
size := 1024
pool := NewByteSlicePool(size)
if pool == nil {
t.Error("Expected NewByteSlicePool to return non-nil pool")
return
}
if pool.size != size {
t.Errorf("Expected pool size %d, got %d", size, pool.size)
}
}
// TestByteSlicePool tests ByteSlicePool functionality
func TestByteSlicePool(t *testing.T) {
size := 512
pool := NewByteSlicePool(size)
// Test Get
buf1 := pool.Get()
if buf1 == nil {
t.Error("Expected Get to return non-nil buffer")
}
// Check size
if len(buf1) != size {
t.Errorf("Expected buffer length %d, got %d", size, len(buf1))
}
// Test Put and reuse
pool.Put(buf1)
// Get another buffer - should reuse the one we just put
buf2 := pool.Get()
if buf2 == nil {
t.Error("Expected Get to return non-nil buffer after Put")
}
// Should be the same size
if len(buf2) != size {
t.Errorf("Expected buffer length %d, got %d", size, len(buf2))
}
}
// TestByteSlicePoolWrongSize tests that wrong-sized buffers are not pooled
func TestByteSlicePoolWrongSize(t *testing.T) {
size := 512
pool := NewByteSlicePool(size)
// Get a buffer
buf := pool.Get()
// Resize it (simulate usage that changes size)
buf = buf[:100] // Change length to 100
// Put it back - should not be pooled due to wrong size
pool.Put(buf)
// Get a new buffer - should be a fresh one with correct size
newBuf := pool.Get()
if len(newBuf) != size {
t.Errorf("Expected fresh buffer with length %d, got %d", size, len(newBuf))
}
}
// TestGlobalStringBuilderPool tests global string builder pool functions
func TestGlobalStringBuilderPool(t *testing.T) {
// Test GetStringBuilder
buf1 := GetStringBuilder()
if buf1 == nil {
t.Error("Expected GetStringBuilder to return non-nil buffer")
}
if cap(buf1) != 256 {
t.Errorf("Expected initial capacity of 256, got %d", cap(buf1))
}
// Test PutStringBuilder and reuse
buf1 = append(buf1, []byte("test data")...)
PutStringBuilder(buf1)
buf2 := GetStringBuilder()
if len(buf2) != 0 {
t.Errorf("Expected buffer length to be reset to 0, got %d", len(buf2))
}
}
// TestGlobalByteSlicePool tests global byte slice pool functions
func TestGlobalByteSlicePool(t *testing.T) {
// Test GetByteSlice
buf1 := GetByteSlice()
if buf1 == nil {
t.Error("Expected GetByteSlice to return non-nil buffer")
}
if len(buf1) != 2048 {
t.Errorf("Expected buffer length of 2048, got %d", len(buf1))
}
// Test PutByteSlice and reuse
PutByteSlice(buf1)
buf2 := GetByteSlice()
if len(buf2) != 2048 {
t.Errorf("Expected buffer length of 2048, got %d", len(buf2))
}
}
// TestPoolsConcurrency tests that pools work correctly under concurrent access
func TestPoolsConcurrency(t *testing.T) {
const numGoroutines = 100
const numOperations = 1000
// Test StringBuilderPool concurrency
t.Run("StringBuilderPool concurrency", func(t *testing.T) {
pool := NewStringBuilderPool()
var wg sync.WaitGroup
for i := 0; i < numGoroutines; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < numOperations; j++ {
buf := pool.Get()
buf = append(buf, []byte("test")...)
pool.Put(buf)
}
}()
}
wg.Wait()
})
// Test ByteSlicePool concurrency
t.Run("ByteSlicePool concurrency", func(t *testing.T) {
pool := NewByteSlicePool(1024)
var wg sync.WaitGroup
for i := 0; i < numGoroutines; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < numOperations; j++ {
buf := pool.Get()
// Simulate some work
for k := 0; k < len(buf); k++ {
buf[k] = byte(k % 256)
}
pool.Put(buf)
}
}()
}
wg.Wait()
})
// Test global pools concurrency
t.Run("Global pools concurrency", func(t *testing.T) {
var wg sync.WaitGroup
for i := 0; i < numGoroutines; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < numOperations; j++ {
// Test string builder pool
strBuf := GetStringBuilder()
strBuf = append(strBuf, []byte("test")...)
PutStringBuilder(strBuf)
// Test byte slice pool
byteBuf := GetByteSlice()
byteBuf[0] = 42
PutByteSlice(byteBuf)
}
}()
}
wg.Wait()
})
}
// TestMemoryReuse tests that pools actually reuse memory
func TestMemoryReuse(t *testing.T) {
// Skip this test when running with race detector as sync.Pool behavior
// is non-deterministic and may not reuse items immediately
if testing.Short() {
t.Skip("Skipping memory reuse test in short mode")
}
t.Run("StringBuilderPool memory reuse", func(t *testing.T) {
pool := NewStringBuilderPool()
// Get a buffer and note its address
buf1 := pool.Get()
if len(buf1) == 0 {
buf1 = append(buf1, 0) // Ensure slice has at least one element
}
addr1 := &buf1[0] // Get address of first element
// Put it back
pool.Put(buf1)
// Force GC to not clear the pool
runtime.GC()
// Get another buffer - should ideally be the same memory
// Note: sync.Pool doesn't guarantee this, especially with race detector
buf2 := pool.Get()
if len(buf2) == 0 {
buf2 = append(buf2, 0)
}
addr2 := &buf2[0]
// This is a best-effort check - sync.Pool may or may not reuse
if addr1 != addr2 {
t.Skip("Pool did not reuse memory (expected behavior with race detector)")
}
})
t.Run("ByteSlicePool memory reuse", func(t *testing.T) {
pool := NewByteSlicePool(1024)
// Get a buffer and note its address
buf1 := pool.Get()
addr1 := &buf1[0]
// Put it back
pool.Put(buf1)
// Force GC to not clear the pool
runtime.GC()
// Get another buffer - should ideally be the same memory
// Note: sync.Pool doesn't guarantee this, especially with race detector
buf2 := pool.Get()
addr2 := &buf2[0]
// This is a best-effort check - sync.Pool may or may not reuse
if addr1 != addr2 {
t.Skip("Pool did not reuse memory (expected behavior with race detector)")
}
})
}
// TestPoolEfficiency tests that pools reduce allocations
func TestPoolEfficiency(t *testing.T) {
if testing.Short() {
t.Skip("Skipping efficiency test in short mode")
}
t.Run("StringBuilderPool allocation reduction", func(t *testing.T) {
pool := NewStringBuilderPool()
// Test with pool
var m1, m2 runtime.MemStats
runtime.GC()
runtime.ReadMemStats(&m1)
for i := 0; i < 10000; i++ {
buf := pool.Get()
buf = append(buf, []byte("test data for benchmarking")...)
pool.Put(buf)
}
runtime.GC()
runtime.ReadMemStats(&m2)
poolAllocs := m2.Mallocs - m1.Mallocs
// Test without pool (creating new buffers each time)
runtime.GC()
runtime.ReadMemStats(&m1)
for i := 0; i < 10000; i++ {
buf := make([]byte, 0, 256)
_ = append(buf, []byte("test data for benchmarking")...)
}
runtime.GC()
runtime.ReadMemStats(&m2)
directAllocs := m2.Mallocs - m1.Mallocs
t.Logf("Pool allocations: %d, Direct allocations: %d", poolAllocs, directAllocs)
// Pool should result in fewer allocations (though this test can be flaky)
if poolAllocs >= directAllocs {
t.Logf("Warning: Pool did not reduce allocations as expected (pool: %d, direct: %d)", poolAllocs, directAllocs)
}
})
}
// TestEdgeCases tests edge cases and error conditions
func TestEdgeCases(t *testing.T) {
t.Run("StringBuilderPool with nil buffer", func(t *testing.T) {
pool := NewStringBuilderPool()
// This should not panic
pool.Put(nil)
// Should still be able to get a buffer
buf := pool.Get()
if buf == nil {
t.Error("Expected Get to return non-nil buffer after putting nil")
}
})
t.Run("ByteSlicePool with nil buffer", func(t *testing.T) {
pool := NewByteSlicePool(1024)
// This should not panic
pool.Put(nil)
// Should still be able to get a buffer
buf := pool.Get()
if buf == nil {
t.Error("Expected Get to return non-nil buffer after putting nil")
}
})
t.Run("Zero-size ByteSlicePool", func(t *testing.T) {
pool := NewByteSlicePool(0)
buf := pool.Get()
if len(buf) != 0 {
t.Errorf("Expected zero-length buffer, got %d", len(buf))
}
// Should not panic
pool.Put(buf)
})
t.Run("Global pools with nil", func(t *testing.T) {
// These should not panic
PutStringBuilder(nil)
PutByteSlice(nil)
// Should still work
buf1 := GetStringBuilder()
buf2 := GetByteSlice()
if buf1 == nil || buf2 == nil {
t.Error("Expected global pools to work after putting nil")
}
})
}
// TestTypeAssertions tests that type assertions in pools work correctly
func TestTypeAssertions(t *testing.T) {
t.Run("StringBuilderPool type assertion", func(t *testing.T) {
pool := NewStringBuilderPool()
// This should not panic
buf := pool.Get()
if buf == nil {
t.Error("Expected non-nil buffer")
}
// Buffer should be usable
buf = append(buf, []byte("test")...)
if string(buf) != "test" {
t.Errorf("Expected 'test', got '%s'", string(buf))
}
})
t.Run("ByteSlicePool type assertion", func(t *testing.T) {
pool := NewByteSlicePool(10)
// This should not panic
buf := pool.Get()
if buf == nil {
t.Error("Expected non-nil buffer")
}
// Buffer should be usable
buf[0] = 42
if buf[0] != 42 {
t.Errorf("Expected 42, got %d", buf[0])
}
})
}
// BenchmarkStringBuilderPool benchmarks the string builder pool
func BenchmarkStringBuilderPool(b *testing.B) {
pool := NewStringBuilderPool()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := pool.Get()
buf = append(buf, []byte("benchmark test data")...)
pool.Put(buf)
}
})
}
// BenchmarkByteSlicePool benchmarks the byte slice pool
func BenchmarkByteSlicePool(b *testing.B) {
pool := NewByteSlicePool(1024)
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := pool.Get()
buf[0] = 42
pool.Put(buf)
}
})
}
// BenchmarkGlobalPools benchmarks the global pools
func BenchmarkGlobalPools(b *testing.B) {
b.Run("StringBuilder", func(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := GetStringBuilder()
buf = append(buf, []byte("test")...)
PutStringBuilder(buf)
}
})
})
b.Run("ByteSlice", func(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := GetByteSlice()
buf[0] = 42
PutByteSlice(buf)
}
})
})
}
// BenchmarkDirectAllocation benchmarks direct allocation for comparison
func BenchmarkDirectAllocation(b *testing.B) {
b.Run("StringBuilder", func(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := make([]byte, 0, 256)
_ = append(buf, []byte("test")...)
}
})
})
b.Run("ByteSlice", func(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
buf := make([]byte, 2048)
buf[0] = 42
}
})
})
}
internal/pool/pool.go
+68
View File
@@ -6,6 +6,8 @@ package pool
import ( import (
"bytes" "bytes"
"compress/gzip" "compress/gzip"
"encoding/json"
"io"
"strings" "strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
@@ -54,6 +56,10 @@ type PoolStats struct {
JWTPuts uint64 JWTPuts uint64
HTTPGets uint64 HTTPGets uint64
HTTPPuts uint64 HTTPPuts uint64
JSONEncoderGets uint64
JSONEncoderPuts uint64
JSONDecoderGets uint64
JSONDecoderPuts uint64
OversizedRejects uint64 OversizedRejects uint64
} }
@@ -378,6 +384,40 @@ func (m *Manager) PutByteSlice(b []byte) {
} }
} }
// GetJSONEncoder returns a JSON encoder from the pool configured for the given writer
func (m *Manager) GetJSONEncoder(w io.Writer) *json.Encoder {
atomic.AddUint64(&m.stats.JSONEncoderGets, 1)
// Since json.Encoder doesn't support resetting, we create new ones each time
encoder := json.NewEncoder(w)
encoder.SetEscapeHTML(false) // Disable HTML escaping for performance
return encoder
}
// PutJSONEncoder returns a JSON encoder to the pool
func (m *Manager) PutJSONEncoder(encoder *json.Encoder) {
if encoder == nil {
return
}
atomic.AddUint64(&m.stats.JSONEncoderPuts, 1)
// JSON encoders can't be reset, so we don't pool them
}
// GetJSONDecoder returns a JSON decoder from the pool configured for the given reader
func (m *Manager) GetJSONDecoder(r io.Reader) *json.Decoder {
atomic.AddUint64(&m.stats.JSONDecoderGets, 1)
// Since json.Decoder doesn't support resetting, we create new ones each time
return json.NewDecoder(r)
}
// PutJSONDecoder returns a JSON decoder to the pool
func (m *Manager) PutJSONDecoder(decoder *json.Decoder) {
if decoder == nil {
return
}
atomic.AddUint64(&m.stats.JSONDecoderPuts, 1)
// JSON decoders can't be reset, so we don't pool them
}
// GetStats returns current pool statistics // GetStats returns current pool statistics
func (m *Manager) GetStats() PoolStats { func (m *Manager) GetStats() PoolStats {
return PoolStats{ return PoolStats{
@@ -391,6 +431,10 @@ func (m *Manager) GetStats() PoolStats {
JWTPuts: atomic.LoadUint64(&m.stats.JWTPuts), JWTPuts: atomic.LoadUint64(&m.stats.JWTPuts),
HTTPGets: atomic.LoadUint64(&m.stats.HTTPGets), HTTPGets: atomic.LoadUint64(&m.stats.HTTPGets),
HTTPPuts: atomic.LoadUint64(&m.stats.HTTPPuts), HTTPPuts: atomic.LoadUint64(&m.stats.HTTPPuts),
JSONEncoderGets: atomic.LoadUint64(&m.stats.JSONEncoderGets),
JSONEncoderPuts: atomic.LoadUint64(&m.stats.JSONEncoderPuts),
JSONDecoderGets: atomic.LoadUint64(&m.stats.JSONDecoderGets),
JSONDecoderPuts: atomic.LoadUint64(&m.stats.JSONDecoderPuts),
OversizedRejects: atomic.LoadUint64(&m.stats.OversizedRejects), OversizedRejects: atomic.LoadUint64(&m.stats.OversizedRejects),
} }
} }
@@ -407,6 +451,10 @@ func (m *Manager) ResetStats() {
atomic.StoreUint64(&m.stats.JWTPuts, 0) atomic.StoreUint64(&m.stats.JWTPuts, 0)
atomic.StoreUint64(&m.stats.HTTPGets, 0) atomic.StoreUint64(&m.stats.HTTPGets, 0)
atomic.StoreUint64(&m.stats.HTTPPuts, 0) atomic.StoreUint64(&m.stats.HTTPPuts, 0)
atomic.StoreUint64(&m.stats.JSONEncoderGets, 0)
atomic.StoreUint64(&m.stats.JSONEncoderPuts, 0)
atomic.StoreUint64(&m.stats.JSONDecoderGets, 0)
atomic.StoreUint64(&m.stats.JSONDecoderPuts, 0)
atomic.StoreUint64(&m.stats.OversizedRejects, 0) atomic.StoreUint64(&m.stats.OversizedRejects, 0)
} }
@@ -471,3 +519,23 @@ func ByteSlice(size int) []byte {
func ReturnByteSlice(b []byte) { func ReturnByteSlice(b []byte) {
Get().PutByteSlice(b) Get().PutByteSlice(b)
} }
// JSONEncoder returns a JSON encoder from the global pool
func JSONEncoder(w io.Writer) *json.Encoder {
return Get().GetJSONEncoder(w)
}
// ReturnJSONEncoder returns a JSON encoder to the global pool
func ReturnJSONEncoder(encoder *json.Encoder) {
Get().PutJSONEncoder(encoder)
}
// JSONDecoder returns a JSON decoder from the global pool
func JSONDecoder(r io.Reader) *json.Decoder {
return Get().GetJSONDecoder(r)
}
// ReturnJSONDecoder returns a JSON decoder to the global pool
func ReturnJSONDecoder(decoder *json.Decoder) {
Get().PutJSONDecoder(decoder)
}
internal/pool/utils.go
+70
View File
@@ -0,0 +1,70 @@
// Package pool provides centralized memory pool management utilities
package pool
import (
"strings"
)
// BuildSessionName efficiently builds session names using pooled string builders
func BuildSessionName(baseName string, index int) string {
sb := StringBuilder()
defer ReturnStringBuilder(sb)
sb.WriteString(baseName)
sb.WriteRune('_')
// Efficient int to string conversion
if index < 10 {
sb.WriteRune('0' + rune(index))
} else {
sb.WriteString(intToString(index))
}
return sb.String()
}
// BuildCacheKey efficiently builds cache keys using pooled string builders
func BuildCacheKey(parts ...string) string {
sb := StringBuilder()
defer ReturnStringBuilder(sb)
for i, part := range parts {
if i > 0 {
sb.WriteRune(':')
}
sb.WriteString(part)
}
return sb.String()
}
// FormatString efficiently formats a string using a pooled string builder
func FormatString(format func(*strings.Builder)) string {
sb := StringBuilder()
defer ReturnStringBuilder(sb)
format(sb)
return sb.String()
}
// intToString converts int to string without allocation (for small numbers)
func intToString(n int) string {
if n < 0 {
return "-" + intToString(-n)
}
if n < 10 {
return string(rune('0' + n))
}
if n < 100 {
return string(rune('0'+n/10)) + string(rune('0'+n%10))
}
// Fall back to standard conversion for larger numbers
buf := make([]byte, 0, 20)
for n > 0 {
buf = append(buf, byte('0'+n%10))
n /= 10
}
// Reverse the buffer
for i, j := 0, len(buf)-1; i < j; i, j = i+1, j-1 {
buf[i], buf[j] = buf[j], buf[i]
}
return string(buf)
}
jwt.go
+33 -22
View File
@@ -1,19 +1,21 @@
package traefikoidc package traefikoidc
import ( import (
"bytes"
"context" "context"
"crypto" "crypto"
"crypto/ecdsa" "crypto/ecdsa"
"crypto/rsa" "crypto/rsa"
"crypto/x509" "crypto/x509"
"encoding/base64" "encoding/base64"
"encoding/json"
"encoding/pem" "encoding/pem"
"fmt" "fmt"
"math/big" "math/big"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/lukaszraczylo/traefikoidc/internal/pool"
) )
// Replay attack protection cache and synchronization primitives. // Replay attack protection cache and synchronization primitives.
@@ -173,28 +175,30 @@ func parseJWT(tokenString string) (*JWT, error) {
return nil, fmt.Errorf("invalid JWT format: expected 3 parts, got %d", len(parts)) return nil, fmt.Errorf("invalid JWT format: expected 3 parts, got %d", len(parts))
} }
pools := GetGlobalMemoryPools() pm := pool.Get()
jwtBuf := pools.GetJWTParsingBuffer() jwtBuf := pm.GetJWTBuffer()
defer pools.PutJWTParsingBuffer(jwtBuf) defer pm.PutJWTBuffer(jwtBuf)
jwt := &JWT{ jwt := &JWT{
Token: tokenString, Token: tokenString,
} }
headerLen := base64.RawURLEncoding.DecodedLen(len(parts[0])) headerLen := base64.RawURLEncoding.DecodedLen(len(parts[0]))
if headerLen > cap(jwtBuf.HeaderBuf) { if headerLen > cap(jwtBuf.Header) {
jwtBuf.HeaderBuf = make([]byte, headerLen) jwtBuf.Header = make([]byte, headerLen)
} else { } else {
jwtBuf.HeaderBuf = jwtBuf.HeaderBuf[:headerLen] jwtBuf.Header = jwtBuf.Header[:headerLen]
} }
n, err := base64.RawURLEncoding.Decode(jwtBuf.HeaderBuf, []byte(parts[0])) n, err := base64.RawURLEncoding.Decode(jwtBuf.Header, []byte(parts[0]))
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid JWT format: failed to decode header: %v", err) return nil, fmt.Errorf("invalid JWT format: failed to decode header: %v", err)
} }
headerBytes := jwtBuf.HeaderBuf[:n] headerBytes := jwtBuf.Header[:n]
if err := json.Unmarshal(headerBytes, &jwt.Header); err != nil { decoder := pm.GetJSONDecoder(bytes.NewReader(headerBytes))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&jwt.Header); err != nil {
return nil, fmt.Errorf("invalid JWT format: failed to unmarshal header: %v", err) return nil, fmt.Errorf("invalid JWT format: failed to unmarshal header: %v", err)
} }
@@ -203,19 +207,21 @@ func parseJWT(tokenString string) (*JWT, error) {
} }
claimsLen := base64.RawURLEncoding.DecodedLen(len(parts[1])) claimsLen := base64.RawURLEncoding.DecodedLen(len(parts[1]))
if claimsLen > cap(jwtBuf.PayloadBuf) { if claimsLen > cap(jwtBuf.Payload) {
jwtBuf.PayloadBuf = make([]byte, claimsLen) jwtBuf.Payload = make([]byte, claimsLen)
} else { } else {
jwtBuf.PayloadBuf = jwtBuf.PayloadBuf[:claimsLen] jwtBuf.Payload = jwtBuf.Payload[:claimsLen]
} }
n, err = base64.RawURLEncoding.Decode(jwtBuf.PayloadBuf, []byte(parts[1])) n, err = base64.RawURLEncoding.Decode(jwtBuf.Payload, []byte(parts[1]))
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid JWT format: failed to decode claims: %v", err) return nil, fmt.Errorf("invalid JWT format: failed to decode claims: %v", err)
} }
claimsBytes := jwtBuf.PayloadBuf[:n] claimsBytes := jwtBuf.Payload[:n]
if err := json.Unmarshal(claimsBytes, &jwt.Claims); err != nil { decoder2 := pm.GetJSONDecoder(bytes.NewReader(claimsBytes))
defer pm.PutJSONDecoder(decoder2)
if err := decoder2.Decode(&jwt.Claims); err != nil {
return nil, fmt.Errorf("invalid JWT format: failed to unmarshal claims: %v", err) return nil, fmt.Errorf("invalid JWT format: failed to unmarshal claims: %v", err)
} }
@@ -224,19 +230,24 @@ func parseJWT(tokenString string) (*JWT, error) {
} }
sigLen := base64.RawURLEncoding.DecodedLen(len(parts[2])) sigLen := base64.RawURLEncoding.DecodedLen(len(parts[2]))
if sigLen > cap(jwtBuf.SignatureBuf) { if sigLen > cap(jwtBuf.Signature) {
jwtBuf.SignatureBuf = make([]byte, sigLen) jwtBuf.Signature = make([]byte, sigLen)
} else { } else {
jwtBuf.SignatureBuf = jwtBuf.SignatureBuf[:sigLen] jwtBuf.Signature = jwtBuf.Signature[:sigLen]
} }
n, err = base64.RawURLEncoding.Decode(jwtBuf.SignatureBuf, []byte(parts[2])) n, err = base64.RawURLEncoding.Decode(jwtBuf.Signature, []byte(parts[2]))
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid JWT format: failed to decode signature: %v", err) return nil, fmt.Errorf("invalid JWT format: failed to decode signature: %v", err)
} }
jwt.Signature = make([]byte, n) // Reuse the signature buffer if it's large enough, otherwise allocate
copy(jwt.Signature, jwtBuf.SignatureBuf[:n]) if cap(jwtBuf.Signature) >= n {
jwt.Signature = jwtBuf.Signature[:n:n] // Use slice trick to prevent aliasing
} else {
jwt.Signature = make([]byte, n)
copy(jwt.Signature, jwtBuf.Signature[:n])
}
return jwt, nil return jwt, nil
} }
memory_pools.go
-264
View File
@@ -1,264 +0,0 @@
package traefikoidc
import (
"bytes"
"strings"
"sync"
)
// MemoryPoolManager provides centralized management of object pools for memory efficiency.
// It maintains pools for frequently allocated objects like buffers for compression, JWT parsing,
// HTTP responses, and string building operations to reduce garbage collection pressure.
type MemoryPoolManager struct {
// compressionBufferPool pools buffers for compression/decompression operations
compressionBufferPool *sync.Pool
// jwtParsingPool pools specialized buffers for JWT token parsing
jwtParsingPool *sync.Pool
// httpResponsePool pools buffers for HTTP response handling
httpResponsePool *sync.Pool
// stringBuilderPool pools string.Builder instances for string operations
stringBuilderPool *sync.Pool
}
// JWTParsingBuffer provides pre-allocated buffers for JWT token parsing.
// Using pooled buffers for the three JWT components (header, payload, signature)
// avoids repeated allocations during token validation, which can significantly
// improve performance under high load.
type JWTParsingBuffer struct {
// HeaderBuf stores the decoded JWT header
HeaderBuf []byte
// PayloadBuf stores the decoded JWT payload/claims
PayloadBuf []byte
// SignatureBuf stores the decoded JWT signature
SignatureBuf []byte
}
// NewMemoryPoolManager creates a new memory pool manager with optimized pool configurations.
// Each pool is initialized with appropriate buffer sizes to balance memory usage with performance benefits.
func NewMemoryPoolManager() *MemoryPoolManager {
return &MemoryPoolManager{
compressionBufferPool: &sync.Pool{
New: func() interface{} {
return bytes.NewBuffer(make([]byte, 0, 4096))
},
},
jwtParsingPool: &sync.Pool{
New: func() interface{} {
return &JWTParsingBuffer{
HeaderBuf: make([]byte, 0, 512),
PayloadBuf: make([]byte, 0, 2048),
SignatureBuf: make([]byte, 0, 512),
}
},
},
httpResponsePool: &sync.Pool{
New: func() interface{} {
buf := make([]byte, 0, 8192)
return &buf
},
},
stringBuilderPool: &sync.Pool{
New: func() interface{} {
var sb strings.Builder
sb.Grow(1024)
return &sb
},
},
}
}
// GetCompressionBuffer retrieves a buffer from the compression pool.
// The buffer should be returned to the pool using PutCompressionBuffer when done.
func (m *MemoryPoolManager) GetCompressionBuffer() *bytes.Buffer {
return m.compressionBufferPool.Get().(*bytes.Buffer)
}
// PutCompressionBuffer returns a compression buffer to the pool.
// The buffer is reset before being returned to prevent data leaks.
// Oversized buffers are discarded to prevent memory bloat.
func (m *MemoryPoolManager) PutCompressionBuffer(buf *bytes.Buffer) {
if buf == nil {
return
}
if buf.Cap() <= 16384 {
buf.Reset()
m.compressionBufferPool.Put(buf)
}
}
// GetJWTParsingBuffer retrieves specialized buffers for JWT parsing.
// Returns a structure with pre-allocated buffers for header, payload, and signature.
func (m *MemoryPoolManager) GetJWTParsingBuffer() *JWTParsingBuffer {
return m.jwtParsingPool.Get().(*JWTParsingBuffer)
}
// PutJWTParsingBuffer returns JWT parsing buffers to the pool.
// All buffer slices are reset to zero length and oversized buffers are discarded.
func (m *MemoryPoolManager) PutJWTParsingBuffer(buf *JWTParsingBuffer) {
if buf == nil {
return
}
if cap(buf.HeaderBuf) <= 2048 && cap(buf.PayloadBuf) <= 8192 && cap(buf.SignatureBuf) <= 2048 {
buf.HeaderBuf = buf.HeaderBuf[:0]
buf.PayloadBuf = buf.PayloadBuf[:0]
buf.SignatureBuf = buf.SignatureBuf[:0]
m.jwtParsingPool.Put(buf)
}
}
// GetHTTPResponseBuffer retrieves a buffer for HTTP response handling.
// Returns a pre-allocated byte slice suitable for HTTP operations.
func (m *MemoryPoolManager) GetHTTPResponseBuffer() []byte {
return *m.httpResponsePool.Get().(*[]byte)
}
// PutHTTPResponseBuffer returns an HTTP response buffer to the pool.
// The buffer slice is reset to zero length and oversized buffers are discarded.
func (m *MemoryPoolManager) PutHTTPResponseBuffer(buf []byte) {
if buf == nil {
return
}
if cap(buf) <= 32768 {
buf = buf[:0]
m.httpResponsePool.Put(&buf)
}
}
// GetStringBuilder retrieves a pre-allocated string builder from the pool.
// The string builder is ready for use with an initial capacity allocation.
func (m *MemoryPoolManager) GetStringBuilder() *strings.Builder {
return m.stringBuilderPool.Get().(*strings.Builder)
}
// PutStringBuilder returns a string builder to the pool.
// The builder is reset and oversized builders are discarded to prevent memory bloat.
func (m *MemoryPoolManager) PutStringBuilder(sb *strings.Builder) {
if sb == nil {
return
}
if sb.Cap() <= 16384 {
sb.Reset()
m.stringBuilderPool.Put(sb)
}
}
// TokenCompressionPool manages specialized memory pools for token compression operations.
// Provides separate pools optimized for compression, decompression, and string building
// to handle the specific memory patterns of token processing workflows.
type TokenCompressionPool struct {
// compressionBuffers pools buffers specifically sized for token compression
compressionBuffers sync.Pool
// decompressionBuffers pools buffers for token decompression with larger capacity
decompressionBuffers sync.Pool
// stringBuilders pools string builders optimized for token operations
stringBuilders sync.Pool
}
// NewTokenCompressionPool creates a specialized memory pool for token operations.
// Initializes pools with buffer sizes optimized for token compression workflows.
func NewTokenCompressionPool() *TokenCompressionPool {
return &TokenCompressionPool{
compressionBuffers: sync.Pool{
New: func() interface{} {
return bytes.NewBuffer(make([]byte, 0, 4096))
},
},
decompressionBuffers: sync.Pool{
New: func() interface{} {
return bytes.NewBuffer(make([]byte, 0, 8192))
},
},
stringBuilders: sync.Pool{
New: func() interface{} {
var sb strings.Builder
sb.Grow(2048)
return &sb
},
},
}
}
// GetCompressionBuffer retrieves a buffer optimized for token compression.
// Returns a buffer with appropriate capacity for typical token sizes.
func (p *TokenCompressionPool) GetCompressionBuffer() *bytes.Buffer {
return p.compressionBuffers.Get().(*bytes.Buffer)
}
// PutCompressionBuffer returns a compression buffer to the pool.
// Resets the buffer and discards oversized buffers to prevent memory bloat.
func (p *TokenCompressionPool) PutCompressionBuffer(buf *bytes.Buffer) {
if buf != nil && buf.Cap() <= 16384 {
buf.Reset()
p.compressionBuffers.Put(buf)
}
}
// GetDecompressionBuffer retrieves a buffer optimized for token decompression.
// Returns a larger buffer suitable for expanded token data.
func (p *TokenCompressionPool) GetDecompressionBuffer() *bytes.Buffer {
return p.decompressionBuffers.Get().(*bytes.Buffer)
}
// PutDecompressionBuffer returns a decompression buffer to the pool.
// Resets the buffer and discards oversized buffers to prevent memory bloat.
func (p *TokenCompressionPool) PutDecompressionBuffer(buf *bytes.Buffer) {
if buf != nil && buf.Cap() <= 32768 {
buf.Reset()
p.decompressionBuffers.Put(buf)
}
}
// GetStringBuilder retrieves a string builder optimized for token operations.
// Returns a pre-allocated builder with capacity suitable for token processing.
func (p *TokenCompressionPool) GetStringBuilder() *strings.Builder {
return p.stringBuilders.Get().(*strings.Builder)
}
// PutStringBuilder returns a string builder to the pool.
// Resets the builder and discards oversized builders to prevent memory bloat.
func (p *TokenCompressionPool) PutStringBuilder(sb *strings.Builder) {
if sb != nil && sb.Cap() <= 16384 {
sb.Reset()
p.stringBuilders.Put(sb)
}
}
// Global memory pool manager instance and synchronization primitives.
// Provides singleton access to memory pools across the entire application.
var (
// globalMemoryPools is the singleton memory pool manager instance
globalMemoryPools *MemoryPoolManager
// memoryPoolOnce ensures single initialization of the global pools
memoryPoolOnce sync.Once
// memoryPoolMutex protects global pool operations
memoryPoolMutex sync.RWMutex
)
// GetGlobalMemoryPools returns the singleton memory pool manager instance.
// Uses sync.Once to ensure thread-safe initialization of the global pools.
func GetGlobalMemoryPools() *MemoryPoolManager {
memoryPoolOnce.Do(func() {
globalMemoryPools = NewMemoryPoolManager()
})
return globalMemoryPools
}
// CleanupGlobalMemoryPools cleans up the global memory pool manager.
// Resets the singleton instance and sync.Once for potential re-initialization.
// It's safe to call multiple times.
func CleanupGlobalMemoryPools() {
memoryPoolMutex.Lock()
defer memoryPoolMutex.Unlock()
if globalMemoryPools != nil {
globalMemoryPools = nil
memoryPoolOnce = sync.Once{}
}
}
profiling.go
+4 -152
View File
@@ -89,7 +89,7 @@ type LeakDetectionConfig struct {
CacheMemoryThreshold uint64 CacheMemoryThreshold uint64
// HTTPClientThreshold sets limit for HTTP client connections // HTTPClientThreshold sets limit for HTTP client connections
HTTPClientThreshold int HTTPClientThreshold int
// TokenCompressionThreshold sets limit for token compression memory // Deprecated: TokenCompressionThreshold is no longer used
TokenCompressionThreshold uint64 TokenCompressionThreshold uint64
} }
@@ -658,157 +658,9 @@ func (hcp *HTTPClientProfiler) AnalyzeLeaks(baseline, current *MemorySnapshot) *
return analysis return analysis
} }
// TokenCompressionProfiler monitors token compression memory usage // Deprecated profilers removed - use internal/pool statistics instead
type TokenCompressionProfiler struct { // The centralized pool manager in internal/pool provides comprehensive
compressionPool *TokenCompressionPool // statistics tracking that replaces these specialized profilers
logger *Logger
}
// NewTokenCompressionProfiler creates a new token compression profiler
func NewTokenCompressionProfiler(pool *TokenCompressionPool, logger *Logger) *TokenCompressionProfiler {
if logger == nil {
logger = GetSingletonNoOpLogger()
}
return &TokenCompressionProfiler{
compressionPool: pool,
logger: logger,
}
}
// TakeSnapshot captures token compression memory statistics
func (tcp *TokenCompressionProfiler) TakeSnapshot() (*MemorySnapshot, error) {
snapshot := &MemorySnapshot{
Timestamp: time.Now(),
CustomMetrics: make(map[string]interface{}),
}
runtime.ReadMemStats(&snapshot.RuntimeStats)
snapshot.CustomMetrics["compression_pool_active"] = true
return snapshot, nil
}
// StartProfiling begins profiling (no-op for compression)
func (tcp *TokenCompressionProfiler) StartProfiling(config ProfilingConfig) error {
return nil
}
// StopProfiling ends profiling
func (tcp *TokenCompressionProfiler) StopProfiling() (*MemorySnapshot, error) {
return tcp.TakeSnapshot()
}
// GetCurrentStats returns current memory statistics
func (tcp *TokenCompressionProfiler) GetCurrentStats() *runtime.MemStats {
stats := &runtime.MemStats{}
runtime.ReadMemStats(stats)
return stats
}
// AnalyzeLeaks analyzes token compression for memory leaks
func (tcp *TokenCompressionProfiler) AnalyzeLeaks(baseline, current *MemorySnapshot) *LeakAnalysis {
analysis := &LeakAnalysis{
SuspectedLeaks: make([]string, 0),
Recommendations: make([]string, 0),
}
if baseline == nil || current == nil {
analysis.LeakDescription = "Insufficient compression data"
return analysis
}
memoryIncrease := current.RuntimeStats.Alloc - baseline.RuntimeStats.Alloc
if memoryIncrease > 2*1024*1024 {
analysis.HasLeak = true
analysis.SuspectedLeaks = append(analysis.SuspectedLeaks,
"Token compression memory usage increased significantly")
analysis.Recommendations = append(analysis.Recommendations,
"Check for compression buffers not being returned to pool")
}
return analysis
}
// MemoryPoolProfiler monitors memory pool usage and detects leaks
type MemoryPoolProfiler struct {
memoryPoolManager *MemoryPoolManager
tokenCompressionPool *TokenCompressionPool
logger *Logger
}
// NewMemoryPoolProfiler creates a new memory pool profiler
func NewMemoryPoolProfiler(memoryPoolManager *MemoryPoolManager, tokenCompressionPool *TokenCompressionPool, logger *Logger) *MemoryPoolProfiler {
if logger == nil {
logger = GetSingletonNoOpLogger()
}
return &MemoryPoolProfiler{
memoryPoolManager: memoryPoolManager,
tokenCompressionPool: tokenCompressionPool,
logger: logger,
}
}
// TakeSnapshot captures memory pool statistics
func (mpp *MemoryPoolProfiler) TakeSnapshot() (*MemorySnapshot, error) {
snapshot := &MemorySnapshot{
Timestamp: time.Now(),
CustomMetrics: make(map[string]interface{}),
}
runtime.ReadMemStats(&snapshot.RuntimeStats)
if mpp.memoryPoolManager != nil {
snapshot.CustomMetrics["memory_pool_active"] = true
}
if mpp.tokenCompressionPool != nil {
snapshot.CustomMetrics["token_compression_pool_active"] = true
}
return snapshot, nil
}
// StartProfiling begins profiling (no-op for memory pools)
func (mpp *MemoryPoolProfiler) StartProfiling(config ProfilingConfig) error {
return nil
}
// StopProfiling ends profiling
func (mpp *MemoryPoolProfiler) StopProfiling() (*MemorySnapshot, error) {
return mpp.TakeSnapshot()
}
// GetCurrentStats returns current memory statistics
func (mpp *MemoryPoolProfiler) GetCurrentStats() *runtime.MemStats {
stats := &runtime.MemStats{}
runtime.ReadMemStats(stats)
return stats
}
// AnalyzeLeaks analyzes memory pools for leaks
func (mpp *MemoryPoolProfiler) AnalyzeLeaks(baseline, current *MemorySnapshot) *LeakAnalysis {
analysis := &LeakAnalysis{
SuspectedLeaks: make([]string, 0),
Recommendations: make([]string, 0),
}
if baseline == nil || current == nil {
analysis.LeakDescription = "Insufficient memory pool data"
return analysis
}
memoryIncrease := current.RuntimeStats.Alloc - baseline.RuntimeStats.Alloc
if memoryIncrease > 5*1024*1024 {
analysis.HasLeak = true
analysis.SuspectedLeaks = append(analysis.SuspectedLeaks,
"Memory pool operations caused significant memory increase")
analysis.Recommendations = append(analysis.Recommendations,
"Check for objects not being returned to memory pools properly")
}
return analysis
}
// Global profiling manager instance // Global profiling manager instance
var globalProfilingManager *ProfilingManager var globalProfilingManager *ProfilingManager
profiling_test.go
+3 -239
View File
@@ -150,17 +150,8 @@ func TestComponentProfilers(t *testing.T) {
t.Fatal("HTTP client snapshot is nil") t.Fatal("HTTP client snapshot is nil")
} }
// Test Token Compression Profiler // Token Compression Profiler removed - use internal/pool statistics instead
compressionPool := NewTokenCompressionPool() t.Log("Token compression profiler deprecated - use internal/pool stats")
tcp := NewTokenCompressionProfiler(compressionPool, logger)
snapshot, err = tcp.TakeSnapshot()
if err != nil {
t.Fatalf("Failed to take compression snapshot: %v", err)
}
if snapshot == nil {
t.Fatal("Compression snapshot is nil")
}
} }
func TestLeakAnalysis(t *testing.T) { func TestLeakAnalysis(t *testing.T) {
@@ -621,232 +612,5 @@ func TestProviderMetadataMemoryLeakDetection(t *testing.T) {
// TestMemoryPoolLeakDetection tests for memory leaks in memory pool operations // TestMemoryPoolLeakDetection tests for memory leaks in memory pool operations
func TestMemoryPoolLeakDetection(t *testing.T) { func TestMemoryPoolLeakDetection(t *testing.T) {
if testing.Short() { t.Skip("Deprecated - memory pool profilers removed. Use internal/pool statistics instead")
t.Skip("Skipping test in short mode")
}
logger := NewLogger("debug")
strictMode := os.Getenv("STRICT_MEMORY_TEST") == "true"
if strictMode {
t.Log("Running in strict memory test mode - will fail on detected leaks")
} else {
t.Log("Running in lenient memory test mode - will log warnings instead of failing")
}
config := LeakDetectionConfig{
EnableLeakDetection: true,
LeakThresholdMB: 10,
}
mto := NewMemoryTestOrchestrator(config, logger)
// Create memory pool manager and token compression pool
memoryPoolManager := NewMemoryPoolManager()
tokenCompressionPool := NewTokenCompressionPool()
// Create profiler for memory pools
profiler := NewMemoryPoolProfiler(memoryPoolManager, tokenCompressionPool, logger)
mto.RegisterComponent("memory_pools", profiler)
// Take initial baseline
baseline, err := profiler.TakeSnapshot()
if err != nil {
t.Fatalf("Failed to take baseline snapshot: %v", err)
}
initialGoroutines := runtime.NumGoroutine()
// Phase 1: Simulate various memory pool operations
t.Log("Phase 1: Testing memory pool operations with various patterns...")
// Test compression buffer pool
for i := 0; i < 100; i++ {
buf := memoryPoolManager.GetCompressionBuffer()
// Simulate some work with the buffer
buf.WriteString(fmt.Sprintf("test data %d", i))
// Properly return buffer to pool
memoryPoolManager.PutCompressionBuffer(buf)
}
// Test JWT parsing buffer pool
for i := 0; i < 50; i++ {
jwtBuf := memoryPoolManager.GetJWTParsingBuffer()
// Simulate JWT parsing operations
jwtBuf.HeaderBuf = append(jwtBuf.HeaderBuf, []byte("header")...)
jwtBuf.PayloadBuf = append(jwtBuf.PayloadBuf, []byte("payload")...)
jwtBuf.SignatureBuf = append(jwtBuf.SignatureBuf, []byte("signature")...)
// Properly return buffer to pool
memoryPoolManager.PutJWTParsingBuffer(jwtBuf)
}
// Test HTTP response buffer pool
for i := 0; i < 75; i++ {
httpBuf := memoryPoolManager.GetHTTPResponseBuffer()
// Simulate HTTP response processing
copy(httpBuf[:min(len(httpBuf), 100)], []byte("http response data"))
// Properly return buffer to pool
memoryPoolManager.PutHTTPResponseBuffer(httpBuf)
}
// Test string builder pool
for i := 0; i < 60; i++ {
sb := memoryPoolManager.GetStringBuilder()
// Simulate string building operations
sb.WriteString(fmt.Sprintf("built string %d", i))
_ = sb.String() // Use the result
// Properly return string builder to pool
memoryPoolManager.PutStringBuilder(sb)
}
// Test token compression pool
for i := 0; i < 40; i++ {
compBuf := tokenCompressionPool.GetCompressionBuffer()
// Simulate compression operations
compBuf.WriteString(fmt.Sprintf("compress data %d", i))
// Properly return buffer to pool
tokenCompressionPool.PutCompressionBuffer(compBuf)
decompBuf := tokenCompressionPool.GetDecompressionBuffer()
// Simulate decompression operations
decompBuf.WriteString(fmt.Sprintf("decompress data %d", i))
// Properly return buffer to pool
tokenCompressionPool.PutDecompressionBuffer(decompBuf)
sb := tokenCompressionPool.GetStringBuilder()
// Simulate string operations
sb.WriteString(fmt.Sprintf("token string %d", i))
_ = sb.String()
// Properly return string builder to pool
tokenCompressionPool.PutStringBuilder(sb)
}
// Take intermediate snapshot
intermediate, err := profiler.TakeSnapshot()
if err != nil {
t.Fatalf("Failed to take intermediate snapshot: %v", err)
}
// Phase 2: Continue with more intensive operations to test sustained usage
t.Log("Phase 2: Testing sustained memory pool usage...")
// Simulate mixed operations with varying patterns
for i := 0; i < 200; i++ {
// Mix different pool operations
switch i % 4 {
case 0:
buf := memoryPoolManager.GetCompressionBuffer()
buf.WriteString("mixed operation data")
memoryPoolManager.PutCompressionBuffer(buf)
case 1:
jwtBuf := memoryPoolManager.GetJWTParsingBuffer()
jwtBuf.HeaderBuf = append(jwtBuf.HeaderBuf, []byte("mixed")...)
memoryPoolManager.PutJWTParsingBuffer(jwtBuf)
case 2:
httpBuf := memoryPoolManager.GetHTTPResponseBuffer()
copy(httpBuf[:min(len(httpBuf), 50)], []byte("mixed http"))
memoryPoolManager.PutHTTPResponseBuffer(httpBuf)
case 3:
sb := memoryPoolManager.GetStringBuilder()
sb.WriteString("mixed string building")
_ = sb.String()
memoryPoolManager.PutStringBuilder(sb)
}
}
// Take final snapshot
current, err := profiler.TakeSnapshot()
if err != nil {
t.Fatalf("Failed to take current snapshot: %v", err)
}
finalGoroutines := runtime.NumGoroutine()
// Analyze for leaks
analysis := profiler.AnalyzeLeaks(baseline, current)
// Assertions for memory leaks
if analysis.HasLeak {
if strictMode {
t.Errorf("Memory leak detected in memory pool operations: %s", analysis.LeakDescription)
for _, leak := range analysis.SuspectedLeaks {
t.Errorf("Suspected leak: %s", leak)
}
} else {
t.Logf("Memory leak warning in memory pool operations: %s", analysis.LeakDescription)
for _, leak := range analysis.SuspectedLeaks {
t.Logf("Suspected leak: %s", leak)
}
}
for _, rec := range analysis.Recommendations {
t.Logf("Recommendation: %s", rec)
}
}
// Check total memory growth
totalMemoryIncrease := current.RuntimeStats.Alloc - baseline.RuntimeStats.Alloc
if totalMemoryIncrease > 15*1024*1024 { // 15MB threshold for entire test
if strictMode {
t.Errorf("Total memory usage increased by %.2f MB during memory pool operations", float64(totalMemoryIncrease)/(1024*1024))
} else {
t.Logf("Total memory usage increased by %.2f MB during memory pool operations", float64(totalMemoryIncrease)/(1024*1024))
}
}
// Check for gradual memory growth patterns
intermediateMemoryIncrease := intermediate.RuntimeStats.Alloc - baseline.RuntimeStats.Alloc
if intermediateMemoryIncrease > 8*1024*1024 { // 8MB threshold for first phase
if strictMode {
t.Errorf("Memory usage increased by %.2f MB during first phase of memory pool operations", float64(intermediateMemoryIncrease)/(1024*1024))
} else {
t.Logf("Memory usage increased by %.2f MB during first phase of memory pool operations", float64(intermediateMemoryIncrease)/(1024*1024))
}
}
// Check goroutine count stability
goroutineIncrease := finalGoroutines - initialGoroutines
if goroutineIncrease > 3 { // Allow small variance for test environment
if strictMode {
t.Errorf("Goroutine count increased by %d during memory pool operations (initial: %d, final: %d)",
goroutineIncrease, initialGoroutines, finalGoroutines)
} else {
t.Logf("Goroutine count increased by %d during memory pool operations (initial: %d, final: %d)",
goroutineIncrease, initialGoroutines, finalGoroutines)
}
}
// Phase 3: Test cleanup verification
t.Log("Phase 3: Testing cleanup verification...")
// Force garbage collection to see if pools are properly managed
runtime.GC()
runtime.GC() // Run twice to ensure cleanup
time.Sleep(GetTestDuration(10 * time.Millisecond)) // Allow cleanup to complete
// Take post-cleanup snapshot
postCleanup, err := profiler.TakeSnapshot()
if err != nil {
t.Fatalf("Failed to take post-cleanup snapshot: %v", err)
}
// Check if memory decreased after cleanup
if postCleanup.RuntimeStats.Alloc < current.RuntimeStats.Alloc {
memoryDecrease := current.RuntimeStats.Alloc - postCleanup.RuntimeStats.Alloc
t.Logf("Memory decreased by %.2f MB after cleanup phase", float64(memoryDecrease)/(1024*1024))
} else if postCleanup.RuntimeStats.Alloc > current.RuntimeStats.Alloc {
memoryIncrease := postCleanup.RuntimeStats.Alloc - current.RuntimeStats.Alloc
if strictMode {
t.Errorf("Memory increased by %.2f MB after cleanup phase - possible cleanup issues", float64(memoryIncrease)/(1024*1024))
} else {
t.Logf("Memory increased by %.2f MB after cleanup phase - possible cleanup issues", float64(memoryIncrease)/(1024*1024))
}
}
t.Logf("Memory pool leak detection test completed")
t.Logf("Memory usage: baseline=%.2f MB, intermediate=%.2f MB, final=%.2f MB, post-cleanup=%.2f MB",
float64(baseline.RuntimeStats.Alloc)/(1024*1024),
float64(intermediate.RuntimeStats.Alloc)/(1024*1024),
float64(current.RuntimeStats.Alloc)/(1024*1024),
float64(postCleanup.RuntimeStats.Alloc)/(1024*1024))
} }
session.go
+7 -6
View File
@@ -18,6 +18,7 @@ import (
"time" "time"
"github.com/gorilla/sessions" "github.com/gorilla/sessions"
"github.com/lukaszraczylo/traefikoidc/internal/pool"
) )
// constantTimeStringCompare performs a constant-time comparison of two strings // constantTimeStringCompare performs a constant-time comparison of two strings
@@ -101,9 +102,9 @@ func compressToken(token string) string {
return token return token
} }
pools := GetGlobalMemoryPools() pm := pool.Get()
b := pools.GetCompressionBuffer() b := pm.GetBuffer(4096)
defer pools.PutCompressionBuffer(b) defer pm.PutBuffer(b)
gz := gzip.NewWriter(b) gz := gzip.NewWriter(b)
@@ -181,9 +182,9 @@ func decompressTokenInternal(compressed string) string {
return compressed return compressed
} }
pools := GetGlobalMemoryPools() pm := pool.Get()
readerBuf := pools.GetHTTPResponseBuffer() readerBuf := pm.GetHTTPResponseBuffer()
defer pools.PutHTTPResponseBuffer(readerBuf) defer pm.PutHTTPResponseBuffer(readerBuf)
gz, err := gzip.NewReader(bytes.NewReader(data)) gz, err := gzip.NewReader(bytes.NewReader(data))
if err != nil { if err != nil {
session_chunk_manager.go
+14 -5
View File
@@ -1,9 +1,9 @@
package traefikoidc package traefikoidc
import ( import (
"bytes"
"context" "context"
"encoding/base64" "encoding/base64"
"encoding/json"
"fmt" "fmt"
"runtime" "runtime"
"strings" "strings"
@@ -12,6 +12,7 @@ import (
"time" "time"
"github.com/gorilla/sessions" "github.com/gorilla/sessions"
"github.com/lukaszraczylo/traefikoidc/internal/pool"
) )
// TokenConfig defines validation and storage parameters for different token types. // TokenConfig defines validation and storage parameters for different token types.
@@ -981,9 +982,13 @@ func (cm *ChunkManager) extractJWTExpiration(token string) (*time.Time, error) {
return nil, fmt.Errorf("failed to decode JWT payload: %w", err) return nil, fmt.Errorf("failed to decode JWT payload: %w", err)
} }
// Parse the JSON payload // Parse the JSON payload using pooled decoder
var claims map[string]interface{} var claims map[string]interface{}
if err := json.Unmarshal(payload, &claims); err != nil { pm := pool.Get()
decoder := pm.GetJSONDecoder(bytes.NewReader(payload))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&claims); err != nil {
return nil, fmt.Errorf("failed to parse JWT claims: %w", err) return nil, fmt.Errorf("failed to parse JWT claims: %w", err)
} }
@@ -1067,9 +1072,13 @@ func (cm *ChunkManager) extractJWTIssuedAt(token string) (*time.Time, error) {
return nil, fmt.Errorf("failed to decode JWT payload: %w", err) return nil, fmt.Errorf("failed to decode JWT payload: %w", err)
} }
// Parse the JSON payload // Parse the JSON payload using pooled decoder
var claims map[string]interface{} var claims map[string]interface{}
if err := json.Unmarshal(payload, &claims); err != nil { pm := pool.Get()
decoder := pm.GetJSONDecoder(bytes.NewReader(payload))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&claims); err != nil {
return nil, fmt.Errorf("failed to parse JWT claims: %w", err) return nil, fmt.Errorf("failed to parse JWT claims: %w", err)
} }
string_builder_pool.go
-109
View File
@@ -1,109 +0,0 @@
package traefikoidc
import (
"strings"
"sync"
)
// StringBuilderPool manages a pool of string builders for efficient string operations
type StringBuilderPool struct {
pool sync.Pool
}
var (
globalStringBuilderPool *StringBuilderPool
globalStringBuilderPoolOnce sync.Once
)
// GetGlobalStringBuilderPool returns the global string builder pool
func GetGlobalStringBuilderPool() *StringBuilderPool {
globalStringBuilderPoolOnce.Do(func() {
globalStringBuilderPool = &StringBuilderPool{
pool: sync.Pool{
New: func() interface{} {
return &strings.Builder{}
},
},
}
})
return globalStringBuilderPool
}
// Get retrieves a string builder from the pool
func (p *StringBuilderPool) Get() *strings.Builder {
sb := p.pool.Get().(*strings.Builder)
sb.Reset() // Ensure it's clean
return sb
}
// Put returns a string builder to the pool
func (p *StringBuilderPool) Put(sb *strings.Builder) {
if sb == nil {
return
}
// Only return to pool if not too large (avoid keeping huge buffers)
if sb.Cap() <= 4096 {
sb.Reset()
p.pool.Put(sb)
}
}
// FormatString efficiently formats a string using the pool
func (p *StringBuilderPool) FormatString(format func(*strings.Builder)) string {
sb := p.Get()
defer p.Put(sb)
format(sb)
return sb.String()
}
// BuildSessionName efficiently builds session names
func BuildSessionName(baseName string, index int) string {
pool := GetGlobalStringBuilderPool()
return pool.FormatString(func(sb *strings.Builder) {
sb.WriteString(baseName)
sb.WriteRune('_')
// Efficient int to string conversion
if index < 10 {
sb.WriteRune('0' + rune(index))
} else {
sb.WriteString(sbIntToString(index))
}
})
}
// BuildCacheKey efficiently builds cache keys
func BuildCacheKey(parts ...string) string {
pool := GetGlobalStringBuilderPool()
return pool.FormatString(func(sb *strings.Builder) {
for i, part := range parts {
if i > 0 {
sb.WriteRune(':')
}
sb.WriteString(part)
}
})
}
// sbIntToString converts int to string without allocation (for small numbers)
func sbIntToString(n int) string {
if n < 0 {
return "-" + sbIntToString(-n)
}
if n < 10 {
return string(rune('0' + n))
}
if n < 100 {
return string(rune('0'+n/10)) + string(rune('0'+n%10))
}
// Fall back to standard conversion for larger numbers
buf := make([]byte, 0, 20)
for n > 0 {
buf = append(buf, byte('0'+n%10))
n /= 10
}
// Reverse the buffer
for i, j := 0, len(buf)-1; i < j; i, j = i+1, j-1 {
buf[i], buf[j] = buf[j], buf[i]
}
return string(buf)
}
token_validator.go
+11 -3
View File
@@ -1,11 +1,13 @@
package traefikoidc package traefikoidc
import ( import (
"bytes"
"encoding/base64" "encoding/base64"
"encoding/json"
"fmt" "fmt"
"strings" "strings"
"time" "time"
"github.com/lukaszraczylo/traefikoidc/internal/pool"
) )
// TokenValidator provides unified token validation functionality // TokenValidator provides unified token validation functionality
@@ -93,7 +95,10 @@ func (v *TokenValidator) validateJWT(token string) TokenValidationResult {
} }
var claims map[string]interface{} var claims map[string]interface{}
if err := json.Unmarshal(payload, &claims); err != nil { pm := pool.Get()
decoder := pm.GetJSONDecoder(bytes.NewReader(payload))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&claims); err != nil {
result.Error = fmt.Errorf("failed to parse JWT claims: %w", err) result.Error = fmt.Errorf("failed to parse JWT claims: %w", err)
return result return result
} }
@@ -233,7 +238,10 @@ func (v *TokenValidator) ExtractClaims(token string) (map[string]interface{}, er
} }
var claims map[string]interface{} var claims map[string]interface{}
if err := json.Unmarshal(payload, &claims); err != nil { pm := pool.Get()
decoder := pm.GetJSONDecoder(bytes.NewReader(payload))
defer pm.PutJSONDecoder(decoder)
if err := decoder.Decode(&claims); err != nil {
return nil, fmt.Errorf("failed to parse claims: %w", err) return nil, fmt.Errorf("failed to parse claims: %w", err)
} }
utilities.go
+2 -2
View File
@@ -286,8 +286,8 @@ func (t *TraefikOidc) Close() error {
taskRegistry.StopAllTasks() taskRegistry.StopAllTasks()
t.safeLogDebug("All global background tasks stopped") t.safeLogDebug("All global background tasks stopped")
CleanupGlobalMemoryPools() // Note: Centralized pool in internal/pool is singleton-managed and doesn't require explicit cleanup
t.safeLogDebug("Global memory pools cleaned up") t.safeLogDebug("Memory pools managed by singleton pattern")
// Force garbage collection to help with memory cleanup after shutdown // Force garbage collection to help with memory cleanup after shutdown
runtime.GC() runtime.GC()