Skip to content

vCPU

The class "vCPU" is an implementation derived from the "PhysicalComponent" class. A "vCPU" object consists of multiple "vCPUCore" instances and a cache that serves as a pool for storing "vProcess" objects. Whenever a new "vProcess" is added to the cache or a "vProcess" is terminated, the "vCPU" class handles the assignment of "vProcess" objects to its corresponding "vCPUCore" based on several factors. These factors include the priority of the "vProcess," the available computational power of each "vCPUCore," and the allowed CPU time of the associated "vContainer." Notably, a single "vProcess" can be assigned to multiple "vCPUCore" instances, allowing for simulated parallel execution. The scheduling of "vProcess" is implemented as an event with only one scheduling event being active for each "vCPU" at any given time.

Bases: PhysicalComponent

Source code in PyCloudSim\entity\v_cpu.py 
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class vCPU(PhysicalComponent):
    _cpu_cores: List[vCPUCore]
    _processes: List[vProcess]

    def __init__(
        self,
        ipc: Union[int, float],
        frequency: Union[int, float],
        num_cores: int,
        tdp: Union[int, float] = 50,
        at: Union[int, float, Callable] = simulation.now,
        after: Optional[Entity | List[Entity]] = None,
        label: Optional[str] = None,
    ):
        """Creates a new vCPU.

        Args:
            ipc (Union[int, float]): instructions per cycle.
            frequency (Union[int, float]): the frequency of cpu core.
            num_cores (int): the number of cores in the cpu.
            tdp (Union[int, float], optional): the TDP, aka power consumption of the cpu. Defaults to 50W.
            at (Union[int, float, Callable], optional): when the cpu should be created. Defaults to simulation.now.
            after (Optional[Entity  |  List[Entity]], optional): the entity that must be terminated before the cpu can be created. Defaults to None.
            label (Optional[str], optional): short description of the cpu. Defaults to None.
        """
        super().__init__(at, after, label)
        self._ipc = ipc
        self._frequency = frequency * 1000000
        self._num_cores = num_cores
        cpu_cores = [
            vCPUCore(
                ipc=self.ipc,
                frequency=self.frequency,
                cpu=self,
                label=f"{self.label}-Core-{i}",
            )
            for i in range(num_cores)
        ]
        self._cpu_cores = cpu_cores
        self._tdp = tdp
        self._processes = list()
        self._process_scheduler: Actor = None  # type: ignore
        simulation.CPUS.append(self)

    def creation(self):
        """Creates the cpu."""
        return super().creation()

    def termination(self):
        """Terminates the cpu."""
        return super().termination()

    def _power_on(self):
        """Power on the cpu and all its cores."""
        super()._power_on()
        for core in self.cpu_cores:
            core.power_on()

    def _power_off(self):
        """Power off the cpu and all its cores."""
        super()._power_off()
        for core in self.cpu_cores:
            core.power_off()

    def cache_process(self, process: vProcess):
        """Cache a process in the cpu and call schedule_process()."""
        if not process.cached:
            self.processes.append(process)
            process._cpu_id = self.id
            process.status.append(CACHED)
            self.schedule_process()

    def schedule_process(self):
        """shcedule processes in the cpu queue."""

        def _schedule_process():
            LOGGER.debug(
                f"{simulation.now:0.2f}:\tvCPU {self.label} is scheduling ... {len(self.processes)} processes"
            )
            self.processes.sort(key=lambda process: process.priority)
            for process in self.processes:
                # if not process.executing and not process.terminated:
                for core in self.cpu_cores:
                    remaining_to_schedule_instruction_length = (
                        process.remaining - process.current_scheduled_length
                    )
                    if process.__class__.__name__ == "vPacketHandler":
                        container_allowed_instruction_length = inf
                    else:
                        container_allowed_instruction_length = (
                            process.container.cpu.available_quantity  # type: ignore
                            / 1000
                            * core.capacity
                        )
                    schedulable_instruction_length = int(
                        min(
                            [
                                remaining_to_schedule_instruction_length,
                                container_allowed_instruction_length,
                                core.availablity,
                            ]
                        )
                    )
                    if schedulable_instruction_length > 0:
                        core.execute_process(process, schedulable_instruction_length)
                        process._current_scheduled_length += (
                            schedulable_instruction_length
                        )
                        scheduled_cpu_time = (
                            schedulable_instruction_length / core.capacity
                        ) * 1000

                        if process.__class__.__name__ != "vPacketHandler":
                            process.container.cpu.distribute(  # type: ignore
                                process, scheduled_cpu_time
                            )

            LOGGER.debug(
                f"{simulation.now:0.2f}:\tvCPU {self.label} scheduled all process within the queue."
            )

            self._process_scheduler = None  # type: ignore

        if self.process_scheduler is None:
            self._process_scheduler = Actor(
                at=simulation.now,
                action=_schedule_process,
                label=f"vCPU {self.label} Schedule vProcess",
                priority=CPU_SCHEDULE_PROCESS,
            )

    @property
    def ipc(self) -> Union[int, float]:
        """return the IPC of the cpu."""
        return self._ipc

    @property
    def frequency(self) -> Union[int, float]:
        """return the frequency of the cpu."""
        return self._frequency

    @property
    def num_cores(self) -> int:
        """return the number of cores of the cpu."""
        return self._num_cores

    @property
    def single_core_capacity(self) -> Union[int, float]:
        """return the single core capacity of the cpu."""
        return (self.ipc * self.frequency) / simulation.cpu_acceleration

    @property
    def cpu_cores(self) -> List[vCPUCore]:
        """return the cpu cores of the cpu."""
        return self._cpu_cores

    @property
    def capacity(self) -> Union[int, float]:
        """return the capacity of the cpu."""
        return sum([core.capacity for core in self.cpu_cores])

    @property
    def availablity(self) -> Union[int, float]:
        """return the availablity of the cpu."""
        return sum([core.availablity for core in self.cpu_cores])

    @property
    def utilization(self) -> Union[int, float]:
        """return the utilization of the cpu."""
        return (self.capacity - self.availablity) / self.capacity * 100

    def utilization_in_past(self, interval: Union[int, float]) -> Union[int, float]:
        """return the utilization of the cpu in the past interval."""
        return (
            sum(
                [
                    core.computational_power.utilization_in_past(interval)
                    for core in self.cpu_cores
                ]
            )
            / self.num_cores
        )

    @property
    def processes(self) -> List[vProcess]:
        """return the processes of the cpu."""
        return self._processes

    @property
    def process_scheduler(self) -> Actor:
        """return the process scheduler of the cpu, if it is runing at the moment."""
        return self._process_scheduler

    @property
    def tdp(self) -> Union[int, float]:
        """return the TDP of the cpu."""
        return self._tdp

