hanahmily commented on code in PR #670:
URL:
https://github.com/apache/skywalking-banyandb/pull/670#discussion_r2113448252
##########
banyand/internal/storage/storage.go:
##########
@@ -119,6 +119,7 @@ type Segment[T TSTable, O any] interface {
GetTimeRange() timestamp.TimeRange
CreateTSTableIfNotExist(shardID common.ShardID) (T, error)
Tables() []T
+ Caches() []*ShardCache
Review Comment:
```suggestion
Tables() ([]T, []*ShardCache)
```
The value of segment.sLst is atomic. Its length may vary if you call it more
than once.
##########
banyand/internal/storage/cache.go:
##########
@@ -0,0 +1,230 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storage
+
+import (
+ "container/heap"
+ "sync"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+type entry struct {
+ value any
+ lastAccess uint64
+}
+
+// EntryKey is the key of an entry in the cache.
+type EntryKey struct {
+ group string
+ PartID uint64
+ Offset uint64
+ segmentID segmentID
+ shardID common.ShardID
+}
+
+type entryIndex struct {
+ *entry
+ key EntryKey
+ index int
+}
+
+type entryIndexHeap []*entryIndex
+
+func (h entryIndexHeap) Len() int { return len(h) }
+
+func (h entryIndexHeap) Less(i, j int) bool {
+ return atomic.LoadUint64(&h[i].entry.lastAccess) <
atomic.LoadUint64(&h[j].entry.lastAccess)
+}
+
+func (h entryIndexHeap) Swap(i, j int) {
+ h[i], h[j] = h[j], h[i]
+ h[i].index = i
+ h[j].index = j
+}
+
+func (h *entryIndexHeap) Push(x interface{}) {
+ n := len(*h)
+ ei := x.(*entryIndex)
+ ei.index = n
+ *h = append(*h, ei)
+}
+
+func (h *entryIndexHeap) Pop() interface{} {
+ old := *h
+ n := len(old)
+ x := old[n-1]
+ *h = old[:n-1]
+ return x
+}
+
+// Cache stores the compressed primary blocks.
+type Cache struct {
+ entry map[EntryKey]*entry
+ entryIndex map[EntryKey]*entryIndex
+ entryIndexHeap *entryIndexHeap
+ stopCh chan struct{}
+ requests uint64
+ misses uint64
+ mu sync.RWMutex
+ wg sync.WaitGroup
+ maxCacheSize uint64
+ cleanupInterval time.Duration
+ idleTimeout time.Duration
+}
+
+// NewCache creates a cache.
+func NewCache() *Cache {
+ h := &entryIndexHeap{}
+ heap.Init(h)
+ c := &Cache{
+ entry: make(map[EntryKey]*entry),
+ entryIndexHeap: h,
+ entryIndex: make(map[EntryKey]*entryIndex),
+ stopCh: make(chan struct{}),
+ wg: sync.WaitGroup{},
+ maxCacheSize: 100 * 1024 * 1024,
+ cleanupInterval: 30 * time.Second,
+ idleTimeout: 2 * time.Minute,
+ }
+ c.wg.Add(1)
+ return c
+}
+
+// Clean periodically cleans the cache.
+func (c *Cache) Clean() {
Review Comment:
```suggestion
func (c *Cache) StartCleaner() {
```
##########
banyand/internal/storage/cache.go:
##########
@@ -0,0 +1,230 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storage
+
+import (
+ "container/heap"
+ "sync"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+type entry struct {
+ value any
+ lastAccess uint64
+}
+
+// EntryKey is the key of an entry in the cache.
+type EntryKey struct {
+ group string
+ PartID uint64
+ Offset uint64
+ segmentID segmentID
+ shardID common.ShardID
+}
+
+type entryIndex struct {
+ *entry
+ key EntryKey
+ index int
+}
+
+type entryIndexHeap []*entryIndex
+
+func (h entryIndexHeap) Len() int { return len(h) }
+
+func (h entryIndexHeap) Less(i, j int) bool {
+ return atomic.LoadUint64(&h[i].entry.lastAccess) <
atomic.LoadUint64(&h[j].entry.lastAccess)
+}
+
+func (h entryIndexHeap) Swap(i, j int) {
+ h[i], h[j] = h[j], h[i]
+ h[i].index = i
+ h[j].index = j
+}
+
+func (h *entryIndexHeap) Push(x interface{}) {
+ n := len(*h)
+ ei := x.(*entryIndex)
+ ei.index = n
+ *h = append(*h, ei)
+}
+
+func (h *entryIndexHeap) Pop() interface{} {
+ old := *h
+ n := len(old)
+ x := old[n-1]
+ *h = old[:n-1]
+ return x
+}
+
+// Cache stores the compressed primary blocks.
+type Cache struct {
+ entry map[EntryKey]*entry
+ entryIndex map[EntryKey]*entryIndex
+ entryIndexHeap *entryIndexHeap
+ stopCh chan struct{}
+ requests uint64
+ misses uint64
+ mu sync.RWMutex
+ wg sync.WaitGroup
+ maxCacheSize uint64
+ cleanupInterval time.Duration
+ idleTimeout time.Duration
+}
+
+// NewCache creates a cache.
+func NewCache() *Cache {
+ h := &entryIndexHeap{}
+ heap.Init(h)
+ c := &Cache{
+ entry: make(map[EntryKey]*entry),
+ entryIndexHeap: h,
+ entryIndex: make(map[EntryKey]*entryIndex),
+ stopCh: make(chan struct{}),
+ wg: sync.WaitGroup{},
+ maxCacheSize: 100 * 1024 * 1024,
+ cleanupInterval: 30 * time.Second,
+ idleTimeout: 2 * time.Minute,
+ }
+ c.wg.Add(1)
Review Comment:
Move it to `Clean`
##########
banyand/internal/storage/storage.go:
##########
Review Comment:
My recommendation is to create an interface `Cache` to conceal the cache
struts. Other modules will refer to this interface and convert all caches to
unexported.
##########
banyand/backup/lifecycle/service.go:
##########
@@ -131,6 +131,7 @@ func (l *lifecycleService) Serve() run.StopNotify {
defer streamSVC.GracefulStop()
}
if measureSVC != nil {
+ measureSVC.Serve()
Review Comment:
```suggestion
```
If you remove `c.wg.Add(1)` from `NewCache`, you won't need to start the
service. `measureSVC` serves as a facade for the metadata repository, which
does not require starting the local cache for metadata.
##########
banyand/internal/storage/cache.go:
##########
@@ -0,0 +1,230 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storage
+
+import (
+ "container/heap"
+ "sync"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+type entry struct {
+ value any
+ lastAccess uint64
+}
+
+// EntryKey is the key of an entry in the cache.
+type EntryKey struct {
+ group string
+ PartID uint64
+ Offset uint64
+ segmentID segmentID
+ shardID common.ShardID
+}
+
+type entryIndex struct {
+ *entry
+ key EntryKey
+ index int
+}
+
+type entryIndexHeap []*entryIndex
+
+func (h entryIndexHeap) Len() int { return len(h) }
+
+func (h entryIndexHeap) Less(i, j int) bool {
+ return atomic.LoadUint64(&h[i].entry.lastAccess) <
atomic.LoadUint64(&h[j].entry.lastAccess)
+}
+
+func (h entryIndexHeap) Swap(i, j int) {
+ h[i], h[j] = h[j], h[i]
+ h[i].index = i
+ h[j].index = j
+}
+
+func (h *entryIndexHeap) Push(x interface{}) {
+ n := len(*h)
+ ei := x.(*entryIndex)
+ ei.index = n
+ *h = append(*h, ei)
+}
+
+func (h *entryIndexHeap) Pop() interface{} {
+ old := *h
+ n := len(old)
+ x := old[n-1]
+ *h = old[:n-1]
+ return x
+}
+
+// Cache stores the compressed primary blocks.
+type Cache struct {
+ entry map[EntryKey]*entry
+ entryIndex map[EntryKey]*entryIndex
+ entryIndexHeap *entryIndexHeap
+ stopCh chan struct{}
+ requests uint64
+ misses uint64
+ mu sync.RWMutex
+ wg sync.WaitGroup
+ maxCacheSize uint64
+ cleanupInterval time.Duration
+ idleTimeout time.Duration
+}
+
+// NewCache creates a cache.
+func NewCache() *Cache {
+ h := &entryIndexHeap{}
+ heap.Init(h)
+ c := &Cache{
+ entry: make(map[EntryKey]*entry),
+ entryIndexHeap: h,
+ entryIndex: make(map[EntryKey]*entryIndex),
+ stopCh: make(chan struct{}),
+ wg: sync.WaitGroup{},
+ maxCacheSize: 100 * 1024 * 1024,
+ cleanupInterval: 30 * time.Second,
+ idleTimeout: 2 * time.Minute,
Review Comment:
You should set them through flags.
##########
banyand/measure/query_test.go:
##########
Review Comment:
Could you please add a benchmark to compare the query performance w/o the
cache?
We can start with over 1,000 series, with each series containing 1,440 data
points (24 hours x 60 minutes). The critical test case is to query 10% of the
total series concurrently, focusing on the latest 30 data points.
##########
banyand/internal/storage/cache.go:
##########
@@ -0,0 +1,230 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storage
+
+import (
+ "container/heap"
+ "sync"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+)
+
+type entry struct {
+ value any
+ lastAccess uint64
+}
+
+// EntryKey is the key of an entry in the cache.
+type EntryKey struct {
+ group string
+ PartID uint64
+ Offset uint64
+ segmentID segmentID
+ shardID common.ShardID
+}
+
+type entryIndex struct {
+ *entry
+ key EntryKey
+ index int
+}
+
+type entryIndexHeap []*entryIndex
+
+func (h entryIndexHeap) Len() int { return len(h) }
+
+func (h entryIndexHeap) Less(i, j int) bool {
+ return atomic.LoadUint64(&h[i].entry.lastAccess) <
atomic.LoadUint64(&h[j].entry.lastAccess)
+}
+
+func (h entryIndexHeap) Swap(i, j int) {
+ h[i], h[j] = h[j], h[i]
+ h[i].index = i
+ h[j].index = j
+}
+
+func (h *entryIndexHeap) Push(x interface{}) {
+ n := len(*h)
+ ei := x.(*entryIndex)
+ ei.index = n
+ *h = append(*h, ei)
+}
+
+func (h *entryIndexHeap) Pop() interface{} {
+ old := *h
+ n := len(old)
+ x := old[n-1]
+ *h = old[:n-1]
+ return x
+}
+
+// Cache stores the compressed primary blocks.
+type Cache struct {
+ entry map[EntryKey]*entry
+ entryIndex map[EntryKey]*entryIndex
+ entryIndexHeap *entryIndexHeap
+ stopCh chan struct{}
+ requests uint64
+ misses uint64
+ mu sync.RWMutex
+ wg sync.WaitGroup
+ maxCacheSize uint64
+ cleanupInterval time.Duration
+ idleTimeout time.Duration
+}
+
+// NewCache creates a cache.
+func NewCache() *Cache {
+ h := &entryIndexHeap{}
+ heap.Init(h)
+ c := &Cache{
+ entry: make(map[EntryKey]*entry),
+ entryIndexHeap: h,
+ entryIndex: make(map[EntryKey]*entryIndex),
+ stopCh: make(chan struct{}),
+ wg: sync.WaitGroup{},
+ maxCacheSize: 100 * 1024 * 1024,
+ cleanupInterval: 30 * time.Second,
+ idleTimeout: 2 * time.Minute,
+ }
+ c.wg.Add(1)
+ return c
+}
+
+// Clean periodically cleans the cache.
+func (c *Cache) Clean() {
+ go func() {
+ defer c.wg.Done()
+ c.clean()
+ }()
+}
+
+func (c *Cache) clean() {
+ ticker := time.NewTicker(c.cleanupInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ now := uint64(time.Now().UnixNano())
+ c.mu.Lock()
+ for key, entry := range c.entry {
+ if now-atomic.LoadUint64(&entry.lastAccess) >
uint64(c.idleTimeout.Nanoseconds()) {
+ delete(c.entry, key)
+ heap.Remove(c.entryIndexHeap,
c.entryIndex[key].index)
+ }
+ }
+ c.mu.Unlock()
+ case <-c.stopCh:
+ return
+ }
+ }
+}
+
+// Close closes the cache.
+func (c *Cache) Close() {
+ close(c.stopCh)
+ c.wg.Wait()
+ c.entry = nil
+ c.entryIndex = nil
+ c.entryIndexHeap = nil
+}
+
+// Get gets the compressed primary block from the cache.
+func (c *Cache) Get(key EntryKey) any {
+ atomic.AddUint64(&c.requests, 1)
+
+ c.mu.RLock()
+ entry := c.entry[key]
+ c.mu.RUnlock()
+
+ if entry != nil {
+ now := uint64(time.Now().UnixNano())
+ if atomic.LoadUint64(&entry.lastAccess) != now {
+ c.mu.Lock()
+ atomic.StoreUint64(&entry.lastAccess, now)
+ if ei := c.entryIndex[key]; ei != nil {
+ heap.Fix(c.entryIndexHeap, ei.index)
+ }
+ c.mu.Unlock()
+ }
+ return entry.value
+ }
+
+ atomic.AddUint64(&c.misses, 1)
+ return nil
+}
+
+// Put puts the compressed primary block into the cache.
+func (c *Cache) Put(key EntryKey, value any) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ for c.size() > c.maxCacheSize && c.len() > 0 {
+ ei := heap.Pop(c.entryIndexHeap).(*entryIndex)
+ delete(c.entry, ei.key)
+ delete(c.entryIndex, ei.key)
+ }
+
+ now := uint64(time.Now().UnixNano())
+ e := &entry{
+ value: value,
+ lastAccess: now,
+ }
+ ei := &entryIndex{
+ key: key,
+ entry: e,
+ }
+ c.entry[key] = e
+ c.entryIndex[key] = ei
+ heap.Push(c.entryIndexHeap, ei)
+}
+
+// Requests returns the number of cache requests.
+func (c *Cache) Requests() uint64 {
+ return atomic.LoadUint64(&c.requests)
+}
+
+// Misses returns the number of cache misses.
+func (c *Cache) Misses() uint64 {
+ return atomic.LoadUint64(&c.misses)
+}
+
+// Len returns the number of entries in the cache.
+func (c *Cache) Len() uint64 {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ return c.len()
+}
+
+func (c *Cache) len() uint64 {
+ return uint64(len(c.entry))
+}
+
+// Size returns the size of the cache.
+func (c *Cache) Size() uint64 {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ return c.size()
+}
Review Comment:
Expose these metrics through the observability module.
##########
banyand/measure/part_iter.go:
##########
@@ -192,6 +208,14 @@ func (pi *partIter) readPrimaryBlock(bms []blockMetadata,
mr *primaryBlockMetada
if err != nil {
return nil, fmt.Errorf("cannot unmarshal index block: %w", err)
}
+ bmPtrs := make([]*blockMetadata, 0)
Review Comment:
```suggestion
bmPtrs := make([]*blockMetadata, 0, len(bms))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
