Copie de fichier par voie d'une zone de transit avec sockets de type flux (sockets TCP) – exemple de solution

Ce qui suit se veut un petit exemple de solution à un problème proposé informellement en classe, soit celui d'écrire un programme (inefficace, mais c'est pour se faire la main) qui consommera des données d'un fichier source à l'aide d'un fil d'exécution, et rendra progressivement ces données disponibles à d'autres à travers un canal sécurisé (une zone de transit) pour qu'un autre fil d'exécution les consomme et les envoie sur un socket de type flux. Par la suite, un fil d'exécution consommera les données du socket pour les rendre disponibles à travers une zone de transit, pendant qu'un autre fil d'exécution les en consommera pour les écrire dans un autre fichier.

Conceptuellement, le code proposé se sépare en deux processus :

J'ai mis le tout dans un même processus pour simplifier la présentation, mais je vous invite bien sûr à finir le travail; cela impliquera un peu de copier/ coller, pas beaucoup plus de travail que ça.

J'ai proposé que vous acomplissiez cette tâche avec un fichier binaire quelconque, et le code ci-dessous fonctionnerait (outre l'affichage en toute fin) tel quel avec un .jpeg ou un .mp3, mais j'ai écrit l'exemple de manière à vous le rendre disponible rapidement alors ma version lit... le code source du programme (fichier Principal.cpp) pour l'écrire dans un autre fichier (sortie.txt), pour enfin en afficher le contenu à la console. Ceci permet de s'assurer qu'il ne manquera aucun caractère en fin de parcours (un réel piège ici).

Puisque les sockets ne sont pas représentés par des types standards de C++ mais bien par des bibliothèques locales aux plateformes, j'ai isolé le code des sockets derrière une interface simple inspirée de l'idiome pImpl. Conséquemment, le code ci-dessous se subdivise en trois fichiers :

Si vous séparez le tout en deux processus, seul Principal.cpp sera affecté.

Fichier socket_tcp.h

J'ai implémenté une couche de type pImpl sur les sockets de type flux pour que leur interface (le fichier socket_tcp.h) soit pleinement portable, et pour éviter de trop faire fuir les détails de la plateforme sous-jacente dans le code client. Notez que le code proposé ici est du code écrit rapidement; j'ai des architectures plus riches (mais plus complexes) pour qui souhaite les explorer, comme par exemple ceci.

Le code des sockets de type flux proposé ici est logé dans l'espace nommé socket_tcp.

 
#ifndef SOCKET_TCP_H
#define SOCKET_TCP_H
#include <memory>
#include <string>
#include <cstdint>
#include <exception>
#include <iterator>
namespace socket_tcp {

J'ai défini une classe incopiable nommée chargeur dont le constructeur chargera les sockets en mémoire et dont le destructeur assurera la finalisation correspondante. En pratique, le programme principal définira localement une instance de chargeur avant de commencer à manipuler des sockets.

Le chargement explicite des sockets est un artéfact de Microsoft Windows; sous Linux, le constructeur et le destruceur de cette classe seraient vides.

 
   struct chargeur {
      chargeur();
      ~chargeur();
      chargeur(const chargeur&) = delete;
      chargeur& operator=(const chargeur&) = delete;
   };

Les implémentations à proprement dit d'un socket de conversation et d'un socket serveur sont abstraites sous des déclarations a priori. Notez que j'ai utilisé deux types distincts ici, même si l'implémentation de chacun reposera en pratique sur un socket de la plateforme, pour éviter une situation absurde comme celle où le code client utiliserait un socket non-serveur pour faire un accept().

Comme on le verra dans socket_tcp.cpp, nous représenterons les liens entre ces types à même l'implémentation.

 
   struct impl;
   struct serveur_impl;

Les classes pImpl que j'ai utilisé jouent deux rôles :

