Upload 12 files

colab_tunnel/_diagnostics.py

@@ -84,15 +84,16 @@ def _stop_test_server(server_bin: Path) -> None:
 def _check_url_latency(
     url: str,
     timeout: float = 10.0,
-    warmup_retries: int = 2,
+    warmup_retries: int = 3,
     warmup_delay: float = 5.0,
 ) -> Optional[float]:
     """
     Проверяет доступность публичного URL и возвращает RTT.
 
-    Делает повторные попытки с паузой — некоторые провайдеры (cloudflared,
-    tunnelite) анонсируют URL раньше, чем маршрут фактически распространяется
-    по их сети. Cloudflared обычно требует 5–10 секунд после старта.
+    Некоторые провайдеры (cloudflared, tunnelite) анонсируют URL раньше,
+    чем маршрут фактически распространяется по их сети. cloudflared
+    обычно требует 5–15 секунд после получения URL.
+    Делаем несколько попыток с паузой между ними.
 
     Returns:
         RTT в секундах, или None если URL недоступен после всех попыток.
@@ -102,7 +103,9 @@ def _check_url_latency(
 
     for attempt in range(warmup_retries):
         if attempt > 0:
-            logger.debug(f'Повтор проверки доступности (попытка {attempt + 1}): {clean_url}')
+            logger.debug(
+                f'Повтор проверки доступности {attempt + 1}/{warmup_retries}: {clean_url}'
+            )
             time.sleep(warmup_delay)
         try:
             t = time.time()
@@ -112,11 +115,12 @@ def _check_url_latency(
                 return elapsed
             logger.debug(f'HTTP {resp.status_code} от {clean_url}')
         except Exception as e:
-            logger.debug(f'Попытка {attempt + 1}/{warmup_retries} не удалась: {e}')
+            logger.debug(f'Попытка {attempt + 1}: {e}')
 
     return None
 
 
+
 def _print_report(results: list[TunnelBenchmarkResult]) -> None:
     """Выводит форматированную таблицу результатов."""
     W = {'name': 14, 'status': 10, 'tunnel': 14, 'latency': 12}

colab_tunnel/_tunnels.py

@@ -37,6 +37,7 @@ def get_revproxy_url(
     write_link: bool = False,
     stdin_input: str | None = None,
     extra_env: dict[str, str] | None = None,
+    read_both_streams: bool = False,
 ) -> str:
     """
     Универсальный запуск бинарного туннельного провайдера.
@@ -56,6 +57,7 @@ def get_revproxy_url(
         stdin_input:      Строка для отправки в stdin процесса.
         extra_env: Дополнительные переменные окружения для процесса.
                    Накладываются поверх os.environ (не заменяют его).
+        read_both_streams: Читать ли оба потока.
 
     Returns:
         Публичный URL туннеля.
@@ -119,6 +121,7 @@ def get_revproxy_url(
             url_pattern=url_pattern,
             timeout=timeout,
             read_from_stderr=read_from_stderr,
+            read_both_streams=read_both_streams,  # ← передаём дальше
         )
     except RuntimeError:
         kill_process_by_name(bin_path.name)
@@ -315,7 +318,6 @@ def get_beeceptor_url(port: int) -> str:
 
 @tunnel_provider('bore')
 def get_bore_url(port: int) -> str:
-    # Для актуальной версии запрашиваем GitHub API, только если бинарника ещё нет
     bin_url = 'https://github.com/ekzhang/bore/releases/download/v0.6.0/bore-v0.6.0-x86_64-unknown-linux-musl.tar.gz'
     if not bore_bin.exists():
         try:
@@ -332,16 +334,17 @@ def get_bore_url(port: int) -> str:
         need_unpack=True,
         bin_path=bore_bin,
         start_commands=[str(bore_bin), 'local', str(port), '--to', 'bore.pub'],
-        read_from_stderr=True,  # Rust логи идут в stderr
+        # read_both_streams: bore пишет в stderr через tracing-subscriber,
+        # но читаем оба потока — на случай если конкретная сборка пишет в stdout
+        read_from_stderr=False,
+        read_both_streams=True,
         url_pattern=r'bore\.pub:\d+',
-        timeout=15.0,
-        # Bore использует Rust-крейт `tracing`. Без RUST_LOG=info строка
-        # "listening at bore.pub:XXXXX" (INFO-уровень) не выводится совсем.
-        extra_env={'RUST_LOG': 'info'},
+        timeout=20.0,
     )
     return f'http://{raw}/'
 
 
