Cours programmation réseau en C++

UDP – Envoi de paquets ordonné fiable

Bousk

Lors du dernier chapitre, nous avons implémenté un envoi de paquets ordonné non fiable.

Cette fois nous allons implémenter un envoi ordonné fiable.

Commentez

- Titre : Cours programmation réseau en C++

- Auteur : Bousk

- Parution : 5 juillet 2018

- Mise à jour : 18 juillet 2019

- Public visé :

- Licence : Les sources présentées sur cette page sont libres de droits et vous pouvez les utiliser à votre convenance. Par contre, la page de présentation constitue une œuvre intellectuelle protégée par les droits d'auteur. Copyright ® 2019 Cyrille (Bousk) Bousquet. Aucune reproduction, même partielle, ne peut être faite de ce site et de l'ensemble de son contenu : textes, documents, images, etc. sans l'autorisation expresse de l'auteur. Sinon vous encourez selon la loi jusqu'à trois ans de prison et jusqu'à 300 000 € de dommages et intérêts.

Sommaire

- I. Réaliser un envoi fiable via un protocole non fiable ?

- II. Comment créer de la fiabilité ?

- II-A. Quand renvoyer un paquet ?

- II-B. Identifier les paquets perdus

- III. Multiplexeur, Démultiplexeur

- IV. Paquets

- V. Multiplexeur

- V-A. Fonctionnement

- V-A-1. État d’envoi d’un paquet

- V-B. Interface

- V-C. Implémentations

- V-C-1. void queue(std::vector<uint8_t>&& data);

- V-C-2. size_t serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId);

- V-C-3. void onDatagramAcked(Datagram::ID datagramId);

- V-C-4. void onDatagramLost(Datagram::ID datagramId);

- VI. Démultiplexeur

- VI-A. Fonctionnement

- VI-B. Interface

- VI-C. Implémentations

- VI-C-1. std::vector<std::vector<uint8_t>> process();

- VII. Attention aux attaques malicieuses : amélioration et optimisation

- VII-A. Adieu vector, bonjour array

- VII-A-1. Quelle taille pour l’array ?

- VII-A-2. Modifications du démultiplexeur

- VII-A-2-a. onPacketReceived(const Packet* pckt)

- VII-A-2-b. process()

- VII-A-3. Modifications du multiplexeur

- VII-A-3-a. serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId)

- VII-A-3-b. onDatagramAcked(Datagram::ID datagramId)

- VIII. Tests

I - Réaliser un envoi fiable via un protocole non fiable ?

UDP est un protocole non fiable. Chaque datagramme envoyé peut être perdu, dupliqué ou désordonné.

Pourtant dans le chapitre précédent nous avons réussi à proposer un protocole, utilisant UDP, et permettant d’échanger des données ordonnées, sans duplication, mais toujours non fiables.

Cette fois nous allons franchir la marche suivante et rendre l’envoi de données ordonnées fiables possible.

La faisabilité ne devrait pas vous surprendre : UDP utilise IP tout comme TCP, et ce dernier est un protocole ordonné fiable alors que IP ne l’est pas.

Il ne s’agit pas de remplacer TCP mais de proposer cette fonctionnalité (envoi ordonné fiable) dans notre moteur, parce qu’il s’agit d’un aspect nécessaire dans certains cas.

Toutefois, la fiabilité de notre protocole sera différente de ce qui est proposé par TCP.

II - Comment créer de la fiabilité ?

Il n’y a pas de secret : pour ajouter de la fiabilité, il va falloir renvoyer les données non reçues.

Dès le 2e chapitre, nous avons introduit un système nous permettant d’acquitter des identifiants et d’informer la machine distante de ceux que nous avons reçus, et donc de déduire lesquels ont été à priori perdus. Cette information est, comme dit alors, la pierre angulaire de notre protocole, et tout particulièrement pour le chapitre actuel.

II-A - Quand renvoyer un paquet ?

Il s’agit ici de la partie critique de ce protocole : à quel moment renvoyer un paquet ?

- Attendre trop longtemps entraîne une latence qui peut être dérangeante.

- Le renvoyer trop rapidement (ou trop souvent) signifie une bande passante plus élevée ;

- Ou retarder l’envoi d’autres données pour éviter un pic d’envoi.

- Ou, dans le pire des cas, provoquer une congestion et une mise en défaut de la connexion.

Il y a donc deux solutions ici

- Soit attendre que le datagramme qui contenait le paquet soit identifié comme perdu ;

- Soit renvoyer plusieurs fois chaque paquet dans des datagrammes différents afin que l’un d’eux parvienne à l’autre machine.

Dans ce chapitre, nous implémenterons la première solution.

Un tel protocole sera idéal pour des phases plus lentes ou des données moins critiques.

Pensez au cas où les joueurs sont dans les menus, par exemple en train de se préparer pour la partie à venir, ou à l’utilisation d’un tchat textuel.

Dans le premier cas, puisqu’il n’y a pas encore de simulation, la quantité de données à envoyer est faible (seules les données du menu sont envoyées, qui sont généralement négligeables comparées à ce qui transite pendant les phases de jeu).

Dans le deuxième cas, si le texte arrive avec quelques frames de retard on estime que ce n’est pas critique pour l’expérience des joueurs.

Dans tous les cas vous voyez ici émerger une des problématiques du réseau dans un jeu vidéo : le compromis.

Puisque notre en-tête contient l’acquittement de 64 datagrammes, si notre vitesse d’envoi est de 30 datagrammes par seconde, nous remarquerons sa perte après ~2s. À 60 datagrammes par seconde, il sera marqué perdu après ~1s.

II-B - Identifier les paquets perdus

La clé de la fiabilité est de pouvoir identifier quels paquets ne sont pas parvenus à la machine distante afin de les renvoyer.

