perf(recrawl): divide the zstd encoders across the writers

ccrawl/capture.go

@@ -231,6 +231,60 @@ type CaptureWriter struct {
 	tmp, path string
 }
 
+// Where the memory of a wide run actually is.
+//
+// A heap profile of a 96 worker run on server2, taken while the encoders still
+// held their buffers, says 750 MB in use and puts 54 percent of it inside zstd:
+// 304 MB in ensureHist, 98 MB in encoderOptions.encoder and 5 MB in blockEnc,
+// against 124 MB of Parquet column buffers and 136 MB of response bodies. That
+// is not the queue, which the same run reported peaking at 25.7 MB of its 134.2
+// MB bound, and it is not the rows.
+//
+// zstd allocates that state per concurrent encoder, and the codec asked for
+// four of them, and every writer has its own codec. So the encoder state is
+// writers times concurrency, and at 6 writers that is 24 encoders holding
+// something like 1.2 GB before a single page is in a buffer. It is most of what
+// killed the 288 worker run.
+//
+// The concurrency was worth having when there was one writer, since it is the
+// only parallelism a single encoder has. It stopped being worth having when
+// --writers arrived, because the parts already compress in parallel and this
+// multiplies against them rather than adding to them. So it is divided by the
+// writer count now, and a run with one writer keeps exactly what it had.
+
+// defaultCodecConcurrency is what a single writer uses, and it is the number
+// that was there before the writers were split out.
+const defaultCodecConcurrency = 4
+
+// codecConcurrency is how many zstd encoders one part opens when the run has n
+// parts. It is a division rather than a cap because the cost is the product:
+// four encoders across six writers is the same twenty four states whatever any
+// single writer thinks it is doing, and the machine has six cores.
+//
+// It floors at one rather than at zero, since a codec with no encoders does not
+// compress, and one encoder per part at six parts is still six going at once.
+func codecConcurrency(n int) int {
+	if n < 1 {
+		n = 1
+	}
+	return max(1, defaultCodecConcurrency/n)
+}
+
+// SetCodecConcurrency sets how many zstd encoders this writer opens.
+//
+// It is a setter rather than a constructor argument because the only caller
+// that knows the answer is the fanout, which is two layers above the
+// constructor, and threading it down would put a number that means nothing to
+// them through NewCaptureSink and NewCaptureFanout. It has to be called before
+// the first row, since the codec is handed to the encoder when a shard opens,
+// and the fanout calls it while it is still building.
+func (w *CaptureWriter) SetCodecConcurrency(n int) {
+	if n < 1 {
+		n = 1
+	}
+	w.codec.Concurrency = uint(n)
+}
+
 // NewCaptureWriter builds a writer over dir. A non-positive target size means
 // one shard that never rotates, which is for a small run and wrong for a large
 // one.
@@ -248,7 +302,7 @@ func NewCaptureWriter(dir, prefix string, target int64) (*CaptureWriter, error)
 		target:    target,
 		// BetterCompression is around zstd level 7. It is a large ratio gain over
 		// the default and still far faster than the network feeding it.
-		codec: &zstd.Codec{Level: zstd.SpeedBetterCompression, Concurrency: 4},
+		codec: &zstd.Codec{Level: zstd.SpeedBetterCompression, Concurrency: defaultCodecConcurrency},
 	}
 	cw.buf = make([]Capture, 0, cw.batchRows)
 	return cw, nil

ccrawl/capture_fanout.go

@@ -67,6 +67,14 @@ func NewCaptureFanout(f CaptureFormat, dir, prefix string, shardSize int64, info
 			}
 			return nil, err
 		}
+		// The parts already compress in parallel, so the encoders inside each of
+		// them are divided by how many parts there are. Four each across six
+		// parts is twenty four zstd encoders, and a heap profile puts more than
+		// half a run's memory in their history buffers. A WARC part has no codec
+		// and ignores this.
+		if cw, ok := p.(*CaptureWriter); ok {
+			cw.SetCodecConcurrency(codecConcurrency(n))
+		}
 		fan.parts = append(fan.parts, p)
 		r, ok := p.(captureRowSink)
 		if !ok {

ccrawl/capture_fanout_test.go

@@ -201,3 +201,61 @@ func TestCaptureFanoutWARCDoesNotTakeRows(t *testing.T) {
 		}
 	}
 }
