first

client/ebpf/measure.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+"""eBPF-based UDP tracer for MP-QUIC performance measurement.
+
+Traces kernel-level UDP send/receive latency and packet drops.
+Requires root privileges and Linux with BCC installed.
+"""
+
+from bcc import BPF
+import time
+import argparse
+import signal
+import sys
+
+bpf_text = """
+#include <uapi/linux/ptrace.h>
+#include <net/sock.h>
+#include <net/inet_sock.h>
+#include <bcc/proto.h>
+
+#define MPQUIC_PORT __TARGET_PORT__
+
+BPF_HASH(send_start, u32, u64);
+BPF_HASH(recv_start, u32, u64);
+
+BPF_PERF_OUTPUT(events);
+
+struct event_t {
+    u64 timestamp_ns;
+    u64 delta_ns;
+    u32 pid;
+    u16 sport;
+    u16 dport;
+    u8  event_type;  // 0=SEND, 1=RECV, 2=DROP
+};
+
+int trace_udp_sendmsg(struct pt_regs *ctx, struct sock *sk) {
+    struct inet_sock *inet = (struct inet_sock *)sk;
+    u16 dport = 0;
+    bpf_probe_read_kernel(&dport, sizeof(dport), &inet->inet_dport);
+    dport = ntohs(dport);
+
+    // Filter: only trace traffic to/from MP-QUIC port
+    u16 sport = 0;
+    bpf_probe_read_kernel(&sport, sizeof(sport), &inet->inet_sport);
+    sport = ntohs(sport);
+
+    if (dport != MPQUIC_PORT && sport != MPQUIC_PORT)
+        return 0;
+
+    u32 pid = bpf_get_current_pid_tgid();
+    u64 ts = bpf_ktime_get_ns();
+    send_start.update(&pid, &ts);
+    return 0;
+}
+
+int trace_udp_sendmsg_ret(struct pt_regs *ctx) {
+    u32 pid = bpf_get_current_pid_tgid();
+    u64 *tsp = send_start.lookup(&pid);
+    if (tsp == 0)
+        return 0;
+
+    u64 delta = bpf_ktime_get_ns() - *tsp;
+    send_start.delete(&pid);
+
+    struct event_t event = {};
+    event.timestamp_ns = bpf_ktime_get_ns();
+    event.delta_ns = delta;
+    event.pid = pid;
+    event.event_type = 0;  // SEND
+    events.perf_submit(ctx, &event, sizeof(event));
+    return 0;
+}
+
+int trace_udp_recvmsg(struct pt_regs *ctx, struct sock *sk) {
+    struct inet_sock *inet = (struct inet_sock *)sk;
+    u16 sport = 0;
+    bpf_probe_read_kernel(&sport, sizeof(sport), &inet->inet_sport);
+    sport = ntohs(sport);
+
+    u16 dport = 0;
+    bpf_probe_read_kernel(&dport, sizeof(dport), &inet->inet_dport);
+    dport = ntohs(dport);
+
+    if (sport != MPQUIC_PORT && dport != MPQUIC_PORT)
+        return 0;
+
+    u32 pid = bpf_get_current_pid_tgid();
+    u64 ts = bpf_ktime_get_ns();
+    recv_start.update(&pid, &ts);
+    return 0;
+}
+
+int trace_udp_recvmsg_ret(struct pt_regs *ctx) {
+    u32 pid = bpf_get_current_pid_tgid();
+    u64 *tsp = recv_start.lookup(&pid);
+    if (tsp == 0)
+        return 0;
+
+    u64 delta = bpf_ktime_get_ns() - *tsp;
+    recv_start.delete(&pid);
+
+    struct event_t event = {};
+    event.timestamp_ns = bpf_ktime_get_ns();
+    event.delta_ns = delta;
+    event.pid = pid;
+    event.event_type = 1;  // RECV
+    events.perf_submit(ctx, &event, sizeof(event));
+    return 0;
+}
+
+TRACEPOINT_PROBE(skb, kfree_skb) {
+    struct event_t event = {};
+    event.timestamp_ns = bpf_ktime_get_ns();
+    event.delta_ns = 0;
+    event.pid = bpf_get_current_pid_tgid();
+    event.event_type = 2;  // DROP
+    events.perf_submit(ctx, &event, sizeof(event));
+    return 0;
+}
+"""
+
+EVENT_TYPES = {0: "SEND_LATENCY", 1: "RECV_LATENCY", 2: "DROP"}
+
+# Counters for live summary
+stats = {"send_count": 0, "recv_count": 0, "drop_count": 0,
+         "send_total_ns": 0, "recv_total_ns": 0}
+
+
+def main():
+    parser = argparse.ArgumentParser(description="eBPF UDP Tracer for MP-QUIC")
+    parser.add_argument("--output", type=str, default="measurement.csv",
+                        help="CSV output file")
+    parser.add_argument("--port", type=int, default=4242,
+                        help="MP-QUIC port to filter on")
+    args = parser.parse_args()
+
+    program = bpf_text.replace("__TARGET_PORT__", str(args.port))
+
+    print(f"Compiling eBPF program (requires root)...")
+    print(f"Filtering on UDP port {args.port}")
+    b = BPF(text=program)
+
+    # Attach kprobes for send and receive paths
+    b.attach_kprobe(event="udp_sendmsg", fn_name="trace_udp_sendmsg")
+    b.attach_kretprobe(event="udp_sendmsg", fn_name="trace_udp_sendmsg_ret")
+    b.attach_kprobe(event="udp_recvmsg", fn_name="trace_udp_recvmsg")
+    b.attach_kretprobe(event="udp_recvmsg", fn_name="trace_udp_recvmsg_ret")
+
+    csv_file = open(args.output, "w")
+    csv_file.write("timestamp,event_type,value_ns,pid\n")
+
+    def handle_event(cpu, data, size):
+        event = b["events"].event(data)
+        etype = EVENT_TYPES.get(event.event_type, "UNKNOWN")
+        csv_file.write(f"{time.time()},{etype},{event.delta_ns},{event.pid}\n")
+
+        # Update live stats
+        if event.event_type == 0:
+            stats["send_count"] += 1
+            stats["send_total_ns"] += event.delta_ns
+        elif event.event_type == 1:
+            stats["recv_count"] += 1
+            stats["recv_total_ns"] += event.delta_ns
+        elif event.event_type == 2:
+            stats["drop_count"] += 1
+
+    b["events"].open_perf_buffer(handle_event)
+
+    def signal_handler(sig, frame):
+        print("\n--- Measurement Summary ---")
+        if stats["send_count"] > 0:
+            avg_send = stats["send_total_ns"] / stats["send_count"] / 1000
+            print(f"  Send events:  {stats['send_count']:>8}  (avg {avg_send:.1f} µs)")
+        if stats["recv_count"] > 0:
+            avg_recv = stats["recv_total_ns"] / stats["recv_count"] / 1000
+            print(f"  Recv events:  {stats['recv_count']:>8}  (avg {avg_recv:.1f} µs)")
+        print(f"  Drop events:  {stats['drop_count']:>8}")
+        print(f"  Output file:  {args.output}")
+        csv_file.close()
+        sys.exit(0)
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    print(f"Tracing... Output → {args.output}. Ctrl-C to stop.")
+    while True:
+        b.perf_buffer_poll()
+
+
+if __name__ == "__main__":
+    main()

