[rp2040] Tune oversized lwIP defaults for ESPHome (#14843)

script/stress_test_connect.py

@@ -0,0 +1,84 @@
+"""Rapid connect/disconnect stress test for ESPHome native API."""
+
+import asyncio
+import sys
+import time
+
+from aioesphomeapi import APIClient
+
+HOST = "192.168.1.100"
+PORT = 6053
+PASSWORD = ""
+NOISE_PSK = None
+ITERATIONS = 500
+CONCURRENCY = 4  # simultaneous connection attempts
+
+
+async def connect_disconnect(client_id: int, iteration: int) -> tuple[int, bool, str]:
+    """Connect and immediately disconnect."""
+    cli = APIClient(HOST, PORT, PASSWORD, noise_psk=NOISE_PSK)
+    try:
+        await asyncio.wait_for(cli.connect(login=True), timeout=10)
+        await cli.disconnect()
+        return iteration, True, ""
+    except Exception as e:
+        return (
+            iteration,
+            False,
+            f"client{client_id} iter{iteration}: {type(e).__name__}: {e}",
+        )
+    finally:
+        await cli.disconnect(force=True)
+
+
+async def main() -> None:
+    iterations = int(sys.argv[1]) if len(sys.argv) > 1 else ITERATIONS
+    concurrency = int(sys.argv[2]) if len(sys.argv) > 2 else CONCURRENCY
+
+    print(f"Stress testing {HOST}:{PORT}")
+    print(f"Iterations: {iterations}, Concurrency: {concurrency}")
+    print()
+
+    success = 0
+    fail = 0
+    errors: list[str] = []
+    start = time.monotonic()
+
+    sem = asyncio.Semaphore(concurrency)
+
+    async def run(client_id: int, iteration: int) -> tuple[int, bool, str]:
+        async with sem:
+            return await connect_disconnect(client_id, iteration)
+
+    tasks = [asyncio.create_task(run(i % concurrency, i)) for i in range(iterations)]
+
+    for coro in asyncio.as_completed(tasks):
+        iteration, ok, err = await coro
+        if ok:
+            success += 1
+        else:
+            fail += 1
+            errors.append(err)
+        total = success + fail
+        if total % 10 == 0 or not ok:
+            elapsed = time.monotonic() - start
+            rate = total / elapsed if elapsed > 0 else 0
+            print(f"[{total}/{iterations}] ok={success} fail={fail} ({rate:.1f}/s)")
+            if err:
+                print(f"  ERROR: {err}")
+
+    elapsed = time.monotonic() - start
+    print()
+    print(f"Done in {elapsed:.1f}s")
+    print(f"Success: {success}, Failed: {fail}, Rate: {iterations / elapsed:.1f}/s")
+
+    if errors:
+        print("\nLast 10 errors:")
+        for e in errors[-10:]:
+            print(f"  {e}")
+
+    sys.exit(1 if fail > 0 else 0)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())