# find/repro.py
# Server端不处理
import threading
import requests
import sys
import time
import socket
from http.server import HTTPServer, BaseHTTPRequestHandler

PORT = 8022

def start_hang_server():
    class HangHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            # Server 端卡死
            time.sleep(30)
    server = HTTPServer(("localhost", PORT), HangHandler)
    server.serve_forever()

threading.Thread(target=start_hang_server).start()
time.sleep(1)

session = requests.Session()
stop_flag = False
lock = threading.Lock()

def open_conn():
    global stop_flag
    try:
        resp = session.get(
            f"http://localhost:{PORT}",
            stream=True,
            timeout=10,
        )
        for v in resp.iter_lines():
            if v:
                print(f"Received: {v.decode('utf-8')}")
    except Exception as e:
        print(f"Got exception in thread: {e!r}")
        # 如果是 Errno 16，就能看到类似 Device or resource busy
        with lock:
            stop_flag = True

count = 0
threads = []
while True:
    if stop_flag:
        print("Encountered failure, exiting loop.")
        break
    t = threading.Thread(target=open_conn)
    t.daemon = True
    t.start()
    threads.append(t)
    count += 1
    print(f"Opened connections: {count}")
    time.sleep(0.1)


for t in threads:
    t.join()

# find/main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time


app = FastAPI()

def generate_stream():
    for i in range(3):
        yield f"{i}\n"
        time.sleep(1)


@app.get("/")
def stream_text():
    return StreamingResponse(generate_stream(), media_type="text/plain")

# find/client.py
import requests
import threading
import time

session = requests.Session()
adapter = requests.adapters.HTTPAdapter(pool_connections=2, pool_maxsize=2)
session.mount("http://", adapter)
session.mount("https://", adapter)

url = "http://localhost:8022"

def make_leak(i):
    try:
        print(f"[Thread-{i}] start")
        resp = session.get(url, stream=True)
        for line in resp.iter_lines():
            print(f"[Thread-{i}] line: {line}")
            if b"2" in line:
                # 模拟异常中断，连接不释放
                raise Exception("simulate disconnect")
    except Exception as e:
        print(f"[Thread-{i}] Exception: {e}")

threads = []
for i in range(100):
    t = threading.Thread(target=make_leak, args=(i,))
    t.start()
    time.sleep(0.1)
    threads.append(t)

for t in threads:
    t.join()

# sentry_sdk/utils.py
def _is_contextvars_broken():
    # type: () -> bool
    """
    Returns whether gevent/eventlet have patched the stdlib in a way where thread locals are now more "correct" than contextvars.
    """
    try:
        import gevent  # type: ignore
        from gevent.monkey import is_object_patched  # type: ignore

        # Get the MAJOR and MINOR version numbers of Gevent
        version_tuple = tuple(
            [int(part) for part in re.split(r"a|b|rc|\.", gevent.__version__)[:2]]
        )
        if is_object_patched("threading", "local"):
            # Gevent 20.9.0 depends on Greenlet 0.4.17 which natively handles switching
            # context vars when greenlets are switched, so, Gevent 20.9.0+ is all fine.
            # Ref: https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/monkey.py#L604-L609
            # Gevent 20.5, that doesn't depend on Greenlet 0.4.17 with native support
            # for contextvars, is able to patch both thread locals and contextvars, in
            # that case, check if contextvars are effectively patched.
            if (
                # Gevent 20.9.0+
                (sys.version_info >= (3, 7) and version_tuple >= (20, 9))
                # Gevent 20.5.0+ or Python < 3.7
                or (is_object_patched("contextvars", "ContextVar"))
            ):
                return False

            return True
    except ImportError:
        pass

    try:
        import greenlet  # type: ignore
        from eventlet.patcher import is_monkey_patched  # type: ignore

        greenlet_version = parse_version(greenlet.__version__)

        if greenlet_version is None:
            logger.error(
                "Internal error in Sentry SDK: Could not parse Greenlet version from greenlet.__version__."
            )
            return False

        if is_monkey_patched("thread") and greenlet_version < (0, 5):
            return True
    except ImportError:
        pass

    return False


