-
Notifications
You must be signed in to change notification settings - Fork 251
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Darren/feat/add subscription cache (#866)
* ehancement: create a cache for block based subscriptions * minor: change function names for subscriptions * test: add unit test for message cache * chore: add license headers * refactor: fix up error handling * fix: remove bad test * fix: PR comments * fix: PR comments - remove block cache * refactor(subscriptions): store structs in cache, not bytes * fix(license): add license header * chore(subscriptions): revert unit test changes * enhancement: resolve pr comments to use simplelru * enhancement: resolve pr comments - use id as key
- Loading branch information
1 parent
ad5bfbd
commit 9d5b515
Showing
9 changed files
with
202 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| // Copyright (c) 2024 The VeChainThor developers | ||
| // | ||
| // Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying | ||
| // file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html> | ||
|
|
||
| package subscriptions | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| "github.com/hashicorp/golang-lru/simplelru" | ||
| "github.com/vechain/thor/v2/thor" | ||
| ) | ||
|
|
||
| // messageCache is a generic cache that stores messages of any type. | ||
| type messageCache[T any] struct { | ||
| cache *simplelru.LRU | ||
| mu sync.RWMutex | ||
| } | ||
|
|
||
| // newMessageCache creates a new messageCache with the specified cache size. | ||
| func newMessageCache[T any](cacheSize uint32) *messageCache[T] { | ||
| if cacheSize > 1000 { | ||
| cacheSize = 1000 | ||
| } | ||
| if cacheSize == 0 { | ||
| cacheSize = 1 | ||
| } | ||
| cache, err := simplelru.NewLRU(int(cacheSize), nil) | ||
| if err != nil { | ||
| // lru.New only throws an error if the number is less than 1 | ||
| panic(fmt.Errorf("failed to create message cache: %v", err)) | ||
| } | ||
| return &messageCache[T]{ | ||
| cache: cache, | ||
| } | ||
| } | ||
|
|
||
| // GetOrAdd returns the message of the block. If the message is not in the cache, | ||
| // it will generate the message and add it to the cache. The second return value | ||
| // indicates whether the message is newly generated. | ||
| func (mc *messageCache[T]) GetOrAdd(id thor.Bytes32, createMessage func() (T, error)) (T, bool, error) { | ||
| mc.mu.RLock() | ||
| msg, ok := mc.cache.Get(id) | ||
| mc.mu.RUnlock() | ||
| if ok { | ||
| return msg.(T), false, nil | ||
| } | ||
|
|
||
| mc.mu.Lock() | ||
| defer mc.mu.Unlock() | ||
| msg, ok = mc.cache.Get(id) | ||
| if ok { | ||
| return msg.(T), false, nil | ||
| } | ||
|
|
||
| newMsg, err := createMessage() | ||
| if err != nil { | ||
| var zero T | ||
| return zero, false, err | ||
| } | ||
| mc.cache.Add(id, newMsg) | ||
| return newMsg, true, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| // Copyright (c) 2024 The VeChainThor developers | ||
| // | ||
| // Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying | ||
| // file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html> | ||
|
|
||
| package subscriptions | ||
|
|
||
| import ( | ||
| "sync" | ||
| "sync/atomic" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/vechain/thor/v2/block" | ||
| ) | ||
|
|
||
| type message struct { | ||
| id string | ||
| } | ||
|
|
||
| func handler(blk *block.Block) func() (message, error) { | ||
| return func() (message, error) { | ||
| msg := message{ | ||
| id: blk.Header().ID().String(), | ||
| } | ||
| return msg, nil | ||
| } | ||
| } | ||
|
|
||
| func TestMessageCache_GetOrAdd(t *testing.T) { | ||
| _, generatedBlocks, _ := initChain(t) | ||
|
|
||
| blk0 := generatedBlocks[0] | ||
| blk1 := generatedBlocks[1] | ||
|
|
||
| cache := newMessageCache[message](10) | ||
|
|
||
| counter := atomic.Int32{} | ||
| wg := sync.WaitGroup{} | ||
| for i := 0; i < 100; i++ { | ||
| wg.Add(1) | ||
| start := time.Now().Add(20 * time.Millisecond) | ||
| go func() { | ||
| defer wg.Done() | ||
| time.Sleep(time.Until(start)) | ||
| _, added, err := cache.GetOrAdd(blk0.Header().ID(), handler(blk0)) | ||
| assert.NoError(t, err) | ||
| if added { | ||
| counter.Add(1) | ||
| } | ||
| }() | ||
| } | ||
| wg.Wait() | ||
| assert.Equal(t, counter.Load(), int32(1)) | ||
|
|
||
| _, added, err := cache.GetOrAdd(blk1.Header().ID(), handler(blk1)) | ||
| assert.NoError(t, err) | ||
| assert.True(t, added) | ||
| assert.Equal(t, cache.cache.Len(), 2) | ||
| } |
Oops, something went wrong.