commit ff23039b517ef3464dcbe40d7910fdf2120ce17b
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-30T15:33:48Z |
|---|
| subject | Concurrency test use builtin to reduce pyendpoints |
|---|
commit ff23039b517ef3464dcbe40d7910fdf2120ce17b
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-30T15:33:48Z
Concurrency test use builtin to reduce pyendpoints
---
mk/common.mk | 1 +
tests/run_tests.py | 2 +-
tests/support.py | 1 +
tests/test_tcp_stress.py | 79 +++++++++++++++++++++++++++++++++++-------------
4 files changed, 61 insertions(+), 22 deletions(-)
diff --git a/mk/common.mk b/mk/common.mk
index effef9b..9fe120e 100644
--- a/mk/common.mk
+++ b/mk/common.mk
@@ -41,6 +41,7 @@ TEST_FLAGS := CONCURRENCY=100 TOTAL=100 FD_LIMIT=512
else ifeq ($(UNAME_S),FreeBSD)
STATIC ?= 0
CFLAGS += -pthread
+CFLAGS += -Wno-gnu-zero-variadic-macro-arguments
CPPFLAGS += -I/usr/local/include
LDFLAGS += -pthread
LDFLAGS += -L/usr/local/lib
diff --git a/tests/run_tests.py b/tests/run_tests.py
index dea7a12..f612992 100644
--- a/tests/run_tests.py
+++ b/tests/run_tests.py
@@ -25,7 +25,6 @@ from . import test_unix_listeners
TCP_PROXY_TEST_MODULES = [
test_tcp_basic,
- test_tcp_stress,
]
UDP_PROXY_TEST_MODULES = [
@@ -33,6 +32,7 @@ UDP_PROXY_TEST_MODULES = [
]
STANDALONE_TEST_MODULES = [
+ test_tcp_stress,
test_unix_listeners,
test_haproxy_proxy_v2,
test_udp_proxy_v2,
diff --git a/tests/support.py b/tests/support.py
index 97060c4..fabc7a7 100644
--- a/tests/support.py
+++ b/tests/support.py
@@ -484,6 +484,7 @@ async def run_default_tcp_tinyproxy(proxy_bin: str):
async with run_tinyproxy_with_conf(
proxy_bin=proxy_bin,
conf_text=conf_text,
+ args=["-w", "2"],
listen_host=LISTEN_HOST,
listen_port=PROXY_PORT,
proto="tcp",
diff --git a/tests/test_tcp_stress.py b/tests/test_tcp_stress.py
index 456b260..1ed7050 100644
--- a/tests/test_tcp_stress.py
+++ b/tests/test_tcp_stress.py
@@ -1,46 +1,83 @@
import asyncio
+import time
import os
-from .support import proxy_roundtrip
+from .support import (
+ LISTEN_HOST,
+ PROXY_PORT,
+ SkipTest,
+ run_tinyproxy_with_conf,
+)
+
+PROXY_PORT = PROXY_PORT + 1
+
+async def close_writer(writer: asyncio.StreamWriter) -> None:
+ try:
+ writer.close()
+ await writer.wait_closed()
+ except ConnectionResetError:
+ pass
async def run_one_concurrent(
i: int,
- payload_size: int,
sem: asyncio.Semaphore,
) -> None:
async with sem:
- prefix = f"worker-{i}-".encode()
- payload = (prefix * ((payload_size // len(prefix)) + 1))[:payload_size]
+ reader, writer = await asyncio.open_connection(LISTEN_HOST, PROXY_PORT)
+
+ try:
+ got = await asyncio.wait_for(reader.read(4096), timeout=10.0)
- got = await proxy_roundtrip(payload, timeout=30.0)
+ if not got:
+ raise AssertionError(f"worker {i} got empty response")
- if got != payload:
- raise AssertionError(
- f"worker {i} mismatch: expected {len(payload)} bytes, got {len(got)}"
- )
+ # Keep this weak on purpose. Other tests cover exact builtin output.
+ if b":" not in got:
+ raise AssertionError(f"worker {i} got invalid client_addr response: {got!r}")
+ finally:
+ await close_writer(writer)
async def test_concurrent_connections() -> None:
+ proxy_bin = os.environ.get("TINYPROXY_BIN")
+ if not proxy_bin:
+ raise SkipTest("TINYPROXY_BIN is not set")
+
total = int(os.environ.get("TOTAL", "10000"))
concurrency = int(os.environ.get("CONCURRENCY", "10000"))
- payload_size = int(os.environ.get("PAYLOAD_SIZE", "1024"))
- sem = asyncio.Semaphore(concurrency)
+ conf_text = (
+ f"listen"
+ f" tcp {LISTEN_HOST}:{PROXY_PORT}"
+ f" builtin client_addr\n"
+ )
+
+ started_at = time.monotonic()
+ async with run_tinyproxy_with_conf(
+ proxy_bin=proxy_bin,
+ conf_text=conf_text,
+ args=["-w", "0"],
+ listen_host=LISTEN_HOST,
+ listen_port=PROXY_PORT,
+ proto="tcp",
+ ):
+ sem = asyncio.Semaphore(concurrency)
- tasks = [
- asyncio.create_task(run_one_concurrent(i, payload_size, sem))
- for i in range(total)
- ]
+ tasks = [
+ asyncio.create_task(run_one_concurrent(i, sem))
+ for i in range(total)
+ ]
- done = 0
+ done = 0
- for task in asyncio.as_completed(tasks):
- await task
- done += 1
+ for task in asyncio.as_completed(tasks):
+ await task
+ done += 1
- if done % 1000 == 0:
- print(f" completed {done}/{total}")
+ if done % 1000 == 0:
+ elapsed = time.monotonic() - started_at
+ print(f" {elapsed:.1f}s completed ({done}/{total})")
TESTS = [