def _get_contextvars():
    # type: () -> Tuple[bool, type]
    """
    Figure out the "right" contextvars installation to use. Returns a
    `contextvars.ContextVar`-like class with a limited API.

    See https://docs.sentry.io/platforms/python/contextvars/ for more information.
    """
    if not _is_contextvars_broken():
        # aiocontextvars is a PyPI package that ensures that the contextvars
        # backport (also a PyPI package) works with asyncio under Python 3.6
        #
        # Import it if available.
        if sys.version_info < (3, 7):
            # `aiocontextvars` is absolutely required for functional
            # contextvars on Python 3.6.
            try:
                from aiocontextvars import ContextVar

                return True, ContextVar
            except ImportError:
                pass
        else:
            # On Python 3.7 contextvars are functional.
            try:
                from contextvars import ContextVar

                return True, ContextVar
            except ImportError:
                pass

    # Fall back to basic thread-local usage.

    from threading import local

    return False, _make_threadlocal_contextvars(local)


HAS_REAL_CONTEXTVARS, ContextVar = _get_contextvars()

import time

def fake_sleep(seconds):
    print(f"pretend sleeping for {seconds} seconds")

# 猴子补丁：替换原来的 time.sleep
time.sleep = fake_sleep

# 原本应该阻塞的代码，现在变成打印
time.sleep(3)   # 输出：pretend sleeping for 3 seconds

# ……其他一些包导入，包含项目文件
import requests

def send_log():
    resp = requests.get("https://www.baidu.com")
    print(f"send log to SLS success.")

# strange/main.py
from threading import Thread
import queue

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import sentry_sdk
import requests


def send_log():
    resp = requests.get("https://www.baidu.com")
    print(f"send log success.")


class Worker:
    def __init__(self):
        sentry_sdk.init(
            dsn="http://example@sentry.io/1",
            auto_enabling_integrations=False
        )
        self.input_queue = queue.Queue()

        def worker():
            while True:
                q, response_queue = self.input_queue.get()
                if q is None:
                    break
                for i in range(3):
                    token = f"token-{i}"
                    response_queue.put(token)
                response_queue.put(None)
        Thread(target=worker, daemon=True).start()
    
    def run(self, q: str):
        response_queue = queue.Queue()
        self.input_queue.put((q, response_queue))
        
        while True:
            token = response_queue.get()
            if token is None:
                break
            yield f"data: {token}\n\n"
        
        Thread(
            target=send_log,
            args=(),
        ).start()


worker = Worker()

app = FastAPI()


@app.post("/", response_model=None)
def llm_svc(q: str):
    return StreamingResponse(
        worker.run(q), 
        media_type="text/event-stream", 
    )

# strange/client.py
import requests

def call_service(q: str):  
    url = "http://127.0.0.1:9321/?q=q"
    response = requests.post(url, stream=True)
    for chunk in response.iter_lines():
        if chunk:
            yield chunk.decode("utf-8")


if __name__ == "__main__":    
    for i in range(10):
        gen = call_service("hello")
        for cont in gen:
            print(f"-->receive cont: {cont!r}")

# strange/main_simple.py
import threading

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

import sentry_sdk
import requests

sentry_sdk.init(
    dsn="http://example@sentry.io/1",
    auto_enabling_integrations=False
)
session = requests.Session()


def send_log():
    resp = session.get("https://www.baidu.com")
    print(f"send log success.")


app = FastAPI()


@app.post("/")
def stream_tokens(q: str):
    def generator():
        for i in range(3):
            yield f"data: token-{i}\n\n"
        threading.Thread(target=send_log).start()
    return StreamingResponse(generator(), media_type="text/event-stream")

# why/mini.py
import os
import threading
import time
import subprocess

import sentry_sdk
import requests


def send_log():
    requests.get("https://www.baidu.com")
    print("send_log done")

def show_fds():
    pid = os.getpid()
    cmd = f"lsof -p {pid} | awk '{{print $5}}' | sort | uniq -c | sort -nr"
    out = subprocess.check_output(cmd, shell=True)
    print(out.decode())