availablity: Union[int, float] property

return the availablity of the cpu.

capacity: Union[int, float] property

return the capacity of the cpu.

cpu_cores: List[vCPUCore] property

return the cpu cores of the cpu.

frequency: Union[int, float] property

return the frequency of the cpu.

ipc: Union[int, float] property

return the IPC of the cpu.

num_cores: int property

return the number of cores of the cpu.

process_scheduler: Actor property

return the process scheduler of the cpu, if it is runing at the moment.

processes: List[vProcess] property

return the processes of the cpu.

single_core_capacity: Union[int, float] property

return the single core capacity of the cpu.

tdp: Union[int, float] property

return the TDP of the cpu.

utilization: Union[int, float] property

return the utilization of the cpu.

__init__(ipc, frequency, num_cores, tdp=50, at=simulation.now, after=None, label=None)

Creates a new vCPU.

Parameters:

Name Type Description Default
ipc Union[int, float]

instructions per cycle.

 required
frequency Union[int, float]

the frequency of cpu core.

 required
num_cores int

the number of cores in the cpu.

 required
tdp Union[int, float]

the TDP, aka power consumption of the cpu. Defaults to 50W.

 50
at Union[int, float, Callable]

when the cpu should be created. Defaults to simulation.now.

 now
after Optional[Entity | List[Entity]]

the entity that must be terminated before the cpu can be created. Defaults to None.

 None
label Optional[str]

short description of the cpu. Defaults to None.

 None