  • Isoler l'implémentation derrière une classe a priori indéfinie (impl pour socket_impl, et serveur_impl pour socket_serveur_impl; ce ne sont pas de très bons noms, mais je n'y ai pas beaucoup réfléchi alors j'implore votre tolérance), et
  • Assurer une gestion RAII de l'implémentation, à travers des unique_ptr.

Notez que j'ai implémenté la sémantique de mouvement, ce qui me permettra de manipuler mes objets un peu comme on manipulerait des classes concrètes ou des types primitifs, mais que mes classes sont incopiables en vertu de leur attribut qui est lui-même incopiable.

Notez aussi que les attributs de ces deux classes sont publics, mais que cela ne m'importe pas beaucoup ici puisque les types des pointés sont inaccessibles au code client, qui ne pourra à peu près rien faire avec eux.

 
   struct socket_impl {
      std::unique_ptr<impl> p;
      socket_impl() = delete;
      socket_impl(std::unique_ptr<impl>&&);
      socket_impl(socket_impl&&);
      ~socket_impl();
   };
   struct socket_serveur_impl {
      std::unique_ptr<serveur_impl> p;
      socket_serveur_impl() = delete;
      socket_serveur_impl(std::unique_ptr<impl>&&);
      socket_serveur_impl(std::unique_ptr<serveur_impl>&&);
      socket_serveur_impl(socket_serveur_impl&&);
      ~socket_serveur_impl();
   };

Pour faciliter la gestion des erreurs, j'ai défini une classe par type d'exception possible. Ne confondez pas ceci avec « une exception par code d'erreur possible », ce qui serait faux ici (il y a des tas de codes d'erreurs possibles pour chaque opération sur un socket). Je me suis intéressé à deux choses, pratiquement :

  • L'opération qui aura échoué, et
  • Le code d'erreur rapporté par l'implémentation sous-jacente

Normalement, je n'aurais pas utilisé std::exception ou des messages d'erreur pour chaque type, mais dans le code de test que j'ai écrit, j'ai limité le traitement des exceptions à un affichage suivi de la conclusion du fil d'exécution ayant détecté le problème. Cette stratégie plutôt naïve ne se veut pas de qualité industrielle, et se limite à rapporter les cas problèmes plutôt que de les gérer.

 
// ...
   struct erreur : std::exception {
      int code;
      erreur(const char*);
   };
   struct erreur_creation : erreur { erreur_creation() : erreur{"socket()"} {} };
   struct erreur_connexion : erreur { erreur_connexion() : erreur{"connect()"} {} };
   struct erreur_lier_port : erreur { erreur_lier_port() : erreur{"bind()"} {} };
   struct erreur_file_attente : erreur { erreur_file_attente() : erreur{"liste()"} {}};
   struct erreur_acceptation : erreur { erreur_acceptation() : erreur{"accept()"} {}};
   struct erreur_envoi : erreur { erreur_envoi() : erreur{"send()"} {}};
   struct erreur_reception : erreur { erreur_reception() : erreur{"recv()"} {}};

Enfin, j'ai exprimé les opérations exposées au code client dans des fonctions opérant sur mes classes pImpl, et faisant essentiellement abstraction des détails d'implémentation. À noter tout de même :

  • Le port est représenté par un entier non-signé sur 16 bits, dans le respect des règles sur Internet
  • Les fonctions de bas niveau pour envoyer et recevoir des données opèrent sur des séquences de bytes bruts (une interface simple permettant d'alléger l'écriture du code client a été offerte pour recevoir())
  • Les types acceptés par les fonctions sont parfois celui d'une implémentation de socket de conversation (incluant les clients), parfois ceui d'une implémentation de socket serveur, parfois les deux. J'ai fait ceci pour réduire les risques d'erreurs à l'exécution

C'est une interface naïve mais opaque et utilisable.

 
// ...
   socket_impl creer_client(const std::string &adresse, std::uint16_t port);
   socket_serveur_impl creer_serveur(std::uint16_t port, std::uint16_t taille_file = 16);
   socket_impl accepter(socket_serveur_impl&);
   int envoyer(socket_impl&, const char*, int);
   int recevoir(socket_impl &sck, char *buf, int cap);
   template <int N>
      int recevoir(socket_impl &sck, char(&buf)[N]) {
         return recevoir(sck, std::begin(buf), N);
      }
   void fermer(socket_impl);
   void fermer(socket_serveur_impl);
}
#endif

