PocketMine-MP 5.21.1 git-2ff647079265e7c600203af4fd902b15e99d49a4
ReceiveReliabilityLayer.php
1<?php
2
3/*
4 * This file is part of RakLib.
5 * Copyright (C) 2014-2022 PocketMine Team <https://github.com/pmmp/RakLib>
6 *
7 * RakLib is not affiliated with Jenkins Software LLC nor RakNet.
8 *
9 * RakLib is free software: you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License as published by
11 * the Free Software Foundation, either version 3 of the License, or
12 * (at your option) any later version.
13 */
14
15declare(strict_types=1);
16
17namespace raklib\generic;
18
25use function array_fill;
26use function assert;
27use function count;
28
30
31 public static int $WINDOW_SIZE = 2048;
32
33 private int $windowStart;
34 private int $windowEnd;
35 private int $highestSeqNumber = -1;
36
38 private array $ACKQueue = [];
40 private array $NACKQueue = [];
41
42 private int $reliableWindowStart;
43 private int $reliableWindowEnd;
45 private array $reliableWindow = [];
46
48 private array $receiveOrderedIndex;
50 private array $receiveSequencedHighestIndex;
52 private array $receiveOrderedPackets;
53
55 private array $splitPackets = [];
56
62 public function __construct(
63 private \Logger $logger,
64 private \Closure $onRecv,
65 private \Closure $sendPacket,
66 private int $maxSplitPacketPartCount = PHP_INT_MAX,
67 private int $maxConcurrentSplitPackets = PHP_INT_MAX
68 ){
69 $this->windowStart = 0;
70 $this->windowEnd = self::$WINDOW_SIZE;
71
72 $this->reliableWindowStart = 0;
73 $this->reliableWindowEnd = self::$WINDOW_SIZE;
74
75 $this->receiveOrderedIndex = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, 0);
76 $this->receiveSequencedHighestIndex = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, 0);
77
78 $this->receiveOrderedPackets = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, []);
79 }
80
81 private function handleEncapsulatedPacketRoute(EncapsulatedPacket $pk) : void{
82 ($this->onRecv)($pk);
83 }
84
90 private function handleSplit(EncapsulatedPacket $packet) : ?EncapsulatedPacket{
91 if($packet->splitInfo === null){
92 return $packet;
93 }
94 $totalParts = $packet->splitInfo->getTotalPartCount();
95 $partIndex = $packet->splitInfo->getPartIndex();
96 if(
97 $totalParts >= $this->maxSplitPacketPartCount or $totalParts < 0 or
98 $partIndex >= $totalParts or $partIndex < 0
99 ){
100 $this->logger->debug("Invalid split packet part, too many parts or invalid split index (part index $partIndex, part count $totalParts)");
101 return null;
102 }
103
104 $splitId = $packet->splitInfo->getId();
105 if(!isset($this->splitPackets[$splitId])){
106 if(count($this->splitPackets) >= $this->maxConcurrentSplitPackets){
107 $this->logger->debug("Ignored split packet part because reached concurrent split packet limit of $this->maxConcurrentSplitPackets");
108 return null;
109 }
110 $this->splitPackets[$splitId] = array_fill(0, $totalParts, null);
111 }elseif(count($this->splitPackets[$splitId]) !== $totalParts){
112 $this->logger->debug("Wrong split count $totalParts for split packet $splitId, expected " . count($this->splitPackets[$splitId]));
113 return null;
114 }
115
116 $this->splitPackets[$splitId][$partIndex] = $packet;
117
118 $parts = [];
119 foreach($this->splitPackets[$splitId] as $splitIndex => $part){
120 if($part === null){
121 return null;
122 }
123 $parts[$splitIndex] = $part;
124 }
125
126 //got all parts, reassemble the packet
127 $pk = new EncapsulatedPacket();
128 $pk->buffer = "";
129
130 $pk->reliability = $packet->reliability;
131 $pk->messageIndex = $packet->messageIndex;
132 $pk->sequenceIndex = $packet->sequenceIndex;
133 $pk->orderIndex = $packet->orderIndex;
134 $pk->orderChannel = $packet->orderChannel;
135
136 for($i = 0; $i < $totalParts; ++$i){
137 $pk->buffer .= $parts[$i]->buffer;
138 }
139
140 unset($this->splitPackets[$splitId]);
141
142 return $pk;
143 }
144
145 private function handleEncapsulatedPacket(EncapsulatedPacket $packet) : void{
146 if($packet->messageIndex !== null){
147 //check for duplicates or out of range
148 if($packet->messageIndex < $this->reliableWindowStart or $packet->messageIndex > $this->reliableWindowEnd or isset($this->reliableWindow[$packet->messageIndex])){
149 return;
150 }
151
152 $this->reliableWindow[$packet->messageIndex] = true;
153
154 if($packet->messageIndex === $this->reliableWindowStart){
155 for(; isset($this->reliableWindow[$this->reliableWindowStart]); ++$this->reliableWindowStart){
156 unset($this->reliableWindow[$this->reliableWindowStart]);
157 ++$this->reliableWindowEnd;
158 }
159 }
160 }
161
162 if(($packet = $this->handleSplit($packet)) === null){
163 return;
164 }
165
166 if(PacketReliability::isSequencedOrOrdered($packet->reliability) and ($packet->orderChannel < 0 or $packet->orderChannel >= PacketReliability::MAX_ORDER_CHANNELS)){
167 //TODO: this should result in peer banning
168 $this->logger->debug("Invalid packet, bad order channel ($packet->orderChannel)");
169 return;
170 }
171
172 if(PacketReliability::isSequenced($packet->reliability)){
173 if($packet->sequenceIndex < $this->receiveSequencedHighestIndex[$packet->orderChannel] or $packet->orderIndex < $this->receiveOrderedIndex[$packet->orderChannel]){
174 //too old sequenced packet, discard it
175 return;
176 }
177
178 $this->receiveSequencedHighestIndex[$packet->orderChannel] = $packet->sequenceIndex + 1;
179 $this->handleEncapsulatedPacketRoute($packet);
180 }elseif(PacketReliability::isOrdered($packet->reliability)){
181 if($packet->orderIndex === $this->receiveOrderedIndex[$packet->orderChannel]){
182 //this is the packet we expected to get next
183 //Any ordered packet resets the sequence index to zero, so that sequenced packets older than this ordered
184 //one get discarded. Sequenced packets also include (but don't increment) the order index, so a sequenced
185 //packet with an order index less than this will get discarded
186 $this->receiveSequencedHighestIndex[$packet->orderChannel] = 0;
187 $this->receiveOrderedIndex[$packet->orderChannel] = $packet->orderIndex + 1;
188
189 $this->handleEncapsulatedPacketRoute($packet);
190 $i = $this->receiveOrderedIndex[$packet->orderChannel];
191 for(; isset($this->receiveOrderedPackets[$packet->orderChannel][$i]); ++$i){
192 $this->handleEncapsulatedPacketRoute($this->receiveOrderedPackets[$packet->orderChannel][$i]);
193 unset($this->receiveOrderedPackets[$packet->orderChannel][$i]);
194 }
195
196 $this->receiveOrderedIndex[$packet->orderChannel] = $i;
197 }elseif($packet->orderIndex > $this->receiveOrderedIndex[$packet->orderChannel]){
198 if(count($this->receiveOrderedPackets[$packet->orderChannel]) >= self::$WINDOW_SIZE){
199 //queue overflow for this channel - we should probably disconnect the peer at this point
200 return;
201 }
202 $this->receiveOrderedPackets[$packet->orderChannel][$packet->orderIndex] = $packet;
203 }else{
204 //duplicate/already received packet
205 }
206 }else{
207 //not ordered or sequenced
208 $this->handleEncapsulatedPacketRoute($packet);
209 }
210 }
211
212 public function onDatagram(Datagram $packet) : void{
213 if($packet->seqNumber < $this->windowStart or $packet->seqNumber > $this->windowEnd or isset($this->ACKQueue[$packet->seqNumber])){
214 $this->logger->debug("Received duplicate or out-of-window packet (sequence number $packet->seqNumber, window " . $this->windowStart . "-" . $this->windowEnd . ")");
215 return;
216 }
217
218 unset($this->NACKQueue[$packet->seqNumber]);
219 $this->ACKQueue[$packet->seqNumber] = $packet->seqNumber;
220 if($this->highestSeqNumber < $packet->seqNumber){
221 $this->highestSeqNumber = $packet->seqNumber;
222 }
223
224 if($packet->seqNumber === $this->windowStart){
225 //got a contiguous packet, shift the receive window
226 //this packet might complete a sequence of out-of-order packets, so we incrementally check the indexes
227 //to see how far to shift the window, and stop as soon as we either find a gap or have an empty window
228 for(; isset($this->ACKQueue[$this->windowStart]); ++$this->windowStart){
229 ++$this->windowEnd;
230 }
231 }elseif($packet->seqNumber > $this->windowStart){
232 //we got a gap - a later packet arrived before earlier ones did
233 //we add the earlier ones to the NACK queue
234 //if the missing packets arrive before the end of tick, they'll be removed from the NACK queue
235 for($i = $this->windowStart; $i < $packet->seqNumber; ++$i){
236 if(!isset($this->ACKQueue[$i])){
237 $this->NACKQueue[$i] = $i;
238 }
239 }
240 }else{
241 assert(false, "received packet before window start");
242 }
243
244 foreach($packet->packets as $pk){
245 $this->handleEncapsulatedPacket($pk);
246 }
247 }
248
249 public function update() : void{
250 $diff = $this->highestSeqNumber - $this->windowStart + 1;
251 assert($diff >= 0);
252 if($diff > 0){
253 //Move the receive window to account for packets we either received or are about to NACK
254 //we ignore any sequence numbers that we sent NACKs for, because we expect the client to resend them
255 //when it gets a NACK for it
256
257 $this->windowStart += $diff;
258 $this->windowEnd += $diff;
259 }
260
261 if(count($this->ACKQueue) > 0){
262 $pk = new ACK();
263 $pk->packets = $this->ACKQueue;
264 ($this->sendPacket)($pk);
265 $this->ACKQueue = [];
266 }
267
268 if(count($this->NACKQueue) > 0){
269 $pk = new NACK();
270 $pk->packets = $this->NACKQueue;
271 ($this->sendPacket)($pk);
272 $this->NACKQueue = [];
273 }
274 }
275
276 public function needsUpdate() : bool{
277 return count($this->ACKQueue) !== 0 or count($this->NACKQueue) !== 0;
278 }
279}
__construct(private \Logger $logger, private \Closure $onRecv, private \Closure $sendPacket, private int $maxSplitPacketPartCount=PHP_INT_MAX, private int $maxConcurrentSplitPackets=PHP_INT_MAX)