Function run_python

pyhooks/pyhooks/execs.py:106–149

exec the code in a separate thread and collect the stdout and stderr. If the code takes longer than timeout seconds to run, then we attempt to kill it. If it takes longer than wait_after_kill seconds to die, then we give up and return the result anyway. (The thread will die when its

(
    code: str,
    timeout_sec: float,
    wait_after_kill=15.0,
    minimum_free_ram_bytes=800_000_000,
    log=False,
)

async def run_python(
    code: str,
    timeout_sec: float,
    wait_after_kill=15.0,
    minimum_free_ram_bytes=800_000_000,
    log=False,
) -> str:
    """
    exec the code in a separate thread and collect the stdout and stderr.
    If the code takes longer than timeout seconds to run, then we attempt to kill it.
    If it takes longer than wait_after_kill seconds to die, then we give up and return the result anyway.
    (The thread will die when its running system call finishes executing.)

    Variables are shared between threads, so e.g. `shared_box[0] += 1` works. Note that `x += 1` won't work.
    """
    from pyhooks import Actions  # type: ignore

    await Actions().check_safety(code)

    async with aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(
            total=60 * 25, sock_connect=60 * 25, sock_read=60 * 25
        )
    ) as session:
        try:
            async with session.post(
                "http://localhost:9712/run_python",
                json={
                    "code": code,
                    "timeout": timeout_sec,
                    "wait_after_kill": wait_after_kill,
                    "minimum_free_ram_bytes": minimum_free_ram_bytes,
                    "log": log,
                },
            ) as resp:
                # encode and decode to prevent errors from unicode surrogate characters
                return (
                    (await resp.json())["result"]
                    .encode("utf-8", errors="replace")
                    .decode("utf-8")
                )
        except Exception as e:
            print(f"Error connecting to python server: {e}", file=sys.stderr)
            return f"Unknown error. May be caused by python code timeout after 25 minutes. Details: {e}"


async def _run_tests():