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: vHardwareComponent

Source code in PyCloudSim\entity\v_cpu.py 
 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
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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
class vCPU(vHardwareComponent):
    def __init__(
        self,
        ipc: int | Callable,
        frequency: int | Callable,
        num_cores: int | Callable,
        tdp: int | float | Callable,
        mode: int,
        host: vHardwareEntity | None = None,
        label: str | None = None,
        create_at: int
        | float
        | Callable[..., int]
        | Callable[..., float]
        | None = None,
        terminate_at: int
        | float
        | Callable[..., int]
        | Callable[..., float]
        | None = None,
        precursor: Entity | List[Entity] | None = None,
    ) -> None:
        """Create a simulated CPU.

        Args:
            ipc (int | Callable): the instructions per cycle (IPC) of the CPU core.
            frequency (int | Callable): the frequency of the CPU core.
            num_cores (int | Callable): the number of cores of the CPU.
            tdp (int | float | Callable): the thermal design power (TDP) of the CPU.
            mode (int): 1 or 2, the mode of the CPU.
            host (vHardwareEntity | None, optional): the host of this CPU. Defaults to None.
            label (str | None, optional): the short description of the CPU. Defaults to None.
            create_at (int | float | Callable[..., int] | Callable[..., float] | None, optional): when this CPU should be created. Defaults to None.
            terminate_at (int | float | Callable[..., int] | Callable[..., float] | None, optional): when this CPU should be terminated. Defaults to None.
            precursor (Entity | List[Entity] | None, optional): the entity that this CPU must not be created before. Defaults to None.
        """
        super().__init__(label, create_at, terminate_at, precursor)

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

        if callable(frequency):
            self._frequency = round(frequency())
        else:
            self._frequency = frequency

        if callable(num_cores):
            self._num_cores = round(num_cores())
        else:
            self._num_cores = num_cores

        if callable(tdp):
            self._tdp = tdp()
        else:
            self._tdp = tdp

        self._mode = mode

        self._process_queue: List[vProcess | vContainerProcess | vDeamon | vDecoder] = EntityList(label=f"{self}-Process Queue")
        self._cores: List[vCPUCore] = EntityList(label=f"vCPU {self}_Cores")
        self._computational_power_reservoir = Resource(
            capacity=1000 * self.num_cores,
            label=f"{self}-Computational Power Reservoir",
        )
        self._host = host

    def on_creation(self):
        # create cores
        for i in range(self._num_cores):
            self.cores.append(
                vCPUCore(
                    self._ipc,
                    self._frequency,
                    label=f"{self.label}-{i}",
                    create_at=simulation.now,
                )
            )

    def on_termination(self):
        # terminate cores
        for core in self._cores[:]:
            core.terminate(simulation.now)

    def on_power_on(self) -> None:
        """Power on the CPU, also starts the process scheduling."""
        # power on all the cores
        for core in self.cores:
            core.power_on(at=simulation.now)

        # start process scheduling
        @self.continuous_event(
            at=simulation.now,
            interval=self.instruction_cycle,
            duration=inf,
            label=f"{self} Scheduling process",
        )
        def _schedule_process():
            # sort the process queue by priority
            self.process_queue.sort(key=lambda process: process.priority, reverse=False)
            for process in self.process_queue:
                # calculate the number of schedulable instructions
                if process.container is None:
                    container_cpu_capacity = inf
                else:
                    container_cpu_capacity = round(
                        ((process.container.cpu_limit - process.container.cpu_usage) / 1000)
                        * (self.ipc * self.frequency)
                    )
                schedulable_instructions = min(
                    [len(process.unscheduled_instructions), container_cpu_capacity]
                )
                if schedulable_instructions <= 0:
                    continue
                logger.debug(
                    f"{simulation.now}:\t{self} is executing {schedulable_instructions} instructions of {process}."
                )
                try:
                    while schedulable_instructions > 0:
                        # get and sort available cores
                        available_cores = [
                            core
                            for core in self.cores
                            if core.computational_power.amount > 0
                        ]
                        if len(available_cores) == 0:
                            break
                        available_cores.sort(
                            key=lambda core: core.computational_power.amount,
                            reverse=True,
                        )
                        # mode 1: assign one instruction to one core, then move on to the next core
                        if self.mode == 1:
                            core = available_cores[0]
                            # get the next instruction
                            instruction = process.unscheduled_instructions.pop(0)
                            # update the host ram usage
                            if process.host is None:
                                raise RuntimeError()
                            # a value error exception will be raised if fail to distribute the ram space
                            instruction.get(process.host.ram, instruction.length)
                            # update the container's ram usage
                            if process.container is not None:
                                process.container._ram_usage += instruction.length
                                # break the loop if container's ram usage exceeds the limit
                                if (
                                    process.container.ram_usage
                                    > process.container.ram_limit
                                ):
                                    raise Exception()
                                # update the container's cpu usage
                                process.container._cpu_usage += (1 * 1000) / (
                                    self.ipc * self.frequency
                                )
                            # cache the instruction to the core
                            core.cache_instruction(instruction)
                            schedulable_instructions -= 1
                        # mode 2: assign as many instructions as possible to one core, then move on to the next core
                        elif self.mode == 2:
                            core = available_cores[0]
                            for _ in range(
                                round(
                                    min(
                                        [
                                            schedulable_instructions,
                                            core.computational_power.amount,
                                        ]
                                    )
                                )
                            ):
                                # get the next instruction
                                instruction = process.unscheduled_instructions.pop(0)
                                # update the host ram usage
                                if process.host is None:
                                    raise RuntimeError()
                                # a value error exception will be raised if fail to distribute the ram space
                                instruction.get(process.host.ram, instruction.length)
                                if process.container is not None:
                                    # update the container's ram usage
                                    process.container._ram_usage += instruction.length
                                    # break the loop if container's ram usage exceeds the limit
                                    if (
                                        process.container.ram_usage
                                        > process.container.ram_limit
                                    ):
                                        raise Exception()
                                    # update the container's cpu usage
                                    process.container._cpu_usage += (1 * 1000) / (
                                        self.ipc * self.frequency
                                    )
                                # cache the instruction to the core
                                core.cache_instruction(instruction)
                                schedulable_instructions -= 1
                except Exception:
                    process.fail(simulation.now)
                    if process.container is not None:
                        process.container.fail(simulation.now)
                    continue                   

    def on_power_off(self) -> None:
        """Power off the CPU, also terminates all unfinished processes."""
        # terminate all unfinished processes
        for process in self.process_queue:
            process.fail(simulation.now)

        # power off all the cores
        for core in self.cores:
            core.power_off(at=simulation.now)

        # cancel process scheduling
        for event in self.events:
            if event.label == f"{self} Scheduling process":
                event.cancel()

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

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

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

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

    @property
    def tdp(self) -> int | float:
        """Returns the thermal design power (TDP) of the CPU."""
        return self._tdp

    @property
    def computational_power_reservoir(self) -> Resource:
        """Returns the computational power reservoir of the CPU."""
        return self._computational_power_reservoir

    @property
    def process_queue(self):
        """Returns the process queue of the CPU."""
        return self._process_queue

    @property
    def cores(self):
        """Returns the cores of the CPU."""
        return self._cores

    @property
    def host(self):
        """Returns the host of the CPU."""
        return self._host

    @property
    def mode(self):
        """Returns the mode of the CPU."""
        return self._mode

    @property
    def instruction_cycle(self):
        """Returns the instruction cycle of the CPU."""
        return 1 / (self.ipc * self.frequency)

    def usage(self, duration: int | float | None = None):
        """Returns the CPU usage."""
        return sum([core.usage(duration) for core in self.cores])

    def utilization(self, duration: int | float | None = None):
        """Returns the CPU utilization."""
        return sum([core.utilization(duration) for core in self.cores]) / len(
            self.cores
        )

