Repository: kopia/kopia
Pull request #3695 (2024-02-29 18:19:15)

fix(providers): Reduce memory allocation (gc thrash)
This change causes Minio libraries to reduce memory allocations.  Specifically, buffer allocations for S3 upload.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
Attention: Patch coverage is `79.48718%` with `8 lines` in your changes are missing coverage. Please review.
> Project coverage is 76.95%. Comparing base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) to head [(`7cc06b5`)](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).
> Report is 35 commits behind head on master.

| [Files](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) | Patch % | Lines |
|---|---|---|
| [internal/gather/gather\_bytes.go](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvZ2F0aGVyL2dhdGhlcl9ieXRlcy5nbw==) | 79.48% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3695      +/-   ##
==========================================
+ Coverage   75.86%   76.95%   +1.08%     
==========================================
  Files         470      471       +1     
  Lines       37301    28652    -8649     
==========================================
- Hits        28299    22048    -6251     
+ Misses       7071     4678    -2393     
+ Partials     1931     1926       -5     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/3695?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).

jkowalski: this looks good but the code is very tricky and needs unit tests before we can merge
aaron-kasten: > this looks good but the code is very tricky and needs unit tests before we can merge

@jkowalski I'm unsure of how to unit test this.
aaron-kasten: <img width="625" alt="image" src="https://github.com/kopia/kopia/assets/55999015/b8f655e3-d446-4918-be34-f02a2580a953">

This is an example of the memory allocation savings for a 5GB snapshot - approximately 3x the memory of the snapshot size in this case.
aaron-kasten: @redgoat650 Lets chat later.  Test coverage has been enlarged to deal with corner cases not for Kopia ReaderAt
aaron-kasten: Im unclear on how the test cases you mention are specific to this implementation vs the Buffer implementation in general.  The thinking is that anything that applies to the original Buffer should be implemented (as its already been accepted).
aaron-kasten: Are you proposing that additional coverage be introduced for the original Buffer implementation? 
redgoat650: Ah! I think I missed that all of the `ReaderAt` cases I mentioned are probably being tested by this line
```go
	require.NoError(t, iotest.TestReader(reader, buf))
```
Is that accurate, @aaron-kasten? If so, that's great. Perhaps we can add a comment to draw attention to it? It was easy to miss when reading through the test case.
aaron-kasten: A majority, at least for compliance with the Go apis. The scatter-gather buffer in Kopia is a different beast though.
Im wondering if it needs additional tests ... which your comments imply ... then it might be worth while expanding them.
Its later on a friday for me.  Ill take a closer look at the code and your comments on Monday and see what can be done.
aaron-kasten: Thank you for your detailed observations ... its a big help having something so specific to work from.
aaron-kasten: > Ah! I think I missed that all of the `ReaderAt` cases I mentioned are probably being tested by this line
> 
> ```go
> 	require.NoError(t, iotest.TestReader(reader, buf))
> ```
> 
> Is that accurate, @aaron-kasten? If so, that's great. Perhaps we can add a comment to draw attention to it? It was easy to miss when reading through the test, and I thought we were missing the obvious cases 😄
> 
> [edit to add the `TestReader` description]:
> 
> ```go
> // TestReader tests that reading from r returns the expected file content.
> // It does reads of different sizes, until EOF.
> // If r implements [io.ReaderAt] or [io.Seeker], TestReader also checks
> // that those operations behave as they should.
> //
> // If TestReader finds any misbehaviors, it returns an error reporting them.
> // The error text may span multiple lines.
> ```

@redgoat650 In the future can you submit this as a change - ideally inline as a comment with a change or a change directly to the branch?  It saves on risking my interpretation of your intended meaning and saves steps (Its implied to be approved by you and does not need to go through another review cycle).

Shrekster: Added keep-open, label. I'd like to test this more thoroughly. Do not merge yet.
Shrekster: > Added keep-open, label. I'd like to test this more thoroughly. Do not merge yet.

done with testing, should be good now; fixed some slice index increment bugs
julio-lopez: I also triggered the storage provider tests. Please keep an eye on those before merging.

https://github.com/kopia/kopia/actions/runs/8504413695/job/23291280771?pr=3695

aaron-kasten: @julio-lopez (@Shrekster ) got an error in the code-coverage test: `EPIPE`.  Something happened when trying to upload coverage results (?).

