d.zayakin/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#1818] writecache: Reuse FSTree flushing code between flushes

Browse source 
Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>
This commit is contained in:
Evgenii Stratonikov 2022-10-18 17:57:50 +03:00 committed by fyrchik
parent 6e2f7e291d
commit b580846630
1 changed files with 55 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
pkg/local_object_storage/writecache/flush.go
View file

@ -137,7 +137,18 @@ func (c *cache) flushBigObjects() {
break break
} }
_ = c.flushFSTree(true)
c.modeMtx.RUnlock()
case <-c.closeCh:
return
}
}
}
func (c *cache) flushFSTree(ignoreErrors bool) error {
var prm common.IteratePrm var prm common.IteratePrm
prm.IgnoreErrors = ignoreErrors
prm.LazyHandler = func(addr oid.Address, f func() ([]byte, error)) error { prm.LazyHandler = func(addr oid.Address, f func() ([]byte, error)) error {
sAddr := addr.EncodeToString() sAddr := addr.EncodeToString()
@ -147,28 +158,35 @@ func (c *cache) flushBigObjects() {
data, err := f() data, err := f()
if err != nil { if err != nil {
if ignoreErrors {
c.log.Error("can't read a file", zap.Stringer("address", addr)) c.log.Error("can't read a file", zap.Stringer("address", addr))
return nil return nil
} }
return err
}
c.mtx.Lock() var obj object.Object
_, compress := c.compressFlags[sAddr] err = obj.Unmarshal(data)
c.mtx.Unlock() if err != nil {
if ignoreErrors {
c.log.Error("can't unmarshal an object", zap.Stringer("address", addr))
return nil
}
return err
}
var prm common.PutPrm var prm common.PutPrm
prm.Address = addr prm.Address = addr
prm.Object = &obj
prm.RawData = data prm.RawData = data
prm.DontCompress = !compress
if _, err := c.blobstor.Put(prm); err != nil { _, err = c.blobstor.Put(prm)
if err != nil {
if ignoreErrors {
c.log.Error("cant flush object to blobstor", zap.Error(err)) c.log.Error("cant flush object to blobstor", zap.Error(err))
return nil return nil
} }
return err
if compress {
c.mtx.Lock()
delete(c.compressFlags, sAddr)
c.mtx.Unlock()
} }
// mark object as flushed // mark object as flushed
@ -177,13 +195,8 @@ func (c *cache) flushBigObjects() {
return nil return nil
} }
_, _ = c.fsTree.Iterate(prm) _, err := c.fsTree.Iterate(prm)
return err
c.modeMtx.RUnlock()
case <-c.closeCh:
return
}
}
} }
// flushWorker writes objects to the main storage. // flushWorker writes objects to the main storage.
@ -241,36 +254,7 @@ func (c *cache) Flush(ignoreErrors bool) error {
} }
func (c *cache) flush(ignoreErrors bool) error { func (c *cache) flush(ignoreErrors bool) error {
var prm common.IteratePrm if err := c.flushFSTree(ignoreErrors); err != nil {
prm.IgnoreErrors = ignoreErrors
prm.LazyHandler = func(addr oid.Address, f func() ([]byte, error)) error {
_, ok := c.flushed.Peek(addr.EncodeToString())
if ok {
return nil
}
data, err := f()
if err != nil {
if ignoreErrors {
return nil
}
return err
}
var obj object.Object
err = obj.Unmarshal(data)
if err != nil {
if ignoreErrors {
return nil
}
return err
}
return c.flushObject(&obj)
}
_, err := c.fsTree.Iterate(prm)
if err != nil {
return err return err
} }