Nous avons un suivi des datagrammes reçus, et manquants, il nous faut donc un moyen d’identifier quels paquets étaient présents dans un datagramme donné pour agir en conséquence.

Il se trouve que nous envoyons nous-même le datagramme, il suffit donc de traquer les paquets que nous y insérons afin de pouvoir savoir, lors de la réception de l’acquittement d’un datagramme donné, quels paquets ont été reçus et perdus.

Il ne s’agit pas de la seule manière de réaliser la fiabilité, mais il s’agira de la méthode introduite et présentée dans ce chapitre.

III - Multiplexeur, Démultiplexeur

Comme dans le chapitre précédent, un multiplexeur pour l’envoi et un démultiplexeur pour la réception seront mis en place. Leurs fonctionnements seront similaires mais leurs implémentations permettront cette fois de créer un protocole ordonné fiable.

IV - Paquets

La structure de paquet du chapitre précédent sera réutilisée.

Afin de ne pas avoir à modifier cette structure, le multiplexeur fera le suivi des datagrammes incluant chaque paquet en interne.

V - Multiplexeur

V-A - Fonctionnement

Le multiplexeur sera le moteur et contrôleur de ce protocole. C’est lui qui prend la décision d’envoyer les données suivantes ou non en coordination avec le démultiplexeur que nous allons voir juste après.

Le principe de base est simple : quand un paquet est envoyé, on note le datagramme dans lequel il a été inclus. Quand ce datagramme est acquitté ou déclaré perdu, on le supprime enfin de la file d’envoi ou on le prépare à être renvoyé.

V-A-1 - État d’envoi d’un paquet

Ajoutons donc un état d’envoi à nos paquets en attente dans le multiplexeur : un simple booléen fera l’affaire.

Ce booléen définira si le paquet doit être (r)envoyé ou non. Quand un paquet est inclus dans un datagramme, l’état passe à false. Quand le datagramme qui l’inclut est perdu, il repasse à true.

Ceci est le fonctionnement de base qui pourra être complexifié par la suite.

Par exemple si un paquet un peu vieux n’a toujours pas été reçu, peut-être serait-ce une bonne idée de le renvoyer plus rapidement afin de ne pas congestionner le protocole (puisque les messages peuvent uniquement être extraits dans l’ordre d’envoi).

V-B - Interface

L’interface du multiplexeur sera quasi identique. On ajoutera une interface pour notifier quand un datagramme a été acquitté ou perdu ainsi que le suivi des paquets inclus dans chaque datagramme.

Partons donc sur cette base :

Multiplexeur

1. class Multiplexer

2. {

3. friend class ReliableOrdered_Multiplexer_Test;

4. public:

5. Multiplexer() = default;

6. ~Multiplexer() = default;

8. void queue(std::vector<uint8_t>&& msgData);

9. size_t serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId);

11. void onDatagramAcked(Datagram::ID datagramId);

12. void onDatagramLost(Datagram::ID datagramId);

13. private:

14. class ReliablePacket

15. {

16. public:

17. Packet& packet() { return mPacket; }

18. const Packet& packet() const { return mPacket; }

20. void onSent(Datagram::ID datagramId) { mDatagramsIncluding.insert(datagramId); }

21. bool isIncludedIn(Datagram::ID datagramId) const { return mDatagramsIncluding.find(datagramId) != mDatagramsIncluding.cend(); }

23. private:

24. Packet mPacket;

25. std::set<Datagram::ID> mDatagramsIncluding;

26. };

27. std::vector<ReliablePacket> mQueue;

28. Packet::Id mNextId{ 0 };

29. };

V-C - Implémentations

V-C-1 - void queue(std::vector<uint8_t>&& data);

L’implémentation de queue sera quasi identique à ce qui a été fait pour le protocole ordonné non fiable du chapitre précédent.

L’unique différence réside dans le type de données que l’on met en file d’envoi puisque nous avons une structure intermédiaire englobant le paquet.

queue

1. void Multiplexer::queue(std::vector<uint8_t>&& msgData)

2. {

3. assert(msgData.size() <= Packet::MaxMessageSize);

4. if (msgData.size() > Packet::DataMaxSize)

5. {

6. size_t queuedSize = 0;

7. while (queuedSize < msgData.size())

8. {

9. const auto fragmentSize = std::min(Packet::DataMaxSize, static_cast<uint16_t>(msgData.size() - queuedSize));

10. mQueue.resize(mQueue.size() + 1);

11. Packet& packet = mQueue.back().packet();

12. packet.header.id = mNextId++;

13. packet.header.type = ((queuedSize == 0) ? Packet::Type::FirstFragment : Packet::Type::Fragment);

14. packet.header.size = fragmentSize;

15. memcpy(packet.data(), msgData.data() + queuedSize, fragmentSize);

16. queuedSize += fragmentSize;

17. }

18. mQueue.back().packet().header.type = Packet::Type::LastFragment;

19. assert(queuedSize == msgData.size());

20. }

21. else

22. {

23. mQueue.resize(mQueue.size() + 1);

24. Packet& packet = mQueue.back().packet();

25. packet.header.id = mNextId++;

26. packet.header.type = Packet::Type::Packet;

27. packet.header.size = static_cast<uint16_t>(msgData.size());

28. memcpy(packet.data(), msgData.data(), msgData.size());

29. }

30. }

V-C-2 - size_t serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId);

Ici aussi l’implémentation sera similaire. Nous itérons sur la liste d’envoi, et sérialisons les paquets qui le peuvent et le doivent dans le tampon fourni. Sans oublier de mettre à jour le paquet avec l’information du datagramme qui le contient quand un paquet a été sérialisé.

Multiplexer::serialize

1. size_t Multiplexer::serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId)