client/main.go

@@ -0,0 +1,96 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/binary"
+	"flag"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	quic "github.com/AeonDave/mp-quic-go"
+	"mpquic-exp/client/scheduler"
+)
+
+func main() {
+	schedName := flag.String("scheduler", "roundrobin", "Scheduler to use")
+	listScheds := flag.Bool("list-schedulers", false, "List available schedulers")
+	addr := flag.String("addr", "127.0.0.1:4242", "Server address")
+	duration := flag.Int("duration", 10, "Duration in seconds")
+	payloadSize := flag.Int("payload-size", 1024, "Size of the payload in bytes (min 20)")
+	flag.Parse()
+
+	if *listScheds {
+		fmt.Println("Available schedulers:")
+		for _, name := range scheduler.List() {
+			fmt.Println(" -", name)
+		}
+		return
+	}
+
+	if *payloadSize < 20 {
+		*payloadSize = 20 // 4 bytes length, 8 bytes seq, 8 bytes timestamp
+	}
+
+	tlsConf := &tls.Config{
+		InsecureSkipVerify: true,
+		NextProtos:         []string{"mpquic-exp"},
+	}
+
+	sched, err := scheduler.Get(*schedName)
+	if err != nil {
+		log.Fatalf("Error: %v. Available: %s", err, strings.Join(scheduler.List(), ", "))
+	}
+
+	quicConfig := &quic.Config{
+		MaxPaths:            4,
+		MultipathController: quic.NewDefaultMultipathController(sched),
+	}
+
+	fmt.Printf("Connecting to %s using %s scheduler...\n", *addr, *schedName)
+	conn, err := quic.DialAddr(context.Background(), *addr, tlsConf, quicConfig)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	stream, err := conn.OpenStreamSync(context.Background())
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	fmt.Printf("Stream opened, sending data for %d seconds (Payload Size: %d bytes)...\n", *duration, *payloadSize)
+
+	end := time.Now().Add(time.Duration(*duration) * time.Second)
+	payload := make([]byte, *payloadSize)
+	
+	// Frame structure:
+	// [0:4] uint32 Total Length
+	// [4:12] uint64 Sequence Number
+	// [12:20] uint64 Send Timestamp (nanoseconds)
+	// [20:] Padding (dummy data)
+	binary.BigEndian.PutUint32(payload[0:4], uint32(*payloadSize))
+
+	var seqNum uint64 = 0
+	totalBytes := 0
+	
+	// Target sending rate: we don't want to lock the CPU entirely in a busy loop.
+	// We yield slightly to allow the network stack to process.
+	// But to measure max throughput we just send as fast as stream.Write allows.
+	for time.Now().Before(end) {
+		seqNum++
+		sendTime := uint64(time.Now().UnixNano())
+		
+		binary.BigEndian.PutUint64(payload[4:12], seqNum)
+		binary.BigEndian.PutUint64(payload[12:20], sendTime)
+		
+		n, err := stream.Write(payload)
+		if err != nil {
+			log.Fatal("Stream write error:", err)
+		}
+		totalBytes += n
+	}
+	
+	fmt.Printf("Finished sending %d packets, %d bytes (%.2f MB)\n", seqNum, totalBytes, float64(totalBytes)/1024/1024)
+}

