Function python_exec_sync

pyhooks/pyhooks/python_server.py:202–262

(
    code: str,
    timeout: float,
    wait_after_kill=2.0,
    minimum_free_ram_bytes=100_000_000,
    log=True,
)

def python_exec_sync(
    code: str,
    timeout: float,
    wait_after_kill=2.0,
    minimum_free_ram_bytes=100_000_000,
    log=True,
) -> str:
    global ipython_shell
    with io.StringIO() as file:
        t = InterruptibleThread(target=worker, args=[code, file, timeout, log])
        t.start()

        started_waiting = time.time()
        n_wait_steps = 0

        # Go until timeout reached or memory cap reached
        ExceptionClass = PythonExecTimeoutException
        while t.is_alive() and time.time() - started_waiting < timeout:
            cur_bytes = get_available_ram_bytes()
            if cur_bytes < minimum_free_ram_bytes:
                ExceptionClass = PythonExecOutOfMemoryException
                ipython_shell = InteractiveShell.instance()
                break
            time.sleep(0.05)
            n_wait_steps += 1
            if n_wait_steps % 80 == 0:
                print(
                    f"python exec still running after {round(time.time()-started_waiting)} seconds {cur_bytes:,} bytes free",
                    # file=stderr,
                )

        # Try to kill it until we succeed or wait_after_kill exceeded
        started_waiting = time.time()
        gave_up = False
        while t.is_alive():
            t.raiseException(ExceptionClass)
            time.sleep(0.05)
            if time.time() - started_waiting > wait_after_kill:
                gave_up = True
                break

        result = file.getvalue()
        ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
        result_cleaned = re.sub(
            r"(^|\n)(Out|In)\[[0-9]+\]: ", r"\1", ansi_escape.sub("", result)
        )
        if gave_up:
            result_cleaned += "\nExecException: python exec timed out but could not be killed and is still going in the background"

        # "fix" ipython bug? causing error formatting exception
        if (
            "Unexpected exception formatting exception. Falling back to standard exception"
            in result_cleaned
        ):
            result_cleaned = result_cleaned.split(
                "During handling of the above exception, another exception occurred"
            )[0].replace(
                "Unexpected exception formatting exception. Falling back to standard exception",