2. {

3. size_t serializedSize = 0;

4. for (auto& packetHolder : mQueue)

5. {

6. if (!packetHolder.shouldSend())

7. continue;

8. const auto& packet = packetHolder.packet();

9. if (serializedSize + packet.size() > buffersize)

10. break; //!< Paquet trop gros

12. memcpy(buffer, packet.buffer(), packet.size());

13. serializedSize += packet.size();

14. buffer += packet.size();

16. //!< Quand un paquet a été sérialisé, on enregistre dans quel datagramme il a été inclus

17. packetHolder.onSent(datagramId);

18. }

19. return serializedSize;

20. }

V-C-3 - void onDatagramAcked(Datagram::ID datagramId);

Ici il s’agit de supprimer de la file d’envoi les paquets qui étaient contenus dans le datagramme #datagramId. Une simple itération sur notre file d’envoi fera l’affaire :

Multiplexer::onDatagramAcked

1. void Multiplexer::onDatagramAcked(Datagram::ID datagramId)

2. {

3. mQueue.erase(std::remove_if(mQueue.begin(), mQueue.end()

4. , [&](const ReliablePacket& packetHolder) { return packetHolder.isIncludedIn(datagramId); })

5. , mQueue.cend());

6. }

V-C-4 - void onDatagramLost(Datagram::ID datagramId);

Similaire à l’acquittement, nous itérons sur la liste d’envoi afin cette fois de remettre les paquets concernés dans un état pour être renvoyés :

void Multiplexer::onDatagramLost(Datagram::ID datagramId)
{
for (auto& packetHolder : mQueue)
{
if (packetHolder.isIncludedIn(datagramId))
packetHolder.resend();
}
}

Avec resend de la forme void resend() { mShouldSend = true; }

Ces implémentations sont loin d’être optimales mais sont faciles à comprendre et traduisent directement le besoin de chaque fonction. Aussi il s’agira de la première version mise en place.

VI - Démultiplexeur

VI-A - Fonctionnement

Le démultiplexeur aura l’exact même fonctionnement que pour le protocole ordonné non fiable : réceptionner les paquets, les réordonner puis reformer les messages complets.

Mais contrairement au protocole non fiable, celui-ci devra sortir les messages de façon ordonnée et fiable. Donc quand un paquet est perdu, il faudra attendre qu’il soit renvoyé et reçu avec succès avant de pouvoir extraire et fournir les messages complets à l’application.

VI-B - Interface

Son interface sera identique à la version pour le protocole ordonné non fiable.

Demultiplexer

1. class Demultiplexer

2. {

3. friend class ReliableOrdered_Demultiplexer_Test;

4. public:

5. Demultiplexer() = default;

6. ~Demultiplexer() = default;

8. void onDataReceived(const uint8_t* data, const size_t datasize);

9. std::vector<std::vector<uint8_t>> process();

11. private:

12. void onPacketReceived(const Packet* pckt);

14. private:

15. std::vector<Packet> mPendingQueue;

16. Packet::Id mLastProcessed{ std::numeric_limits<Packet::Id>::max() };

17. };

VI-C - Implémentations

Seule l’implémentation de process sera différente puisque, contrairement au protocole ordonné non fiable, celui-ci ne pourra plus passer un paquet et devra les traiter dans l’ordre sans interruption.

Vous pouvez retrouver les implémentations dans le chapitre précédent.

VI-C-1 - std::vector<std::vector<uint8_t>> process();

L’implémentation est extrêmement similaire au protocole ordonné non fiable.

L’unique différence est, pour avoir la fiabilité, qu’on ne peut plus sauter les paquets non reçus. À la place, il faut maintenant interrompre le traitement.

Demultiplexer::process

1. std::vector<std::vector<uint8_t>> Demultiplexer::process()

2. {

3. std::vector<std::vector<uint8_t>> messagesReady;

5. auto itPacket = mPendingQueue.cbegin();

6. auto itEnd = mPendingQueue.cend();

7. std::vector<Packet>::const_iterator newestProcessedPacket;

8. Packet::Id expectedPacketId = mLastProcessed + 1;

9. //!< La queue est ordonnée, il suffit d’itérer et réassembler les paquets

10. while (itPacket != itEnd && itPacket->id() == expectedPacketId)

11. {

12. if (itPacket->type() == Packet::Type::Packet)

13. {

14. //!< Paquet-Message, on recopie ses données

15. std::vector<uint8_t> msg(itPacket->data(), itPacket->data() + itPacket->datasize());

16. messagesReady.push_back(std::move(msg));

17. newestProcessedPacket = itPacket;

18. ++itPacket;

19. ++expectedPacketId;

20. }

21. else if (itPacket->type() == Packet::Type::FirstFragment)

22. {

23. //!< Vérifier si le message est prêt (tous les fragments ont été reçus)

24. std::vector<uint8_t> msg = [&]()

25. {

26. std::vector<uint8_t> msg(itPacket->data(), itPacket->data() + itPacket->datasize());

27. ++itPacket;

28. ++expectedPacketId;

29. while (itPacket != itEnd && itPacket->id() == expectedPacketId)

30. {

31. if (itPacket->type() == Packet::Type::LastFragment)

32. {

33. //!< Dernier fragment reçu, le message est complet

34. msg.insert(msg.cend(), itPacket->data(), itPacket->data() + itPacket->datasize());

35. return msg;

36. }

37. else if (itPacket->type() != Packet::Type::Fragment)

38. {

39. //!< Si on rentre ici, nous avons reçu un paquet mal formé ou mal intentionné

40. msg.clear();

41. return msg;

42. }

44. msg.insert(msg.cend(), itPacket->data(), itPacket->data() + itPacket->datasize());

45. ++itPacket;

46. ++expectedPacketId;

47. }

48. msg.clear();

49. return msg;

50. }();

51. if (!msg.empty())

52. {

53. //!< Nous avons un message

54. messagesReady.push_back(std::move(msg));

55. newestProcessedPacket = itPacket;

56. //!< On déplace l’itérateur de parcours de la liste après le dernier paquet traité, qui est le dernier paquet du message

57. ++itPacket;

58. }

59. }

60. else

61. {

62. //!< Si on arrive ici, il s’agit d’un message incomplet

63. break;

64. }

65. }

67. //!< Si des messages ont été extraits, on doit mettre à jour notre état interne en supprimant les paquets traités de la file

68. if (!messagesReady.empty())

69. {

70. mLastProcessed = newestProcessedPacket->id();

71. mPendingQueue.erase(mPendingQueue.cbegin(), std::next(newestProcessedPacket));

72. }

74. return messagesReady;

75. }

