<commit_before># SPDX-License-Identifier: Apache-2.0
#
# Copyright (C) 2015, ARM Limited and contributors.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from env import TestEnv
from wlgen import RTA, Periodic, Step
import trappy
import shutil
import os
import unittest
import logging
import json
import logging

logging.basicConfig(level=logging.INFO)
# Read the config file and update the globals
CONF_FILE = os.path.join(
    os.path.dirname(
        os.path.abspath(__file__)),
    "hmp_parity.config")

with open(CONF_FILE, "r") as fh:
    conf_vars = json.load(fh)
    globals().update(conf_vars)


def local_setup(env):
    env.target.cpufreq.set_all_governors("performance")

    if ENABLE_EAS:
        env.target.execute(
            "echo ENERGY_AWARE > /sys/kernel/debug/sched_features")


def between_threshold_pct(a, b):
    THRESHOLD_PERCENT = 3
    lower = b - THRESHOLD_PERCENT
    upper = b + THRESHOLD_PERCENT

    if a >= lower and a <= upper:
        return True
    return False


def between_threshold_abs(a, b):
    THRESHOLD = 0.25
    lower = b - THRESHOLD
    upper = b + THRESHOLD

    if a >= lower and a <= upper:
        return True
    return False


SMALL_WORKLOAD = {

    "duty_cycle_pct": SMALL_DCYCLE,
    "duration_s": WORKLOAD_DURATION_S,
    "period_ms": WORKLOAD_PERIOD_MS,
}

BIG_WORKLOAD = {

    "duty_cycle_pct": BIG_DCYCLE,
    "duration_s": WORKLOAD_DURATION_S,
    "period_ms": WORKLOAD_PERIOD_MS,
}

STEP_WORKLOAD = {

    "start_pct": STEP_LOW_DCYCLE,
    "end_pct": STEP_HIGH_DCYCLE,
    "time_s": WORKLOAD_DURATION_S,
    "loops": 2
}

from bart.sched.SchedAssert import SchedAssert
from bart.sched.SchedMultiAssert import SchedMultiAssert
import operator
import json


def log_result(data, log_fh):
    result_str = json.dumps(data, indent=3)
    logging.info(result_str)
    log_fh.write(result_str)


