broker: multiple members for gzip compression

Each active gzip writer introduces a small but significant amount of memory
overhead, on the order of several hundred KB. When there are many active
journals being written, this can add up to a large amount of memory usage.

This adds a threshold where incremental compression will occur only if there is
at least 1 MB of data to compress, and creates a new gzip writing mechanism that
allows closing & creating new gzip members, concatenated into the same output
file. The spool logic uses this mechanism to create a new gzip member for every
batch of incremental compression, eliminating the need to hold a gzip writer in
memory for the entire lifetime of the fragment file.

This change only applies to standard gzip compression with client-side
decompression. If decompression offloading is used, gzip files will continue to
be written in a single stream, as there are issues with some object stores
truncating multi-member gzip content after the first member.

Author: williamhbaker

broker/codecs/codecs.go

@@ -40,7 +40,9 @@ func NewCodecWriter(w io.Writer, codec pb.CompressionCodec) (Compressor, error)
 	switch codec {
 	case pb.CompressionCodec_NONE:
 		return nopWriteCloser{w}, nil
-	case pb.CompressionCodec_GZIP, pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION:
+	case pb.CompressionCodec_GZIP:
+		return &GzipMultiMemberWriter{w: w}, nil
+	case pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION:
 		return gzip.NewWriter(w), nil
 	case pb.CompressionCodec_SNAPPY:
 		return snappy.NewBufferedWriter(w), nil
@@ -51,6 +53,34 @@ func NewCodecWriter(w io.Writer, codec pb.CompressionCodec) (Compressor, error)
 	}
 }
 
+// GzipMultiMemberWriter allows for batching multiple writes into a single gzip
+// member, which are concatenated into a single file per RFC 1952. Members are
+// terminated by calling Close, with a new gzip writer initialized on the next
+// Write.
+type GzipMultiMemberWriter struct {
+	w  io.Writer
+	gz *gzip.Writer
+}
+
+func (gzb *GzipMultiMemberWriter) Write(p []byte) (n int, err error) {
+	if gzb.gz == nil {
+		gzb.gz = gzip.NewWriter(gzb.w)
+	}
+
+	return gzb.gz.Write(p)
+}
+
+func (gzb *GzipMultiMemberWriter) Close() error {
+	if gzb.gz == nil {
+		return nil
+	} else if err := gzb.gz.Close(); err != nil {
+		return err
+	}
+	gzb.gz = nil
+
+	return nil
+}
+
 type nopWriteCloser struct{ io.Writer }
 
 func (nopWriteCloser) Close() error { return nil }

broker/fragment/spool.go

@@ -14,6 +14,11 @@ import (
 	pb "go.gazette.dev/core/broker/protocol"
 )
 
+// Minimum size of accumulated uncompressed data before performing incremental
+// compression. Useful for GZIP, which creates a new member per compression
+// invocation.
+var compressionBatchSize = 1024 * 1024
+
 // Spool is a Fragment which is in the process of being created, backed by a
 // local *os.File. As commits occur and the file extent is updated, the Spool
 // Fragment is also updated to reflect the new committed extent. At all