client/scheduler/custom_example.go

@@ -0,0 +1,100 @@
+package scheduler
+
+// This file demonstrates how to add a custom scheduler.
+// To create your own:
+//   1. Copy this file and rename it (e.g., weighted_rtt.go)
+//   2. Implement the three PathScheduler methods
+//   3. Update the init() to register with your scheduler name
+//
+// The scheduler will automatically appear in --list-schedulers.
+
+import (
+	"sync"
+
+	quic "github.com/AeonDave/mp-quic-go"
+)
+
+func init() {
+	Register("weighted", func() quic.PathScheduler {
+		return NewWeightedScheduler()
+	})
+}
+
+// WeightedScheduler is an example custom scheduler that weighs paths
+// by a combination of RTT and available congestion window.
+type WeightedScheduler struct {
+	mu     sync.Mutex
+	quotas map[quic.PathID]uint64
+}
+
+// NewWeightedScheduler creates a new weighted scheduler.
+func NewWeightedScheduler() *WeightedScheduler {
+	return &WeightedScheduler{
+		quotas: make(map[quic.PathID]uint64),
+	}
+}
+
+// SelectPath picks the path with the best weighted score.
+// Score = (CongestionWindow - BytesInFlight) / (1 + SmoothedRTT_ms)
+// Higher score = more capacity available per unit latency.
+func (s *WeightedScheduler) SelectPath(paths []quic.SchedulerPathInfo, hasRetransmission bool) *quic.SchedulerPathInfo {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if len(paths) == 0 {
+		return nil
+	}
+	if len(paths) == 1 {
+		if !hasRetransmission && !paths[0].SendingAllowed {
+			return nil
+		}
+		return &paths[0]
+	}
+
+	var best *quic.SchedulerPathInfo
+	var bestScore float64 = -1
+
+	for i := range paths {
+		p := &paths[i]
+		if !hasRetransmission && !p.SendingAllowed {
+			continue
+		}
+		if p.PotentiallyFailed {
+			continue
+		}
+
+		// Available window
+		available := float64(0)
+		if p.CongestionWindow > p.BytesInFlight {
+			available = float64(p.CongestionWindow - p.BytesInFlight)
+		}
+
+		// RTT factor (ms, minimum 1 to avoid division by zero)
+		rttMs := float64(1)
+		if p.SmoothedRTT.Milliseconds() > 0 {
+			rttMs = float64(p.SmoothedRTT.Milliseconds())
+		}
+
+		score := available / rttMs
+		if score > bestScore {
+			bestScore = score
+			best = p
+		}
+	}
+
+	return best
+}
+
+// UpdateQuota tracks per-path send counts.
+func (s *WeightedScheduler) UpdateQuota(pathID quic.PathID, packetSize quic.ByteCount) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.quotas[pathID]++
+}
+
+// Reset clears all state.
+func (s *WeightedScheduler) Reset() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.quotas = make(map[quic.PathID]uint64)
+}

