Compare commits

..
6 Commits
Author SHA1 Message Date
lukaszraczyloandgithub-actions[bot] d2453fe2f5 Update go.mod and go.sum
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2025-01-28 03:04:14 +00:00
lukaszraczylo 28eec46f4c General improvements:
Added sharded cache to reduce lock contention
Optimized buffer usage with pre-sized pools
Improved HTTP client with connection pooling and timeouts
Reduced allocations in JSON handling
2025-01-08 11:16:51 +00:00
lukaszraczyloandgithub-actions[bot] 63e449ef77 Update go.mod and go.sum
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2025-01-07 03:09:25 +00:00
lukaszraczyloandgithub-actions[bot] 50cfc5f98d Update go.mod and go.sum
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2025-01-05 03:10:35 +00:00
lukaszraczyloandgithub-actions[bot] 4a575ac9c8 Update go.mod and go.sum
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2024-12-19 03:12:34 +00:00
lukaszraczyloandgithub-actions[bot] f54cedaeb3 Update go.mod and go.sum
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2024-12-12 03:17:28 +00:00
6 changed files with 164 additions and 76 deletions
+60 -33
View File
@@ -3,6 +3,7 @@ package libpack_cache
import (
"bytes"
"compress/gzip"
"hash/fnv"
"io"
"sync"
"time"
@@ -13,12 +14,25 @@ type CacheEntry struct {
Value []byte
}
const shardCount = 256 // Must be power of 2
type shard struct {
entries map[string]CacheEntry
sync.RWMutex
}
type Cache struct {
compressPool sync.Pool
decompressPool sync.Pool
entries sync.Map
globalTTL time.Duration
sync.RWMutex
shards [shardCount]*shard
globalTTL time.Duration
}
// getShard returns the appropriate shard for a given key
func (c *Cache) getShard(key string) *shard {
hash := fnv.New32a()
hash.Write([]byte(key))
return c.shards[hash.Sum32()%shardCount]
}
func New(globalTTL time.Duration) *Cache {
@@ -32,13 +46,19 @@ func New(globalTTL time.Duration) *Cache {
},
decompressPool: sync.Pool{
New: func() interface{} {
// Ensure that new is returning a new reader initialized with an empty byte buffer
r, _ := gzip.NewReader(bytes.NewReader([]byte{}))
return r
},
},
}
// Initialize shards
for i := 0; i < shardCount; i++ {
cache.shards[i] = &shard{
entries: make(map[string]CacheEntry),
}
}
go cache.cleanupRoutine(globalTTL)
return cache
}
@@ -52,33 +72,43 @@ func (c *Cache) cleanupRoutine(globalTTL time.Duration) {
}
}
func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
c.Lock() // use the lock
defer c.Unlock()
expiresAt := time.Now().Add(ttl)
shard := c.getShard(key)
shard.Lock()
defer shard.Unlock()
compressedValue, err := c.compress(value)
if err != nil {
return
}
entry := CacheEntry{
shard.entries[key] = CacheEntry{
Value: compressedValue,
ExpiresAt: expiresAt,
ExpiresAt: time.Now().Add(ttl),
}
c.entries.Store(key, entry)
}
func (c *Cache) Get(key string) ([]byte, bool) {
c.RLock() // use the read lock
defer c.RUnlock()
entry, ok := c.entries.Load(key)
if !ok || entry.(CacheEntry).ExpiresAt.Before(time.Now()) {
shard := c.getShard(key)
shard.RLock()
entry, ok := shard.entries[key]
if !ok {
shard.RUnlock()
return nil, false
}
compressedValue := entry.(CacheEntry).Value
value, err := c.decompress(compressedValue)
if entry.ExpiresAt.Before(time.Now()) {
shard.RUnlock()
// Clean up expired entry in background
go func() {
shard.Lock()
delete(shard.entries, key)
shard.Unlock()
}()
return nil, false
}
shard.RUnlock()
value, err := c.decompress(entry.Value)
if err != nil {
return nil, false
}
@@ -87,26 +117,23 @@ func (c *Cache) Get(key string) ([]byte, bool) {
}
func (c *Cache) Delete(key string) {
c.Lock()
defer c.Unlock()
_, ok := c.entries.Load(key)
if !ok {
return
}
c.entries.Delete(key)
shard := c.getShard(key)
shard.Lock()
delete(shard.entries, key)
shard.Unlock()
}
func (c *Cache) CleanExpiredEntries() {
now := time.Now()
c.entries.Range(func(key, value interface{}) bool {
entry := value.(CacheEntry)
if entry.ExpiresAt.Before(now) {
c.entries.Delete(key)
for _, shard := range c.shards {
shard.Lock()
for key, entry := range shard.entries {
if entry.ExpiresAt.Before(now) {
delete(shard.entries, key)
}
}
return true
})
shard.Unlock()
}
}
func (c *Cache) compress(data []byte) ([]byte, error) {
+31 -11
View File
@@ -7,7 +7,6 @@ import (
"fmt"
"io"
"net/http"
"sync"
"time"
"github.com/avast/retry-go/v4"
@@ -15,11 +14,22 @@ import (
libpack_logger "github.com/lukaszraczylo/go-simple-graphql/logging"
)
var bufferPool = sync.Pool{
New: func() interface{} {
return new(bytes.Buffer)
},
}
var (
// Shared HTTP transport with optimized settings
defaultTransport = &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 100,
IdleConnTimeout: 90 * time.Second,
DisableCompression: false, // Enable compression for responses
ForceAttemptHTTP2: true,
}
// Shared HTTP client with timeouts
defaultClient = &http.Client{
Transport: defaultTransport,
Timeout: 30 * time.Second,
}
)
func (qe *QueryExecutor) executeQuery() ([]byte, error) {
// Reuse buffer from pool to avoid allocations
@@ -62,7 +72,12 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
httpRequest.Body = io.NopCloser(bytes.NewReader(buf.Bytes()))
httpRequest.ContentLength = int64(buf.Len())
httpResponse, err := qe.client.Do(httpRequest)
// Use default client if custom client is not set
client := qe.client
if client == nil {
client = defaultClient
}
httpResponse, err := client.Do(httpRequest)
if err != nil {
return err
}
@@ -88,14 +103,18 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
reader = httpResponse.Body
}
// Read all response body
body, err := io.ReadAll(reader)
// Use buffer pool for reading response
respBuf := bufferPool.Get().(*bytes.Buffer)
respBuf.Reset()
defer bufferPool.Put(respBuf)
_, err = io.Copy(respBuf, reader)
if err != nil {
return fmt.Errorf("error reading HTTP response: %w", err)
}
// Unmarshal response
err = json.Unmarshal(body, &queryResult)
// Unmarshal response directly from buffer
err = json.Unmarshal(respBuf.Bytes(), &queryResult)
if err != nil {
return fmt.Errorf("error unmarshalling HTTP response: %w", err)
}
@@ -111,6 +130,7 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
retry.Attempts(uint(retriesMax)),
retry.DelayType(retry.BackOffDelay),
retry.Delay(time.Duration(qe.retries_delay)),
retry.MaxDelay(10 * time.Second),
retry.LastErrorOnly(true),
)
if err != nil {
+4 -4
View File
@@ -4,11 +4,11 @@ go 1.21
require (
github.com/avast/retry-go/v4 v4.6.0
github.com/goccy/go-json v0.10.3
github.com/goccy/go-json v0.10.5
github.com/goccy/go-reflect v1.2.0
github.com/gookit/goutil v0.6.18
github.com/stretchr/testify v1.9.0
golang.org/x/net v0.32.0
golang.org/x/net v0.34.0
)
require (
@@ -17,8 +17,8 @@ require (
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
golang.org/x/sync v0.10.0 // indirect
golang.org/x/sys v0.28.0 // indirect
golang.org/x/term v0.27.0 // indirect
golang.org/x/sys v0.29.0 // indirect
golang.org/x/term v0.28.0 // indirect
golang.org/x/text v0.21.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+8 -8
View File
@@ -2,8 +2,8 @@ github.com/avast/retry-go/v4 v4.6.0 h1:K9xNA+KeB8HHc2aWFuLb25Offp+0iVRXEvFx8IinR
github.com/avast/retry-go/v4 v4.6.0/go.mod h1:gvWlPhBVsvBbLkVGDg/KwvBv0bEkCOLRRSHKIr2PyOE=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
github.com/goccy/go-reflect v1.2.0 h1:O0T8rZCuNmGXewnATuKYnkL0xm6o8UNOJZd/gOkb9ms=
github.com/goccy/go-reflect v1.2.0/go.mod h1:n0oYZn8VcV2CkWTxi8B9QjkCoq6GTtCEdfmR66YhFtE=
github.com/gookit/color v1.5.4 h1:FZmqs7XOyGgCAxmWyPslpiok1k05wmY3SJTytgvYFs0=
@@ -18,14 +18,14 @@ github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavM
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561 h1:MDc5xs78ZrZr3HMQugiXOAkSZtfTpbJLDr/lwfgO53E=
golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561/go.mod h1:cyybsKvd6eL0RnXn6p/Grxp8F5bW7iYuBgsNCOHpMYE=
golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI=
golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs=
golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek=
golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+24
View File
@@ -0,0 +1,24 @@
package gql
import (
"bytes"
"sync"
)
var (
// Pre-sized buffer pool for better performance
bufferPool = sync.Pool{
New: func() interface{} {
// Pre-allocate buffer with 4KB capacity
b := bytes.NewBuffer(make([]byte, 0, 4096))
return b
},
}
// Pre-allocate error maps to reduce allocations
errPairsPool = sync.Pool{
New: func() interface{} {
return make(map[string]interface{}, 1)
},
}
)
+37 -20
View File
@@ -11,56 +11,73 @@ import (
)
func (b *BaseClient) convertToJSON(v any) []byte {
// Reuse buffer to reduce allocations
buf := bufferPool.Get().(*bytes.Buffer)
defer bufferPool.Put(buf)
buf.Reset()
// Use json.Marshal to avoid adding extra newline
jsonData, err := json.Marshal(v)
if err != nil {
// Use json.NewEncoder for better performance with buffers
enc := json.NewEncoder(buf)
enc.SetEscapeHTML(false) // Reduce unnecessary escaping
if err := enc.Encode(v); err != nil {
errPairs := errPairsPool.Get().(map[string]interface{})
errPairs["error"] = err.Error()
b.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't convert to JSON",
Pairs: map[string]interface{}{"error": err.Error()},
Pairs: errPairs,
})
errPairsPool.Put(errPairs)
return nil
}
// Copy the bytes to the buffer
buf.Write(jsonData)
// Get the buffer bytes directly, trimming the trailing newline
bytes := buf.Bytes()
if len(bytes) > 0 && bytes[len(bytes)-1] == '\n' {
bytes = bytes[:len(bytes)-1]
}
// Copy the bytes to a new slice before returning
result := make([]byte, buf.Len())
copy(result, buf.Bytes())
// Make a copy of the bytes since the buffer will be reused
result := make([]byte, len(bytes))
copy(result, bytes)
return result
}
func (b *BaseClient) compileQuery(queryPartials ...any) *Query {
var query string
var variables map[string]interface{}
for _, partial := range queryPartials {
switch val := partial.(type) {
case string:
query = val
case map[string]interface{}:
variables = val
// Pre-allocate the query with an estimated size
if len(queryPartials) > 0 {
if str, ok := queryPartials[0].(string); ok {
query = str
}
}
// Only allocate variables map if we have more than one partial
if len(queryPartials) > 1 {
if vars, ok := queryPartials[1].(map[string]interface{}); ok {
variables = vars
}
}
if query == "" {
errPairs := errPairsPool.Get().(map[string]interface{})
errPairs["error"] = "query is empty"
b.Logger.Error(&libpack_logger.LogMessage{
Message: "Can't compile query",
Pairs: map[string]interface{}{"error": "query is empty"},
Pairs: errPairs,
})
errPairsPool.Put(errPairs)
return nil
}
jsonQuery := b.convertToJSON(&Query{Query: query, Variables: variables})
return &Query{
// Construct query object once
q := &Query{
Query: query,
Variables: variables,
JsonQuery: jsonQuery,
}
q.JsonQuery = b.convertToJSON(q)
return q
}
func (b *BaseClient) Query(query string, variables map[string]interface{}, headers map[string]interface{}) (any, error) {