Nous voici maintenant avec un couple multiplexeur & démultiplexeur capable d’échanger des paquets de façon ordonnée et fiable. Si vous avez lu le chapitre précédent, vous auriez sans doute pu y parvenir de vous-même : le code est en grande partie identique.

Allons donc un peu plus loin pour améliorer ces nouveaux objets.

VII - Attention aux attaques malicieuses : amélioration et optimisation

J’attire votre attention sur un premier point faible des protocoles créés : le vector qui accueille les paquets reçus.

Imaginez ce qu’il se passe si, volontairement ou non (paquet mal formé), un client reçoit de nombreux fragments, mais jamais le dernier afin de compléter un message : le vector grandit indéfiniment, jusqu’à ce que le programme crash par manque de mémoire.

Il existe un moyen relativement simple d’éviter ceci en complexifiant un peu le multiplexeur et le démultiplexeur et, qui plus est, améliore les performances du démultiplexeur : utiliser un tableau statique. Grâce à ce dernier, nous avons le contrôle de la mémoire utilisée, aucune allocation, aucune réallocation et donc aucune fragmentation et un parcours plus efficace des paquets reçus.

VII-A - Adieu vector, bonjour array

Afin de réaliser cette amélioration, le démultiplexeur n’utilisera plus de vector pour les paquets reçus, mais un array.

VII-A-1 - Quelle taille pour l’array ?

L’array, contrairement au vector, a une taille définie à la compilation. Et bien sûr cette taille doit être suffisamment grande pour pouvoir supporter le message le plus grand et fragmenté possible défini.

Cette taille est totalement arbitraire et va induire la mémoire utilisée par le protocole, mais aura aussi une incidence sur la latence de réception des messages en cas de perte.

Définissons cette taille au double du max de fragments d’un message. Ainsi, un message de taille maximale pourra être contenu, et il reste de la place pour d’autres messages plus petits afin de ne pas engorger l’envoi de cet unique gros message.

ReliableOrdered.hpp

1. class Demultiplexer

2. {

3. std::array<Packet, 2 * Packet::MaxPacketsPerMessage> mPendingQueue;

4. };

Il reste un dernier problème : puisque notre file de réception est maintenant statique, tous les paquets qu’elle contient ne sont pas forcément légitimes. Un paquet qui n’a pas été reçu via le réseau est toutefois alloué et présent dans le tableau. Ce n’est pas pour autant un paquet valide et il ne doit donc pas être traité.

Pour repérer qu’un paquet est légitime, plusieurs astuces sont possibles :

- Utiliser un tuple<bool, Packet> dont la valeur du booléen mise à false par défaut indique s’il s’agit d’un paquet correct ou non ;

- Utiliser un masque de bits pour marquer les paquets corrects.

Mais je vais présenter ma petite astuce, qui ne nécessite aucune donnée supplémentaire : utiliser la taille des données du paquet. Si le paquet n’a aucune donnée, il n’a pas été initialisé par le réseau. Pour que cela fonctionne, il faut bien entendu que Packet::Header::size soit initialisée par défaut à 0 – ce qui n’était pas le cas jusqu’à présent.

VII-A-2 - Modifications du démultiplexeur

Mettons maintenant à jour le démultiplexeur.

La liste des paquets reçus est maintenant de taille fixe. Elle est donc définie comme std::array<Packet, QueueSize> avec QueueSize étant static constexpr size_t QueueSize = 2 * Packet::MaxPacketsPerMessage;.

VII-A-2-a - onPacketReceived(const Packet* pckt)

Maintenant à la réception d’un paquet, nous n’avons plus besoin de parcourir notre liste pour le positionner à la bonne place. L’opération devient beaucoup plus simple :

- On définit l’index dans notre tableau en fonction de l’identifiant du paquet et la taille du tableau ;

- On place le paquet à cet index si l’emplacement est libre ;

- S’il est non vide, alors il devrait déjà contenir notre paquet et nous avons reçu un doublon.

void Demultiplexer::onPacketReceived(const Packet* pckt)

1. void Demultiplexer::onPacketReceived(const Packet* pckt)

2. {

3. if (!Utils::IsSequenceNewer(pckt->id(), mLastProcessed))

4. return; //!< Paquet obsolète

6. //!< Calcul de l’index dans le tableau

7. const size_t index = pckt->id() % mPendingQueue.size();

8. Packet& pendingPacket = mPendingQueue[index];

9. if (pendingPacket.datasize() == 0)

10. {

11. // Emplacement disponible, copier simplement les données du réseau dans notre tableau

12. pendingPacket = *pckt;

13. }

14. else

15. {

16. // Emplacement NON disponible, s’assurer qu’il contient déjà notre paquet, sinon il y a un problème

17. assert(pendingPacket.id() == pckt->id() && pendingPacket.datasize() == pckt->datasize());

18. }

19. }

VII-A-2-b - process()

Il reste à mettre à jour le processus d’extraction des messages.