Have either of you seen this before?  Is there a standard resolution?

```
gpg: directory '/home/runner/.gnupg' created
gpg: keybox '/home/runner/.gnupg/pubring.kbx' created
gpg: /home/runner/.gnupg/trustdb.gpg: trustdb created
gpg: key 806BB28AED779869: public key "Codecov Uploader (Codecov Uploader Verification Key) <security@codecov.io>" imported
gpg: Total number processed: 1
gpg:               imported: 1

Error: write EPIPE
    at afterWriteDispatched (node:internal/stream_base_commons:160:15)
    at writeGeneric (node:internal/stream_base_commons:151:3)
    at Socket._writeGeneric (node:net:952:11)
    at Socket._write (node:net:964:8)
    at writeOrBuffer (node:internal/streams/writable:447:12)
    at _write (node:internal/streams/writable:389:10)
    at Socket.Writable.end (node:internal/streams/writable:665:17)
    at Socket.end (node:net:722:31)
    at module.exports (/home/runner/work/_actions/codecov/codecov-action/54bcd8715eee62d40e33596ef5e8f0f48dbbccab/node_modules/gpg/lib/spawnGPG.js:50:1)
    at Object.call (/home/runner/work/_actions/codecov/codecov-action/54bcd8715eee62d40e33596ef5e8f0f48dbbccab/node_modules/gpg/lib/gpg.js:28:1)
```
aaron-kasten: @Shrekster @jkowalski @julio-lopez I made some final changes. Please take a look.
julio-lopez: > @julio-lopez (@Shrekster ) got an error in the code-coverage test: `EPIPE`. Something happened when trying to upload coverage results (?).

Likely an infra issue on their side?? dunno. 🤷🏼 

Patch:
diff --git a/internal/gather/gather_bytes.go b/internal/gather/gather_bytes.go
index 5caf11a1..9b5c9f58 100644
--- a/internal/gather/gather_bytes.go
+++ b/internal/gather/gather_bytes.go
@@ -10,8 +10,12 @@
 	"github.com/pkg/errors"
 )
 