+
 @tunnel_provider('boredigital')
 def get_boredigital_url(port: int) -> str:
     return get_revproxy_url(

colab_tunnel/_utils.py

@@ -192,84 +192,120 @@ def read_until_pattern(
     read_both_streams: bool = False,
 ) -> tuple[str, str]:
     """
-    Неблокирующее чтение вывода процесса до нахождения URL-паттерна.
+    Читает вывод процесса с таймаутом, ища URL-паттерн.
 
-    Использует select() + os.set_blocking() вместо readline(),
-    что полностью исключает зависание при нехватке строк вывода.
+    Использует блокирующие read() в daemon-потоках вместо select() +
+    неблокирующих read(). select()-подход даёт гонку при быстрых коротких
+    записях (< 1мс): select сигнализирует «readable», но к моменту вызова
+    read() данные ещё не попали в буфер пайпа — read возвращает пусто.
+    Блокирующий read в потоке этой гонки не имеет: он просто ждёт данных.
+
+    После нахождения URL потоки автоматически переходят в режим дренирования:
+    продолжают читать и отбрасывать данные, предотвращая переполнение
+    системного буфера пайпа (обычно 64 КБ) за время жизни туннеля.
 
     Args:
         process:           Запущенный Popen с открытыми потоками PIPE.
-        url_pattern:       Регулярное выражение для поиска URL.
-        timeout:           Максимальное время ожидания в секундах.
+        url_pattern:       Regex для поиска URL.
+        timeout:           Максимальное время ожидания (сек).
         read_from_stderr:  Читать из stderr (иначе stdout).
-        read_both_streams: Читать из обоих потоков (для SSH).
+        read_both_streams: Читать из обоих потоков одновременно (для SSH).
 
     Returns:
-        Кортеж (matched_url, full_output).
+        Кортеж (matched_url, full_output_so_far).
 
     Raises:
-        RuntimeError: Паттерн не найден за время timeout.
+        RuntimeError: Паттерн не найден за timeout секунд.
     """
+    import queue as _queue
+
     if read_both_streams:
-        active_streams = [s for s in (process.stdout, process.stderr) if s]
+        active = [s for s in (process.stdout, process.stderr) if s]
     elif read_from_stderr:
-        active_streams = [process.stderr] if process.stderr else []
+        active = [process.stderr] if process.stderr else []
     else:
-        active_streams = [process.stdout] if process.stdout else []
+        active = [process.stdout] if process.stdout else []
 
-    if not active_streams:
+    if not active:
         raise RuntimeError('Потоки вывода процесса недоступны.')
 
-    raws: list = []
-    for stream in active_streams:
+    q: _queue.Queue[bytes | None] = _queue.Queue()
+    drain_mode = threading.Event()
+
+    def _reader(stream) -> None:
+        """
+        Блокирующий читатель в daemon-потоке.
+        Пока URL не найден — пушит чанки в очередь.
+        После установки drain_mode — читает и отбрасывает (дренирование).
+        """
         raw = getattr(stream, 'raw', stream)
         try:
-            os.set_blocking(raw.fileno(), False)
+            # Явно возвращаем в блокирующий режим на случай,
+            # если где-то раньше FD был переведён в неблокирующий
+            os.set_blocking(raw.fileno(), True)
+        except Exception:
+            pass
+        try:
+            while True:
+                chunk = raw.read(4096)
+                if not chunk:       # EOF — процесс закрыл поток
+                    break
+                if drain_mode.is_set():
+                    pass            # Дренирование: читаем и отбрасываем
+                else:
+                    q.put(chunk)    # Ищем URL: передаём в очередь
         except Exception:
             pass
-        raws.append(raw)
+        finally:
+            # Сигнал EOF только если мы всё ещё ищем URL
+            if not drain_mode.is_set():
+                q.put(None)
+
+    for stream in active:
+        t = threading.Thread(target=_reader, args=(stream,), daemon=True)
+        t.start()
 
     buffer = b''
     deadline = time.time() + timeout
+    n_sentinels = 0
 
     while time.time() < deadline:
+        remaining = max(0.01, deadline - time.time())
         try:
-            ready, _, _ = select.select(raws, [], [], 0.1)
-        except Exception:
-            break
+            item = q.get(timeout=min(remaining, 0.5))
+        except _queue.Empty:
+            if process.poll() is not None:
+                # Процесс завершился — дочитываем остатки из очереди
+                while True:
+                    try:
+                        item = q.get_nowait()
+                    except _queue.Empty:
+                        break
+                    if item is None:
+                        n_sentinels += 1
+                    else:
+                        buffer += item
+                break
+            continue
 
-        for raw in ready:
-            try:
-                chunk = raw.read(4096)
-                if chunk:
-                    buffer += chunk
-            except (BlockingIOError, ValueError):
-                pass
-            except Exception:
-                pass
+        if item is None:
+            n_sentinels += 1
+            if n_sentinels >= len(active):
+                break       # Все читатели достигли EOF
+            continue
 
+        buffer += item
         text = buffer.decode('utf-8', errors='ignore')
         match = re.search(url_pattern, text)
         if match:
+            drain_mode.set()    # Переключаем потоки в режим дренирования
             return match.group(), text
 
-        if process.poll() is not None:
-            # Процесс завершился — финальное чтение остатков
-            for raw in raws:
-                try:
-                    remaining = raw.read(65536)
-                    if remaining:
-                        buffer += remaining
-                except Exception:
-                    pass
-            text = buffer.decode('utf-8', errors='ignore')
-            match = re.search(url_pattern, text)
-            if match:
-                return match.group(), text
-            break
+    raise RuntimeError(
+        f'URL не найден за {timeout} сек. Вывод процесса:\n'
+        + buffer.decode('utf-8', errors='ignore')
+    )
 
-    output_str = buffer.decode('utf-8', errors='ignore')
-    raise RuntimeError(f'URL не найден за {timeout} сек. Вывод процесса:\n{output_str}')
 
 
 def move_path(old_path: Path | str, new_path: Path | str) -> None: