diff --git a/go.mod b/go.mod index a8cce3b..4e138c3 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/bodgit/sevenzip -go 1.25.0 +go 1.25.5 require ( github.com/andybalholm/brotli v1.2.2 @@ -12,7 +12,7 @@ require ( github.com/spf13/afero v1.15.0 github.com/stangelandcl/ppmd v0.1.1 github.com/stretchr/testify v1.11.1 - github.com/ulikunitz/xz v0.5.15 + github.com/unxed/xz v0.1.8 go4.org v0.0.0-20260112195520-a5071408f32f golang.org/x/sync v0.22.0 golang.org/x/text v0.40.0 diff --git a/go.sum b/go.sum index 0881f04..b161002 100644 --- a/go.sum +++ b/go.sum @@ -29,8 +29,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/ulikunitz/xz v0.5.15 h1:9DNdB5s+SgV3bQ2ApL10xRc35ck0DuIX/isZvIk+ubY= -github.com/ulikunitz/xz v0.5.15/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14= +github.com/unxed/xz v0.1.8 h1:0ETuOJkpMtk1dX9/C+1a+HGDgdzAHa/PiJqhyyecKcw= +github.com/unxed/xz v0.1.8/go.mod h1:+i9oua9YYxBJGNLh9s4itPLtCpCXs8X7M8/SNU4FeZc= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= go4.org v0.0.0-20260112195520-a5071408f32f h1:ziUVAjmTPwQMBmYR1tbdRFJPtTcQUI12fH9QQjfb0Sw= diff --git a/internal/lzma/reader.go b/internal/lzma/reader.go index b8d277c..548d9a6 100644 --- a/internal/lzma/reader.go +++ b/internal/lzma/reader.go @@ -8,7 +8,7 @@ import ( "fmt" "io" - "github.com/ulikunitz/xz/lzma" + "github.com/unxed/xz/lzma" ) type readCloser struct { @@ -26,12 +26,20 @@ func (rc *readCloser) Close() error { return errAlreadyClosed } - if err := rc.c.Close(); err != nil { - return fmt.Errorf("lzma: error closing: %w", err) + var errs []error + // Important: close the lzma.Reader to release buffers to the pool + if closer, ok := rc.r.(io.Closer); ok { + errs = append(errs, closer.Close()) } + errs = append(errs, rc.c.Close()) + rc.c, rc.r = nil, nil + if err := errors.Join(errs...); err != nil { + return fmt.Errorf("lzma: error closing: %w", err) + } + return nil } diff --git a/internal/lzma2/reader.go b/internal/lzma2/reader.go index 59cca7b..746f5b9 100644 --- a/internal/lzma2/reader.go +++ b/internal/lzma2/reader.go @@ -6,9 +6,34 @@ import ( "fmt" "io" - "github.com/ulikunitz/xz/lzma" + "github.com/unxed/xz" + "github.com/unxed/xz/lzma" ) +type seekReaderAt interface { + io.ReaderAt + io.Seeker +} + +func streamSizeBySeeking(s io.Seeker) (int64, error) { + curr, err := s.Seek(0, io.SeekCurrent) + if err != nil { + return 0, fmt.Errorf("seek current: %w", err) + } + + size, err := s.Seek(0, io.SeekEnd) + if err != nil { + return 0, fmt.Errorf("seek end: %w", err) + } + + _, err = s.Seek(curr, io.SeekStart) + if err != nil { + return size, fmt.Errorf("seek start: %w", err) + } + + return size, nil +} + type readCloser struct { c io.Closer r io.Reader @@ -26,12 +51,21 @@ func (rc *readCloser) Close() error { return errAlreadyClosed } - if err := rc.c.Close(); err != nil { - return fmt.Errorf("lzma2: error closing: %w", err) + var errs []error + + // We close the reader from the xz library to return the buffers to the pool and stop the goroutines + if closer, ok := rc.r.(io.Closer); ok { + errs = append(errs, closer.Close()) } + errs = append(errs, rc.c.Close()) + rc.c, rc.r = nil, nil + if err := errors.Join(errs...); err != nil { + return fmt.Errorf("lzma2: error closing: %w", err) + } + return nil } @@ -48,6 +82,45 @@ func (rc *readCloser) Read(p []byte) (int, error) { return n, err } +func tryParallelReader(config lzma.Reader2Config, readers []io.ReadCloser) (io.ReadCloser, bool) { + sra, ok := readers[0].(seekReaderAt) + if !ok { + return nil, false + } + + currentOffset, err := sra.Seek(0, io.SeekCurrent) + if err != nil { + return nil, false + } + + size, err := streamSizeBySeeking(sra) + if err != nil { + return nil, false + } + + var rAt io.ReaderAt = sra + + streamSize := size + + if currentOffset > 0 { + rAt = io.NewSectionReader(sra, currentOffset, size-currentOffset) + streamSize = size - currentOffset + } + + // Use the parallel reader from github.com/unxed/xz + pconfig := xz.ReaderConfig{DictCap: config.DictCap} + + pr, err := pconfig.NewParallelReader(rAt, streamSize) + if err != nil { + return nil, false + } + + return &readCloser{ + c: readers[0], + r: pr, + }, true +} + // NewReader returns a new LZMA2 io.ReadCloser. func NewReader(p []byte, _ uint64, readers []io.ReadCloser) (io.ReadCloser, error) { if len(readers) != 1 { @@ -70,6 +143,11 @@ func NewReader(p []byte, _ uint64, readers []io.ReadCloser) (io.ReadCloser, erro return nil, fmt.Errorf("lzma2: error verifying config: %w", err) } + // Try parallel decompression if the input is seekable + if pr, ok := tryParallelReader(config, readers); ok { + return pr, nil + } + lr, err := config.NewReader2(readers[0]) if err != nil { return nil, fmt.Errorf("lzma2: error creating reader: %w", err) diff --git a/internal/lzma2/reader_test.go b/internal/lzma2/reader_test.go index b6a743e..13f3ecb 100644 --- a/internal/lzma2/reader_test.go +++ b/internal/lzma2/reader_test.go @@ -42,3 +42,124 @@ func TestNewReader(t *testing.T) { t.Errorf("unexpected errInvalidProperties for valid property, got %v", err) } } + +type mockSeekReaderAt struct { + io.Reader + seekStartFail bool + seekCurrentFail bool + seekEndFail bool + currentOffset int64 +} + +func (m mockSeekReaderAt) ReadAt(_ []byte, _ int64) (int, error) { + return 0, io.EOF +} + +func (m mockSeekReaderAt) Seek(_ int64, whence int) (int64, error) { + if whence == io.SeekStart && m.seekStartFail { + return 0, errors.New("seek start failed") + } + + if whence == io.SeekCurrent { + if m.seekCurrentFail { + return 0, errors.New("seek current failed") + } + + return m.currentOffset, nil + } + + if whence == io.SeekEnd { + if m.seekEndFail { + return 0, errors.New("seek end failed") + } + + return 100, nil // Simulate the presence of an end for the ParallelReader trigger + } + + return 0, nil +} + +func TestNewReader_InterfaceLogic(t *testing.T) { + // 1. Test with the standard Reader (should select NewReader2) + p := []byte{0} // Properties + r1 := dummyReadCloser{bytes.NewReader([]byte{0, 0, 0, 0, 0})} + + rc1, err := NewReader(p, 0, []io.ReadCloser{r1}) + if err != nil { + t.Fatalf("Failed to create basic reader: %v", err) + } + + _ = rc1.Close() + + // 2. Test with Seeker (should attempt to launch ParallelReader and roll back) + r2 := dummyReadCloser{mockSeekReaderAt{Reader: bytes.NewReader([]byte{0, 0, 0, 0, 0})}} + + rc2, err := NewReader(p, 0, []io.ReadCloser{r2}) + if err != nil { + t.Fatalf("Failed to create seekable reader: %v", err) + } + + // We verify that Close() does not panic after all attempts and rollbacks. + if err := rc2.Close(); err != nil { + t.Errorf("Close failed: %v", err) + } +} + +func TestStreamSizeBySeeking_Errors(t *testing.T) { + s := mockSeekReaderAt{Reader: bytes.NewReader(nil), seekCurrentFail: true} + + _, err := streamSizeBySeeking(s) + if err == nil { + t.Error("expected error for seek current failure") + } + + s = mockSeekReaderAt{Reader: bytes.NewReader(nil), seekEndFail: true} + + _, err = streamSizeBySeeking(s) + if err == nil { + t.Error("expected error for seek end failure") + } + + s = mockSeekReaderAt{Reader: bytes.NewReader(nil), seekStartFail: true} + + _, err = streamSizeBySeeking(s) + if err == nil { + t.Error("expected error for seek start failure") + } +} + +func TestTryParallelReader_Offsets(t *testing.T) { + p := []byte{0} + r2 := dummyReadCloser{mockSeekReaderAt{Reader: bytes.NewReader([]byte{0, 0, 0, 0, 0}), currentOffset: 10}} + + rc2, err := NewReader(p, 0, []io.ReadCloser{r2}) + if err != nil { + t.Fatalf("Failed to create seekable reader: %v", err) + } + + _ = rc2.Close() +} + +func TestTryParallelReader_Errors(t *testing.T) { + p := []byte{0} + + // Test failure at SeekCurrent (should fallback to NewReader2 and succeed) + r1 := dummyReadCloser{mockSeekReaderAt{Reader: bytes.NewReader([]byte{0, 0, 0, 0, 0}), seekCurrentFail: true}} + + rc1, err := NewReader(p, 0, []io.ReadCloser{r1}) + if err != nil { + t.Fatalf("Failed to create reader despite SeekCurrent error: %v", err) + } + + _ = rc1.Close() + + // Test failure at streamSizeBySeeking (should fallback to NewReader2 and succeed) + r2 := dummyReadCloser{mockSeekReaderAt{Reader: bytes.NewReader([]byte{0, 0, 0, 0, 0}), seekEndFail: true}} + + rc2, err := NewReader(p, 0, []io.ReadCloser{r2}) + if err != nil { + t.Fatalf("Failed to create reader despite SeekEnd error: %v", err) + } + + _ = rc2.Close() +}