112 lines
3.5 KiB
Go
112 lines
3.5 KiB
Go
package store
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/lbryio/lbry.go/v2/extras/errors"
|
|
"github.com/lbryio/lbry.go/v2/stream"
|
|
"github.com/lbryio/reflector.go/shared"
|
|
log "github.com/sirupsen/logrus"
|
|
|
|
"github.com/lbryio/reflector.go/internal/metrics"
|
|
)
|
|
|
|
// CachingStore combines two stores, typically a local and a remote store, to improve performance.
|
|
// Accessed blobs are stored in and retrieved from the cache. If they are not in the cache, they
|
|
// are retrieved from the origin and cached. Puts are cached and also forwarded to the origin.
|
|
type CachingStore struct {
|
|
origin, cache BlobStore
|
|
component string
|
|
}
|
|
|
|
// NewCachingStore makes a new caching disk store and returns a pointer to it.
|
|
func NewCachingStore(component string, origin, cache BlobStore) *CachingStore {
|
|
return &CachingStore{
|
|
component: component,
|
|
origin: WithSingleFlight(component, origin),
|
|
cache: WithSingleFlight(component, cache),
|
|
}
|
|
}
|
|
|
|
const nameCaching = "caching"
|
|
|
|
// Name is the cache type name
|
|
func (c *CachingStore) Name() string { return nameCaching }
|
|
|
|
// Has checks the cache and then the origin for a hash. It returns true if either store has it.
|
|
func (c *CachingStore) Has(hash string) (bool, error) {
|
|
has, err := c.cache.Has(hash)
|
|
if has || err != nil {
|
|
return has, err
|
|
}
|
|
return c.origin.Has(hash)
|
|
}
|
|
|
|
// Get tries to get the blob from the cache first, falling back to the origin. If the blob comes
|
|
// from the origin, it is also stored in the cache.
|
|
func (c *CachingStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
|
|
start := time.Now()
|
|
blob, trace, err := c.cache.Get(hash)
|
|
if err == nil || !errors.Is(err, ErrBlobNotFound) {
|
|
metrics.CacheHitCount.With(metrics.CacheLabels(c.cache.Name(), c.component)).Inc()
|
|
rate := float64(len(blob)) / 1024 / 1024 / time.Since(start).Seconds()
|
|
metrics.CacheRetrievalSpeed.With(map[string]string{
|
|
metrics.LabelCacheType: c.cache.Name(),
|
|
metrics.LabelComponent: c.component,
|
|
metrics.LabelSource: "cache",
|
|
}).Set(rate)
|
|
return blob, trace.Stack(time.Since(start), c.Name()), err
|
|
}
|
|
|
|
metrics.CacheMissCount.With(metrics.CacheLabels(c.cache.Name(), c.component)).Inc()
|
|
|
|
blob, trace, err = c.origin.Get(hash)
|
|
if err != nil {
|
|
return nil, trace.Stack(time.Since(start), c.Name()), err
|
|
}
|
|
// there is no need to wait for the blob to be stored before we return it
|
|
// TODO: however this should be refactored to limit the amount of routines that the process can spawn to avoid a possible DoS
|
|
metrics.RoutinesQueue.WithLabelValues("store", "cache-put").Inc()
|
|
go func() {
|
|
defer metrics.RoutinesQueue.WithLabelValues("store", "cache-put").Dec()
|
|
err = c.cache.Put(hash, blob)
|
|
if err != nil {
|
|
log.Errorf("error saving blob to underlying cache: %s", errors.FullTrace(err))
|
|
}
|
|
}()
|
|
return blob, trace.Stack(time.Since(start), c.Name()), nil
|
|
}
|
|
|
|
// Put stores the blob in the origin and the cache
|
|
func (c *CachingStore) Put(hash string, blob stream.Blob) error {
|
|
err := c.origin.Put(hash, blob)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.cache.Put(hash, blob)
|
|
}
|
|
|
|
// PutSD stores the sd blob in the origin and the cache
|
|
func (c *CachingStore) PutSD(hash string, blob stream.Blob) error {
|
|
err := c.origin.PutSD(hash, blob)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.cache.PutSD(hash, blob)
|
|
}
|
|
|
|
// Delete deletes the blob from the origin and the cache
|
|
func (c *CachingStore) Delete(hash string) error {
|
|
err := c.origin.Delete(hash)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.cache.Delete(hash)
|
|
}
|
|
|
|
// Shutdown shuts down the store gracefully
|
|
func (c *CachingStore) Shutdown() {
|
|
c.origin.Shutdown()
|
|
c.cache.Shutdown()
|
|
return
|
|
}
|