Function execute_for_runtime

tools/_experimental/evaluate_runtime.py:19–67

(
    code: str, inputs: List, warmups: List, entry_point: str
)

def execute_for_runtime(
    code: str, inputs: List, warmups: List, entry_point: str
) -> Union[str, float]:
    def unsafe_execute():
        with create_tempdir():
            # These system calls are needed when cleaning up tempdir.
            import os
            import shutil

            rmtree = shutil.rmtree
            rmdir = os.rmdir
            chdir = os.chdir
            # Disable functionalities that can make destructive changes to the test.
            reliability_guard()
            # load functions
            exec_globals = {}
            exec(code, exec_globals)
            fn = exec_globals[entry_point]
            try:
                # warmup calls
                for warmup in warmups:
                    with swallow_io():
                        fn(*warmup)

                start_time = time.time()
                # real call
                with swallow_io():
                    with time_limit(3):
                        fn(*inputs)
                duration = time.time() - start_time

                result.append(duration)
            except TimeoutException:
                result.append("timed out")
            except BaseException as e:
                result.append("thrown exception")
            # Needed for cleaning up.
            shutil.rmtree = rmtree
            os.rmdir = rmdir
            os.chdir = chdir

    manager = multiprocessing.Manager()
    result = manager.list()
    p = multiprocessing.Process(target=unsafe_execute)
    p.start()
    p.join(timeout=3 + 1)
    if p.is_alive():
        p.kill()
    return result[0]


def test_solution_runtime(