Fichier socket_tcp.cpp

Sans surprises, l'implémentation des types et des fonctions déclarées dans socket_tcp.h repose en partie sur du code propre à la plateforme, donc éminemment non-portable. Le code proposé ici vaut pour Microsoft Windows, et peut bien entendu être adapté au besoin pour les autres plateformes. Considérez cela comme un exercice.

 
#include "socket_tcp.h"
#include <WinSock2.h>
#include <WS2tcpip.h>
#include <memory>
#include <string>
#include <cstdint>
using namespace std;
#pragma comment(lib, "ws2_32.lib")
namespace socket_tcp {

J'ai implémenté une classe parent nommée erreur pour mes cas d'exceptions, ceci dans l'optique d'alléger mon propre travail à peu de frais. Chaque erreur dérive de std::exception et lui relaie un bref message expliquant la nature de l'opération ayant échoué. Un code d'erreur logé à même erreur, implémenté sous forme de constante d'instance (initialisée à la construction) est exposé de manière publique et permet de comprendre plus en détail la nature du problème s'étant manifesté.

 
   erreur::erreur(const char *msg)
      : exception{ msg }, code{ WSAGetLastError() } {
   }

La classe chargeur réalise un chargement et un déchargement RAII du module de sockets de Microsoft Windows. Sous Linux, on laissera ces deux fonctions vides, tout simplement.

 
// ...
   chargeur::chargeur() {
      WSADATA data { };
      auto res = WSAStartup(MAKEWORD(2,2), &data);
   }
   chargeur::~chargeur() {
      WSACleanup();
   }

La classe impl isole le détail d'implémentation qu'est un socket pour la plateforme. J'y ai essentiellement implémenté les éléments constitutifs de la règle de cinq que j'estimais pertinents, la rendant entre autres incopiable, de même qu'une méthode pour transférer explicitrement un socket vers un tiers.

 
// ...
   struct impl {
      SOCKET sck;
      impl(SOCKET sck) : sck{ sck } {
      }
      impl(impl && other) : sck{ other.sck } {
         other.sck = INVALID_SOCKET;
      }
      impl(const impl&) = delete;
      impl& operator=(const impl&) = delete;
      ~impl() {
         if (sck != INVALID_SOCKET) {
            shutdown(sck, SD_BOTH);
            closesocket(sck);
         }
      }
      SOCKET transfer() {
         return std::exchange(sck, INVALID_SOCKET);
      }
   };

La classe pImpl servant à encapsuler un impl se limite à implémenter le mouvement, ce qui lui suffit.

 
// ...
   socket_impl::socket_impl(unique_ptr<impl> &&p) : p{ std::move(p) } {
   }
   socket_impl::socket_impl(socket_impl &&src) = default;
   socket_impl::~socket_impl() = default;

La classe serveur_impl définit un enrobage typé autour d'un impl. J'utilise cette classe pour faire en sorte que le code client doive passer par des types distincts selon la nature des opérations à réaliser sur un socket.

Notez l'absence de relation hiérarchique entre serveur_impl et impl, car elle est délibérée : je ne veux pas que l'une puisse être utilisée comme une abstraction de l'autre. Je veux par contre qu'un impl puisse être transféré (par mouvement) à un serveur_impl, en particulier dans le cas où un socket nouvellement créé (et représenté par un impl) devrait être utiliser comme un serveur (une sorte de « promotion », pour illustrer la démarche).

 
// ...
   struct serveur_impl {
      impl p;
      serveur_impl(impl &&p) : p{ std::move(p) } {
      }
   };

La classe pImpl servant à encapsuler un serveur_impl se limite pour l'essentiel à implémenter le mouvement, ce qui lui suffit. Elle déroge dans un cas, soit celui où elle absorberait un unique_ptr<impl>, réalisant une sorte de « mouvement manuel ».

 
// ...
   socket_serveur_impl::socket_serveur_impl(unique_ptr<serveur_impl> &&p)
      : p{ std::move(p) } {
   }
   socket_serveur_impl::socket_serveur_impl(socket_serveur_impl&&)= default;
   socket_serveur_impl::socket_serveur_impl(unique_ptr<impl> &&src)
      : p{ make_unique<serveur_impl>(src->sck) } {
      src.release();
   }
   socket_serveur_impl::~socket_serveur_impl() = default;

La fonction creer_socket() est pour usage interne seulement, n'étant pas exposée dans socket_tcp.h.

 
// ...
   socket_impl creer_socket() {
      auto sck = make_unique<impl>(socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
      if (sck->sck == INVALID_SOCKET) throw erreur_creation{};
      return { std::move(sck) };
   }

La fonction creer_client() retourne un socket client connecté à un serveur. Elle encapsule les détails techniques reliés à une connexion bloquante par socket de type flux vers un serveur situé à une adresse et à un port spéficiés en paramètre. Notez que cette implémentation utilise l'adressage IPv4.

 
// ...
   socket_impl creer_client(const string &adresse, uint16_t port) {
      auto sck = creer_socket();
      sockaddr_in serveur { };
      inet_pton(AF_INET, adresse.c_str(), &serveur.sin_addr.S_un.S_addr);
      serveur.sin_family = AF_INET;
      serveur.sin_port = htons(port);
      auto res = connect(sck.p->sck, reinterpret_cast<const sockaddr*>(&serveur), sizeof(serveur));
      if (res == SOCKET_ERROR) throw erreur_connexion{};
      return sck;
   }

De manière analogue, la fonction creer_serveur() retourne un socket serveur lié à un port spéficié en paramètre. Il reste encore un peu de nettoyage à faire ici (le return est trop complexe) mais le temps me manque.

 
// ...
   socket_serveur_impl creer_serveur(uint16_t port, uint16_t taille_file) {
      auto sck = creer_socket();
      sockaddr_in serveur { };
      serveur.sin_family = AF_INET;
      serveur.sin_port = htons(port);
      auto res = bind(sck.p->sck, reinterpret_cast<const sockaddr*>(&serveur), sizeof(serveur));
      if (res == SOCKET_ERROR) throw erreur_lier_port{};
      res = listen(sck.p->sck, taille_file);
      if (res == SOCKET_ERROR) throw erreur_file_attente{};
      return socket_serveur_impl{ make_unique<serveur_impl>(sck.p->transfer()) };
   }

La fonction accepter() accepte une demande de connexion et retourne le socket de conversation correspondant.

 
// ...
   socket_impl accepter(socket_serveur_impl &serveur) {
      sockaddr_in client { };
      int len = sizeof(client);
      auto s = serveur.p->p.sck;
      auto sck = accept(s, reinterpret_cast<sockaddr*>(&client), &len);
      if (sck == INVALID_SOCKET) throw erreur_acceptation{};
      return socket_impl{ make_unique<impl>(sck) };
   }

La fonction envoyer() émet une séquence de bytes sur un socket de conversation et retourne le nombre de bytes réellement envoyés.

 
// ...
   int envoyer(socket_impl &sck, const char *data, int len) {
      auto n = send(sck.p->sck, data, len, 0);
      if (n == SOCKET_ERROR) throw erreur_envoi{};
      return n;
   }
}

La fonction recevoir() reçoit une séquence de bytes d'un socket de conversation, jusqu'à concurrence de la capacité d'un tampon en entrée, et retourne de nombre de bytes réellement reçus.

 
// ...
   int recevoir(socket_impl &sck, char *buf, int cap) {
      auto n = recv(sck.p->sck, buf, cap, 0);
      if (n == SOCKET_ERROR) throw erreur_reception{};
      return n;
   }

Enfin, de manière amusante, les fonctions fermer(), sans l'annoncer ouvertement, ne font rien : ce sont ce qu'on nomme des fonctions Sink, donc des fonctions à qui l'on confie un paramètre qui ne ressortira plus.

Ici, les types acceptés ne peuvent être passés que par mouvement, et sont tous deux des types RAII, ce qui fait qu'on n'a même pas à nommer les paramètres : le simple fait de les recevoir localement mènera à leur éventuelle destruction, et à leur finalisation en propre.

 
// ...
   void fermer(socket_impl) { // sink
   }
   void fermer(socket_serveur_impl) { // sink
   }
}