La principale modification consiste à libérer l’emplacement d’un paquet une fois consommé pour créer un message. Puisque nous avons établi plus haut qu’un paquet est valide quand sa taille est non nulle, libérer un paquet va donc simplement consister à réinitialiser sa taille à 0.

std::vector<std::vector<uint8_t>> Demultiplexer::process()

1. std::vector<std::vector<uint8_t>> Demultiplexer::process()

2. {

3. //!< Fonction de réinitialisation d’un paquet

4. auto ResetPacket = [](Packet& pckt) { pckt.header.size = 0; };

5. std::vector<std::vector<uint8_t>> messagesReady;

7. Packet::Id expectedPacketId = mLastProcessed + 1;

8. //!< Il faut itérer sur notre tableau en commençant par le paquet attendu, qui peut ne pas être en index 0

9. const size_t startIndexOffset = expectedPacketId % mPendingQueue.size();

10. for (size_t i = 0; i < mPendingQueue.size(); ++i, ++expectedPacketId)

11. {

12. //!< On calcule l’index dans notre tableau du prochain paquet à traiter

13. const size_t packetIndex = (i + startIndexOffset) % mPendingQueue.size();

14. Packet& packet = mPendingQueue[packetIndex];

15. if (packet.type() == Packet::Type::Packet)

16. {

17. //!< Message complet

18. std::vector<uint8_t> msg(packet.data(), packet.data() + packet.datasize());

19. mLastProcessed = packet.id();

20. ResetPacket(packet);

21. messagesReady.push_back(std::move(msg));

22. }

23. else if (packet.type() == Packet::Type::FirstFragment)

24. {

25. //!< Vérifier que le message est prêt

26. const bool isMessageFull = [=]() mutable

27. {

28. // On saute le premier fragment déjà traité par la boucle sur i

29. ++i;

30. ++expectedPacketId;

31. // On itère sur les paquets restants pour vérifier que notre message soit complet

32. for (size_t j = i; j < mPendingQueue.size(); ++j, ++expectedPacketId)

33. {

34. const size_t idx = (j + startIndexOffset) % mPendingQueue.size();

35. const Packet& pckt = mPendingQueue[idx];

36. if (pckt.id() != expectedPacketId || pckt.datasize() == 0)

37. break; // Un paquet est manquant

38. if (pckt.type() == Packet::Type::LastFragment)

39. {

40. //!< Nous avons atteint et reçu le dernier fragment, le message est complet

41. return true;

42. }

43. else if (pckt.type() != Packet::Type::Fragment)

44. {

45. //!< Si nous arrivons ici nous avons probablement reçu un paquet mal formé ou malicieux

46. break;

47. }

48. }

49. return false;

50. }();

51. if (!isMessageFull)

52. break; // Protocole ordonné fiable : si le message suivant à extraire est incomplet, nous pouvons arrêter le processus d’extraction

54. // Nous avons un message fragmenté complet, nous pouvons maintenant extraire les données et réinitialiser chaque paquet utilisé

55. std::vector<uint8_t> msg(packet.data(), packet.data() + packet.datasize());

56. ++i;

57. ++expectedPacketId;

58. // Itération sur les paquets restants pour compléter le message

59. for (size_t j = i; j < mPendingQueue.size(); ++i, ++j, ++expectedPacketId)

60. {

61. const size_t idx = (j + startIndexOffset) % mPendingQueue.size();

62. Packet& pckt = mPendingQueue[idx];

64. if (pckt.type() == Packet::Type::LastFragment)

65. {

66. //!< Dernier fragment du message maintenant complet

67. msg.insert(msg.cend(), pckt.data(), pckt.data() + pckt.datasize());

68. mLastProcessed = pckt.id();

69. ResetPacket(pckt);

70. messagesReady.push_back(std::move(msg));

71. break;

72. }

73. else if (pckt.type() != Packet::Type::Fragment)

74. {

75. //!< Paquet mal formé ou malicieux

76. break;

77. }

79. msg.insert(msg.cend(), pckt.data(), pckt.data() + pckt.datasize());

80. ResetPacket(pckt);

81. }

82. }

83. else

84. {

85. // Protocole ordonné fiable : si le message suivant à extraire est incomplet, nous pouvons arrêter le processus d’extraction

86. break;

87. }

88. }

89. return messagesReady;

90. }

Dans cette implémentation il est très important que l’application appelle process(); avant de réaliser l’envoi de données.

En effet, le multiplexeur du client A va gérer ses envois en fonction des datagrammes reçus par le client B. Il est donc important que B extraie ses messages avant que A ne soit informé de quelles données ont été reçues et commence à envoyer les suivantes.

Si A envoie des données alors que B n’a pas libéré le tableau de réception des paquets, B finira en état de deadlock : des nouveaux messages vont écraser les précédents, alors qu’il essayera d’extraire le message précédent puisqu’il s’agit d’un protocole ordonné fiable. Le résultat sera que B ne recevra plus aucune donnée sur ce canal.

VII-A-3 - Modifications du multiplexeur

Il faut ensuite mettre à jour le multiplexeur pour qu’il fonctionne de pair avec le démultiplexeur.

Le changement consiste à limiter quels paquets peuvent être envoyés, en fonction de ceux reçus.

Par exemple, si notre démultiplexeur avait une file de réception pouvant contenir deux paquets, alors si le dernier paquet reçu est le paquet #4, nous pouvons envoyer les paquets 5 et 6 uniquement.

Pour cela, commençons par ajouter un membre au multiplexeur : l’identifiant du paquet le plus ancien autorisé à être envoyé, nommé mFirstAllowedPacket, et qui sera initialisé à 0.

VII-A-3-a - serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId)

Il s’agira d’ajouter un test afin de s’assurer que le paquet que l’on veut sérialiser, soit dans les bornes d’identifiants pouvant être envoyés.

size_t Multiplexer::serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId)

