Function get_instruction_count_shared_mem

evalplus/perf/profile.py:70–129

(
    profiler: Callable,
    func_code: str,
    entry_point: str,
    test_inputs: List[Any],
    timeout_second_per_test: float,
    memory_bound_gb: int,
    warmup_inputs: Optional[List[Any]],
    # shared memory
    compute_cost,  # Value("d", 0.0),
    progress,  # Value("i", 0),
)

def get_instruction_count_shared_mem(
    profiler: Callable,
    func_code: str,
    entry_point: str,
    test_inputs: List[Any],
    timeout_second_per_test: float,
    memory_bound_gb: int,
    warmup_inputs: Optional[List[Any]],
    # shared memory
    compute_cost,  # Value("d", 0.0),
    progress,  # Value("i", 0),
) -> Optional[float]:

    error = None

    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.
        maximum_memory_bytes = memory_bound_gb * 1024 * 1024 * 1024
        reliability_guard(maximum_memory_bytes=maximum_memory_bytes)
        exec_globals = {}

        # run (eval) the func def
        exec(func_code, exec_globals)
        fn = exec_globals[entry_point]

        # warmup the function
        if warmup_inputs:
            for _ in range(3):
                fn(*warmup_inputs)

        progress.value = _STAT_START
        try:  # run the function
            with time_limit(timeout_second_per_test):
                with swallow_io():
                    compute_cost.value = profiler(fn, test_inputs)
                    progress.value = _STAT_SUCC
        except TimeoutException:
            print("[Warning] Profiling hits TimeoutException")
        except MemoryError:
            print("[Warning] Profiling hits MemoryError")
        except:
            print("[CRITICAL] ! Unknown exception during profiling !")
            error = format_exc()
            print(error)

        if progress.value != _STAT_SUCC:
            progress.value = _STAT_ERROR

        # Needed for cleaning up.
        shutil.rmtree = rmtree