Fichier Principal.cpp

Remarquez tout d'abord que le code proposé ici est totalement portable à toute plateforme, dans la mesure où un compilateur C++ 11 y est disponible (et dans la mesure où socket_tcp.h est portable, bien sûr).

 
#include "socket_tcp.h"
#include <fstream>
#include <mutex>
#include <thread>
#include <vector>
#include <algorithm>
#include <atomic>
#include <iterator>
#include <iostream>
#include <cstdint>
using namespace std;

J'ai implémenté une zone de transit générique relativement simple. Remarquez l'implémentation de la méthode extraire(), qui est plus efficace que celle (plus naïve) proposée en classe. Comprenez-vous la manoeuvre?

J'en profite pour vous rappeler l'importance de conserver un mutex juste assez longtemps, donc pas moins... mais pas plus, car moins votre code sera ralenti par la synchronisation et meilleur en sera le débit de traitement.

 
template <class T>
   class zone_transit {
      vector<T> data;
      mutable mutex m;
   public:
      template <class It>
         void ajouter(It debut, It fin) {
            lock_guard<mutex> _{ m };
            data.insert(end(data), debut, fin);
         }
      vector<T> extraire() {
         lock_guard<mutex> _{ m };
         vector<T> temp;
         temp.swap(data);
         return temp;
      }
   };