client/scheduler/lowlatency.go

@@ -0,0 +1,9 @@
+package scheduler
+
+import quic "github.com/AeonDave/mp-quic-go"
+
+func init() {
+	Register("lowlatency", func() quic.PathScheduler {
+		return quic.NewLowLatencyScheduler()
+	})
+}

client/scheduler/minrtt.go

@@ -0,0 +1,11 @@
+package scheduler
+
+import quic "github.com/AeonDave/mp-quic-go"
+
+func init() {
+	Register("minrtt", func() quic.PathScheduler {
+		// rttBias=0.8 favors lower RTT paths while still considering load balance.
+		// Adjust this value: 1.0=pure RTT, 0.0=pure load balancing.
+		return quic.NewMinRTTScheduler(0.8)
+	})
+}

client/scheduler/registry.go

@@ -0,0 +1,60 @@
+// Package scheduler provides a modular registry for MP-QUIC path schedulers.
+//
+// Adding a new scheduler:
+//  1. Create a new file in this package (e.g., myscheduler.go)
+//  2. Implement quic.PathScheduler (SelectPath, UpdateQuota, Reset)
+//  3. In an init() function, call Register("myscheduler", factory)
+//
+// See custom_example.go for a complete example.
+package scheduler
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+
+	quic "github.com/AeonDave/mp-quic-go"
+)
+
+// Factory is a function that creates a new PathScheduler instance.
+type Factory func() quic.PathScheduler
+
+var (
+	mu        sync.RWMutex
+	factories = make(map[string]Factory)
+)
+
+// Register adds a scheduler factory to the registry.
+// Call this from init() in each scheduler file.
+func Register(name string, factory Factory) {
+	mu.Lock()
+	defer mu.Unlock()
+	if _, exists := factories[name]; exists {
+		panic(fmt.Sprintf("scheduler: duplicate registration for %q", name))
+	}
+	factories[name] = factory
+}
+
+// Get creates a new instance of the named scheduler.
+// Returns an error if the name is not registered.
+func Get(name string) (quic.PathScheduler, error) {
+	mu.RLock()
+	defer mu.RUnlock()
+	factory, exists := factories[name]
+	if !exists {
+		return nil, fmt.Errorf("scheduler: unknown scheduler %q (available: %v)", name, List())
+	}
+	return factory(), nil
+}
+
+// List returns sorted names of all registered schedulers.
+func List() []string {
+	mu.RLock()
+	defer mu.RUnlock()
+	names := make([]string, 0, len(factories))
+	for name := range factories {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}

client/scheduler/roundrobin.go

@@ -0,0 +1,9 @@
+package scheduler
+
+import quic "github.com/AeonDave/mp-quic-go"
+
+func init() {
+	Register("roundrobin", func() quic.PathScheduler {
+		return quic.NewRoundRobinScheduler()
+	})
+}

client/scripts/run.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+# Build the client
+echo "Building client..."
+go build -o client_bin main.go
+
+echo "Starting client..."
+./client_bin "$@"

client/scripts/setup.sh

@@ -0,0 +1,94 @@
+#!/bin/bash
+# Setup script for MP-QUIC experiment environment.
+# Run on each Linux machine (server + client) before experiments.
+set -euo pipefail
+
+echo "╔═══════════════════════════════════════════╗"
+echo "║  MP-QUIC Experiment Environment Setup     ║"
+echo "╚═══════════════════════════════════════════╝"
+
+# --- OS Check ---
+if [[ ! -f /etc/os-release ]]; then
+    echo "❌ Not running on Linux. eBPF tracing requires Linux."
+    echo "   You can still build and run the Go binaries on this machine."
+    exit 1
+fi
+
+. /etc/os-release
+echo "📋 Detected OS: $PRETTY_NAME"
+
+# --- Install dependencies ---
+if [[ "$ID" == "ubuntu" || "$ID" == "debian" ]]; then
+    echo ""
+    echo "📦 Installing packages..."
+    sudo apt-get update -qq
+    sudo apt-get install -y -qq \
+        bpfcc-tools \
+        python3-bpfcc \
+        linux-headers-$(uname -r) \
+        golang-go \
+        iperf3 \
+        net-tools \
+        iproute2
+
+    # Ensure Python3 BCC bindings work
+    python3 -c "from bcc import BPF; print('   ✓ BCC Python bindings OK')" 2>/dev/null || {
+        echo "   ⚠  BCC Python import failed. Try: sudo apt install python3-bpfcc"
+    }
+elif [[ "$ID" == "fedora" || "$ID" == "rhel" || "$ID" == "centos" ]]; then
+    echo ""
+    echo "📦 Installing packages (DNF)..."
+    sudo dnf install -y \
+        bcc-tools \
+        python3-bcc \
+        kernel-devel-$(uname -r) \
+        golang \
+        iperf3 \
+        iproute
+else
+    echo "⚠  Unsupported distro: $ID"
+    echo "   Please install manually: bcc-tools, python3-bcc, golang, linux-headers"
+fi
+
+# --- Verify Go ---
+echo ""
+if command -v go &>/dev/null; then
+    echo "✓ Go $(go version | awk '{print $3}')"
+else
+    echo "❌ Go not found. Install from https://go.dev/dl/"
+    exit 1
+fi
+
+# --- Kernel config check ---
+echo ""
+echo "🔍 Checking kernel eBPF support..."
+if [[ -d /sys/kernel/debug/tracing ]]; then
+    echo "   ✓ debugfs mounted"
+else
+    echo "   ⚠  debugfs not mounted. Run: sudo mount -t debugfs debugfs /sys/kernel/debug"
+fi
+
+if grep -q CONFIG_BPF=y /boot/config-$(uname -r) 2>/dev/null; then
+    echo "   ✓ CONFIG_BPF enabled"
+else
+    echo "   ⚠  Could not verify CONFIG_BPF (may still work)"
+fi
+
+# --- Network tuning (optional) ---
+echo ""
+echo "🔧 Applying network tuning..."
+sudo sysctl -w net.core.rmem_max=26214400 2>/dev/null && echo "   ✓ rmem_max=25MB" || true
+sudo sysctl -w net.core.wmem_max=26214400 2>/dev/null && echo "   ✓ wmem_max=25MB" || true
+sudo sysctl -w net.core.rmem_default=1048576 2>/dev/null || true
+sudo sysctl -w net.core.wmem_default=1048576 2>/dev/null || true
+
+echo ""
+echo "╔═══════════════════════════════════════════╗"
+echo "║  ✓ Setup complete!                        ║"
+echo "║                                           ║"
+echo "║  Quick start:                             ║"
+echo "║    python3 run_experiment.py --duration 10 ║"
+echo "║                                           ║"
+echo "║  With eBPF (needs sudo):                  ║"
+echo "║    python3 run_experiment.py --ebpf        ║"
+echo "╚═══════════════════════════════════════════╝"