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 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.

Author: williamhbaker

broker/codecs/codecs.go

@@ -41,7 +41,7 @@ func NewCodecWriter(w io.Writer, codec pb.CompressionCodec) (Compressor, error)
 	case pb.CompressionCodec_NONE:
 		return nopWriteCloser{w}, nil
 	case pb.CompressionCodec_GZIP, pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION:
-		return gzip.NewWriter(w), nil
+		return &GzipBatcher{w: w}, nil
 	case pb.CompressionCodec_SNAPPY:
 		return snappy.NewBufferedWriter(w), nil
 	case pb.CompressionCodec_ZSTANDARD:
@@ -51,6 +51,33 @@ func NewCodecWriter(w io.Writer, codec pb.CompressionCodec) (Compressor, error)
 	}
 }
 
+// GzipBatcher allows for batching multiple writes into a single gzip member,
+// concatenated per RFC 1952. Members are terminated by calling Close, with a
+// new gzip writer initialized on the next Write.
+type GzipBatcher struct {
+	w  io.Writer
+	gz *gzip.Writer
+}
+
+func (gzb *GzipBatcher) Write(p []byte) (n int, err error) {
+	if gzb.gz == nil {
+		gzb.gz = gzip.NewWriter(gzb.w)
+	}
+
+	return gzb.gz.Write(p)
+}
+
+func (gzb *GzipBatcher) 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
+	// 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,7 @@ 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 {
+		if primary && s.CompressionCodec != pb.CompressionCodec_NONE && int(r.Proposal.End-s.compressedTo) >= compressionBatchSize {
 			s.compressThrough(r.Proposal.End)
 		}
 		s.Fragment.Fragment = *r.Proposal
@@ -233,6 +240,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 +249,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.compressedTo
 
 		if _, err = io.CopyBuffer(s.compressor, io.NewSectionReader(s.File, offset, delta), buf); err == nil {
+			if s.CompressionCodec == pb.CompressionCodec_GZIP || s.CompressionCodec == pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION {
+				err = s.compressor.Close()
+			}
+		}
+		if err == nil {
+			s.compressedTo = end
 			return // Done.
 		}
 		err = fmt.Errorf("while incrementally compressing: %s", err)
@@ -281,7 +295,15 @@ func (s *Spool) compressThrough(end int64) {
 			s.compressor = nil
 			continue
 		}
+		if s.CompressionCodec == pb.CompressionCodec_GZIP || s.CompressionCodec == pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION {
+			if err = s.compressor.Close(); err != nil {
+				err = fmt.Errorf("flushing gzip batch compressor: %s", err)
+				s.compressor = nil
+				continue
+			}
+		}
 
+		s.compressedTo = end
 		break // Success.
 	}
 }

broker/fragment/spool_test.go

@@ -1,6 +1,7 @@
 package fragment
 
 import (
+	"bytes"
 	"errors"
 	"io"
 	"testing"
@@ -117,6 +118,102 @@ 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(&pb.ReplicateRequest{
+		Proposal: &pb.Fragment{
+			Journal:          "a/journal",
+			Begin:            0,
+			End:              0,
+			CompressionCodec: pb.CompressionCodec_GZIP,
+		},
+		Registers: &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, and the fourth commit triggers
+	// a third member.
+	for _, req := range []pb.ReplicateRequest{
+		{Content: []byte("first write ")},
+		{Content: []byte("second ")},
+		{Content: []byte("third ")},
+		{Content: []byte("fourth write ")},
+	} {
+		var resp, err = spool.Apply(&req, true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+		var proposal = spool.Next()
+		resp, err = spool.Apply(&pb.ReplicateRequest{Proposal: &proposal}, true)
+		require.NoError(t, err)
+		require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+	}
+
+	// Complete the spool.
+	resp, err = spool.Apply(&pb.ReplicateRequest{Proposal: &pb.Fragment{
+		Journal:          "a/journal",
+		Begin:            12 + 7 + 6 + 13,
+		End:              12 + 7 + 6 + 13,
+		CompressionCodec: pb.CompressionCodec_GZIP,
+	},
+		Registers: &regEmpty,
+	}, true)
+	require.NoError(t, err)
+	require.Equal(t, pb.ReplicateResponse{Status: pb.Status_OK}, resp)
+
+	require.Len(t, obv.commits, 4)
+	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 "
+	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, 3)
+	require.Equal(t, "first write ", parts[0])
+	require.Equal(t, "second third ", parts[1])
+	require.Equal(t, "fourth write ", parts[2])
+}
+
 func TestRejectRollBeforeCurrentEnd(t *testing.T) {
 	var obv testSpoolObserver
 	var spool = NewSpool("a/journal", &obv)