Skip to content

vPacket

Bases: vSoftwareEntity

Source code in PyCloudSim\entity\v_packet.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
class vPacket(vSoftwareEntity):
    def __init__(
        self,
        src: vContainer | vUser,
        dst: vContainer | vUser,
        size: int | Callable[..., int],
        priority: int | Callable[..., int],
        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 packet."""
        super().__init__(label, create_at, terminate_at, precursor)
        self._src = src
        self._src_host = src.host
        self._dst = dst
        self._dst_host = dst.host
        if callable(size):
            self._size = size()
        else:
            self._size = size
        if callable(priority):
            self._priority = priority()
        else:
            self._priority = priority
        self._content = bytes()
        self._path: List[vHardwareEntity | vGateway] = []
        self._current_hop: vHardwareEntity | vGateway = None  # type: ignore
        self._next_hop: vHardwareEntity | vGateway = None  # type: ignore

    def on_creation(self):
        """Creation procedure of the simulated packet."""
        super().on_creation()
        logger.info(f"{simulation.now}:\t{self} is created.")
        self.initiate(simulation.now)

    def on_initiate(self) -> None:
        """The initiation procedure of the simulated packet."""
        super().on_initiate()
        # generate the random bytes content
        self._content = randbytes(self.size)
        # find the shortest path from src to dst
        self._path = simulation.network.route(self.src_host, self.dst_host)  # type: ignore
        if len(self.path) == 1:
            self._path = [self.path[0], self.path[0]]
        self._current_hop = self.path[0]
        self._next_hop = self.path[1]
        # inject the packet to its src
        try:
            self.get(self.src_host.ram, self.size)
        except:
            # drop the packet if its src does not have enough ram
            self.drop()
            return
        self.state.append(Constants.DECODED)
        self.src_host.packet_queue.append(self)
        logger.info(f"{simulation.now}:\t{self} is initiated.")

    def on_termination(self):
        """Termination procedure of the simulated packet."""
        super().on_termination()
        logger.info(f"{simulation.now}:\tPacket {self.label} is terminated.")

    def on_fail(self):
        """Failure procedure of the simulated packet."""
        super().on_fail()
        logger.info(f"{simulation.now}:\tPacket {self.label} is dropped.")

    def drop(self):
        """Drop the simulated packet which calls the fail procedure."""
        self.fail(simulation.now)

    @property
    def src(self):
        """return the source of the packet, could be a simulated user or container."""
        return self._src

    @property
    def src_host(self):
        """return the source host of the packet, could be a simulated host or gateway."""
        return self._src_host

    @property
    def dst(self):
        """return the destination of the packet, could be a simulated user or container."""
        return self._dst

    @property
    def dst_host(self):
        """return the destination host of the packet, could be a simulated host or gateway."""
        return self._dst_host

    @property
    def path(self):
        """return the path of the packet."""
        return self._path

    @property
    def size(self) -> int:
        """return the size of the packet."""
        return self._size

    @property
    def content(self) -> bytes:
        """return the content of the packet."""
        return self._content

    @property
    def priority(self) -> int:
        """return the priority of the packet."""
        return self._priority

    @property
    def current_hop(self):
        """return the current hop of the packet."""
        return self._current_hop

    @property
    def next_hop(self):
        """return the next hop of the packet."""
        return self._next_hop

    @property
    def decoded(self) -> bool:
        """return True if the packet is decoded. This happens after the asscoiated decoder process is executed."""
        return Constants.DECODED in self.state

    @property
    def in_transmission(self) -> bool:
        """return True if the packet is in transmission."""
        return Constants.INTRANSMISSION in self.state

content: bytes property

return the content of the packet.

current_hop property

return the current hop of the packet.

decoded: bool property

return True if the packet is decoded. This happens after the asscoiated decoder process is executed.

dst property

return the destination of the packet, could be a simulated user or container.

dst_host property

return the destination host of the packet, could be a simulated host or gateway.

in_transmission: bool property

return True if the packet is in transmission.

next_hop property

return the next hop of the packet.

path property

return the path of the packet.

priority: int property

return the priority of the packet.

size: int property

return the size of the packet.

src property

return the source of the packet, could be a simulated user or container.

src_host property

return the source host of the packet, could be a simulated host or gateway.

__init__(src, dst, size, priority, label=None, create_at=None, terminate_at=None, precursor=None)

Create a simulated packet.

Source code in PyCloudSim\entity\v_packet.py 
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
def __init__(
    self,
    src: vContainer | vUser,
    dst: vContainer | vUser,
    size: int | Callable[..., int],
    priority: int | Callable[..., int],
    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 packet."""
    super().__init__(label, create_at, terminate_at, precursor)
    self._src = src
    self._src_host = src.host
    self._dst = dst
    self._dst_host = dst.host
    if callable(size):
        self._size = size()
    else:
        self._size = size
    if callable(priority):
        self._priority = priority()
    else:
        self._priority = priority
    self._content = bytes()
    self._path: List[vHardwareEntity | vGateway] = []
    self._current_hop: vHardwareEntity | vGateway = None  # type: ignore
    self._next_hop: vHardwareEntity | vGateway = None  # type: ignore

drop()

Drop the simulated packet which calls the fail procedure.

Source code in PyCloudSim\entity\v_packet.py 
97
98
99
def drop(self):
    """Drop the simulated packet which calls the fail procedure."""
    self.fail(simulation.now)

on_creation()

Creation procedure of the simulated packet.

Source code in PyCloudSim\entity\v_packet.py 
59
60
61
62
63
def on_creation(self):
    """Creation procedure of the simulated packet."""
    super().on_creation()
    logger.info(f"{simulation.now}:\t{self} is created.")
    self.initiate(simulation.now)

on_fail()

Failure procedure of the simulated packet.

Source code in PyCloudSim\entity\v_packet.py 
92
93
94
95
def on_fail(self):
    """Failure procedure of the simulated packet."""
    super().on_fail()
    logger.info(f"{simulation.now}:\tPacket {self.label} is dropped.")

on_initiate()

The initiation procedure of the simulated packet.

Source code in PyCloudSim\entity\v_packet.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def on_initiate(self) -> None:
    """The initiation procedure of the simulated packet."""
    super().on_initiate()
    # generate the random bytes content
    self._content = randbytes(self.size)
    # find the shortest path from src to dst
    self._path = simulation.network.route(self.src_host, self.dst_host)  # type: ignore
    if len(self.path) == 1:
        self._path = [self.path[0], self.path[0]]
    self._current_hop = self.path[0]
    self._next_hop = self.path[1]
    # inject the packet to its src
    try:
        self.get(self.src_host.ram, self.size)
    except:
        # drop the packet if its src does not have enough ram
        self.drop()
        return
    self.state.append(Constants.DECODED)
    self.src_host.packet_queue.append(self)
    logger.info(f"{simulation.now}:\t{self} is initiated.")

on_termination()

Termination procedure of the simulated packet.

Source code in PyCloudSim\entity\v_packet.py 
87
88
89
90
def on_termination(self):
    """Termination procedure of the simulated packet."""
    super().on_termination()
    logger.info(f"{simulation.now}:\tPacket {self.label} is terminated.")