computational_power_reservoir: Resource property

Returns the computational power reservoir of the CPU.

cores property

Returns the cores of the CPU.

frequency: int property

Returns the frequency of the CPU core.

host property

Returns the host of the CPU.

instruction_cycle property

Returns the instruction cycle of the CPU.

ipc: int property

Returns the number of instructions per cycle (IPC) of the CPU core.

mode property

Returns the mode of the CPU.

num_cores: int property

Returns the number of CPU cores.

process_queue property

Returns the process queue of the CPU.

tdp: int | float property

Returns the thermal design power (TDP) of the CPU.

__init__(ipc, frequency, num_cores, tdp, mode, host=None, label=None, create_at=None, terminate_at=None, precursor=None)

Create a simulated CPU.

Parameters:

Name Type Description Default
ipc int | Callable

the instructions per cycle (IPC) of the CPU core.

 required
frequency int | Callable

the frequency of the CPU core.

 required
num_cores int | Callable

the number of cores of the CPU.

 required
tdp int | float | Callable

the thermal design power (TDP) of the CPU.

 required
mode int

1 or 2, the mode of the CPU.

 required
host vHardwareEntity | None

the host of this CPU. Defaults to None.

 None
label str | None

the short description of the CPU. Defaults to None.

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

when this CPU should be created. Defaults to None.

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