class ForkMigration(unittest.TestCase):
    """
    Goal
    ====

    Check that newly created threads start on a big CPU

    Detailed Description
    ====================

    The test spawns as many threads as there are little cores.  It
    then checks that all threads started on a big core.

    Expected Behaviour
    ==================

    The threads start on a big core.
    """

    @classmethod
    def setUpClass(cls):
        cls.params = {}
        cls.task_prefix = "fmig"
        cls.env = TestEnv(test_conf=TEST_CONF)
        cls.trace_file = os.path.join(cls.env.res_dir, "fork_migration.dat")
        cls.log_file = os.path.join(cls.env.res_dir, "fork_migration.json")
        cls.populate_params()
        cls.tasks = cls.params.keys()
        cls.num_tasks = len(cls.tasks)
        local_setup(cls.env)
        cls.run_workload()
        cls.log_fh = open(os.path.join(cls.env.res_dir, cls.log_file), "w")

    @classmethod
    def tearDownClass(cls):
        cls.log_fh.close()

    @classmethod
    def populate_params(cls):
        for idx in range(len(cls.env.target.bl.bigs)):
            task = cls.task_prefix + str(idx)
            cls.params[task] = Periodic(**BIG_WORKLOAD).get()

        for idx in range(len(cls.env.target.bl.littles)):
            task = cls.task_prefix + str(idx)
            cls.params[task] = Periodic(**SMALL_WORKLOAD).get()

    @classmethod
    def run_workload(cls):
        wload = RTA(
            cls.env.target,
            "fork_migration",<commit_msg>tests/eas/hmp_parity: don't clobber big tasks<commit_after># SPDX-License-Identifier: Apache-2.0
#
# Copyright (C) 2015, ARM Limited and contributors.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from env import TestEnv
from wlgen import RTA, Periodic, Step
import trappy
import shutil
import os
import unittest
import logging
import json
import logging

logging.basicConfig(level=logging.INFO)
# Read the config file and update the globals
CONF_FILE = os.path.join(
    os.path.dirname(
        os.path.abspath(__file__)),
    "hmp_parity.config")

with open(CONF_FILE, "r") as fh:
    conf_vars = json.load(fh)
    globals().update(conf_vars)


def local_setup(env):
    env.target.cpufreq.set_all_governors("performance")

    if ENABLE_EAS:
        env.target.execute(
            "echo ENERGY_AWARE > /sys/kernel/debug/sched_features")


def between_threshold_pct(a, b):
    THRESHOLD_PERCENT = 3
    lower = b - THRESHOLD_PERCENT
    upper = b + THRESHOLD_PERCENT

    if a >= lower and a <= upper:
        return True
    return False


def between_threshold_abs(a, b):
    THRESHOLD = 0.25
    lower = b - THRESHOLD
    upper = b + THRESHOLD

    if a >= lower and a <= upper:
        return True
    return False


SMALL_WORKLOAD = {

    "duty_cycle_pct": SMALL_DCYCLE,
    "duration_s": WORKLOAD_DURATION_S,
    "period_ms": WORKLOAD_PERIOD_MS,
}

BIG_WORKLOAD = {

    "duty_cycle_pct": BIG_DCYCLE,
    "duration_s": WORKLOAD_DURATION_S,
    "period_ms": WORKLOAD_PERIOD_MS,
}

STEP_WORKLOAD = {

    "start_pct": STEP_LOW_DCYCLE,
    "end_pct": STEP_HIGH_DCYCLE,
    "time_s": WORKLOAD_DURATION_S,
    "loops": 2
}

from bart.sched.SchedAssert import SchedAssert
from bart.sched.SchedMultiAssert import SchedMultiAssert
import operator
import json


def log_result(data, log_fh):
    result_str = json.dumps(data, indent=3)
    logging.info(result_str)
    log_fh.write(result_str)


class ForkMigration(unittest.TestCase):
    """
    Goal
    ====

    Check that newly created threads start on a big CPU

    Detailed Description
    ====================

    The test spawns as many threads as there are cores in the system.
    It then checks that all threads started on a big core.

    Expected Behaviour
    ==================

    The threads start on a big core.
    """

    @classmethod
    def setUpClass(cls):
        cls.params = {}
        cls.task_prefix = "fmig"
        cls.env = TestEnv(test_conf=TEST_CONF)
        cls.trace_file = os.path.join(cls.env.res_dir, "fork_migration.dat")
        cls.log_file = os.path.join(cls.env.res_dir, "fork_migration.json")
        cls.populate_params()
        cls.tasks = cls.params.keys()
        cls.num_tasks = len(cls.tasks)
        local_setup(cls.env)
        cls.run_workload()
        cls.log_fh = open(os.path.join(cls.env.res_dir, cls.log_file), "w")

    @classmethod
    def tearDownClass(cls):
        cls.log_fh.close()

    @classmethod
    def populate_params(cls):
        big_prefix = cls.task_prefix + "_big"
        for idx in range(len(cls.env.target.bl.bigs)):
            task = big_prefix + str(idx)
            cls.params[task] = Periodic(**BIG_WORKLOAD).get()

        little_prefix = cls.task_prefix + "_little"
        for idx in range(len(cls.env.target.bl.littles)):
            task = little_prefix + str(idx)
            cls.params[task] = Periodic(**SMALL_WORKLOAD).get()

    @classmethod
    def run_workload(cls):
        wload = RTA(
            cls.env.target,
            "fork_migration",
{'fraction_non_alphanumeric': 0.10722906873332487, 'fraction_numerical': 0.004065557108372506, 'mean_word_length': 3.0327868852459017, 'pattern_counts': {'":': 20, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7114866', 'n_tokens_mistral': 2711, 'n_tokens_neox': 2587, 'n_words': 673}