Skip to content

vNIC

The class "vNIC" serves as the implementation of the network interface card in the simulation. It includes a queue specifically designed to store "vPacket" objects and has the capability to establish connections with other "vNIC" instances. The "vNIC" class is equipped with two resources: uplink bandwidth and downlink bandwidth. These resources are allocated to the transmission and reception of "vPacket" objects respectively. The transmission and reception functionalities of "vPacket" are implemented as member functions within the class. Upon receiving a "vPacket", a specialized "vProcess" called PacketHandler is created to simulate the decoding process and the associated processing delay. If there is insufficient uplink or downlink bandwidth available, the "vPacket" will be kept in the queue until the necessary resources become available.

Bases: PhysicalComponent

Source code in PyCloudSim\entity\v_nic.py 
 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
class vNIC(PhysicalComponent):
    def __init__(
        self,
        host: Union[vHost, vRouter, vSwitch, vGateway],
        connected_to: Optional[Union[vHost, vRouter, vSwitch, vGateway]] = None,
        bandwidth: int = 1000,
        delay: float = 0.02,
        ip: Optional[IPv4Address] = None,
        at: Union[int, float, Callable] = simulation.now,
        after: Optional[Entity | List[Entity]] = None,
        label: Optional[str] = None,
    ):
        """Create a vNIC object.

        Args:
            host (Union[vHost, vRouter, vSwitch, vGateway]): the host that the vNIC is attached to.
            connected_to (Optional[Union[vHost, vRouter, vSwitch, vGateway]], optional): the device that the vNIC is connected to. Defaults to None.
            bandwidth (int, optional): the bandwidth in MBps of the vNIC. Defaults to 1000.
            delay (float, optional): the processing delay of the vNIC. Defaults to 0.02.
            ip (Optional[IPv4Address], optional): the IP address of the vNIC. Defaults to None.
            at (Union[int, float, Callable], optional): same as Entity. Defaults to simulation.now.
            after (Optional[Entity  |  List[Entity]], optional): same as Entity. Defaults to None.
            label (Optional[str], optional): Same as Entity. Defaults to None.
        """
        super().__init__(at, after, label)
        self._host = host
        self._connected_to = connected_to
        self._bandwidth = MiB(bandwidth)
        self._delay = delay
        self._ip = ip
        if host.__class__.__name__ == "vHost" or host.__class__.__name__ == "vRouter":
            self._type = "Interface"
        else:
            self._type = "Port"
        self._uplink = Resource(
            capacity=self.bandwidth, label=f"vNIC {self.label} Uplink"
        )
        self._downlink = Resource(
            capacity=self.bandwidth, label=f"vNIC {self.label} Downlink"
        )
        simulation.NICS.append(self)

    def creation(self):
        """Creation process of the vNIC"""
        return super().creation()

    def termination(self):
        """Termination process of the vNIC"""
        return super().termination()

    def _power_on(self):
        """Power on the vNIC"""
        return super()._power_on()

    def _power_off(self):
        """Power off the vNIC"""
        return super()._power_off()

    def receive_packet(self, packet: vPacket, delay: float = 0.0):
        """Rceive a vPacket, the vPacket will be dropped if the attached host does not have enough RAM.

        Args:
            packet (vPacket): the vPacket to be recevived.
            delay (float, optional): the delay for receiving this vPacket in term of transmitting time. Defaults to 0.0.
        """
        self.uplink.distribute(packet, packet.size)
        LOGGER.debug(
            f"{simulation.now:0.2f}:\tvPacket {packet.label} is using {packet.size}/{self.uplink.available_quantity}/{self.uplink.capacity} bytes of vNIC {self.label} uplink."
        )

        def _received_packet():
            self.uplink.release(packet)
            packet.status.remove(TRANSMITTING)
            packet.status.remove(DECODED)
            try:
                self.host.cache_packet(packet)
                LOGGER.info(
                    f"{simulation.now:0.2f}:\tvPacket {packet.label} is received by {self.host.__class__.__name__} {self.host.label}"
                )
            except:
                packet.drop()
                LOGGER.info(
                    f"{simulation.now:0.2f}:\tvPacket {packet.label} is droped by {self.host.__class__.__name__} {self.host.label}"
                )

        Actor(
            at=simulation.now + delay,
            action=_received_packet,
            label=f"vNIC {self.label} Receive Packet",
            priority=CORE_EXECUTE_PROCESS,
        )

    def send_packet(self, packet: vPacket, delay: float = 0.0):
        """Send a vPacket.

        Args:
            packet (vPacket): the vPacket to be sent.
            delay (float, optional): the delay for sending this packet in term of transmitting time. Defaults to 0.0.
        """
        self.downlink.distribute(packet, packet.size)
        LOGGER.debug(
            f"{simulation.now:0.2f}:\tvPacket {packet.label} is using {packet.size}/{self.downlink.available_quantity}/{self.downlink.capacity} bytes of vNIC {self.label} downlink."
        )
        self.host.packets.remove(packet)
        packet.status.append(TRANSMITTING)
        packet.status.remove(QUEUED)

        def _sent_packet():
            self.downlink.release(packet)
            if self.host.__class__.__name__ == "vGateway":
                pass
            else:
                self.host.ram.release(packet)  # type: ignore
            LOGGER.info(
                f"{simulation.now:0.2f}:\tvPacket {packet.label} is sent by {self.host.__class__.__name__} {self.host.label}"
            )
            Actor(
                at=simulation.now,
                action=self.host.send_packets,
                label=f"{self.__class__.__name__} {self.label} Send Packets",
                priority=HOST_SCHEDULE_PACKET,
            )

        Actor(
            at=simulation.now + delay,
            action=_sent_packet,
            label=f"vNIC {self.label} Send Packet",
            priority=CORE_EXECUTE_PROCESS,
        )

    @property
    def host(self) -> Union[vHost, vRouter, vSwitch, vGateway]:
        """The attached host."""
        return self._host

    @property
    def connected_to(self) -> Optional[Union[vHost, vRouter, vSwitch, vGateway]]:
        """The connected device."""
        return self._connected_to

    @property
    def bandwidth(self) -> float:
        """The bandwidth of the vNIC in MB/s."""
        return self._bandwidth.bytes

    @property
    def delay(self) -> float:
        return self._delay

    @property
    def ip(self) -> Optional[IPv4Address]:
        """The IP address of the vNIC."""
        if self.type == "Interface":
            return self._ip
        else:
            return None

    @property
    def type(self) -> str:
        """The type of the vNIC."""
        return self._type

    @property
    def uplink(self) -> Resource:
        """The uplink of the vNIC."""
        return self._uplink

    @property
    def downlink(self) -> Resource:
        """The downlink of the vNIC."""
        return self._downlink

    def downlink_utilization(self, inertval: float = 0.1) -> float:
        """The downlink utilization of the vNIC in percentage."""
        return self.downlink.utilization_in_past(inertval) * 100

    def uplink_utilization(self, inertval: float = 0.1) -> float:
        """The uplink utilization of the vNIC in percentage."""
        return self.uplink.utilization_in_past(inertval) * 100

