writecache: Fix small object flush #726
2 changed files with 12 additions and 9 deletions
|
@ -21,7 +21,7 @@ type cache struct {
|
|||
modeMtx sync.RWMutex
|
||||
|
||||
// flushCh is a channel with objects to flush.
|
||||
flushCh chan *objectSDK.Object
|
||||
flushCh chan objectInfo
|
||||
// scheduled4Flush contains objects scheduled for flush via flushCh
|
||||
// helps to avoid multiple flushing of one object
|
||||
scheduled4Flush map[oid.Address]struct{}
|
||||
|
@ -52,7 +52,7 @@ const (
|
|||
// New creates new writecache instance.
|
||||
func New(opts ...Option) writecache.Cache {
|
||||
c := &cache{
|
||||
flushCh: make(chan *objectSDK.Object),
|
||||
flushCh: make(chan objectInfo),
|
||||
mode: mode.ReadWrite,
|
||||
scheduled4Flush: map[oid.Address]struct{}{},
|
||||
|
||||
|
|
|
@ -85,7 +85,11 @@ func (c *collector) Send(buf *z.Buffer) error {
|
|||
c.cache.scheduled4FlushMtx.Unlock()
|
||||
c.scheduled++
|
||||
select {
|
||||
case c.cache.flushCh <- obj:
|
||||
case c.cache.flushCh <- objectInfo{
|
||||
addr: addr,
|
||||
data: val,
|
||||
obj: obj,
|
||||
}:
|
||||
case <-c.cache.closeCh:
|
||||
c.cancel()
|
||||
return nil
|
||||
|
@ -175,22 +179,21 @@ func (c *cache) reportFlushError(msg string, addr string, err error) {
|
|||
func (c *cache) workerFlushSmall() {
|
||||
defer c.wg.Done()
|
||||
|
||||
var obj *objectSDK.Object
|
||||
var objInfo objectInfo
|
||||
for {
|
||||
// Give priority to direct put.
|
||||
select {
|
||||
case obj = <-c.flushCh:
|
||||
case objInfo = <-c.flushCh:
|
||||
case <-c.closeCh:
|
||||
return
|
||||
}
|
||||
|
||||
addr := objectCore.AddressOf(obj)
|
||||
err := c.flushObject(context.TODO(), obj, nil, writecache.StorageTypeDB)
|
||||
err := c.flushObject(context.TODO(), objInfo.obj, objInfo.data, writecache.StorageTypeDB)
|
||||
if err == nil {
|
||||
c.deleteFromDB([]internalKey{addr2key(addr)})
|
||||
c.deleteFromDB([]internalKey{addr2key(objInfo.addr)})
|
||||
}
|
||||
c.scheduled4FlushMtx.Lock()
|
||||
delete(c.scheduled4Flush, addr)
|
||||
delete(c.scheduled4Flush, objInfo.addr)
|
||||
c.scheduled4FlushMtx.Unlock()
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue