Skip to content

vCPUCore

The class "vCPUCore" serves as the implementation of CPU cores within the simulation. It includes two mandatory attributes: frequency and instructions-per-cycle, which determine the computational capacity of the core. The computational capacity is represented as a "Resource" object from the "Akatosh" library, enabling withdrawal or return of the available amount during the simulation. The "vCPUCore" is responsible for allocating the computational power to the assigned processes and reclaiming the distributed amount once the execution of the assigned process is complete. In the event that the "vCPUCore" is arbitrarily powered off during the simulation, all processes currently in execution will be considered as failed.

Bases: vHardwareComponent

Source code in PyCloudSim\entity\v_cpu_core.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class vCPUCore(vHardwareComponent):
    def __init__(
        self,
        ipc: int | Callable,
        frequency: int | Callable,
        label: str | None = None,
        create_at: int | float | Callable[..., Any] | None = None,
        terminate_at: int | float | Callable[..., Any] | None = None,
        precursor: Entity | List[Entity] | None = None,
    ) -> None:
        """Create a simulated CPU core.

        Args:
            ipc (int | Callable): instructions per cycle.
            frequency (int | Callable): the frequency of the CPU core.
            label (str | None, optional): the short description of the CPU Core. Defaults to None.
            create_at (int | float | Callable[..., Any] | None, optional): when this CPU core should be created. Defaults to None.
            terminate_at (int | float | Callable[..., Any] | None, optional): when this CPU core should be terminated. Defaults to None.
            precursor (Entity | List[Entity] | None, optional): the entity that this CPU core must not be created before. Defaults to None.
        """
        super().__init__(label, create_at, terminate_at, precursor)

        if callable(ipc):
            self._ipc = round(ipc())
        else:
            self._ipc = ipc

        if callable(frequency):
            self._frequency = round(frequency())
        else:
            self._frequency = frequency
        self._instruction_cycle = 1 / (self._ipc * self._frequency)
        self._computational_power = Resource(
            capacity=self.ipc * self.frequency,
            label=f"{self} Computational Power",
        )
        self._instructions_queue: List[vInstruction] = EntityList(
            label=f"{self} InstructionsQueue"
        )
        self._clock: Event = None  # type: ignore

    def on_power_on(self):
        """Create the execution clock of the CPU core."""

        @self.continuous_event(
            at=simulation.now + self.instruction_cycle,
            interval=self.instruction_cycle,
            duration=inf,
            label=f"{self.label} Clock",
        )
        def _execute_instruction():
            if len(self.instructions_queue) > 0:
                instruction = self.instructions_queue[0]
                instruction.terminate(at=simulation.now)
                logger.debug(
                    f"{simulation.now}:\t{self} executed {instruction}, current capacity: {self.computational_power.amount}, queue length: {len(self.instructions_queue)}."
                )
                if instruction.process.deamon:
                    instruction = vInstruction(
                        process=instruction.process,
                        create_at=simulation.now,
                    )

    def cache_instruction(self, instruction: vInstruction) -> None:
        """Cache instructions from a process to the CPU core."""
        self.instructions_queue.append(instruction)
        instruction.get(self.computational_power, 1)

    def on_power_off(self) -> None:
        """Power off the CPU core."""
        # terminate the clock of the CPU core.
        for event in self.events:
            if event.label == f"{self.label} Clock":
                event.cancel()

        # find impacted process.
        impacted_process: List[vProcess] = []
        for instruction in self.instructions_queue:
            if instruction.process not in impacted_process:
                impacted_process.append(instruction.process)
        # fail impacted process.
        for process in impacted_process:
            process.fail(simulation.now)

    def on_fail(self) -> None:
        self.power_off(simulation.now)

    @property
    def ipc(self) -> int:
        """Returns the number of instructions per cycle of the CPU core."""
        return self._ipc

    @property
    def frequency(self) -> int:
        """Returns the frequency of the CPU core."""
        return self._frequency

    @property
    def instruction_cycle(self) -> float:
        """Returns the time taken to execute one instruction."""
        return self._instruction_cycle

    @property
    def instructions_queue(self):
        """Returns the instructions queue of the CPU core."""
        return self._instructions_queue

    @property
    def computational_power(self) -> Resource:
        """Returns the computational power of the CPU core."""
        return self._computational_power

    @property
    def clock(self):
        """Returns the clock of the CPU core, which is the continuous event that consumes instructions."""
        return self._clock

    def usage(self, duration: int | float | None = None):
        """Returns the usage of the CPU core."""
        return self.computational_power.usage(duration)

    def utilization(self, duration: int | float | None = None):
        """Returns the utilization of the CPU core."""
        return self.computational_power.utilization(duration)

clock property

Returns the clock of the CPU core, which is the continuous event that consumes instructions.

computational_power: Resource property

Returns the computational power of the CPU core.

frequency: int property

Returns the frequency of the CPU core.

instruction_cycle: float property

Returns the time taken to execute one instruction.

instructions_queue property

