⚡️ Speed up method Tunnel._read_url_from_tunnel_stream by 20%

[codeflash-ai]

📄 20% (0.20x) speedup for Tunnel._read_url_from_tunnel_stream in gradio/tunneling.py

⏱️ Runtime : 2.23 milliseconds → 1.86 milliseconds (best of 5 runs)

📝 Explanation and details

The optimized code achieves a 19% speedup through several key micro-optimizations that reduce overhead in the tight polling loop:

Key Optimizations:

1. Pre-compiled regex: Moving re.compile(r"start proxy success: (.+)\n") outside the loop eliminates repeated compilation overhead. The line profiler shows this saves time on the 24 regex search operations.

2. Reduced attribute lookups: Caching self.proc.stdout as proc_stdout eliminates repeated attribute access in the hot loop, which executes millions of times according to the profiler.

3. Optimized empty line handling: The new logic checks if not line_bytes before decoding, avoiding unnecessary UTF-8 decoding for empty responses. Additionally, it strips the line once and reuses log_line for all substring checks.

4. CPU-friendly polling: When proc_stdout is None, the code now uses time.sleep(0.01) instead of busy-waiting, reducing CPU usage while waiting for the process to initialize.

5. Early empty line filtering: By checking if not log_line after stripping, the code skips processing completely empty lines more efficiently.

Performance Impact by Test Case:

• Large-scale tests show the biggest gains: 11.6-34.7% faster on tests with many irrelevant/empty lines, where the optimizations compound over thousands of loop iterations
• Basic cases remain comparable: Small overhead (1-5%) on simple cases due to the regex compilation setup cost
• Edge cases benefit moderately: 6-23% improvement in various error conditions

The optimizations are particularly effective for this tunneling use case where the function may need to process hundreds of log lines before finding the success message, making the per-iteration savings significant.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 67 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import re
import sys
import time

# imports
import pytest  # used for our unit tests
from gradio.tunneling import Tunnel

# function to test
TUNNEL_TIMEOUT_SECONDS = 30
TUNNEL_ERROR_MESSAGE = (
    "Could not create share URL. "
    "Please check the appended log from frpc for more information:"
)

class DummyStdout:
    """A dummy stdout object to simulate proc.stdout.readline()."""
    def __init__(self, lines):
        # lines: list of bytes objects to be returned by readline
        self._lines = lines
        self._index = 0

    def readline(self):
        if self._index < len(self._lines):
            line = self._lines[self._index]
            self._index += 1
            return line
        return b''  # Simulate EOF or no output

class DummyProc:
    """A dummy proc object to simulate process with stdout."""
    def __init__(self, lines=None, stdout=None):
        # lines: list of bytes objects to be returned by stdout.readline
        # stdout: override stdout directly
        if stdout is not None:
            self.stdout = stdout
        elif lines is not None:
            self.stdout = DummyStdout(lines)
        else:
            self.stdout = None
from gradio.tunneling import Tunnel

# unit tests

# ---------------------- Basic Test Cases ----------------------

def test_basic_success():
    """Test that a normal stream with a valid 'start proxy success' line returns the URL."""
    success_line = b"start proxy success: https://share.url/\n"
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=[success_line])
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 9.08μs -> 9.55μs (4.91% slower)

