Use a managed dict for the _io types #161184

"""Real-world code paths that run Python methods on io subclasses."""
import gzip, io, sys, tempfile, zipfile, pyperf
import http.client
from urllib3.response import HTTPResponse as U3Response

CHUNK = 100
N = 1000
BODY = bytes(range(256)) * (CHUNK * N // 256)          # ~100 KB
LINES = b"".join(b"%06d: some text on this line, to be read\n" % i for i in range(N))

class FakeSocket:
    def __init__(self, data):
        self.data = data
    def makefile(self, mode):
        return io.BytesIO(self.data)
    def close(self):
        pass

HEADERS = (b"HTTP/1.1 200 OK\r\nServer: bench\r\nDate: Mon, 02 Nov 2026 10:00:00 GMT\r\n"
           b"Content-Type: text/plain\r\nCache-Control: no-cache\r\n"
           b"Connection: keep-alive\r\nX-Request-Id: 0123456789abcdef\r\n")

def chunked_body(data, size):
    out = []
    for i in range(0, len(data), size):
        c = data[i:i + size]
        out.append(b"%x\r\n%s\r\n" % (len(c), c))
    out.append(b"0\r\n\r\n")
    return b"".join(out)

RESP_CL = HEADERS + b"Content-Length: %d\r\n\r\n" % len(BODY) + BODY
RESP_LINES = HEADERS + b"Content-Length: %d\r\n\r\n" % len(LINES) + LINES
RESP_CHUNKED = HEADERS + b"Transfer-Encoding: chunked\r\n\r\n" + chunked_body(BODY, CHUNK)
RESP_SMALL = HEADERS + b"Content-Length: 2\r\n\r\nok"

def _response(raw):
    r = http.client.HTTPResponse(FakeSocket(raw), method="GET")
    r.begin()
    return r

def httpclient_begin(loops):
    """HTTPResponse(sock).begin(): one per HTTP request."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        _response(RESP_SMALL).read()
    return pyperf.perf_counter() - t0

def httpclient_read_chunked(loops):
    """read() of a chunked body with 1000 100-byte chunks."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        _response(RESP_CHUNKED).read()
    return pyperf.perf_counter() - t0

def httpclient_read_small(loops):
    """read(100) x 1000 on a Content-Length body."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        resp = _response(RESP_CL)
        read = resp.read
        while read(CHUNK):
            pass
    return pyperf.perf_counter() - t0

def httpclient_iter_lines(loops):
    """for line in response: (1000 lines)."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        for line in _response(RESP_LINES):
            pass
    return pyperf.perf_counter() - t0

# --- zipfile ---------------------------------------------------------------
_zbuf = io.BytesIO()
with zipfile.ZipFile(_zbuf, "w", zipfile.ZIP_DEFLATED) as zf:
    for i in range(200):
        zf.writestr("member%03d.txt" % i, b"x" * 100)
    zf.writestr("lines.txt", LINES)
    zf.writestr("body.bin", BODY)
ZF = zipfile.ZipFile(_zbuf)
MEMBERS = [n for n in ZF.namelist() if n.startswith("member")]

def zipfile_read_members(loops):
    """zf.read(name) for 200 small members (ZipExtFile created per member)."""
    r = range(loops); read = ZF.read; t0 = pyperf.perf_counter()
    for _ in r:
        for name in MEMBERS:
            read(name)
    return pyperf.perf_counter() - t0

def zipfile_iter_lines(loops):
    """for line in zf.open(name): (1000 lines)."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        with ZF.open("lines.txt") as f:
            for line in f:
                pass
    return pyperf.perf_counter() - t0

def zipfile_read_small(loops):
    """f.read(100) x 1000 from a deflated member."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        with ZF.open("body.bin") as f:
            read = f.read
            while read(CHUNK):
                pass
    return pyperf.perf_counter() - t0

# --- gzip -------------------------------------------------------------------
GZ = gzip.compress(BODY)

def gzip_read_small(loops):
    """GzipFile.read(100) x 1000."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        with gzip.GzipFile(fileobj=io.BytesIO(GZ)) as f:
            read = f.read
            while read(CHUNK):
                pass
    return pyperf.perf_counter() - t0

def gzip_write_small(loops):
    """GzipFile.write(100 bytes) x 1000."""
    r = range(loops); piece = BODY[:CHUNK]; t0 = pyperf.perf_counter()
    for _ in r:
        with gzip.GzipFile(fileobj=io.BytesIO(), mode="wb") as f:
            write = f.write
            for _ in range(N):
                write(piece)
    return pyperf.perf_counter() - t0

# --- tempfile ---------------------------------------------------------------
def spooled_write_small(loops):
    """SpooledTemporaryFile.write(100 bytes) x 1000 (stays in memory)."""
    r = range(loops); piece = BODY[:CHUNK]; t0 = pyperf.perf_counter()
    for _ in r:
        with tempfile.SpooledTemporaryFile(max_size=10**7) as f:
            write = f.write
            for _ in range(N):
                write(piece)
    return pyperf.perf_counter() - t0

# --- urllib3 ----------------------------------------------------------------
def _u3():
    return U3Response(body=io.BytesIO(BODY), headers={"content-length": str(len(BODY))},
                      status=200, preload_content=False, decode_content=True)

def urllib3_read_small(loops):
    """urllib3 HTTPResponse.read(100) x 1000."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        resp = _u3(); read = resp.read
        while read(CHUNK):
            pass
    return pyperf.perf_counter() - t0

def urllib3_stream(loops):
    """for chunk in resp.stream(100): (1000 chunks)."""
    r = range(loops); t0 = pyperf.perf_counter()
    for _ in r:
        for chunk in _u3().stream(CHUNK):
            pass
    return pyperf.perf_counter() - t0

def control_list(loops):
    """Unaffected code: noise floor of the two-binary comparison."""
    r = range(loops); q = range(20000); t0 = pyperf.perf_counter()
    for _ in r:
        list(q); list(q); list(q); list(q); list(q)
    return pyperf.perf_counter() - t0

BENCHES = (httpclient_begin, httpclient_read_chunked, httpclient_read_small, httpclient_iter_lines,
           zipfile_read_members, zipfile_iter_lines, zipfile_read_small,
           gzip_read_small, gzip_write_small, spooled_write_small,
           urllib3_read_small, urllib3_stream, control_list)

if __name__ == "__main__":
    if "--check" in sys.argv:
        for f in BENCHES:
            print(f.__name__, "%.3f ms" % (f(1) * 1e3))
        sys.exit()
    runner = pyperf.Runner()
    for f in BENCHES:
        runner.bench_time_func(f.__name__, f)