bandwidth: float property

The bandwidth of the vNIC in MB/s.

connected_to: Optional[Union[vHost, vRouter, vSwitch, vGateway]] property

The connected device.

The downlink of the vNIC.

host: Union[vHost, vRouter, vSwitch, vGateway] property

The attached host.

ip: Optional[IPv4Address] property

The IP address of the vNIC.

type: str property

The type of the vNIC.

The uplink of the vNIC.

__init__(host, connected_to=None, bandwidth=1000, delay=0.02, ip=None, at=simulation.now, after=None, label=None)

Create a vNIC object.

Parameters:

Name Type Description Default
host Union[vHost, vRouter, vSwitch, vGateway]

the host that the vNIC is attached to.

 required
connected_to Optional[Union[vHost, vRouter, vSwitch, vGateway]]

the device that the vNIC is connected to. Defaults to None.

 None
bandwidth int

the bandwidth in MBps of the vNIC. Defaults to 1000.

 1000
delay float

the processing delay of the vNIC. Defaults to 0.02.

 0.02
ip Optional[IPv4Address]

the IP address of the vNIC. Defaults to None.

 None
at Union[int, float, Callable]

same as Entity. Defaults to simulation.now.

 now
after Optional[Entity | List[Entity]]

same as Entity. Defaults to None.

 None
label Optional[str]

Same as Entity. Defaults to None.

 None
Source code in PyCloudSim\entity\v_nic.py 
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
def __init__(
    self,
    host: Union[vHost, vRouter, vSwitch, vGateway],
    connected_to: Optional[Union[vHost, vRouter, vSwitch, vGateway]] = None,
    bandwidth: int = 1000,
    delay: float = 0.02,
    ip: Optional[IPv4Address] = None,
    at: Union[int, float, Callable] = simulation.now,
    after: Optional[Entity | List[Entity]] = None,
    label: Optional[str] = None,
):
    """Create a vNIC object.

    Args:
        host (Union[vHost, vRouter, vSwitch, vGateway]): the host that the vNIC is attached to.
        connected_to (Optional[Union[vHost, vRouter, vSwitch, vGateway]], optional): the device that the vNIC is connected to. Defaults to None.
        bandwidth (int, optional): the bandwidth in MBps of the vNIC. Defaults to 1000.
        delay (float, optional): the processing delay of the vNIC. Defaults to 0.02.
        ip (Optional[IPv4Address], optional): the IP address of the vNIC. Defaults to None.
        at (Union[int, float, Callable], optional): same as Entity. Defaults to simulation.now.
        after (Optional[Entity  |  List[Entity]], optional): same as Entity. Defaults to None.
        label (Optional[str], optional): Same as Entity. Defaults to None.
    """
    super().__init__(at, after, label)
    self._host = host
    self._connected_to = connected_to
    self._bandwidth = MiB(bandwidth)
    self._delay = delay
    self._ip = ip
    if host.__class__.__name__ == "vHost" or host.__class__.__name__ == "vRouter":
        self._type = "Interface"
    else:
        self._type = "Port"
    self._uplink = Resource(
        capacity=self.bandwidth, label=f"vNIC {self.label} Uplink"
    )
    self._downlink = Resource(
        capacity=self.bandwidth, label=f"vNIC {self.label} Downlink"
    )
    simulation.NICS.append(self)

