From 8c746a914ac917a16b0ab3098e70fda7e15dcbb0 Mon Sep 17 00:00:00 2001 From: Dmitrii Stepanov Date: Mon, 14 Apr 2025 16:18:08 +0300 Subject: [PATCH] [#1715] compression: Decouple Config and Compressor Refactoring. Change-Id: Ide2e1378f30c39045d4bacd13a902331bd4f764f Signed-off-by: Dmitrii Stepanov --- .../blobstor/blobovniczatree/blobovnicza.go | 4 ++-- .../blobstor/blobovniczatree/option.go | 2 +- pkg/local_object_storage/blobstor/blobstor.go | 4 ++-- .../blobstor/common/storage.go | 4 ++-- .../blobstor/compression/bench_test.go | 16 +++++++++----- .../blobstor/compression/compress.go | 22 +++++++++++-------- .../blobstor/fstree/fstree.go | 18 +++++++-------- .../blobstor/memstore/control.go | 4 ++-- .../blobstor/memstore/option.go | 2 +- .../blobstor/teststore/option.go | 8 +++---- .../blobstor/teststore/teststore.go | 4 ++-- .../writecache/writecache.go | 2 +- 12 files changed, 49 insertions(+), 41 deletions(-) diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go index d9e99d0d13..3e8b9f07b6 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go @@ -158,11 +158,11 @@ func (b *Blobovniczas) Path() string { } // SetCompressor implements common.Storage. -func (b *Blobovniczas) SetCompressor(cc *compression.Config) { +func (b *Blobovniczas) SetCompressor(cc *compression.Compressor) { b.compression = cc } -func (b *Blobovniczas) Compressor() *compression.Config { +func (b *Blobovniczas) Compressor() *compression.Compressor { return b.compression } diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/option.go b/pkg/local_object_storage/blobstor/blobovniczatree/option.go index 0e1b2022e9..2f6d31b4ea 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/option.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/option.go @@ -19,7 +19,7 @@ type cfg struct { openedCacheSize int blzShallowDepth uint64 blzShallowWidth uint64 - compression *compression.Config + compression *compression.Compressor blzOpts []blobovnicza.Option reportError func(context.Context, string, error) // reportError is the function called when encountering disk errors. metrics Metrics diff --git a/pkg/local_object_storage/blobstor/blobstor.go b/pkg/local_object_storage/blobstor/blobstor.go index 7a9568aff9..cf67c6fe94 100644 --- a/pkg/local_object_storage/blobstor/blobstor.go +++ b/pkg/local_object_storage/blobstor/blobstor.go @@ -41,7 +41,7 @@ type SubStorageInfo struct { type Option func(*cfg) type cfg struct { - compression compression.Config + compression compression.Compressor log *logger.Logger storage []SubStorage metrics Metrics @@ -158,6 +158,6 @@ func WithMetrics(m Metrics) Option { } } -func (b *BlobStor) Compressor() *compression.Config { +func (b *BlobStor) Compressor() *compression.Compressor { return &b.compression } diff --git a/pkg/local_object_storage/blobstor/common/storage.go b/pkg/local_object_storage/blobstor/common/storage.go index 6ecef48cdc..e35c35e602 100644 --- a/pkg/local_object_storage/blobstor/common/storage.go +++ b/pkg/local_object_storage/blobstor/common/storage.go @@ -18,8 +18,8 @@ type Storage interface { Path() string ObjectsCount(ctx context.Context) (uint64, error) - SetCompressor(cc *compression.Config) - Compressor() *compression.Config + SetCompressor(cc *compression.Compressor) + Compressor() *compression.Compressor // SetReportErrorFunc allows to provide a function to be called on disk errors. // This function MUST be called before Open. diff --git a/pkg/local_object_storage/blobstor/compression/bench_test.go b/pkg/local_object_storage/blobstor/compression/bench_test.go index 9f70f8ec2a..445a0494bc 100644 --- a/pkg/local_object_storage/blobstor/compression/bench_test.go +++ b/pkg/local_object_storage/blobstor/compression/bench_test.go @@ -11,7 +11,7 @@ import ( ) func BenchmarkCompression(b *testing.B) { - c := Config{Enabled: true} + c := Compressor{Config: Config{Enabled: true}} require.NoError(b, c.Init()) for _, size := range []int{128, 1024, 32 * 1024, 32 * 1024 * 1024} { @@ -33,7 +33,7 @@ func BenchmarkCompression(b *testing.B) { } } -func benchWith(b *testing.B, c Config, data []byte) { +func benchWith(b *testing.B, c Compressor, data []byte) { b.ResetTimer() b.ReportAllocs() for range b.N { @@ -56,8 +56,10 @@ func BenchmarkCompressionRealVSEstimate(b *testing.B) { b.Run("estimate", func(b *testing.B) { b.ResetTimer() - c := &Config{ - Enabled: true, + c := &Compressor{ + Config: Config{ + Enabled: true, + }, } require.NoError(b, c.Init()) @@ -76,8 +78,10 @@ func BenchmarkCompressionRealVSEstimate(b *testing.B) { b.Run("compress", func(b *testing.B) { b.ResetTimer() - c := &Config{ - Enabled: true, + c := &Compressor{ + Config: Config{ + Enabled: true, + }, } require.NoError(b, c.Init()) diff --git a/pkg/local_object_storage/blobstor/compression/compress.go b/pkg/local_object_storage/blobstor/compression/compress.go index 8a86b69826..efe84ea2a3 100644 --- a/pkg/local_object_storage/blobstor/compression/compress.go +++ b/pkg/local_object_storage/blobstor/compression/compress.go @@ -19,6 +19,13 @@ const ( LevelSmallestSize Level = "smallest_size" ) +type Compressor struct { + Config + + encoder *zstd.Encoder + decoder *zstd.Decoder +} + // Config represents common compression-related configuration. type Config struct { Enabled bool @@ -27,9 +34,6 @@ type Config struct { UseCompressEstimation bool CompressEstimationThreshold float64 - - encoder *zstd.Encoder - decoder *zstd.Decoder } // zstdFrameMagic contains first 4 bytes of any compressed object @@ -37,7 +41,7 @@ type Config struct { var zstdFrameMagic = []byte{0x28, 0xb5, 0x2f, 0xfd} // Init initializes compression routines. -func (c *Config) Init() error { +func (c *Compressor) Init() error { var err error if c.Enabled { @@ -84,7 +88,7 @@ func (c *Config) NeedsCompression(obj *objectSDK.Object) bool { // Decompress decompresses data if it starts with the magic // and returns data untouched otherwise. -func (c *Config) Decompress(data []byte) ([]byte, error) { +func (c *Compressor) Decompress(data []byte) ([]byte, error) { if len(data) < 4 || !bytes.Equal(data[:4], zstdFrameMagic) { return data, nil } @@ -93,7 +97,7 @@ func (c *Config) Decompress(data []byte) ([]byte, error) { // Compress compresses data if compression is enabled // and returns data untouched otherwise. -func (c *Config) Compress(data []byte) []byte { +func (c *Compressor) Compress(data []byte) []byte { if c == nil || !c.Enabled { return data } @@ -107,7 +111,7 @@ func (c *Config) Compress(data []byte) []byte { return c.compress(data) } -func (c *Config) compress(data []byte) []byte { +func (c *Compressor) compress(data []byte) []byte { maxSize := c.encoder.MaxEncodedSize(len(data)) compressed := c.encoder.EncodeAll(data, make([]byte, 0, maxSize)) if len(data) < len(compressed) { @@ -117,7 +121,7 @@ func (c *Config) compress(data []byte) []byte { } // Close closes encoder and decoder, returns any error occurred. -func (c *Config) Close() error { +func (c *Compressor) Close() error { var err error if c.encoder != nil { err = c.encoder.Close() @@ -135,7 +139,7 @@ func (c *Config) HasValidCompressionLevel() bool { c.Level == LevelSmallestSize } -func (c *Config) compressionLevel() zstd.EncoderLevel { +func (c *Compressor) compressionLevel() zstd.EncoderLevel { switch c.Level { case LevelDefault, LevelOptimal: return zstd.SpeedDefault diff --git a/pkg/local_object_storage/blobstor/fstree/fstree.go b/pkg/local_object_storage/blobstor/fstree/fstree.go index 031b385b2d..112741ab41 100644 --- a/pkg/local_object_storage/blobstor/fstree/fstree.go +++ b/pkg/local_object_storage/blobstor/fstree/fstree.go @@ -45,7 +45,7 @@ type FSTree struct { log *logger.Logger - *compression.Config + compressor *compression.Compressor Depth uint64 DirNameLen int @@ -82,7 +82,7 @@ func New(opts ...Option) *FSTree { Permissions: 0o700, RootPath: "./", }, - Config: nil, + compressor: nil, Depth: 4, DirNameLen: DirNameLen, metrics: &noopMetrics{}, @@ -196,7 +196,7 @@ func (t *FSTree) iterate(ctx context.Context, depth uint64, curPath []string, pr } if err == nil { - data, err = t.Decompress(data) + data, err = t.compressor.Decompress(data) } if err != nil { if prm.IgnoreErrors { @@ -405,7 +405,7 @@ func (t *FSTree) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, err return common.PutRes{}, err } if !prm.DontCompress { - prm.RawData = t.Compress(prm.RawData) + prm.RawData = t.compressor.Compress(prm.RawData) } size = len(prm.RawData) @@ -448,7 +448,7 @@ func (t *FSTree) Get(ctx context.Context, prm common.GetPrm) (common.GetRes, err } } - data, err = t.Decompress(data) + data, err = t.compressor.Decompress(data) if err != nil { return common.GetRes{}, err } @@ -597,12 +597,12 @@ func (t *FSTree) Path() string { } // SetCompressor implements common.Storage. -func (t *FSTree) SetCompressor(cc *compression.Config) { - t.Config = cc +func (t *FSTree) SetCompressor(cc *compression.Compressor) { + t.compressor = cc } -func (t *FSTree) Compressor() *compression.Config { - return t.Config +func (t *FSTree) Compressor() *compression.Compressor { + return t.compressor } // SetReportErrorFunc implements common.Storage. diff --git a/pkg/local_object_storage/blobstor/memstore/control.go b/pkg/local_object_storage/blobstor/memstore/control.go index 95a916662d..3df96a1c36 100644 --- a/pkg/local_object_storage/blobstor/memstore/control.go +++ b/pkg/local_object_storage/blobstor/memstore/control.go @@ -16,7 +16,7 @@ func (s *memstoreImpl) Init() error func (s *memstoreImpl) Close(context.Context) error { return nil } func (s *memstoreImpl) Type() string { return Type } func (s *memstoreImpl) Path() string { return s.rootPath } -func (s *memstoreImpl) SetCompressor(cc *compression.Config) { s.compression = cc } -func (s *memstoreImpl) Compressor() *compression.Config { return s.compression } +func (s *memstoreImpl) SetCompressor(cc *compression.Compressor) { s.compression = cc } +func (s *memstoreImpl) Compressor() *compression.Compressor { return s.compression } func (s *memstoreImpl) SetReportErrorFunc(func(context.Context, string, error)) {} func (s *memstoreImpl) SetParentID(string) {} diff --git a/pkg/local_object_storage/blobstor/memstore/option.go b/pkg/local_object_storage/blobstor/memstore/option.go index 97a03993d5..7605af4e59 100644 --- a/pkg/local_object_storage/blobstor/memstore/option.go +++ b/pkg/local_object_storage/blobstor/memstore/option.go @@ -7,7 +7,7 @@ import ( type cfg struct { rootPath string readOnly bool - compression *compression.Config + compression *compression.Compressor } func defaultConfig() *cfg { diff --git a/pkg/local_object_storage/blobstor/teststore/option.go b/pkg/local_object_storage/blobstor/teststore/option.go index fb11887517..3a38ecf82f 100644 --- a/pkg/local_object_storage/blobstor/teststore/option.go +++ b/pkg/local_object_storage/blobstor/teststore/option.go @@ -17,8 +17,8 @@ type cfg struct { Type func() string Path func() string - SetCompressor func(cc *compression.Config) - Compressor func() *compression.Config + SetCompressor func(cc *compression.Compressor) + Compressor func() *compression.Compressor SetReportErrorFunc func(f func(context.Context, string, error)) Get func(common.GetPrm) (common.GetRes, error) @@ -45,11 +45,11 @@ func WithClose(f func() error) Option { return func(c *cfg) { c func WithType(f func() string) Option { return func(c *cfg) { c.overrides.Type = f } } func WithPath(f func() string) Option { return func(c *cfg) { c.overrides.Path = f } } -func WithSetCompressor(f func(*compression.Config)) Option { +func WithSetCompressor(f func(*compression.Compressor)) Option { return func(c *cfg) { c.overrides.SetCompressor = f } } -func WithCompressor(f func() *compression.Config) Option { +func WithCompressor(f func() *compression.Compressor) Option { return func(c *cfg) { c.overrides.Compressor = f } } diff --git a/pkg/local_object_storage/blobstor/teststore/teststore.go b/pkg/local_object_storage/blobstor/teststore/teststore.go index 626ba00231..190b6a8760 100644 --- a/pkg/local_object_storage/blobstor/teststore/teststore.go +++ b/pkg/local_object_storage/blobstor/teststore/teststore.go @@ -116,7 +116,7 @@ func (s *TestStore) Path() string { } } -func (s *TestStore) SetCompressor(cc *compression.Config) { +func (s *TestStore) SetCompressor(cc *compression.Compressor) { s.mu.RLock() defer s.mu.RUnlock() switch { @@ -129,7 +129,7 @@ func (s *TestStore) SetCompressor(cc *compression.Config) { } } -func (s *TestStore) Compressor() *compression.Config { +func (s *TestStore) Compressor() *compression.Compressor { s.mu.RLock() defer s.mu.RUnlock() switch { diff --git a/pkg/local_object_storage/writecache/writecache.go b/pkg/local_object_storage/writecache/writecache.go index 70b17eb8e6..7ed5113189 100644 --- a/pkg/local_object_storage/writecache/writecache.go +++ b/pkg/local_object_storage/writecache/writecache.go @@ -52,7 +52,7 @@ type Cache interface { // MainStorage is the interface of the underlying storage of Cache implementations. type MainStorage interface { - Compressor() *compression.Config + Compressor() *compression.Compressor Exists(context.Context, common.ExistsPrm) (common.ExistsRes, error) Put(context.Context, common.PutPrm) (common.PutRes, error) }