[#1268] blobstor: Cleanup zstd encoders/decoders
Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
parent
4253931699
commit
ad92493b86
3 changed files with 25 additions and 8 deletions
|
@ -78,6 +78,8 @@ type blobovniczas struct {
|
|||
// list of active (opened, non-filled) blobovniczas
|
||||
activeMtx sync.RWMutex
|
||||
active map[string]blobovniczaWithIndex
|
||||
|
||||
onClose []func()
|
||||
}
|
||||
|
||||
type blobovniczaWithIndex struct {
|
||||
|
@ -800,14 +802,21 @@ func (b *blobovniczas) updateAndGet(p string, old *uint64) (blobovniczaWithIndex
|
|||
func (b *blobovniczas) init() error {
|
||||
b.log.Debug("initializing Blobovnicza's")
|
||||
|
||||
zstdC, err := zstdCompressor()
|
||||
enc, zstdC, err := zstdCompressor()
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not create zstd compressor: %v", err)
|
||||
}
|
||||
zstdD, err := zstdDecompressor()
|
||||
b.onClose = append(b.onClose, func() {
|
||||
if err := enc.Close(); err != nil {
|
||||
b.log.Debug("can't close zstd compressor", zap.String("err", err.Error()))
|
||||
}
|
||||
})
|
||||
|
||||
dec, zstdD, err := zstdDecompressor()
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not create zstd decompressor: %v", err)
|
||||
}
|
||||
b.onClose = append(b.onClose, dec.Close)
|
||||
|
||||
// Compression is always done based on config settings.
|
||||
if b.compressionEnabled {
|
||||
|
@ -875,6 +884,10 @@ func (b *blobovniczas) close() error {
|
|||
|
||||
b.activeMtx.Unlock()
|
||||
|
||||
for i := range b.onClose {
|
||||
b.onClose[i]()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
@ -16,24 +16,24 @@ func noOpDecompressor(data []byte) ([]byte, error) {
|
|||
return data, nil
|
||||
}
|
||||
|
||||
func zstdCompressor() (func([]byte) []byte, error) {
|
||||
func zstdCompressor() (*zstd.Encoder, func([]byte) []byte, error) {
|
||||
enc, err := zstd.NewWriter(nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return func(data []byte) []byte {
|
||||
return enc, func(data []byte) []byte {
|
||||
return enc.EncodeAll(data, make([]byte, 0, len(data)))
|
||||
}, nil
|
||||
}
|
||||
|
||||
func zstdDecompressor() (func([]byte) ([]byte, error), error) {
|
||||
func zstdDecompressor() (*zstd.Decoder, func([]byte) ([]byte, error), error) {
|
||||
dec, err := zstd.NewReader(nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return func(data []byte) ([]byte, error) {
|
||||
return dec, func(data []byte) ([]byte, error) {
|
||||
return dec.DecodeAll(data, nil)
|
||||
}, nil
|
||||
}
|
||||
|
|
|
@ -161,6 +161,10 @@ func TestIterate_IgnoreErrors(t *testing.T) {
|
|||
require.NotEqual(t, "", p, "expected to not have at least 1 blobovnicza in cache")
|
||||
require.NoError(t, os.Chmod(p, 0))
|
||||
|
||||
require.NoError(t, b.Close())
|
||||
require.NoError(t, bs.Open())
|
||||
require.NoError(t, bs.Init())
|
||||
|
||||
var prm IteratePrm
|
||||
prm.SetIterationHandler(func(e IterationElement) error {
|
||||
return nil
|
||||
|
|
Loading…
Reference in a new issue