def test_basic_success_with_leading_lines():
    """Test that the function skips irrelevant lines before the success line."""
    lines = [
        b"random log line\n",
        b"another message\n",
        b"start proxy success: https://abc.def/\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 8.49μs -> 8.59μs (1.16% slower)

def test_basic_success_with_trailing_lines():
    """Test that the function stops reading after finding the success line."""
    lines = [
        b"start proxy success: https://url.com/\n",
        b"should not be read\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 6.58μs -> 6.59μs (0.091% slower)

def test_basic_success_with_whitespace():
    """Test that the function works with whitespace in the line."""
    lines = [
        b"   start proxy success: https://white.space/\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 6.72μs -> 6.80μs (1.12% slower)

def test_basic_success_multiple_success_lines():
    """Test that the function returns the first URL when multiple success lines are present."""
    lines = [
        b"start proxy success: https://first.url/\n",
        b"start proxy success: https://second.url/\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 6.55μs -> 6.38μs (2.62% faster)

# ---------------------- Edge Test Cases ----------------------

def test_edge_no_proc():
    """Test that the function raises AssertionError if proc is None."""
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = None
    with pytest.raises(AssertionError):
        t._read_url_from_tunnel_stream() # 2.53μs -> 3.29μs (23.2% slower)

def test_edge_proc_stdout_none():
    """Test that the function continues if proc.stdout is None."""
    # We'll simulate that after some time, proc.stdout becomes available
    class FlippingProc:
        def __init__(self, lines):
            self._lines = lines
            self._index = 0
            self.stdout = None
            self.calls = 0
        def flip(self):
            # After 2 calls, enable stdout
            self.calls += 1
            if self.calls == 3:
                self.stdout = DummyStdout(self._lines)

    t = Tunnel("host", 1, "localhost", 2, "token", None)
    flipping_proc = FlippingProc([b"start proxy success: https://url.com/\n"])
    t.proc = flipping_proc

    # Patch the method to flip stdout after a few calls
    orig_read_url = t._read_url_from_tunnel_stream
    def patched_read_url():
        start_timestamp = time.time()
        log = []
        url = ""
        def _raise_tunnel_error():
            log_text = "\n".join(log)
            print(log_text, file=sys.stderr)
            raise ValueError(f"{TUNNEL_ERROR_MESSAGE}\n{log_text}")
        while url == "":
            if time.time() - start_timestamp >= TUNNEL_TIMEOUT_SECONDS:
                _raise_tunnel_error()
            if t.proc.stdout is None:
                t.proc.flip()
                continue
            line = t.proc.stdout.readline()
            line = line.decode("utf-8")
            if line == "":
                continue
            log.append(line.strip())
            if "start proxy success" in line:
                result = re.search("start proxy success: (.+)\n", line)
                if result is None:
                    _raise_tunnel_error()
                else:
                    url = result.group(1)
            elif "login to server failed" in line:
                _raise_tunnel_error()
        return url
    t._read_url_from_tunnel_stream = patched_read_url
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 10.7μs -> 9.91μs (7.49% faster)

def test_edge_empty_lines():
    """Test that the function skips empty lines and continues reading."""
    lines = [
        b"\n",
        b"    \n",
        b"start proxy success: https://empty.com/\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 8.83μs -> 8.36μs (5.70% faster)

def test_edge_login_failed():
    """Test that the function raises ValueError if 'login to server failed' is found."""
    lines = [
        b"random log\n",
        b"login to server failed: error details\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    with pytest.raises(ValueError) as excinfo:
        t._read_url_from_tunnel_stream() # 16.9μs -> 18.0μs (5.96% slower)

def test_edge_success_line_malformed():
    """Test that the function raises ValueError if the success line is malformed."""
    lines = [
        b"start proxy success: \n",  # No URL
        b"start proxy success:\n",   # No URL, no space
        b"start proxy success\n",    # No colon, no URL
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    with pytest.raises(ValueError) as excinfo:
        t._read_url_from_tunnel_stream() # 14.4μs -> 14.3μs (1.24% faster)

def test_edge_timeout(monkeypatch):
    """Test that the function raises ValueError on timeout."""
    lines = [b"\n"] * 10  # Only empty lines, never a success
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    # Monkeypatch time.time to simulate timeout
    start = [0]
    def fake_time():
        start[0] += TUNNEL_TIMEOUT_SECONDS + 1
        return start[0]
    monkeypatch.setattr(time, "time", fake_time)
    with pytest.raises(ValueError) as excinfo:
        t._read_url_from_tunnel_stream() # 9.42μs -> 11.9μs (20.8% slower)

def test_edge_non_utf8_line():
    """Test that the function raises UnicodeDecodeError if a non-UTF8 line is read."""
    lines = [
        b"\xff\xfe\xfd",  # Invalid UTF-8
        b"start proxy success: https://url.com/\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    with pytest.raises(UnicodeDecodeError):
        t._read_url_from_tunnel_stream() # 7.48μs -> 8.23μs (9.13% slower)

def test_edge_success_line_with_extra_text():
    """Test that the function extracts the URL even if extra text follows."""
    lines = [
        b"start proxy success: https://url.com/ extra stuff\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 7.29μs -> 6.98μs (4.53% faster)

def test_edge_success_line_with_newline_in_url():
    """Test that the function extracts the URL even if it contains newline (shouldn't happen, but test anyway)."""
    lines = [
        b"start proxy success: https://url.com/\nmore text\n"
    ]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 6.54μs -> 6.44μs (1.66% faster)

# ---------------------- Large Scale Test Cases ----------------------

def test_large_scale_many_irrelevant_lines():
    """Test that the function can handle many irrelevant lines before the success line."""
    irrelevant = [f"irrelevant line {i}\n".encode("utf-8") for i in range(500)]
    lines = irrelevant + [b"start proxy success: https://large.url/\n"]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 273μs -> 245μs (11.6% faster)

def test_large_scale_many_empty_lines():
    """Test that the function can handle many empty lines before the success line."""
    lines = [b"\n"] * 800 + [b"start proxy success: https://empty.large.url/\n"]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 406μs -> 301μs (34.7% faster)

def test_large_scale_multiple_success_lines():
    """Test that the function returns the first URL when many success lines are present."""
    lines = [b"start proxy success: https://url%d.com/\n" % i for i in range(100)]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 6.59μs -> 6.91μs (4.68% slower)

def test_large_scale_long_url():
    """Test that the function works with a very long URL."""
    long_url = "https://long.url/" + "a" * 900
    lines = [f"start proxy success: {long_url}\n".encode("utf-8")]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output # 7.24μs -> 7.87μs (8.03% slower)

def test_large_scale_all_edge_cases():
    """Test that the function can handle a mix of edge cases in a large stream."""
    lines = []
    # Add 100 irrelevant lines
    lines += [f"not important {i}\n".encode("utf-8") for i in range(100)]
    # Add some empty lines
    lines += [b"\n"] * 100
    # Add some malformed success lines
    lines += [b"start proxy success:\n"] * 10
    # Add a valid success line
    lines += [b"start proxy success: https://final.url/\n"]
    t = Tunnel("host", 1, "localhost", 2, "token", None)
    t.proc = DummyProc(lines=lines)
    codeflash_output = t._read_url_from_tunnel_stream(); url = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import re
import sys
import time

# imports
import pytest
from gradio.tunneling import Tunnel

# function to test
TUNNEL_TIMEOUT_SECONDS = 30
TUNNEL_ERROR_MESSAGE = (
    "Could not create share URL. "
    "Please check the appended log from frpc for more information:"
)

class DummyStdout:
    """
    Dummy stdout to simulate the behavior of a process's stdout stream.
    """
    def __init__(self, lines):
        # lines should be a list of bytes objects to simulate .readline()
        self.lines = lines
        self.index = 0

    def readline(self):
        if self.index < len(self.lines):
            line = self.lines[self.index]
            self.index += 1
            return line
        else:
            return b""  # Simulate EOF or no new output

class DummyProc:
    """
    Dummy process to simulate the proc object with stdout.
    """
    def __init__(self, lines=None, stdout=None):
        # lines: list of bytes objects to simulate .readline()
        if stdout is not None:
            self.stdout = stdout
        elif lines is not None:
            self.stdout = DummyStdout(lines)
        else:
            self.stdout = None
from gradio.tunneling import Tunnel

# unit tests

# ---------- Basic Test Cases ----------

def test_basic_success_single_line():
    """
    Test: Basic success case with a single valid 'start proxy success' line.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    # Simulate stdout with one valid line
    url = "https://example.com"
    line = f"start proxy success: {url}\n".encode("utf-8")
    tunnel.proc = DummyProc([line])
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 7.39μs -> 7.83μs (5.64% slower)

def test_basic_success_with_irrelevant_lines():
    """
    Test: Success case with irrelevant lines before the valid url line.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://abc.com"
    lines = [
        b"some random log\n",
        b"another message\n",
        f"start proxy success: {url}\n".encode("utf-8"),
    ]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 9.02μs -> 8.53μs (5.70% faster)

def test_basic_success_with_multiple_success_lines():
    """
    Test: Multiple 'start proxy success' lines; should return the first URL.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url1 = "https://first.com"
    url2 = "https://second.com"
    lines = [
        f"start proxy success: {url1}\n".encode("utf-8"),
        f"start proxy success: {url2}\n".encode("utf-8"),
    ]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 7.41μs -> 6.69μs (10.8% faster)

# ---------- Edge Test Cases ----------

def test_no_proc_raises_assertion():
    """
    Test: If self.proc is None, should raise AssertionError.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    tunnel.proc = None
    with pytest.raises(AssertionError):
        tunnel._read_url_from_tunnel_stream() # 2.50μs -> 3.24μs (23.0% slower)


def test_login_failed_raises_valueerror():
    """
    Test: If 'login to server failed' line is read, should raise ValueError.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    lines = [
        b"connecting...\n",
        b"login to server failed: invalid token\n",
    ]
    tunnel.proc = DummyProc(lines)
    with pytest.raises(ValueError) as excinfo:
        tunnel._read_url_from_tunnel_stream() # 20.3μs -> 22.1μs (7.93% slower)

def test_malformed_success_line_raises_valueerror():
    """
    Test: If 'start proxy success' line is present but malformed, should raise ValueError.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    lines = [
        b"start proxy success: \n",  # No URL present
    ]
    tunnel.proc = DummyProc(lines)
    with pytest.raises(ValueError) as excinfo:
        tunnel._read_url_from_tunnel_stream() # 14.5μs -> 14.3μs (1.46% faster)

def test_empty_lines_are_skipped():
    """
    Test: Empty lines should be skipped and not affect url extraction.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://skip.com"
    lines = [
        b"\n",
        b"",
        b"   \n",
        f"start proxy success: {url}\n".encode("utf-8"),
    ]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 10.7μs -> 10.0μs (6.71% faster)


def test_success_line_with_extra_whitespace():
    """
    Test: Success line with extra whitespace before/after URL.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://whitespace.com"
    line = f"start proxy success:   {url}   \n".encode("utf-8")
    tunnel.proc = DummyProc([line])
    # The regex expects no extra whitespace, so this should fail
    with pytest.raises(ValueError):
        tunnel._read_url_from_tunnel_stream()

def test_success_line_with_newline_in_url():
    """
    Test: Success line with embedded newline in URL should fail.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://new\nline.com"
    line = f"start proxy success: {url}\n".encode("utf-8")
    tunnel.proc = DummyProc([line])
    # Embedded newline in URL should not match regex
    with pytest.raises(ValueError):
        tunnel._read_url_from_tunnel_stream()

# ---------- Large Scale Test Cases ----------

def test_large_number_of_irrelevant_lines():
    """
    Test: Large number of irrelevant lines before the success line.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://large.com"
    # 999 irrelevant lines, then the success line
    lines = [b"noise\n"] * 999 + [f"start proxy success: {url}\n".encode("utf-8")]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 533μs -> 468μs (14.0% faster)

def test_large_number_of_empty_lines():
    """
    Test: Large number of empty lines before the success line.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    url = "https://empty.com"
    lines = [b"\n"] * 999 + [f"start proxy success: {url}\n".encode("utf-8")]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 510μs -> 388μs (31.5% faster)

def test_large_number_of_success_lines():
    """
    Test: Large number of valid success lines; should return the first URL.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    urls = [f"https://site{i}.com" for i in range(1000)]
    lines = [f"start proxy success: {url}\n".encode("utf-8") for url in urls]
    tunnel.proc = DummyProc(lines)
    codeflash_output = tunnel._read_url_from_tunnel_stream(); result = codeflash_output # 7.08μs -> 7.18μs (1.43% slower)

def test_large_scale_login_failed_among_noise():
    """
    Test: Large number of irrelevant lines, then a login failed line.
    Should raise ValueError.
    """
    tunnel = Tunnel("host", 1, "localhost", 2, "token", None)
    lines = [b"noise\n"] * 500 + [b"login to server failed: something\n"]
    tunnel.proc = DummyProc(lines)
    with pytest.raises(ValueError) as excinfo:
        tunnel._read_url_from_tunnel_stream() # 293μs -> 239μs (22.5% faster)

To edit these changes git checkout codeflash/optimize-Tunnel._read_url_from_tunnel_stream-mhlf8tor and push.