Returns the instructions queue of the CPU core.

ipc: int property

Returns the number of instructions per cycle of the CPU core.

__init__(ipc, frequency, label=None, create_at=None, terminate_at=None, precursor=None)

Create a simulated CPU core.

Parameters:

Name Type Description Default
ipc int | Callable

instructions per cycle.

 required
frequency int | Callable

the frequency of the CPU core.

 required
label str | None

the short description of the CPU Core. Defaults to None.

 None
create_at int | float | Callable[..., Any] | None

when this CPU core should be created. Defaults to None.

 None
terminate_at int | float | Callable[..., Any] | None

when this CPU core should be terminated. Defaults to None.

 None
precursor Entity | List[Entity] | None

the entity that this CPU core must not be created before. Defaults to None.

 None
Source code in PyCloudSim\entity\v_cpu_core.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    ipc: int | Callable,
    frequency: int | Callable,
    label: str | None = None,
    create_at: int | float | Callable[..., Any] | None = None,
    terminate_at: int | float | Callable[..., Any] | None = None,
    precursor: Entity | List[Entity] | None = None,
) -> None:
    """Create a simulated CPU core.

    Args:
        ipc (int | Callable): instructions per cycle.
        frequency (int | Callable): the frequency of the CPU core.
        label (str | None, optional): the short description of the CPU Core. Defaults to None.
        create_at (int | float | Callable[..., Any] | None, optional): when this CPU core should be created. Defaults to None.
        terminate_at (int | float | Callable[..., Any] | None, optional): when this CPU core should be terminated. Defaults to None.
        precursor (Entity | List[Entity] | None, optional): the entity that this CPU core must not be created before. Defaults to None.
    """
    super().__init__(label, create_at, terminate_at, precursor)

    if callable(ipc):
        self._ipc = round(ipc())
    else:
        self._ipc = ipc

    if callable(frequency):
        self._frequency = round(frequency())
    else:
        self._frequency = frequency
    self._instruction_cycle = 1 / (self._ipc * self._frequency)
    self._computational_power = Resource(
        capacity=self.ipc * self.frequency,
        label=f"{self} Computational Power",
    )
    self._instructions_queue: List[vInstruction] = EntityList(
        label=f"{self} InstructionsQueue"
    )
    self._clock: Event = None  # type: ignore

cache_instruction(instruction)

Cache instructions from a process to the CPU core.

Source code in PyCloudSim\entity\v_cpu_core.py 
78
79
80
81
def cache_instruction(self, instruction: vInstruction) -> None:
    """Cache instructions from a process to the CPU core."""
    self.instructions_queue.append(instruction)
    instruction.get(self.computational_power, 1)

on_power_off()

Power off the CPU core.

Source code in PyCloudSim\entity\v_cpu_core.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def on_power_off(self) -> None:
    """Power off the CPU core."""
    # terminate the clock of the CPU core.
    for event in self.events:
        if event.label == f"{self.label} Clock":
            event.cancel()

    # find impacted process.
    impacted_process: List[vProcess] = []
    for instruction in self.instructions_queue:
        if instruction.process not in impacted_process:
            impacted_process.append(instruction.process)
    # fail impacted process.
    for process in impacted_process:
        process.fail(simulation.now)

on_power_on()

Create the execution clock of the CPU core.

Source code in PyCloudSim\entity\v_cpu_core.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def on_power_on(self):
    """Create the execution clock of the CPU core."""

    @self.continuous_event(
        at=simulation.now + self.instruction_cycle,
        interval=self.instruction_cycle,
        duration=inf,
        label=f"{self.label} Clock",
    )
    def _execute_instruction():
        if len(self.instructions_queue) > 0:
            instruction = self.instructions_queue[0]
            instruction.terminate(at=simulation.now)
            logger.debug(
                f"{simulation.now}:\t{self} executed {instruction}, current capacity: {self.computational_power.amount}, queue length: {len(self.instructions_queue)}."
            )
            if instruction.process.deamon:
                instruction = vInstruction(
                    process=instruction.process,
                    create_at=simulation.now,
                )

usage(duration=None)

Returns the usage of the CPU core.

Source code in PyCloudSim\entity\v_cpu_core.py 
132
133
134
def usage(self, duration: int | float | None = None):
    """Returns the usage of the CPU core."""
    return self.computational_power.usage(duration)

utilization(duration=None)

Returns the utilization of the CPU core.

Source code in PyCloudSim\entity\v_cpu_core.py 
136
137
138
def utilization(self, duration: int | float | None = None):
    """Returns the utilization of the CPU core."""
    return self.computational_power.utilization(duration)

Usage Example

from PyCloudSim.entity import vCPUCore
core = vCPUCore(
        1, #IPC
        4000, #Frequency in Mhz
        "Core 0", # Label
        0, # Created at 0s
    )
# power on CPU core
core.power_on(at=0)
# power off CPU core
core.powere_off(at=10)
# terminate the CPU core
core.terminate(at=12)