if __name__ == "__main__":
    print("初始 FD：")
    show_fds()

    # 多线程触发
    threads = [threading.Thread(target=send_log) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    time.sleep(1)
    print("\n触发请求后 FD：")
    show_fds()

from eventlet.patcher import is_monkey_patched

# 换成导入eventlet也是一样的
import eventlet
# from eventlet.patcher import is_monkey_patched 
import urllib3


def send_log():
    try:
        urllib3.request("GET", "127.0.0.1")
    except Exception as e:
        pass
    print("send_log done")

# from eventlet.patcher import is_monkey_patched
# from eventlet import convenience
# from eventlet.green import socket
# from eventlet.support import greendns
# from eventlet import greenio
# from eventlet import patcher

# eventlet/support/greendns.py

from eventlet import patcher
from eventlet.green import _socket_nodns


def import_patched(module_name):
    # Import cycle note: it's crucial to use _socket_nodns here because
    # regular evenlet.green.socket imports *this* module and if we imported
    # it back we'd end with an import cycle (socket -> greendns -> socket).
    # We break this import cycle by providing a restricted socket module.
    
    modules = {
        'select': select,
        'time': time,
        'os': os,
        # 'socket': _socket_nodns,
        'ssl': ssl,
    }
    return patcher.import_patched(module_name, **modules)


dns = import_patched('dns')

for pkg in dns.__all__:
    if pkg == 'rdtypes':
        continue
    # if pkg in ["asyncresolver", "asyncquery", "e164", "query", "resolver"]:
    #     continue
    setattr(dns, pkg, import_patched('dns.' + pkg))

# why/mini2.py
import os, sys
import threading
import time
import subprocess

import eventlet
import urllib3
print("we import urllib3:", urllib3.socket)


def send_log():
    try:
        urllib3.request("GET", "127.0.0.1")
    except Exception as e:
        pass


def show_fds():
    pid = os.getpid()
    cmd = f"lsof -p {pid} | awk '{{print $5}}' | sort | uniq -c | sort -nr"
    out = subprocess.check_output(cmd, shell=True)
    print(out.decode())


if __name__ == "__main__":
    print("初始 FD：")
    show_fds()

    # 多线程触发
    threads = [threading.Thread(target=send_log) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    time.sleep(1)
    print("\n触发请求后 FD：")
    show_fds()

# eventlet/support/greendns.py

print("greendns import dns: import_patched(dns) start")
dns = import_patched('dns')
print("greendns import dns: import_patched(dns) end")

for pkg in dns.__all__:
    if pkg == 'rdtypes':
        continue
    if pkg == "query":
        print("greendns setattr start")
    setattr(dns, pkg, import_patched('dns.' + pkg))
    print("greendns setattr end")


# dns/__init__.py 最后加一行

import socket
print("dns import socket: ", socket)

# dns/query.py
try:
	print("`dns` import urllib3")
    import urllib3
    # import requests
    # from requests_toolbelt.adapters.source import SourceAddressAdapter
    # from requests_toolbelt.adapters.host_header_ssl import HostHeaderSSLAdapter


# urllib3/__init__.py 加一行
import socket
print("urlib3 import socket: ", socket)

import eventlet
eventlet.monkey_patch()

eventlet.monkey_patch(thread=True)

# eventlet/patcher.py
accepted_args = {
    "os",
    "select",
    "socket",
    "thread",
    "time",
    "psycopg",
    "MySQLdb",
    "builtins",
    "subprocess",
}

# why/mini_selectors.py
import os
import subprocess
import threading
import selectors # select的高层抽象模块
import time


def show_fds():
    pid = os.getpid()
    cmd = f"lsof -p {pid} | awk '{{print $5}}' | sort | uniq -c | sort -nr"
    out = subprocess.check_output(cmd, shell=True)
    print(out.decode())

def create_epoll():
    sel = selectors.DefaultSelector()
    # 这里不关闭
    # sel.close()

if __name__ == "__main__":
    print("初始 FD：")
    show_fds()

    threads = [threading.Thread(target=create_epoll) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    time.sleep(0.5)
    print("\n触发请求后 FD：")
    show_fds()

# who_import_me.py
import builtins
import traceback

original_import = builtins.__import__

def traced_import(name, globals=None, locals=None, fromlist=(), level=0):
    stack = "".join(traceback.format_stack(limit=5))
    print(f"Importing {name}, called from:\n{stack}")
    return original_import(name, globals, locals, fromlist, level)

builtins.__import__ = traced_import

import sentry_sdk