1. size_t Multiplexer::serialize(uint8_t* buffer, const size_t buffersize, Datagram::ID datagramId)

2. {

3. size_t serializedSize = 0;

4. for (auto& packetHolder : mQueue)

5. {

6. //!< S’assurer avant tout que le paquet est dans les bornes d’envoi, sinon on peut arrêter d’itérer sur notre file

7. if (!(Utils::SequenceDiff(packetHolder.packet().id(), mFirstAllowedPacket) < Demultiplexer::QueueSize))

8. break;

9. if (!packetHolder.shouldSend())

10. continue;

11. const auto& packet = packetHolder.packet();

12. if (serializedSize + packet.size() > buffersize)

13. continue; //!< Si le paquet est trop gros, essayons d’inclure les suivants

15. memcpy(buffer, packet.buffer(), packet.size());

16. serializedSize += packet.size();

17. buffer += packet.size();

19. packetHolder.onSent(datagramId);

20. }

21. return serializedSize;

22. }

VII-A-3-b - onDatagramAcked(Datagram::ID datagramId)

Apportons maintenant un micro changement à onDatagramAcked(Datagram::ID datagramId) afin de mettre à jour notre champ mNewestAck si nécessaire.

void Multiplexer::onDatagramAcked(Datagram::ID datagramId)

1. void Multiplexer::onDatagramAcked(Datagram::ID datagramId)

2. {

3. if (mQueue.empty())

4. return;

6. mQueue.erase(std::remove_if(mQueue.begin(), mQueue.end()

7. , [&](const ReliablePacket& packetHolder) { return packetHolder.isIncludedIn(datagramId); })

8. , mQueue.cend());

9. if (mQueue.empty())

10. mFirstAllowedPacket = mNextId; //!< Si la file est maintenant vide, la borne commence au prochain paquet mis en file

11. else if (Utils::IsSequenceNewer(mQueue.front().packet().id(), mFirstAllowedPacket))

12. mFirstAllowedPacket = mQueue.front().packet().id(); // Sinon, on déplace les bornes d’envoi au plus ancien paquet en file

13. }

Ce type d’amélioration est bien entendu applicable au protocole ordonné non fiable du chapitre précédent également. Le cours ne présente pas cette partie, mais vous pouvez utiliser ce chapitre pour améliorer le code du chapitre précédent.

VIII - Tests

Vu que ce chapitre a fait la part belle à la réutilisation du chapitre précédent, les tests seront également très fortement inspirés de celui-ci.

ReliableOrdered_Test.hpp

1. #pragma once

3. #include "Tester.hpp"

4. #include "UDP/Protocoles/ReliableOrdered.hpp"

6. #include <limits>

7. #include <chrono>

8. #include <random>

10. class ReliableOrdered_Multiplexer_Test

11. {

12. public:

13. static void Test();

14. };

16. void ReliableOrdered_Multiplexer_Test::Test()