-//nolint:gochecknoglobals
-var invalidSliceBuf = []byte(uuid.NewString())
+var (
+	//nolint:gochecknoglobals
+	invalidSliceBuf = []byte(uuid.NewString())
+	// ErrInvalidOffset checkable error for supplying an invalid offset.
+	ErrInvalidOffset = errors.Errorf("invalid offset")
+)
 
 // Bytes represents a sequence of bytes split into slices.
 type Bytes struct {
@@ -44,13 +48,13 @@ func (b *Bytes) AppendSectionTo(w io.Writer, offset, size int) error {
 	// find the index of starting slice
 	sliceNdx := -1
 
-	for i, p := range b.Slices {
-		if offset < len(p) {
+	for i, bs := range b.Slices {
+		if offset < len(bs) {
 			sliceNdx = i
 			break
 		}
 
-		offset -= len(p)
+		offset -= len(bs)
 	}
 
 	// not found
@@ -120,6 +124,85 @@ type bytesReadSeekCloser struct {
 	offset int
 }
 
+func (b *bytesReadSeekCloser) ReadAt(bs []byte, off int64) (int, error) {
+	b.b.assertValid()
+	// cache "b.b.Slices" - slice parameters will stay constant for duration of
+	// function.  Locking is left to the calling function
+	slices := b.b.Slices
+
+	// source data that is read will be written to w, the buffer backed by p.
+	offset := off
+
+	maxBsIndex := len(bs)
+
+	// negative offsets result in an error
+	if offset < 0 {
+		return 0, ErrInvalidOffset
+	}
+
+	sliceNdx := -1
+
+	// find the index of starting slice
+	for i, slicesBuf := range slices {
+		if offset < int64(len(slicesBuf)) {
+			sliceNdx = i
+			break
+		}
+
+		// update offset to be relative to the sliceNdx slice
+		offset -= int64(len(slicesBuf))
+	}
+
+	// no slice found if sliceNdx is still negative
+	if sliceNdx == -1 {
+		// return no bytes read if the buffer has no length
+		if maxBsIndex == 0 {
+			return 0, nil
+		}
+
+		return 0, io.EOF
+	}
+
+	// save off our working slice as curSlice
+	curSlice := slices[sliceNdx]
+
+	// copy the requested bytes from curSlice into bs (reader output)
+	m := copy(bs, curSlice[offset:])
+	// accounting: keep track of total number of bytes written in n and
+	// number of bytes written from the current slice in m
+	n := m
+
+	// move on to next and then check if all slices were consumed
+	sliceNdx++
+
+	slicesN := len(slices)
+
+	// while there are more bytes to read (maxBsIndex > n) and there are more
+	// slices left (sliceNdx < slicesN)
+	for maxBsIndex > n && sliceNdx < slicesN {
+		// get a new working slice
+		curSlice = slices[sliceNdx]
+
+		// copy what we can from the current slice into our destination.
+		// (no need to keep track of offset within curSlice)
+		m = copy(bs[n:], curSlice)
+		// keep track of total number of bytes written in n and
+		// number of bytes written from the current slice in m
+		n += m
+
+		// move on to next and then check if all slices were consumed
+		sliceNdx++
+	}
+
+	// if we have run out of slices but the input buffer is still not
+	// consumed completely then it means we have hit an EOF
+	if sliceNdx == slicesN && m == len(curSlice) {
+		return n, io.EOF
+	}
+
+	return n, nil
+}
+
 func (b *bytesReadSeekCloser) Close() error {
 	return nil
 }


Test patch:
diff --git a/internal/gather/gather_bytes_test.go b/internal/gather/gather_bytes_test.go
index c80cd026..44cabfaf 100644
--- a/internal/gather/gather_bytes_test.go
+++ b/internal/gather/gather_bytes_test.go
@@ -2,7 +2,9 @@
 
 import (
 	"bytes"
+	"fmt"
 	"io"
+	"math"
 	"testing"
 	"testing/iotest"
 
@@ -161,6 +163,13 @@ func TestGatherBytesReadSeeker(t *testing.T) {
 	reader := tmp.inner.Reader()
 	defer reader.Close() //nolint:errcheck
 
+	// TestReader tests that reading from r returns the expected file content.
+	// It does reads of different sizes, until EOF.
+	// If r implements [io.ReaderAt] or [io.Seeker], TestReader also checks
+	// that those operations behave as they should.
+	//
+	// If TestReader finds any misbehaviors, it returns an error reporting them.
+	// The error text may span multiple lines.
 	require.NoError(t, iotest.TestReader(reader, buf))
 
 	_, err := reader.Seek(-3, io.SeekStart)
@@ -173,6 +182,156 @@ func TestGatherBytesReadSeeker(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestGatherBytesReaderAtErrorResponses(t *testing.T) {
+	// 3.7 times the internal chunk size
+	contentBuf := make([]byte, int(float64(defaultAllocator.chunkSize)*3.7))
+	for i := range contentBuf {
+		contentBuf[i] = uint8(i % math.MaxUint8)
+	}
+
+	tcs := []struct {
+		inBsLen   int
+		inOff     int64
+		expectErr error
+		expectN   int
+	}{
+		{
+			inBsLen:   1 << 10,
+			inOff:     -1,
+			expectErr: ErrInvalidOffset,
+			expectN:   0,
+		},
+		{
+			inBsLen:   1 << 10,
+			inOff:     math.MaxInt64,
+			expectErr: io.EOF,
+			expectN:   0,
+		},
+		{
+			inBsLen:   0,
+			inOff:     -1,
+			expectErr: ErrInvalidOffset,
+			expectN:   0,
+		},
+		{
+			inBsLen: 0,
+			inOff:   math.MaxInt64,
+			expectN: 0,
+		},
+	}
+	for i, tc := range tcs {
+		t.Run(fmt.Sprintf("%d: %d %d %d", i, tc.inBsLen, tc.inOff, tc.expectN), func(t *testing.T) {
+			// tmp is an empty buffer that will supply some bytes
+			// for testing
+			var wrt WriteBuffer
+			defer wrt.Close()
+
+			wrt.Append(contentBuf)
+			require.Equalf(t, defaultAllocator.chunkSize, wrt.alloc.chunkSize,
+				"this test expects that the default-allocator will be used, but we are using: %#v", wrt.alloc)
+
+			// get the reader out of the WriteBuffer so we can read what was written
+			// (presume all 0s)
+			reader := wrt.inner.Reader()
+			defer reader.Close() //nolint:errcheck
+
+			// get the reader as a ReaderAt
+			readerAt := reader.(io.ReaderAt)
+
+			// make an output buffer of the required length
+			bs := make([]byte, tc.inBsLen)
+
+			n, err := readerAt.ReadAt(bs, tc.inOff)
+			require.ErrorIs(t, err, tc.expectErr)
+			require.Equal(t, tc.expectN, n)
+		})
+	}
+}
+
+func TestGatherBytesReaderAtVariableInputBufferSizes(t *testing.T) {
+	const inputBufferMaxMultiplier = 4.0 // maximum number of times the internal chunk size
+
+	contentBuf := make([]byte, defaultAllocator.chunkSize*inputBufferMaxMultiplier)
+	for i := range contentBuf {
+		contentBuf[i] = uint8(i % math.MaxUint8)
+	}
+
+	type testCase struct {
+		name            string
+		inputBufferSize int
+	}
+
+	// Test some interesting input buffer sizes from a 1-byte buffer to many
+	// multiples of the internal allocator chunk size.
+	testCases := []testCase{
+		{"1", 1},
+		{"0.5x", int(0.5 * float64(defaultAllocator.chunkSize))},
+
+		{"x-1", defaultAllocator.chunkSize - 1},
+		{"x", defaultAllocator.chunkSize},
+		{"x+1", defaultAllocator.chunkSize + 1},
+		{"1.5x", int(1.5 * float64(defaultAllocator.chunkSize))},
+
+		{"2x-1", 2*defaultAllocator.chunkSize - 1},
+		{"2x", 2 * defaultAllocator.chunkSize},
+		{"2x+1", 2*defaultAllocator.chunkSize + 1},
+		{"2.5x", int(2.5 * float64(defaultAllocator.chunkSize))},
+
+		{"3x-1", 3*defaultAllocator.chunkSize - 1},
+		{"3x", 3 * defaultAllocator.chunkSize},
+		{"3x+1", 3*defaultAllocator.chunkSize + 1},
+
+		{"4x-1", 4*defaultAllocator.chunkSize - 1},
+		{"4x", 4 * defaultAllocator.chunkSize},
+	}
+
+	// Test the third buffer slice. The idea here is to exercise the part of
+	// the buffer ReaderAt implementation where it has a longer buffer size
+	// than the size of the internal chunks of the buffer implementation. When
+	// we do this, the ReaderAt is forced to draw more data than it actually
+	// can from the first slice it found after searching for the current
+	// pointer in read cycle. Finally, it should increment the read index
+	// correctly.
+	//
+	// x.1 ... x.9
+	for chunkSizeMultiplier := inputBufferMaxMultiplier - 0.9; chunkSizeMultiplier < inputBufferMaxMultiplier; chunkSizeMultiplier += 0.1 {
+		testCases = append(testCases, testCase{
+			fmt.Sprintf("%.1fx", chunkSizeMultiplier),
+			int(float64(defaultAllocator.chunkSize) * chunkSizeMultiplier),
+		},
+		)
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// each test should have its own writer because t.Run() can be
+			// parallelized
+			var preWrt WriteBuffer
+			defer preWrt.Close()
+
+			// assert some preconditions that the reader conforms to ReaderAt
+			buf := contentBuf[:tc.inputBufferSize]
+
+			// write the generated data
+			n, err := preWrt.Write(buf)
+			require.NoErrorf(t, err, "Write() faiiled, inputBufferSize: %8", tc.inputBufferSize)
+			require.Equalf(t, defaultAllocator.chunkSize, preWrt.alloc.chunkSize,
+				"this test expects that the default-allocator will be used, but we are using: %#v", preWrt.alloc)
+
+			require.Lenf(t, buf, n, "unexpected size of data written, inputBufferSize: %d", tc.inputBufferSize)
+
+			// get the reader out of the WriteBuffer so we can read what was written
+			preRdr := preWrt.inner.Reader()
+			_, ok := preRdr.(io.ReaderAt)
+			require.Truef(t, ok, "MUST implement io.ReaderAt, inputBufferSize: %d", tc.inputBufferSize)
+
+			// execute standard ReadAt tests.
+			require.NoErrorf(t, iotest.TestReader(preRdr, buf),
+				"iotest failed, inputBufferSize: %d", tc.inputBufferSize)
+		})
+	}
+}
+
 func TestGatherBytesPanicsOnClose(t *testing.T) {
 	var tmp WriteBuffer