@@ -31,6 +36,8 @@ type Spool struct {
 	// Length of compressed content written to |compressedFile|. Set only after
 	// the compressor is finalized.
 	compressedLength int64
+	// Spool file offset through which content has been compressed.
+	compressedTo int64
 	// Compressor of |compressedFile|.
 	compressor codecs.Compressor
 
@@ -119,7 +126,7 @@ func (s *Spool) applyCommit(r *pb.ReplicateRequest, primary bool) pb.ReplicateRe
 	if r.Proposal.End > s.Fragment.End+s.delta ||
 		(r.Proposal.End == s.Fragment.End && r.Proposal.ContentLength() == 0) {
 
-		if s.compressor != nil {
+		if s.compressor != nil || primary && s.CompressionCodec != pb.CompressionCodec_NONE {
 			s.finishCompression()
 		}
 		if s.ContentLength() != 0 {
@@ -166,7 +173,8 @@ func (s *Spool) applyCommit(r *pb.ReplicateRequest, primary bool) pb.ReplicateRe
 		spoolCommitsTotal.Inc()
 		spoolCommitBytesTotal.Add(float64(s.delta))
 
-		if primary && s.CompressionCodec != pb.CompressionCodec_NONE {
+		var uncompressedBytes = (r.Proposal.End - s.Fragment.Begin) - s.compressedTo
+		if primary && s.CompressionCodec != pb.CompressionCodec_NONE && int(uncompressedBytes) >= compressionBatchSize {
 			s.compressThrough(r.Proposal.End)
 		}
 		s.Fragment.Fragment = *r.Proposal
@@ -233,6 +241,7 @@ func (s *Spool) compressThrough(end int64) {
 	if s.CompressionCodec == pb.CompressionCodec_NONE {
 		panic("expected CompressionCodec != NONE")
 	}
+
 	var err error
 
 	var buf = bufferPool.Get().([]byte)
@@ -241,9 +250,15 @@ func (s *Spool) compressThrough(end int64) {
 	// Garden path: we've already compressed all content of the current Fragment,
 	// and now incrementally compress through |end|.
 	if s.compressor != nil {
-		var offset, delta = s.Fragment.ContentLength(), end - s.Fragment.End
+		var offset, delta = s.compressedTo, (end - s.Fragment.Begin) - s.compressedTo
 
 		if _, err = io.CopyBuffer(s.compressor, io.NewSectionReader(s.File, offset, delta), buf); err == nil {
+			if s.CompressionCodec == pb.CompressionCodec_GZIP {
+				err = s.compressor.Close()
+			}
+		}
+		if err == nil {
+			s.compressedTo = end - s.Fragment.Begin
 			return // Done.
 		}
 		err = fmt.Errorf("while incrementally compressing: %s", err)
@@ -281,7 +296,15 @@ func (s *Spool) compressThrough(end int64) {
 			s.compressor = nil
 			continue
 		}
+		if s.CompressionCodec == pb.CompressionCodec_GZIP {
+			if err = s.compressor.Close(); err != nil {
+				err = fmt.Errorf("flushing gzip batch compressor: %s", err)
+				s.compressor = nil
+				continue
+			}
+		}
 
+		s.compressedTo = end - s.Fragment.Begin
 		break // Success.
 	}
 }
@@ -294,9 +317,13 @@ func (s *Spool) finishCompression() {
 	}
 	var err error
 
-	if s.compressor == nil {
+	if s.compressedTo < s.Fragment.End-s.Fragment.Begin {
 		s.compressThrough(s.Fragment.End)
+	} else if s.compressor == nil {
+		// Empty fragment.
+		return
 	}
+
 	for {
 		if err != nil {
 			log.WithField("err", err).Error("failed to finishCompression (will retry)")

broker/fragment/spool_test.go

@@ -1,6 +1,7 @@
 package fragment
 
 import (
+	"bytes"
 	"errors"
 	"io"
 	"testing"
@@ -117,6 +118,166 @@ func TestCompressionNotPrimary(t *testing.T) {
 		contentString(t, obv.completes[0], pb.CompressionCodec_GZIP))
 }
 
+func TestGzipBatcherMultipleMembers(t *testing.T) {
+	var origBatchSize = compressionBatchSize
+	compressionBatchSize = 10
+	defer func() { compressionBatchSize = origBatchSize }()
+
+	var obv testSpoolObserver
+	var spool = NewSpool("a/journal", &obv)
+
+	var resp, err = spool.Apply(newProposal(pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            0,
+		End:              0,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	}, regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.Status_OK, resp.Status)
+
+	// Commit some data. The compression batch size has been artificially
+	// lowered so the first commit is compressed, the second commit is buffered,
+	// the third commit triggers a second member, the fourth commit triggers
+	// a third member, and the fifth commit is smaller than compressionBatchSize
+	// and will not trigger compression until spool completion.
+	for _, req := range []pb.ReplicateRequest{
+		{Content: []byte("first write ")},
+		{Content: []byte("second ")},
+		{Content: []byte("third ")},
+		{Content: []byte("fourth write ")},
+		{Content: []byte("final")},
+	} {
+		var resp, err = spool.Apply(&req, true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+		resp, err = spool.Apply(newProposal(spool.Next(), regEmpty), true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+	}
+
+	// Complete the spool.
+	resp, err = spool.Apply(newProposal(pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            spool.Fragment.End,
+		End:              spool.Fragment.End,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	}, regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+	require.Len(t, obv.commits, 5)
+	require.Len(t, obv.completes, 1)
+	require.NotNil(t, obv.completes[0].compressedFile)
+	require.NotEqual(t, int64(0), obv.completes[0].compressedLength)
+
+	var expected = "first write second third fourth write final"
+	var actual = contentString(t, obv.completes[0], pb.CompressionCodec_GZIP)
+	require.Equal(t, expected, actual)
+
+	// Decompress and verify each member.
+	var parts []string
+	var compressedData = make([]byte, obv.completes[0].compressedLength)
+	_, err = obv.completes[0].compressedFile.ReadAt(compressedData, 0)
+	require.NoError(t, err)
+	var gzipHeader = []byte{0x1f, 0x8b}
+	for start := bytes.Index(compressedData, gzipHeader); start != -1; {
+		var next = bytes.Index(compressedData[start+len(gzipHeader):], gzipHeader)
+		var end int
+		if next == -1 {
+			end = len(compressedData)
+		} else {
+			end = start + len(gzipHeader) + next
+		}
+
+		var memberData = compressedData[start:end]
+		var reader, readerErr = codecs.NewCodecReader(bytes.NewReader(memberData), pb.CompressionCodec_GZIP)
+		require.NoError(t, readerErr)
+		var decompressed, readErr = io.ReadAll(reader)
+		require.NoError(t, readErr)
+		require.NoError(t, reader.Close())
+
+		parts = append(parts, string(decompressed))
+		if next == -1 {
+			break
+		}
+		start = end
+	}
+
+	require.Len(t, parts, 4)
+	require.Equal(t, "first write ", parts[0])
+	require.Equal(t, "second third ", parts[1])
+	require.Equal(t, "fourth write ", parts[2])
+	require.Equal(t, "final", parts[3])
+
+	// Now test that compression offset tracking works correctly across spool
+	// rolls by verifying that multiple commits written to the new spool can be
+	// read.
+	for _, req := range []pb.ReplicateRequest{
+		{Content: []byte("abc")},
+		{Content: []byte("xyz")},
+	} {
+		var resp, err = spool.Apply(&req, true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+		resp, err = spool.Apply(newProposal(spool.Next(), regEmpty), true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+	}
+
+	// Complete the second spool.
+	resp, err = spool.Apply(newProposal(pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            spool.Fragment.End,
+		End:              spool.Fragment.End,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	}, regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.Status_OK, resp.Status)
+
+	require.Len(t, obv.completes, 2)
+	require.Equal(t, "abcxyz", contentString(t, obv.completes[1], pb.CompressionCodec_GZIP))
+
+	// Test case where the first and only write is below the compression
+	// threshold, which will require initializing compression of the spool in
+	// finishCompression.
+	resp, err = spool.Apply(&pb.ReplicateRequest{Content: []byte("small")}, true)
+	require.NoError(t, err)
+	require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+	resp, err = spool.Apply(newProposal(spool.Next(), regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+	// Complete the third spool.
+	resp, err = spool.Apply(newProposal(pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            spool.Fragment.End,
+		End:              spool.Fragment.End,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	}, regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.Status_OK, resp.Status)
+
+	require.Len(t, obv.completes, 3)
+	require.Equal(t, "small", contentString(t, obv.completes[2], pb.CompressionCodec_GZIP))
+
+	// Test case where an empty fragment with compression is completed.
+	// This should not panic due to nil compressor. Empty fragments don't
+	// trigger completion events (no ContentLength), so completes stays at 3.
+	resp, err = spool.Apply(newProposal(pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            spool.Fragment.End,
+		End:              spool.Fragment.End,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	}, regEmpty), true)
+	require.NoError(t, err)
+	require.Equal(t, pb.Status_OK, resp.Status)
+
+	require.Len(t, obv.completes, 3)
+}
+
 func TestRejectRollBeforeCurrentEnd(t *testing.T) {
 	var obv testSpoolObserver
 	var spool = NewSpool("a/journal", &obv)