17. {

18. Bousk::Network::UDP::Protocoles::ReliableOrdered::Multiplexer mux;

19. CHECK(mux.mQueue.size() == 0);

20. CHECK(mux.mNextId == 0);

21. CHECK(mux.mFirstAllowedPacket == 0);

22. Bousk::Network::UDP::Datagram::ID datagramId = 0;

23. {

24. const std::array<uint8_t, 5> arr{ 'T', 'o', 't', 'o', '\0' };

25. std::vector<uint8_t> data(arr.cbegin(), arr.cend());

26. mux.queue(std::move(data));

27. CHECK(mux.mQueue.size() == 1);

28. CHECK(mux.mNextId == 1);

29. CHECK(mux.mFirstAllowedPacket == 0);

31. const Bousk::Network::UDP::Datagram::ID sentDatagramId = datagramId;

32. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

33. const size_t serializedData = mux.serialize(buffer.data(), buffer.size(), datagramId++);

34. CHECK(serializedData == Bousk::Network::UDP::Packet::HeaderSize + arr.size());

35. const Bousk::Network::UDP::Packet* packet = reinterpret_cast<const Bousk::Network::UDP::Packet*>(buffer.data());

36. CHECK(packet->datasize() == arr.size());

37. CHECK(memcmp(packet->data(), arr.data(), packet->datasize()) == 0);

38. //!< Le message reste en file d’envoi jusqu’à acquittement

39. CHECK(mux.mQueue.size() == 1);

40. //!< Acquittement de ce datagramme

41. mux.onDatagramAcked(sentDatagramId);

42. CHECK(mux.mQueue.size() == 0);

43. CHECK(mux.mFirstAllowedPacket == 1);

44. }

45. {

46. //!< Envoi d’un message en 3 fragments

47. std::vector<uint8_t> data(Bousk::Network::UDP::Packet::DataMaxSize * 3, 0);

48. const unsigned int seed = static_cast<unsigned int>(std::chrono::system_clock::now().time_since_epoch().count());

49. std::default_random_engine generator(seed);

50. std::uniform_int_distribution<unsigned int> distribution(0, 100);

51. for (uint8_t& d : data)

52. d = distribution(generator);

54. const auto datacopy = data;

55. mux.queue(std::move(data));

56. CHECK(mux.mQueue.size() == 3);

57. CHECK(mux.mNextId == 4);

58. size_t totalDataSize = 0;

59. const Bousk::Network::UDP::Datagram::ID firstSentDatagramId = datagramId;

60. const size_t nbPackets = mux.mQueue.size();

61. for (;;)

62. {

63. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

64. const size_t serializedData = mux.serialize(buffer.data(), buffer.size(), datagramId++);

65. if (serializedData == 0)

66. break;

67. const Bousk::Network::UDP::Packet* packet = reinterpret_cast<const Bousk::Network::UDP::Packet*>(buffer.data());

68. CHECK(memcmp(packet->data(), datacopy.data() + totalDataSize, packet->datasize()) == 0);

69. totalDataSize += serializedData - Bousk::Network::UDP::Packet::HeaderSize;

70. }

71. CHECK(totalDataSize == datacopy.size());

72. CHECK(mux.mQueue.size() == 3);

73. //!< Acquittement de chacun des datagrammes

74. size_t acked = 0;

75. for (auto i = 0; i < nbPackets; ++i)

76. {

77. const Bousk::Network::UDP::Datagram::ID dgramId = firstSentDatagramId + i;

78. mux.onDatagramAcked(dgramId);

79. ++acked;

80. CHECK(mux.mQueue.size() == 3 - acked);

81. }

82. CHECK(mux.mFirstAllowedPacket == 4);

83. }

84. {

85. //!< Envoi d’un message qui va être perdu

86. const std::array<uint8_t, 5> arr{ 'T', 'i', 't', 'i', '\0' };

87. std::vector<uint8_t> data(arr.cbegin(), arr.cend());

88. mux.queue(std::move(data));

89. CHECK(mux.mQueue.size() == 1);

90. const Bousk::Network::UDP::Datagram::ID lostDatagramId = datagramId;

92. {

93. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

94. const size_t serializedData = mux.serialize(buffer.data(), buffer.size(), datagramId++);

95. CHECK(serializedData == Bousk::Network::UDP::Packet::HeaderSize + arr.size());

96. const Bousk::Network::UDP::Packet* packet = reinterpret_cast<const Bousk::Network::UDP::Packet*>(buffer.data());

97. CHECK(packet->datasize() == arr.size());

98. CHECK(memcmp(packet->data(), arr.data(), packet->datasize()) == 0);

99. }

100. CHECK(mux.mQueue.size() == 1);

101. CHECK(mux.mFirstAllowedPacket == 4);

103. //!< Le paquet n’a pas encore été perdu, on ne devrait rien avoir à sérialiser

104. {

105. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

106. const size_t serializedData = mux.serialize(buffer.data(), buffer.size(), datagramId++);

107. CHECK(serializedData == 0);

108. }

109. CHECK(mux.mQueue.size() == 1);

110. mux.onDatagramLost(lostDatagramId);

111. CHECK(mux.mFirstAllowedPacket == 4);

113. const Bousk::Network::UDP::Datagram::ID datagramIdToAck = datagramId;

114. {

115. //!< Renvoi

116. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

117. const size_t serializedData = mux.serialize(buffer.data(), buffer.size(), datagramId++);

118. CHECK(serializedData == Bousk::Network::UDP::Packet::HeaderSize + arr.size());

119. const Bousk::Network::UDP::Packet* packet = reinterpret_cast<const Bousk::Network::UDP::Packet*>(buffer.data());

120. CHECK(packet->datasize() == arr.size());

121. CHECK(memcmp(packet->data(), arr.data(), packet->datasize()) == 0);

122. }

123. CHECK(mux.mQueue.size() == 1);

124. mux.onDatagramAcked(datagramIdToAck);

125. //!< Il est maintenant acquitté

126. CHECK(mux.mQueue.size() == 0);

127. CHECK(mux.mFirstAllowedPacket == 5);

128. }

129. {

130. //!< 3 messages, perte du second

131. std::vector<uint8_t> data(Bousk::Network::UDP::Packet::DataMaxSize * 3, 0);

132. const unsigned int seed = static_cast<unsigned int>(std::chrono::system_clock::now().time_since_epoch().count());

133. std::default_random_engine generator(seed);

134. std::uniform_int_distribution<unsigned int> distribution(0, 100);

135. for (uint8_t& d : data)

136. d = distribution(generator);

138. const auto datacopy = data;

139. mux.queue(std::move(data));

140. CHECK(mux.mQueue.size() == 3);

141. CHECK(mux.mFirstAllowedPacket == 5);

143. const auto datagram1 = datagramId;

144. std::array<uint8_t, Bousk::Network::UDP::Packet::PacketMaxSize> buffer;

145. mux.serialize(buffer.data(), buffer.size(), datagramId++);

146. const auto datagram2 = datagramId;

147. mux.serialize(buffer.data(), buffer.size(), datagramId++);

148. const auto datagram3 = datagramId;

149. mux.serialize(buffer.data(), buffer.size(), datagramId++);

150. CHECK(mux.mFirstAllowedPacket == 5);

151. mux.onDatagramAcked(datagram3);

152. CHECK(mux.mFirstAllowedPacket == 5);

153. mux.onDatagramAcked(datagram2);

154. CHECK(mux.mFirstAllowedPacket == 5);

155. mux.onDatagramAcked(datagram1);

156. CHECK(mux.mFirstAllowedPacket == 8);

157. }

158. }

160. class ReliableOrdered_Demultiplexer_Test

161. {

162. public:

163. static void Test();

164. };

166. void ReliableOrdered_Demultiplexer_Test::Test()

