forked from TrueCloudLab/frostfs-node
[#1367] writecache: Drop bbolt DB
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
66e17f4b8e
commit
5f6c7cbdb1
12 changed files with 82 additions and 415 deletions
|
|
@ -4,6 +4,9 @@ import (
|
|||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
||||
|
|
@ -16,7 +19,6 @@ import (
|
|||
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
|
||||
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||
"github.com/mr-tron/base58"
|
||||
"go.etcd.io/bbolt"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
|
|
@ -24,10 +26,6 @@ import (
|
|||
)
|
||||
|
||||
const (
|
||||
// flushBatchSize is amount of keys which will be read from cache to be flushed
|
||||
// to the main storage. It is used to reduce contention between cache put
|
||||
// and cache persist.
|
||||
flushBatchSize = 512
|
||||
// defaultFlushWorkersCount is number of workers for putting objects in main storage.
|
||||
defaultFlushWorkersCount = 20
|
||||
// defaultFlushInterval is default time interval between successive flushes.
|
||||
|
|
@ -41,112 +39,11 @@ func (c *cache) runFlushLoop(ctx context.Context) {
|
|||
if c.disableBackgroundFlush {
|
||||
return
|
||||
}
|
||||
for range c.workersCount {
|
||||
c.wg.Add(1)
|
||||
go c.workerFlushSmall(ctx)
|
||||
}
|
||||
|
||||
c.wg.Add(1)
|
||||
go func() {
|
||||
c.workerFlushBig(ctx)
|
||||
c.wg.Done()
|
||||
}()
|
||||
|
||||
c.wg.Add(1)
|
||||
go func() {
|
||||
defer c.wg.Done()
|
||||
|
||||
tt := time.NewTimer(defaultFlushInterval)
|
||||
defer tt.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-tt.C:
|
||||
c.flushSmallObjects(ctx)
|
||||
tt.Reset(defaultFlushInterval)
|
||||
c.estimateCacheSize()
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (c *cache) flushSmallObjects(ctx context.Context) {
|
||||
var lastKey []byte
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
var m []objectInfo
|
||||
|
||||
c.modeMtx.RLock()
|
||||
if c.readOnly() {
|
||||
c.modeMtx.RUnlock()
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
|
||||
// We put objects in batches of fixed size to not interfere with main put cycle a lot.
|
||||
_ = c.db.View(func(tx *bbolt.Tx) error {
|
||||
b := tx.Bucket(defaultBucket)
|
||||
cs := b.Cursor()
|
||||
|
||||
var k, v []byte
|
||||
|
||||
if len(lastKey) == 0 {
|
||||
k, v = cs.First()
|
||||
} else {
|
||||
k, v = cs.Seek(lastKey)
|
||||
if bytes.Equal(k, lastKey) {
|
||||
k, v = cs.Next()
|
||||
}
|
||||
}
|
||||
|
||||
for ; k != nil && len(m) < flushBatchSize; k, v = cs.Next() {
|
||||
if len(lastKey) == len(k) {
|
||||
copy(lastKey, k)
|
||||
} else {
|
||||
lastKey = bytes.Clone(k)
|
||||
}
|
||||
|
||||
m = append(m, objectInfo{
|
||||
addr: string(k),
|
||||
data: bytes.Clone(v),
|
||||
})
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
var count int
|
||||
for i := range m {
|
||||
obj := objectSDK.New()
|
||||
if err := obj.Unmarshal(m[i].data); err != nil {
|
||||
continue
|
||||
}
|
||||
m[i].obj = obj
|
||||
|
||||
count++
|
||||
select {
|
||||
case c.flushCh <- m[i]:
|
||||
case <-ctx.Done():
|
||||
c.modeMtx.RUnlock()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
c.modeMtx.RUnlock()
|
||||
if count == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
c.log.Debug(logs.WritecacheTriedToFlushItemsFromWritecache,
|
||||
zap.Int("count", count),
|
||||
zap.String("start", base58.Encode(lastKey)))
|
||||
}
|
||||
}
|
||||
|
||||
func (c *cache) workerFlushBig(ctx context.Context) {
|
||||
|
|
@ -197,9 +94,6 @@ func (c *cache) flushFSTree(ctx context.Context, ignoreErrors bool) error {
|
|||
|
||||
err = c.flushObject(ctx, &obj, e.ObjectData, StorageTypeFSTree)
|
||||
if err != nil {
|
||||
if ignoreErrors {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -211,29 +105,6 @@ func (c *cache) flushFSTree(ctx context.Context, ignoreErrors bool) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// workerFlushSmall writes small objects to the main storage.
|
||||
func (c *cache) workerFlushSmall(ctx context.Context) {
|
||||
defer c.wg.Done()
|
||||
|
||||
var objInfo objectInfo
|
||||
for {
|
||||
// Give priority to direct put.
|
||||
select {
|
||||
case objInfo = <-c.flushCh:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
err := c.flushObject(ctx, objInfo.obj, objInfo.data, StorageTypeDB)
|
||||
if err != nil {
|
||||
// Error is handled in flushObject.
|
||||
continue
|
||||
}
|
||||
|
||||
c.deleteFromDB(objInfo.addr, true)
|
||||
}
|
||||
}
|
||||
|
||||
// flushObject is used to write object directly to the main storage.
|
||||
func (c *cache) flushObject(ctx context.Context, obj *objectSDK.Object, data []byte, st StorageType) error {
|
||||
var err error
|
||||
|
|
@ -300,13 +171,33 @@ func (c *cache) Flush(ctx context.Context, ignoreErrors, seal bool) error {
|
|||
}
|
||||
|
||||
func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
|
||||
if err := c.flushFSTree(ctx, ignoreErrors); err != nil {
|
||||
return err
|
||||
return c.flushFSTree(ctx, ignoreErrors)
|
||||
}
|
||||
|
||||
type batchItem struct {
|
||||
data []byte
|
||||
address string
|
||||
}
|
||||
|
||||
func (c *cache) flushAndDropBBoltDB(ctx context.Context) error {
|
||||
_, err := os.Stat(filepath.Join(c.path, dbName))
|
||||
if err != nil && os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not check write-cache database existence: %w", err)
|
||||
}
|
||||
db, err := OpenDB(c.path, true, os.OpenFile, c.pageSize)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not open write-cache database: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
_ = db.Close()
|
||||
}()
|
||||
|
||||
var last string
|
||||
for {
|
||||
batch, err := c.readNextDBBatch(ignoreErrors, last)
|
||||
batch, err := c.readNextDBBatch(db, last)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -316,32 +207,27 @@ func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
|
|||
for _, item := range batch {
|
||||
var obj objectSDK.Object
|
||||
if err := obj.Unmarshal(item.data); err != nil {
|
||||
c.reportFlushError(logs.FSTreeCantDecodeDBObjectAddress, item.address, metaerr.Wrap(err))
|
||||
if ignoreErrors {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
return fmt.Errorf("unmarshal object from database: %w", err)
|
||||
}
|
||||
|
||||
if err := c.flushObject(ctx, &obj, item.data, StorageTypeDB); err != nil {
|
||||
return err
|
||||
return fmt.Errorf("flush object from database: %w", err)
|
||||
}
|
||||
c.deleteFromDB(item.address, false)
|
||||
}
|
||||
last = batch[len(batch)-1].address
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
return fmt.Errorf("close write-cache database: %w", err)
|
||||
}
|
||||
if err := os.Remove(filepath.Join(c.path, dbName)); err != nil {
|
||||
return fmt.Errorf("remove write-cache database: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type batchItem struct {
|
||||
data []byte
|
||||
address string
|
||||
}
|
||||
|
||||
func (c *cache) readNextDBBatch(ignoreErrors bool, last string) ([]batchItem, error) {
|
||||
func (c *cache) readNextDBBatch(db *bbolt.DB, last string) ([]batchItem, error) {
|
||||
const batchSize = 100
|
||||
var batch []batchItem
|
||||
err := c.db.View(func(tx *bbolt.Tx) error {
|
||||
err := db.View(func(tx *bbolt.Tx) error {
|
||||
var addr oid.Address
|
||||
|
||||
b := tx.Bucket(defaultBucket)
|
||||
|
|
@ -352,11 +238,7 @@ func (c *cache) readNextDBBatch(ignoreErrors bool, last string) ([]batchItem, er
|
|||
continue
|
||||
}
|
||||
if err := addr.DecodeString(sa); err != nil {
|
||||
c.reportFlushError(logs.FSTreeCantDecodeDBObjectAddress, sa, metaerr.Wrap(err))
|
||||
if ignoreErrors {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
return fmt.Errorf("decode address from database: %w", err)
|
||||
}
|
||||
|
||||
batch = append(batch, batchItem{data: bytes.Clone(data), address: sa})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue