V14

core/cmd/dragontcp-client/chunk.go

@@ -27,6 +27,7 @@ type chunkClientOptions struct {
 	pollDelay      time.Duration
 	txnTimeout     time.Duration
 	tcpBuffer      int
+	minPipeline    int
 	maxPipeline    int
 }
 
@@ -526,6 +527,7 @@ type chunkConn struct {
 	consumedOffset uint64
 	eof            bool
 	pipeline       int
+	minPipeline    int
 	maxPipeline    int
 
 	closeOnce sync.Once
@@ -556,6 +558,12 @@ func openChunkTunnel(serverAddr, token, targetHost string, targetPort int, opts
 	if opts.maxPipeline > 256 {
 		opts.maxPipeline = 256
 	}
+	if opts.minPipeline < 1 {
+		opts.minPipeline = 1
+	}
+	if opts.minPipeline > opts.maxPipeline {
+		opts.minPipeline = opts.maxPipeline
+	}
 
 	profile := getPathProfile(serverAddr, token, opts)
 	reconnect := opts.reconnectEvery
@@ -620,8 +628,11 @@ func openChunkTunnel(serverAddr, token, targetHost string, targetPort int, opts
 		downloadLane: newRequestLane(serverAddr, opts.tcpBuffer, reconnect, opts.txnTimeout),
 		// Start at the user-configured ceiling. On transport failures the
 		// pipeline is halved; successful data responses grow it back by one,
-		// always staying inside 1..maxPipeline. A ceiling of 1 is fixed.
+		// always staying inside minPipeline..maxPipeline. When the two bounds
+		// are equal the depth is pinned and never adapts, which is what paths
+		// that only work at one specific batch size need.
 		pipeline:    opts.maxPipeline,
+		minPipeline: opts.minPipeline,
 		maxPipeline: opts.maxPipeline,
 	}
 	c.upSizer = newAdaptiveSizer("upload", upStart, opts)
@@ -629,6 +640,53 @@ func openChunkTunnel(serverAddr, token, targetHost string, targetPort int, opts
 	return c, nil
 }
 
+// pinnedBatch reports whether the batch depth is fixed. A pinned depth never
+// grows or shrinks: some paths only deliver correctly at one specific number of
+// records per request, so the adaptive controller must stay out of the way.
+func (c *chunkConn) pinnedBatch() bool { return c.minPipeline >= c.maxPipeline }
+
+// batchCount is how many records the next download request will ask for.
+func (c *chunkConn) batchCount(chunk int) int {
+	count := c.pipeline
+	if count < c.minPipeline {
+		count = c.minPipeline
+	}
+	if count > c.maxPipeline {
+		count = c.maxPipeline
+	}
+	// Bound each batch to roughly 1 MiB of useful data, but never below the
+	// configured floor: a pinned depth is a path requirement, not a hint.
+	if maxCount := (1024 * 1024) / maxInt(chunk, 1); maxCount < count {
+		count = maxInt(maxCount, c.minPipeline)
+	}
+	return count
+}
+
+// growPipeline widens the batch by one after a successful data response.
+func (c *chunkConn) growPipeline() {
+	if c.pinnedBatch() {
+		return
+	}
+	if c.pipeline < c.maxPipeline {
+		c.pipeline++
+	}
+}
+
+// shrinkPipeline halves the batch after a transport failure. It reports the old
+// and new depth, and whether anything actually changed; when it returns false
+// the caller should shrink the record size instead.
+func (c *chunkConn) shrinkPipeline() (int, int, bool) {
+	if c.pinnedBatch() || c.pipeline <= c.minPipeline {
+		return c.pipeline, c.pipeline, false
+	}
+	old := c.pipeline
+	c.pipeline /= 2
+	if c.pipeline < c.minPipeline {
+		c.pipeline = c.minPipeline
+	}
+	return old, c.pipeline, old != c.pipeline
+}
+
 func (c *chunkConn) fillReadBuffer() error {
 	if c.eof {
 		return io.EOF
@@ -636,17 +694,7 @@ func (c *chunkConn) fillReadBuffer() error {
 	minFailures := 0
 	for len(c.readBuf) == 0 && !c.eof {
 		chunk := c.downSizer.Current()
-		count := c.pipeline
-		if count < 1 {
-			count = 1
-		}
-		if count > c.maxPipeline {
-			count = c.maxPipeline
-		}
-		// Bound each batch to roughly 1 MiB of useful data.
-		if maxCount := (1024 * 1024) / maxInt(chunk, 1); maxCount < count {
-			count = maxInt(maxCount, 1)
-		}
+		count := c.batchCount(chunk)
 
 		data, status, err := c.downloadLane.download(c.sid, c.downloadOffset, c.consumedOffset, chunk, count)
 		for _, part := range data {
@@ -655,20 +703,16 @@ func (c *chunkConn) fillReadBuffer() error {
 		}
 		if len(data) > 0 {
 			c.downSizer.Success(chunk)
-			if c.pipeline < c.maxPipeline {
-				c.pipeline++
-			}
+			c.growPipeline()
 			minFailures = 0
 		}
 		if err != nil {
-			if c.pipeline > 1 {
-				old := c.pipeline
-				c.pipeline /= 2
-				if c.pipeline < 1 {
-					c.pipeline = 1
-				}
-				if c.opts.adaptLog && old != c.pipeline {
-					fmt.Printf("adaptive download pipeline: %d -> %d after transport failure\n", old, c.pipeline)
+			// Shrink the batch first, then the record size. When the batch is
+			// pinned (min == max) the depth is left alone entirely and only the
+			// record size adapts.
+			if old, next, shrank := c.shrinkPipeline(); shrank {
+				if c.opts.adaptLog {
+					fmt.Printf("adaptive download batch: %d -> %d after transport failure\n", old, next)
 				}
 			} else {
 				old, next := c.downSizer.Failure(chunk)

core/cmd/dragontcp-client/chunk_test.go

@@ -23,6 +23,76 @@ func TestAdaptiveSizerRecoversFromMinimum(t *testing.T) {
 	}
 }
 
+func newTestConn(min, max int) *chunkConn {
+	return &chunkConn{pipeline: max, minPipeline: min, maxPipeline: max}
+}
+
+func TestPinnedBatchNeverAdapts(t *testing.T) {
+	c := newTestConn(5, 5)
+	if got := c.batchCount(1400); got != 5 {
+		t.Fatalf("pinned batch should request 5 records, got %d", got)
+	}
+	for i := 0; i < 10; i++ {
+		if _, _, shrank := c.shrinkPipeline(); shrank {
+			t.Fatal("pinned batch shrank on transport failure")
+		}
+		c.growPipeline()
+	}
+	if c.pipeline != 5 {
+		t.Fatalf("pinned batch drifted to %d", c.pipeline)
+	}
+	if got := c.batchCount(1400); got != 5 {
+		t.Fatalf("pinned batch should still request 5 records, got %d", got)
+	}
+}
+
+func TestPinnedBatchSurvivesOneMiBCap(t *testing.T) {
+	// 8 x 1 MiB records exceed the ~1 MiB useful-data cap. A pinned depth must
+	// win anyway, otherwise a path that needs exactly 8 records is broken by
+	// an unrelated size heuristic.
+	c := newTestConn(8, 8)
+	if got := c.batchCount(1024 * 1024); got != 8 {
+		t.Fatalf("pinned batch should ignore the 1 MiB cap, got %d", got)
+	}
+	// An unpinned batch is still capped.
+	c = newTestConn(1, 8)
+	if got := c.batchCount(1024 * 1024); got != 1 {
+		t.Fatalf("unpinned batch should be capped to 1, got %d", got)
+	}
+}
+
+func TestAdaptiveBatchStopsAtFloor(t *testing.T) {
+	c := newTestConn(4, 32)
+	seen := map[int]bool{}
+	for i := 0; i < 12; i++ {
+		_, next, _ := c.shrinkPipeline()
+		seen[next] = true
+	}
+	if c.pipeline != 4 {
+		t.Fatalf("batch fell to %d, want the floor 4", c.pipeline)
+	}
+	if !seen[16] || !seen[8] {
+		t.Fatalf("expected halving through 16 and 8, saw %v", seen)
+	}
+	for i := 0; i < 100; i++ {
+		c.growPipeline()
+	}
+	if c.pipeline != 32 {
+		t.Fatalf("batch grew to %d, want the ceiling 32", c.pipeline)
+	}
+}
+
+func TestSingleBatchIsFixed(t *testing.T) {
+	c := newTestConn(1, 1)
+	if !c.pinnedBatch() {
+		t.Fatal("a 1..1 batch must be treated as pinned")
+	}
+	c.growPipeline()
+	if c.pipeline != 1 {
+		t.Fatalf("batch of 1 grew to %d", c.pipeline)
+	}
+}
+
 func TestReconnectZeroMeansPersistent(t *testing.T) {
 	lane := newRequestLane("127.0.0.1:1", 0, 0, 0)
 	if lane.reconnectEvery != 0 {

core/cmd/dragontcp-client/main.go

@@ -374,7 +374,8 @@ func main() {
 		chunkAdaptLog    = flag.Bool("chunk-adapt-log", true, "print adaptive chunk size changes")
 		chunkSizeLegacy  = flag.Int("chunk-size", 0, "legacy fixed chunk size; nonzero disables adaptation")
 		chunkPollers     = flag.Int("chunk-pollers", 1, "reserved compatibility setting; binary transport uses one download worker")
-		chunkConcurrency = flag.Int("chunk-concurrency", 1, "maximum adaptive download pipeline depth (1-256); 1 keeps concurrency fixed at one")
+		chunkConcurrency = flag.Int("chunk-concurrency", 1, "maximum download records per request (1-256)")
+		chunkConcurrencyMin = flag.Int("chunk-concurrency-min", 1, "minimum download records per request (1-256); equal to --chunk-concurrency pins the depth and disables batch adaptation")
 		chunkReconnect   = flag.Int("chunk-reconnect-every", 0, "force reconnect after N logical requests; 0 = persistent/automatic")
 		chunkPollDelay   = flag.Duration("chunk-poll-delay", 2*time.Millisecond, "delay after an empty chunk poll")
 		chunkTimeout     = flag.Duration("chunk-timeout", 2*time.Second, "per-record transaction timeout before adaptive shrink")
@@ -417,6 +418,14 @@ func main() {
 		fmt.Fprintln(os.Stderr, "--chunk-concurrency must be between 1 and 256")
 		os.Exit(2)
 	}
+	if *chunkConcurrencyMin < 1 || *chunkConcurrencyMin > 256 {
+		fmt.Fprintln(os.Stderr, "--chunk-concurrency-min must be between 1 and 256")
+		os.Exit(2)
+	}
+	if *chunkConcurrencyMin > *chunkConcurrency {
+		fmt.Fprintln(os.Stderr, "--chunk-concurrency-min must not exceed --chunk-concurrency")
+		os.Exit(2)
+	}
 	if *chunkReconnect < 0 {
 		fmt.Fprintln(os.Stderr, "--chunk-reconnect-every must be 0 or greater")
 		os.Exit(2)
@@ -433,6 +442,7 @@ func main() {
 		pollDelay:      *chunkPollDelay,
 		txnTimeout:     *chunkTimeout,
 		tcpBuffer:      *tcpBuffer,
+		minPipeline:    *chunkConcurrencyMin,
 		maxPipeline:    *chunkConcurrency,
 	}
 
@@ -450,15 +460,21 @@ func main() {
 	fmt.Printf("remote DragonTCP endpoint=%s\n", serverAddr)
 	fmt.Printf("max_connections=%d transport=%s tcp_buffer=%d\n", *maxConnections, *transport, *tcpBuffer)
 	if *transport == "chunk" {
+		batchMode := "adaptive"
+		if *chunkConcurrencyMin == *chunkConcurrency {
+			batchMode = "pinned"
+		}
 		fmt.Printf(
-			"adaptive_chunk=%v start=%d min=%d max=%d grow_after=%d pollers=%d concurrency=%d reconnect_every=%d timeout=%s\n",
+			"adaptive_chunk=%v start=%d min=%d max=%d grow_after=%d pollers=%d batch=%d-%d(%s) reconnect_every=%d timeout=%s\n",
 			*chunkAdaptive,
 			*chunkStart,
 			*chunkMin,
 			*chunkMax,
 			*chunkSuccesses,
 			*chunkPollers,
+			*chunkConcurrencyMin,
 			*chunkConcurrency,
+			batchMode,
 			*chunkReconnect,
 			chunkTimeout.String(),
 		)