creation()

Creation process of the vNIC

Source code in PyCloudSim\entity\v_nic.py 
68
69
70
def creation(self):
    """Creation process of the vNIC"""
    return super().creation()

The downlink utilization of the vNIC in percentage.

Source code in PyCloudSim\entity\v_nic.py 
198
199
200
def downlink_utilization(self, inertval: float = 0.1) -> float:
    """The downlink utilization of the vNIC in percentage."""
    return self.downlink.utilization_in_past(inertval) * 100

receive_packet(packet, delay=0.0)

Rceive a vPacket, the vPacket will be dropped if the attached host does not have enough RAM.

Parameters:

Name Type Description Default
packet vPacket

the vPacket to be recevived.

 required
delay float

the delay for receiving this vPacket in term of transmitting time. Defaults to 0.0.

 0.0
Source code in PyCloudSim\entity\v_nic.py 
 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
def receive_packet(self, packet: vPacket, delay: float = 0.0):
    """Rceive a vPacket, the vPacket will be dropped if the attached host does not have enough RAM.

    Args:
        packet (vPacket): the vPacket to be recevived.
        delay (float, optional): the delay for receiving this vPacket in term of transmitting time. Defaults to 0.0.
    """
    self.uplink.distribute(packet, packet.size)
    LOGGER.debug(
        f"{simulation.now:0.2f}:\tvPacket {packet.label} is using {packet.size}/{self.uplink.available_quantity}/{self.uplink.capacity} bytes of vNIC {self.label} uplink."
    )

    def _received_packet():
        self.uplink.release(packet)
        packet.status.remove(TRANSMITTING)
        packet.status.remove(DECODED)
        try:
            self.host.cache_packet(packet)
            LOGGER.info(
                f"{simulation.now:0.2f}:\tvPacket {packet.label} is received by {self.host.__class__.__name__} {self.host.label}"
            )
        except:
            packet.drop()
            LOGGER.info(
                f"{simulation.now:0.2f}:\tvPacket {packet.label} is droped by {self.host.__class__.__name__} {self.host.label}"
            )

    Actor(
        at=simulation.now + delay,
        action=_received_packet,
        label=f"vNIC {self.label} Receive Packet",
        priority=CORE_EXECUTE_PROCESS,
    )

send_packet(packet, delay=0.0)

Send a vPacket.

Parameters:

Name Type Description Default
packet vPacket

the vPacket to be sent.

 required
delay float

the delay for sending this packet in term of transmitting time. Defaults to 0.0.

 0.0
Source code in PyCloudSim\entity\v_nic.py 
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
def send_packet(self, packet: vPacket, delay: float = 0.0):
    """Send a vPacket.

    Args:
        packet (vPacket): the vPacket to be sent.
        delay (float, optional): the delay for sending this packet in term of transmitting time. Defaults to 0.0.
    """
    self.downlink.distribute(packet, packet.size)
    LOGGER.debug(
        f"{simulation.now:0.2f}:\tvPacket {packet.label} is using {packet.size}/{self.downlink.available_quantity}/{self.downlink.capacity} bytes of vNIC {self.label} downlink."
    )
    self.host.packets.remove(packet)
    packet.status.append(TRANSMITTING)
    packet.status.remove(QUEUED)

    def _sent_packet():
        self.downlink.release(packet)
        if self.host.__class__.__name__ == "vGateway":
            pass
        else:
            self.host.ram.release(packet)  # type: ignore
        LOGGER.info(
            f"{simulation.now:0.2f}:\tvPacket {packet.label} is sent by {self.host.__class__.__name__} {self.host.label}"
        )
        Actor(
            at=simulation.now,
            action=self.host.send_packets,
            label=f"{self.__class__.__name__} {self.label} Send Packets",
            priority=HOST_SCHEDULE_PACKET,
        )

    Actor(
        at=simulation.now + delay,
        action=_sent_packet,
        label=f"vNIC {self.label} Send Packet",
        priority=CORE_EXECUTE_PROCESS,
    )

termination()

Termination process of the vNIC

Source code in PyCloudSim\entity\v_nic.py 
72
73
74
def termination(self):
    """Termination process of the vNIC"""
    return super().termination()

The uplink utilization of the vNIC in percentage.

Source code in PyCloudSim\entity\v_nic.py 
202
203
204
def uplink_utilization(self, inertval: float = 0.1) -> float:
    """The uplink utilization of the vNIC in percentage."""
    return self.uplink.utilization_in_past(inertval) * 100