+
+// TestCodecConcurrencyIsDividedNotRepeated is the arithmetic that a heap
+// profile made the case for. The cost of the encoders is writers times
+// concurrency, so a constant per writer multiplies against the writers rather
+// than adding to them, and at six writers that was twenty four zstd encoders
+// holding more than half the run's memory.
+func TestCodecConcurrencyIsDividedNotRepeated(t *testing.T) {
+	cases := map[int]int{0: 4, 1: 4, 2: 2, 3: 1, 4: 1, 6: 1, 16: 1}
+	for writers, want := range cases {
+		if got := codecConcurrency(writers); got != want {
+			t.Fatalf("%d writers get %d encoders each and should get %d", writers, got, want)
+		}
+	}
+	// The product never grows past what one writer used to cost on its own,
+	// which is the property worth holding rather than any single row above.
+	for n := 1; n <= 16; n++ {
+		if total := n * codecConcurrency(n); total > defaultCodecConcurrency*n {
+			t.Fatalf("%d writers open %d encoders between them", n, total)
+		}
+	}
+}
+
+func TestFanoutDividesTheEncodersAcrossItsParts(t *testing.T) {
+	dir := t.TempDir()
+	sink, err := NewCaptureFanout(FormatParquet, dir, "captures", 1<<20, WARCInfo{}, 4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sink.Close() }()
+
+	rf, ok := sink.(rowFanout)
+	if !ok {
+		t.Fatalf("a Parquet fanout of 4 came back as %T", sink)
+	}
+	fan := rf.fanoutSink
+	for i, p := range fan.parts {
+		cw, ok := p.(*CaptureWriter)
+		if !ok {
+			t.Fatalf("part %d is %T", i, p)
+		}
+		if cw.codec.Concurrency != 1 {
+			t.Fatalf("part %d of 4 opens %d zstd encoders and should open 1", i, cw.codec.Concurrency)
+		}
+	}
+}
+
+func TestOneWriterKeepsWhatItHad(t *testing.T) {
+	// A run that was never split should not quietly get slower because of a
+	// change made for runs that were.
+	cw, err := NewCaptureWriter(t.TempDir(), "captures", 1<<20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = cw.Close() }()
+	if cw.codec.Concurrency != defaultCodecConcurrency {
+		t.Fatalf("a single writer opens %d zstd encoders and should open %d", cw.codec.Concurrency, defaultCodecConcurrency)
+	}
+}

docs/content/guides/recrawl-fleet.md

@@ -235,6 +235,12 @@ Give the two runs being compared different names and diff them with `go tool ppr
 The heap profile is the one to reach for when a run is using more memory than the queue bound explains, which is most of the time: the queue is bounded and the encoder buffers are not, and a shard buffers until it is sealed.
 Profiling is off unless asked for and no unit sets it, since it costs a few percent and writes tens of megabytes.
 
+The first thing that profile found is worth knowing before setting `CCRAWL_WRITERS`.
+A heap profile of a 96 worker run showed 750 MB in use with 54 percent of it inside zstd, against 124 MB of Parquet column buffers and 136 MB of response bodies.
+zstd holds that state per concurrent encoder, every writer used to open four of them, so the encoders were writers times four rather than four in total, and at six writers that is twenty four of them holding somewhere over a gigabyte before a single page is in a buffer.
+The encoders are divided across the writers now, so the product stays where it was at one writer and raising `CCRAWL_WRITERS` costs the column buffers and not the compressor as well.
+The memory table further down was measured before that change and is the pessimistic reading of what a wide run costs.
+
 So the width has to be set against the memory the box actually has at the moment, and that means checking `free -g` before raising `CCRAWL_WORKERS` rather than copying a number from another machine.
 `install.sh` writes a `MemoryHigh` of 70 percent of installed RAM into a drop-in per machine.
 `MemoryHigh` throttles rather than kills, which is the right end for a crawler: a run that slows down is one that carries on, and a run that is OOM killed repeats its batch and may do it again.