when this CPU should be terminated. Defaults to None.

 None
precursor Entity | List[Entity] | None

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

 None
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
def __init__(
    self,
    ipc: int | Callable,
    frequency: int | Callable,
    num_cores: int | Callable,
    tdp: int | float | Callable,
    mode: int,
    host: vHardwareEntity | None = None,
    label: str | None = None,
    create_at: int
    | float
    | Callable[..., int]
    | Callable[..., float]
    | None = None,
    terminate_at: int
    | float
    | Callable[..., int]
    | Callable[..., float]
    | None = None,
    precursor: Entity | List[Entity] | None = None,
) -> None:
    """Create a simulated CPU.

    Args:
        ipc (int | Callable): the instructions per cycle (IPC) of the CPU core.
        frequency (int | Callable): the frequency of the CPU core.
        num_cores (int | Callable): the number of cores of the CPU.
        tdp (int | float | Callable): the thermal design power (TDP) of the CPU.
        mode (int): 1 or 2, the mode of the CPU.
        host (vHardwareEntity | None, optional): the host of this CPU. Defaults to None.
        label (str | None, optional): the short description of the CPU. Defaults to None.
        create_at (int | float | Callable[..., int] | Callable[..., float] | None, optional): when this CPU should be created. Defaults to None.
        terminate_at (int | float | Callable[..., int] | Callable[..., float] | None, optional): when this CPU should be terminated. Defaults to None.
        precursor (Entity | List[Entity] | None, optional): the entity that this CPU must not be created before. Defaults to None.
    """
    super().__init__(label, create_at, terminate_at, precursor)

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

    if callable(frequency):
        self._frequency = round(frequency())
    else:
        self._frequency = frequency

    if callable(num_cores):
        self._num_cores = round(num_cores())
    else:
        self._num_cores = num_cores

    if callable(tdp):
        self._tdp = tdp()
    else:
        self._tdp = tdp

    self._mode = mode

    self._process_queue: List[vProcess | vContainerProcess | vDeamon | vDecoder] = EntityList(label=f"{self}-Process Queue")
    self._cores: List[vCPUCore] = EntityList(label=f"vCPU {self}_Cores")
    self._computational_power_reservoir = Resource(
        capacity=1000 * self.num_cores,
        label=f"{self}-Computational Power Reservoir",
    )
    self._host = host

on_power_off()

Power off the CPU, also terminates all unfinished processes.

Source code in PyCloudSim\entity\v_cpu.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def on_power_off(self) -> None:
    """Power off the CPU, also terminates all unfinished processes."""
    # terminate all unfinished processes
    for process in self.process_queue:
        process.fail(simulation.now)

    # power off all the cores
    for core in self.cores:
        core.power_off(at=simulation.now)

    # cancel process scheduling
    for event in self.events:
        if event.label == f"{self} Scheduling process":
            event.cancel()

on_power_on()

Power on the CPU, also starts the process scheduling.