167. {

168. //!< Utiliser le multiplexeur pour facilement gérer la file d’envoi

169. //!< Il a été testé auparavant, donc il est fiable d’utilisation

170. Bousk::Network::UDP::Protocoles::ReliableOrdered::Multiplexer mux;

171. Bousk::Network::UDP::Protocoles::ReliableOrdered::Demultiplexer demux;

172. CHECK(demux.mLastProcessed == std::numeric_limits<Bousk::Network::UDP::Packet::Id>::max());

173. Bousk::Network::UDP::Datagram::ID datagramId = 0;

174. {

175. const std::array<uint8_t, 5> arr0{ 'T', 'o', 't', 'o', '\0' };

176. std::vector<uint8_t> data0(arr0.cbegin(), arr0.cend());

177. //!< Réception paquets 0 & 1

178. {

179. mux.queue(std::move(data0));

180. Bousk::Network::UDP::Packet packet;

181. mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

182. CHECK(packet.id() == 0);

183. CHECK(packet.type() == Bousk::Network::UDP::Packet::Type::Packet);

184. CHECK(packet.datasize() == static_cast<uint16_t>(arr0.size()));

185. CHECK(memcmp(packet.data(), arr0.data(), arr0.size()) == 0);

186. demux.onDataReceived(packet.buffer(), packet.size());

187. }

188. const std::array<uint8_t, 5> arr1{ 'T', 'a', 't', 'a', '\0' };

189. std::vector<uint8_t> data1(arr1.cbegin(), arr1.cend());

190. {

191. mux.queue(std::move(data1));

192. Bousk::Network::UDP::Packet packet;

193. mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

194. CHECK(packet.id() == 1);

195. CHECK(packet.type() == Bousk::Network::UDP::Packet::Type::Packet);

196. CHECK(packet.datasize() == static_cast<uint16_t>(arr1.size()));

197. CHECK(memcmp(packet.data(), arr1.data(), arr1.size()) == 0);

198. demux.onDataReceived(packet.buffer(), packet.size());

199. }

200. const std::vector<std::vector<uint8_t>> packets = demux.process();

201. CHECK(packets.size() == 2);

202. CHECK(demux.mLastProcessed == 1);

203. CHECK(packets[0].size() == data0.size());

204. CHECK(packets[0] == data0);

205. CHECK(packets[1].size() == data1.size());

206. CHECK(packets[1] == data1);

207. }

208. {

209. //!< Réception d’un message fragmenté : 3 fragments

210. std::vector<uint8_t> data(Bousk::Network::UDP::Packet::DataMaxSize * 3, 0);

211. const unsigned int seed = static_cast<unsigned int>(std::chrono::system_clock::now().time_since_epoch().count());

212. std::default_random_engine generator(seed);

213. std::uniform_int_distribution<unsigned int> distribution(0, 100);

214. for (uint8_t& d : data)

215. d = distribution(generator);

217. const auto datacopy = data;

218. mux.queue(std::move(data));

219. {

220. const auto sentDatagramdId = datagramId;

221. Bousk::Network::UDP::Packet packet;

222. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

223. demux.onDataReceived(packet.buffer(), packet.size());

224. CHECK(demux.process().empty());

225. mux.onDatagramAcked(sentDatagramdId);

226. }

227. {

228. const auto sentDatagramdId = datagramId;

229. Bousk::Network::UDP::Packet packet;

230. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

231. demux.onDataReceived(packet.buffer(), packet.size());

232. CHECK(demux.process().empty());

233. mux.onDatagramAcked(sentDatagramdId);

234. }

235. {

236. const auto sentDatagramdId = datagramId;

237. Bousk::Network::UDP::Packet packet;

238. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

239. demux.onDataReceived(packet.buffer(), packet.size());

240. mux.onDatagramAcked(sentDatagramdId);

241. }

242. {

243. const auto sentDatagramdId = datagramId;

244. Bousk::Network::UDP::Packet packet;

245. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

246. CHECK(serializedData == 0);

247. mux.onDatagramAcked(sentDatagramdId);

248. }

249. const std::vector<std::vector<uint8_t>> packets = demux.process();

250. CHECK(packets.size() == 1);

251. CHECK(demux.mLastProcessed == 4);

252. CHECK(packets[0].size() == datacopy.size());

253. CHECK(packets[0] == datacopy);

254. }

255. {

256. //!< Réception d’un message fragmenté : 3 fragments, perte du second fragment

257. std::vector<uint8_t> data(Bousk::Network::UDP::Packet::DataMaxSize * 3, 0);

258. const unsigned int seed = static_cast<unsigned int>(std::chrono::system_clock::now().time_since_epoch().count());

259. std::default_random_engine generator(seed);

260. std::uniform_int_distribution<unsigned int> distribution(0, 100);

261. for (uint8_t& d : data)

262. d = distribution(generator);

264. const auto datacopy = data;

265. mux.queue(std::move(data));

266. {

267. const auto sentDatagramdId = datagramId;

268. Bousk::Network::UDP::Packet packet;

269. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

270. demux.onDataReceived(packet.buffer(), packet.size());

271. CHECK(demux.process().empty());

272. mux.onDatagramAcked(sentDatagramdId);

273. }

274. const auto lostDatagramdId = datagramId;

275. {

276. Bousk::Network::UDP::Packet packet;

277. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

278. }

279. {

280. Bousk::Network::UDP::Packet packet;

281. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

282. demux.onDataReceived(packet.buffer(), packet.size());

283. }

284. {

285. Bousk::Network::UDP::Packet packet;

286. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

287. CHECK(serializedData == 0);

288. }

289. {

290. mux.onDatagramLost(lostDatagramdId);

291. Bousk::Network::UDP::Packet packet;

292. const size_t serializedData = mux.serialize(reinterpret_cast<uint8_t*>(&packet), Bousk::Network::UDP::Packet::PacketMaxSize, datagramId++);

293. demux.onDataReceived(packet.buffer(), packet.size());

294. }

295. const std::vector<std::vector<uint8_t>> packets = demux.process();

296. CHECK(packets.size() == 1);

297. CHECK(demux.mLastProcessed == 7);

298. CHECK(packets[0].size() == datacopy.size());

299. CHECK(packets[0] == datacopy);

300. }

301. }

303. class ReliableOrdered_Test

304. {

305. public:

306. static void Test()

307. {

308. ReliableOrdered_Multiplexer_Test::Test();

309. ReliableOrdered_Demultiplexer_Test::Test();

310. }

311. };

Récupérez le code source sous le tag V1.0.6 dans le dépôt Mercurial à l’adresse suivante.

Article précédent

Article suivant

<< UDP – Découpage et unification de paquets

UDP – Combiner tous les protocoles : les canaux de communication >>