DragonTCP/core/internal/xorchunk/chunk_test.go

package xorchunk

import (
	"bytes"
	"net"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"dragontcp/internal/protocol"
)

func TestAdaptiveSizerShrinkBudget(t *testing.T) {
	opts := Options{
		startSize:   1024,
		minSize:     32,
		maxSize:     1024,
		adaptive:    true,
		shrinkAfter: 3,
	}
	s := newAdaptiveSizer("test", opts.startSize, opts.maxSize, opts)
	for i := 1; i <= 2; i++ {
		_, next := s.Failure(1024)
		if next != 1024 {
			t.Fatalf("failure %d reduced early to %d", i, next)
		}
	}
	s.Success(1024)
	for i := 1; i <= 2; i++ {
		_, next := s.Failure(1024)
		if next != 1024 {
			t.Fatalf("post-success failure %d reduced early to %d", i, next)
		}
	}
	_, next := s.Failure(1024)
	if next != 512 {
		t.Fatalf("third consecutive failure reduced to %d, want 512", next)
	}
}

func startCalibrationTestServer(t *testing.T, threshold int) (string, func()) {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	stop := make(chan struct{})
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			go func(c net.Conn) {
				defer c.Close()
				for {
					id, _, payload, err := protocol.ReadRequestFrameProfile(c, 0)
					if err != nil {
						return
					}
					if bytes.HasPrefix(payload, []byte("CIPERFUP ")) {
						parts := bytes.SplitN(payload, []byte(" "), 4)
						if len(parts) != 4 {
							_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad upload"))
							continue
						}
						size, _ := strconv.Atoi(string(parts[2]))
						if size > threshold || len(parts[3]) != size || !bytes.Equal(parts[3], calibrationPattern(size)) {
							_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
							continue
						}
						_ = protocol.WriteResponseFrame(c, id, []byte("IPERFOK"))
						continue
					}
					if bytes.HasPrefix(payload, []byte("CIPERFDW ")) {
						parts := strings.Fields(string(payload))
						if len(parts) != 3 {
							_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad download"))
							continue
						}
						size, _ := strconv.Atoi(parts[2])
						if size > threshold {
							_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
							continue
						}
						_ = protocol.WriteResponseFrame(c, id, calibrationPattern(size))
						continue
					}
					if bytes.HasPrefix(payload, []byte("CPROBE ")) {
						_ = protocol.WriteResponseFrame(c, id, []byte("PROBEOK"))
						continue
					}
					_ = protocol.WriteResponseFrame(c, id, []byte("ERR unsupported"))
				}
			}(conn)
		}
	}()
	return ln.Addr().String(), func() {
		close(stop)
		_ = ln.Close()
	}
}

func TestXCalibrationRefinesUploadAndDownloadTo32Bytes(t *testing.T) {
	const threshold = 731237
	addr, closeServer := startCalibrationTestServer(t, threshold)
	defer closeServer()
	opts := Options{
		startSize:  1024 * 1024,
		minSize:    32,
		maxSize:    1024 * 1024,
		txnTimeout: time.Second,
	}
	for _, tc := range []struct {
		name     string
		download bool
	}{
		{"upload", false},
		{"download", true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			got := calibrateMaximum(addr, "", opts, tc.download, 32)
			if got > threshold {
				t.Fatalf("calibrated size=%d exceeds threshold=%d", got, threshold)
			}
			if threshold-got > 32 {
				t.Fatalf("calibrated size=%d is more than 32 bytes below threshold=%d", got, threshold)
			}
		})
	}
}

func TestXCalibrationRetriesTransientFailure(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer ln.Close()

	const (
		threshold      = 4096
		transientChunk = 2048
	)
	var transient atomic.Int32
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			go func(c net.Conn) {
				defer c.Close()
				for {
					id, _, payload, err := protocol.ReadRequestFrameProfile(c, 0)
					if err != nil {
						return
					}
					if !bytes.HasPrefix(payload, []byte("CIPERFUP ")) {
						_ = protocol.WriteResponseFrame(c, id, []byte("ERR unsupported"))
						continue
					}
					parts := bytes.SplitN(payload, []byte(" "), 4)
					if len(parts) != 4 {
						_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad upload"))
						continue
					}
					size, _ := strconv.Atoi(string(parts[2]))
					if size == transientChunk && transient.CompareAndSwap(0, 1) {
						_ = protocol.WriteResponseFrame(c, id, []byte("ERR transient"))
						return
					}
					if size > threshold || len(parts[3]) != size || !bytes.Equal(parts[3], calibrationPattern(size)) {
						_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
						continue
					}
					_ = protocol.WriteResponseFrame(c, id, []byte("IPERFOK"))
				}
			}(conn)
		}
	}()

	opts := Options{
		startSize:  16 * 1024,
		minSize:    32,
		maxSize:    16 * 1024,
		txnTimeout: time.Second,
	}
	got := calibrateMaximum(ln.Addr().String(), "", opts, false, 32)
	if transient.Load() != 1 {
		t.Fatalf("transient failure count=%d, want 1", transient.Load())
	}
	if got < transientChunk {
		t.Fatalf("X calibration collapsed below transiently failed %d-byte probe: got %d", transientChunk, got)
	}
	if got > threshold || threshold-got > 32 {
		t.Fatalf("X calibrated size=%d, want within 32 bytes below threshold=%d", got, threshold)
	}
}

func TestWithCalibratedChunksLocksIndependentXSizes(t *testing.T) {
	opts := Options{minSize: 32, maxSize: 1024 * 1024, startSize: 1024 * 1024, adaptive: true}
	opts = opts.WithCalibratedChunks(900000, 500000)
	if opts.adaptive {
		t.Fatal("X runtime adaptation remained enabled after calibration")
	}

	up := newAdaptiveSizer("upload", opts.uploadStartSize, opts.uploadMaxSize, opts)
	down := newAdaptiveSizer("download", opts.downloadStartSize, opts.downloadMaxSize, opts)
	if up.Current() != 900000 || up.max != 900000 {
		t.Fatalf("upload current/max=%d/%d, want 900000", up.Current(), up.max)
	}
	if down.Current() != 500000 || down.max != 500000 {
		t.Fatalf("download current/max=%d/%d, want 500000", down.Current(), down.max)
	}

	// The calibrated sizes are immutable during the X session. Neither a
	// transport failure nor a long run of successes may move them.
	if old, next := up.Failure(900000); old != 900000 || next != 900000 {
		t.Fatalf("upload failure changed calibrated chunk: %d -> %d", old, next)
	}
	for i := 0; i < 1000; i++ {
		up.Success(900000)
		down.Success(500000)
	}
	if up.Current() != 900000 {
		t.Fatalf("upload success changed calibrated chunk to %d", up.Current())
	}
	if old, next := down.Failure(500000); old != 500000 || next != 500000 {
		t.Fatalf("download failure changed calibrated chunk: %d -> %d", old, next)
	}
	if down.Current() != 500000 {
		t.Fatalf("download calibrated chunk changed to %d", down.Current())
	}
}