Le programme principal se séparera en quatre fils d'exécution :

En plus des zones de transit, les fils d'exécution se partageront des signaux de fin. Pourquoi? Parce qu'un fil d'exécution consommateur ne peut se baser sur la présence ou l'absence de données dans la zone de transit à travers laquelle son fournisseur l'alimente pour déterminer si son travail est terminé. En effet, si le producteur (le fil d'exécution qui lit du disque et remplit la zone de transit) est moins rapide, au moins de temps à autres, que le consommateur (le fil d'exécution qui lit de la zone de transit pour écrire sur disque), alors il arrivera que le consommateur prenne de l'avance sur le producteur et qu'aucune donnée ne soit disponible pour lui dans la zone de transit... et ce ne sera qu'une situation normale, sans plus.

Comme il se doit, nous ne donnons à chaque fil d'exécution que ce dont il a réellement besoin pour faire son travail, sans plus.

 
//
// Pour un plus joli programme, séparer en deux programmes :)
//
int main() {
   using namespace socket_tcp;
   chargeur ch;
   static constexpr const uint16_t ZE_PORT = 4567;
   auto srv = creer_serveur(ZE_PORT);
   zone_transit<char> zt0, zt1;
   atomic<bool> fini_lecture { false },
                fini_reception { false };

Le fil d'exécution lecture, producteur au sens de la zone de transit zt0, consommera des données du disque et les insérera en bloc dans zt0 (pour éviter une synchronisation trop fréquente). La taille du tampon local (constante N) est à déterminer de manière empirique; pour les besoins de l'exemple, j'ai mis une valeur prise à peu près au hasard.

Le signal de fin, donc l'instruction fini_lecture = true;, sera donné par le producteur quand il saura que plus aucune donnée ne sera insérée dans zt0, et pas avant.

Notez l'ajout dans la zone de transit après la répétitive de lecture, qui permet d'y insérer les données résiduelles du dernier tampon si celui-ci n'est pas rempli. Sans elle, il manquera des données en fin de parcours. 

   thread lecture{[&zt0, &fini_lecture]() {
      ifstream in{ "Principal.cpp", ios::binary };
      enum { N = 1'000 };
      char buf[N];
      int n = 0;
      for (char c; in.get(c);) {
         buf[n++] = c;
         if (n == N) {
            zt0.ajouter(begin(buf), end(buf));
            n = 0;
         }
      }
      zt0.ajouter(begin(buf), begin(buf) + n);
      fini_lecture = true;
   }};

Le fil d'exécution envoi, qui consomme de zt0 pour insérer dans un socket de type flux, crée tout d'abord un socket client vers le port que j'ai choisi (arbitrairement) pour les besoins de la cause. Par la suite, il consomme tout ce qu'il peut de zt0 pour l'envoyer de son mieux sur le socket (ce tronçon pourrait être plus efficace).

Une dernière consommation de zt0 suit le constat de fin (changement à fini_lecture) pour s'assurer de ne pas oublier les dernières insertions qui auront été faites à cette zone de transit.

Aucun signal de fin n'est donné ici; l'implémentation de sockets que j'ai utilisée est RAII et ferme correctement le socket sous-jacent sur destruction; le récepteur constatera la fin de l'émission de données grâce à une éventuelle erreur de réception.

 
      thread envoi{[&zt0,&fini_lecture]() {
         try {
            auto sck = creer_client("127.0.0.1", ZE_PORT);
            while (!fini_lecture) {
               for(auto v = zt0.extraire(); !v.empty(); ) { // bof
                  auto n = envoyer(sck, &v[0], v.size());
                  v.erase(begin(v), begin(v) + n);
               }
            }
            for(auto v = zt0.extraire(); !v.empty(); ) { // bof
               auto n = envoyer(sck, &v[0], v.size());
               v.erase(begin(v), begin(v) + n);
            }
         } catch (erreur &e) {
            cerr << "Erreur dans envoi, operation " << e.what() << ", code " << e.code << endl;
         }
     } };

Le fil d'exécution reception, de son côté, accepte la connexion du client créé par envoi, puis consomme du socket de conversation résultant toutes les données qui y transiteront. La sortie dans le code à droite se fait lors d'une levée d'exception ou lors d'un appel à recevoir() retournant zéro, ce qui représenterait une fermeture propre par l'homologue émetteur. Les données ainsi consommées sont insérées dans la zone de transit zt1.

Une variable nommée fin_reception permet de signaler à ecriture, fil d'exécution consommateur de zt1 que reception ne lui fournira plus de données.

 
      thread reception{ [&srv,&zt1,&fini_reception]() {
         try {
            auto sck = accepter(srv);
            enum { N = 1'024 }; // bof
            for (;;) {
               char buf[N] { };
               auto n = recevoir(sck, buf);
               if (!n) break;
               zt1.ajouter(buf + 0, buf + n);
            }
         } catch (erreur &e) {
            cerr << "Erreur dans reception, operation " << e.what() << ", code " << e.code << endl;
         }
         fini_reception = true;
      } };

Enfin, le fil d'exécution ecriture, consommateur au sens de la zone de transit zt1, extraira le contenu de cette dernière et l'écrira sur disque tant et aussi longtemps que le signal de fin représenté par fini_reception n'aura pas été donné.

Notez la dernière extraction après la répétitive, qui permet de vider la zone de transit de toute données ayant été insérée entre le moment de la plus récente extraction et le constat par le consommateur que le signal de fin a été donné. Sans elle, règle générale, il manquera des données dans le fichier en fin de parcours.

 
// ...
   thread ecriture{[&zt1, &fini_reception]() {
      ofstream out{ "sortie.txt", ios::binary };
      while (!fini_reception) {
         auto v = zt1.extraire();
         copy(begin(v), end(v), ostreambuf_iterator<char>{ out });
      }
      auto v = zt1.extraire();
      copy(begin(v), end(v), ostreambuf_iterator<char>{ out });
   }};

Avant de fermer le programme, nous attendons la complétion de l'exécution des quatre fils d'exécution, puis (pour fins de débogage) nous affichons le contenu du fichier de destination à la console.

Si vous avez copié un fichier contenant des données autres que du texte, présumant que l'extension du fichier de destination corresponde à son contenu, il suffit (du moins avec Microsoft Windows) de remplacer l'appel à copy() par un appel à system() en lui passant le nom du fichier en question, ce qui démarrera le programme associé à ce type de fichier.

 
// ...
   ecriture.join();
   envoi.join();
   reception.join();
   lecture.join();
   ifstream in{ "sortie.txt" };
   copy(istreambuf_iterator<char>{ifstream{ "sortie.txt" }}, istreambuf_iterator<char>{},
        ostream_iterator<char>{cout});
}

Voilà.

Valid XHTML 1.0 Transitional

CSS Valide !