Source code in PyCloudSim\entity\v_cpu.py 
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
def __init__(
    self,
    ipc: Union[int, float],
    frequency: Union[int, float],
    num_cores: int,
    tdp: Union[int, float] = 50,
    at: Union[int, float, Callable] = simulation.now,
    after: Optional[Entity | List[Entity]] = None,
    label: Optional[str] = None,
):
    """Creates a new vCPU.

    Args:
        ipc (Union[int, float]): instructions per cycle.
        frequency (Union[int, float]): the frequency of cpu core.
        num_cores (int): the number of cores in the cpu.
        tdp (Union[int, float], optional): the TDP, aka power consumption of the cpu. Defaults to 50W.
        at (Union[int, float, Callable], optional): when the cpu should be created. Defaults to simulation.now.
        after (Optional[Entity  |  List[Entity]], optional): the entity that must be terminated before the cpu can be created. Defaults to None.
        label (Optional[str], optional): short description of the cpu. Defaults to None.
    """
    super().__init__(at, after, label)
    self._ipc = ipc
    self._frequency = frequency * 1000000
    self._num_cores = num_cores
    cpu_cores = [
        vCPUCore(
            ipc=self.ipc,
            frequency=self.frequency,
            cpu=self,
            label=f"{self.label}-Core-{i}",
        )
        for i in range(num_cores)
    ]
    self._cpu_cores = cpu_cores
    self._tdp = tdp
    self._processes = list()
    self._process_scheduler: Actor = None  # type: ignore
    simulation.CPUS.append(self)

cache_process(process)

Cache a process in the cpu and call schedule_process().

Source code in PyCloudSim\entity\v_cpu.py 
84
85
86
87
88
89
90
def cache_process(self, process: vProcess):
    """Cache a process in the cpu and call schedule_process()."""
    if not process.cached:
        self.processes.append(process)
        process._cpu_id = self.id
        process.status.append(CACHED)
        self.schedule_process()

creation()

Creates the cpu.

Source code in PyCloudSim\entity\v_cpu.py 
64
65
66
def creation(self):
    """Creates the cpu."""
    return super().creation()

schedule_process()

shcedule processes in the cpu queue.

Source code in PyCloudSim\entity\v_cpu.py 
 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
139
140
141
142
143
144
145
146
147
148
149
def schedule_process(self):
    """shcedule processes in the cpu queue."""

    def _schedule_process():
        LOGGER.debug(
            f"{simulation.now:0.2f}:\tvCPU {self.label} is scheduling ... {len(self.processes)} processes"
        )
        self.processes.sort(key=lambda process: process.priority)
        for process in self.processes:
            # if not process.executing and not process.terminated:
            for core in self.cpu_cores:
                remaining_to_schedule_instruction_length = (
                    process.remaining - process.current_scheduled_length
                )
                if process.__class__.__name__ == "vPacketHandler":
                    container_allowed_instruction_length = inf
                else:
                    container_allowed_instruction_length = (
                        process.container.cpu.available_quantity  # type: ignore
                        / 1000
                        * core.capacity
                    )
                schedulable_instruction_length = int(
                    min(
                        [
                            remaining_to_schedule_instruction_length,
                            container_allowed_instruction_length,
                            core.availablity,
                        ]
                    )
                )
                if schedulable_instruction_length > 0:
                    core.execute_process(process, schedulable_instruction_length)
                    process._current_scheduled_length += (
                        schedulable_instruction_length
                    )
                    scheduled_cpu_time = (
                        schedulable_instruction_length / core.capacity
                    ) * 1000

                    if process.__class__.__name__ != "vPacketHandler":
                        process.container.cpu.distribute(  # type: ignore
                            process, scheduled_cpu_time
                        )

        LOGGER.debug(
            f"{simulation.now:0.2f}:\tvCPU {self.label} scheduled all process within the queue."
        )

        self._process_scheduler = None  # type: ignore

    if self.process_scheduler is None:
        self._process_scheduler = Actor(
            at=simulation.now,
            action=_schedule_process,
            label=f"vCPU {self.label} Schedule vProcess",
            priority=CPU_SCHEDULE_PROCESS,
        )

termination()

Terminates the cpu.

Source code in PyCloudSim\entity\v_cpu.py 
68
69
70
def termination(self):
    """Terminates the cpu."""
    return super().termination()

utilization_in_past(interval)

return the utilization of the cpu in the past interval.

Source code in PyCloudSim\entity\v_cpu.py 
191
192
193
194
195
196
197
198
199
200
201
def utilization_in_past(self, interval: Union[int, float]) -> Union[int, float]:
    """return the utilization of the cpu in the past interval."""
    return (
        sum(
            [
                core.computational_power.utilization_in_past(interval)
                for core in self.cpu_cores
            ]
        )
        / self.num_cores
    )