Source code in PyCloudSim\entity\v_cpu.py 
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
217
218
def on_power_on(self) -> None:
    """Power on the CPU, also starts the process scheduling."""
    # power on all the cores
    for core in self.cores:
        core.power_on(at=simulation.now)

    # start process scheduling
    @self.continuous_event(
        at=simulation.now,
        interval=self.instruction_cycle,
        duration=inf,
        label=f"{self} Scheduling process",
    )
    def _schedule_process():
        # sort the process queue by priority
        self.process_queue.sort(key=lambda process: process.priority, reverse=False)
        for process in self.process_queue:
            # calculate the number of schedulable instructions
            if process.container is None:
                container_cpu_capacity = inf
            else:
                container_cpu_capacity = round(
                    ((process.container.cpu_limit - process.container.cpu_usage) / 1000)
                    * (self.ipc * self.frequency)
                )
            schedulable_instructions = min(
                [len(process.unscheduled_instructions), container_cpu_capacity]
            )
            if schedulable_instructions <= 0:
                continue
            logger.debug(
                f"{simulation.now}:\t{self} is executing {schedulable_instructions} instructions of {process}."
            )
            try:
                while schedulable_instructions > 0:
                    # get and sort available cores
                    available_cores = [
                        core
                        for core in self.cores
                        if core.computational_power.amount > 0
                    ]
                    if len(available_cores) == 0:
                        break
                    available_cores.sort(
                        key=lambda core: core.computational_power.amount,
                        reverse=True,
                    )
                    # mode 1: assign one instruction to one core, then move on to the next core
                    if self.mode == 1:
                        core = available_cores[0]
                        # get the next instruction
                        instruction = process.unscheduled_instructions.pop(0)
                        # update the host ram usage
                        if process.host is None:
                            raise RuntimeError()
                        # a value error exception will be raised if fail to distribute the ram space
                        instruction.get(process.host.ram, instruction.length)
                        # update the container's ram usage
                        if process.container is not None:
                            process.container._ram_usage += instruction.length
                            # break the loop if container's ram usage exceeds the limit
                            if (
                                process.container.ram_usage
                                > process.container.ram_limit
                            ):
                                raise Exception()
                            # update the container's cpu usage
                            process.container._cpu_usage += (1 * 1000) / (
                                self.ipc * self.frequency
                            )
                        # cache the instruction to the core
                        core.cache_instruction(instruction)
                        schedulable_instructions -= 1
                    # mode 2: assign as many instructions as possible to one core, then move on to the next core
                    elif self.mode == 2:
                        core = available_cores[0]
                        for _ in range(
                            round(
                                min(
                                    [
                                        schedulable_instructions,
                                        core.computational_power.amount,
                                    ]
                                )
                            )
                        ):
                            # get the next instruction
                            instruction = process.unscheduled_instructions.pop(0)
                            # update the host ram usage
                            if process.host is None:
                                raise RuntimeError()
                            # a value error exception will be raised if fail to distribute the ram space
                            instruction.get(process.host.ram, instruction.length)
                            if process.container is not None:
                                # update the container's ram usage
                                process.container._ram_usage += instruction.length
                                # break the loop if container's ram usage exceeds the limit
                                if (
                                    process.container.ram_usage
                                    > process.container.ram_limit
                                ):
                                    raise Exception()
                                # update the container's cpu usage
                                process.container._cpu_usage += (1 * 1000) / (
                                    self.ipc * self.frequency
                                )
                            # cache the instruction to the core
                            core.cache_instruction(instruction)
                            schedulable_instructions -= 1
            except Exception:
                process.fail(simulation.now)
                if process.container is not None:
                    process.container.fail(simulation.now)
                continue

usage(duration=None)

Returns the CPU usage.

Source code in PyCloudSim\entity\v_cpu.py 
288
289
290
def usage(self, duration: int | float | None = None):
    """Returns the CPU usage."""
    return sum([core.usage(duration) for core in self.cores])

utilization(duration=None)

Returns the CPU utilization.

Source code in PyCloudSim\entity\v_cpu.py 
292
293
294
295
296
def utilization(self, duration: int | float | None = None):
    """Returns the CPU utilization."""
    return sum([core.utilization(duration) for core in self.cores]) / len(
        self.cores
    )