gperf_heap

gperf_heap

Classes:

Name	Description
GperfHeapEnvironment	GperfHeapEnvironment is a class that sets up and manages the execution environment for gperf heap profiling.

panther.plugins.environments.execution_environment.gperf_heap.gperf_heap.GperfHeapEnvironment

GperfHeapEnvironment(env_config_to_test: GperfHeapConfig, output_dir: str, env_type: str, env_sub_type: str, event_manager: EventManager)

Bases: IExecutionEnvironment, ABC

GperfHeapEnvironment is a class that sets up and manages the execution environment for gperf heap profiling.

Attributes:

Name	Type	Description
global_config	GlobalConfig	The global configuration for the environment.
env_config_to_test	GperfHeapConfig	The specific configuration for the environment to test.
services_managers	list[IServiceManager]	List of service managers.
test_config	TestConfig	The test configuration.
plugin_loader	PluginLoader	The plugin loader.
logger	Logger	Logger for debugging and information.

Methods:

Name	Description
setup_environment	list[IServiceManager], test_config: TestConfig, global_config: GlobalConfig, timestamp: str, plugin_loader: PluginLoader): Sets up the environment with the provided service managers, test configuration, global configuration, timestamp, and plugin loader.
to_command	str) -> str: Generates the gperf command based on the configuration.
__repr__	Returns a string representation of the GperfHeapEnvironment instance.

Methods:

Name	Description
is_network_environment	Returns True if the plugin is an network environment.
setup_environment
teardown_environment	Tears down the environment after experiments are completed.
to_command	Generate the gperf command based on the configuration.

Source code in panther/plugins/environments/execution_environment/gperf_heap/gperf_heap.py

def __init__(
    self,
    env_config_to_test: GperfHeapConfig,
    output_dir: str,
    env_type: str,
    env_sub_type: str,
    event_manager: EventManager,
):
    super().__init__(
        env_config_to_test, output_dir, env_type, env_sub_type, event_manager
    )
    self.global_config = None
    self.env_config_to_test = env_config_to_test

is_network_environment

is_network_environment()

Returns True if the plugin is an network environment.

Source code in panther/plugins/environments/execution_environment/execution_environment_interface.py

def is_network_environment(self):
    """
    Returns True if the plugin is an network environment.
    """
    return False

setup_environment

setup_environment(services_managers: list[IServiceManager], test_config: TestConfig, global_config: GlobalConfig, timestamp: str, plugin_loader: PluginLoader)

Source code in panther/plugins/environments/execution_environment/gperf_heap/gperf_heap.py

def setup_environment(
    self,
    services_managers: list[IServiceManager],
    test_config: TestConfig,
    global_config: GlobalConfig,
    timestamp: str,
    plugin_loader: PluginLoader,
):
    """ """
    self.services_managers: list[IServiceManager] = services_managers
    self.test_config = test_config
    self.plugin_loader = plugin_loader
    self.global_config = global_config
    self.logger.debug("Setup environment with:")
    self.logger.debug(f"Services config: {self.env_config_to_test}")

    for service in self.services_managers:
        self.logger.debug(f"Service cmds: {service.run_cmd}")
        if service.service_config_to_test.implementation.gperf_compatible:
            service.environments["GPERF"] = True
            service.run_cmd["run_cmd"]["command_env"][
                "HEAPPROFILE"
            ] = f"/app/logs/{service.service_name}_heap.prof"
            # service.run_cmd["pre_run_cmds"] = service.run_cmd["pre_run_cmds"] + [self.to_command(service.service_name)]
            service.run_cmd["post_run_cmds"] = service.run_cmd["post_run_cmds"] + [
                f"pprof --pdf /app/logs/{service.service_name}_heap.prof > /app/logs/{service.service_name}_heap.pdf"
            ]
            self.logger.debug(f"Service cmds: {service.run_cmd}")
        else:
            self.logger.debug(f"Service {service} is not gperf compatible")

    self.logger.debug(f"Test Config: {OmegaConf.to_yaml(self.test_config)}")
    self.logger.debug(f"Global Config: {OmegaConf.to_yaml(self.global_config)}")

teardown_environment

teardown_environment()

Tears down the environment after experiments are completed.

Source code in panther/plugins/environments/execution_environment/execution_environment_interface.py

def teardown_environment(self):
    """
    Tears down the environment after experiments are completed.
    """
    pass

to_command

to_command(service_name: str) -> str

Generate the gperf command based on the configuration.

Source code in panther/plugins/environments/execution_environment/gperf_heap/gperf_heap.py

def to_command(self, service_name: str) -> str:
    """
    Generate the gperf command based on the configuration.
    """
    conf = GperfHeapConfig(
        # input_file="keywords.txt",
        # output_file="output.c",
        # language="C++",
        # keyword_only=True,
        # readonly_tables=True,
        # includes=["my_header.h"],
        # other_flags=["--ignore-case"]
    )
    command = ["gperf"]

    # Input and output files
    if conf.input_file:
        command.append(f'"{conf.input_file}"')
    if conf.output_file:
        command.append(f'--output-file="{conf.output_file}"')

    # Language option
    if conf.language:
        command.append(f"--language={conf.language}")

    # Flags
    if conf.keyword_only:
        command.append("--keyword-only")
    if conf.readonly_tables:
        command.append("--readonly-tables")
    if conf.switch:
        command.append("--switch")
    if conf.compare_strncmp:
        command.append("--compare-strncmp")

    # Custom functions
    if conf.hash_function:
        command.append(f'--hash-function="{conf.hash_function}"')
    if conf.compare_function:
        command.append(f'--compare-function="{conf.compare_function}"')

    # Includes
    for include in conf.includes:
        command.append(f'--include="{include}"')

    # Other flags
    command.extend(conf.other_flags)

    # Join and return the command
    return " ".join(command)