[gelöst] [boost::asio] TCP-Verbindung verlustbehaftet wenn zu schnell gesendet wird



  • Hallo!

    Ich habe mit boost::asio einen Server und einen Client geschrieben. Dabei habe ich festgestellt, dass es im TCP-Datenstrom vom Client zum Server zu Verlusten kommt, sobald der Client zu schnell sendet. Sowas darf ja bei TCP eigentlich nicht passieren. Der Client sendet mittels boost::asio::async_write() und der Server empfängt die Daten mittels boost::asio::async_read(). Da die zulässige Buffergröße pro async_write() ja begrenzt ist, teile ich die Daten in Häppchen à 16384 Bytes auf.

    Zu Debugging-Zwecken speichere ich alles, was der Client sendet, unmittelbar vor dem Aufruf von async_write() in einer Datei. Auf die gleiche Weise zeichne ich alles, was der Server empfängt, auf. Von den ca. 60 MB, die ich verschicke, kommen stets nur ca. 2,5 MB beim Server an. Gesendet wird an 127.0.0.1, also localhost. Physikalisch bedingte Verluste sind also nicht zu erwarten.

    Jetzt kommt's: Lasse ich den Client nach jeder Sendeoperation eine Millisekunde schlafen, läuft die Übertragung verlustfrei. (1 ms nach jeder fünften oder auch 2 ms nach jeder zehnten Operation gehen ebenfalls, aber nur 1 ms nach jeder zehnten sind zu wenig). Siehe Zeile 127 in gameserver.cpp.

    Verwundern tut mich weiterhin, dass ich von Boost weder eine Erfolgs- noch eine Fehlermeldung über die async_write()-Vorgänge erhalte (per Haltepunkt und std::cout-Ausgabe überprüft). Dies ist unabhängig davon, ob die Übertragung verlustfrei läuft oder nicht. Siehe Zeile 55 in gameserver.cpp.

    Den Code für Client und Server habe ich angehängt. Dafür habe ich ihn auf ein Mindestmaß geschrumpft, um euch unnötige Lesearbeit zu ersparen. Nichtsdestotrotz sind es noch gut 400 Zeilen.

    Wäre dankbar für jede Hilfe!

    server_main.cpp:

    #include "../gameserver.h"
    
    int main(int, char**)
    {
      boost::asio::io_service ioservice;
    
      GameServer server(ioservice, 5555, false);
      server.startAccepting();
    
      ioservice.run();
      return 0;
    }
    

    client_main.cpp:

    #include <iostream>
    #include <memory>
    
    #include "../gameserver.h"
    
    using namespace std;
    
    void iothread_proc(boost::asio::io_service* ioservice)
    {
      ioservice->run();
      cout << "iothread is now terminating.\n";
    }
    
    void printUsage()
    {
      cout << "usage:\n";
      cout << "client filename\n";
    }
    
    int main(int argc, char** argv)
    {
      if(argc != 2) {
        printUsage();
        return 255;
      }
    
      // construct
      thread iothread;
      auto ioservice = make_shared<boost::asio::io_service>();
      auto connection = make_shared<GameServerConnection>(*ioservice, nullptr);
    
      // connect
      const string address = "127.0.0.1";
      const uint16_t port = 5555;
      if(connection->connect(address, port) == false) {
        cerr << "Could not connect.\n";
        return 2;
      }
      else
        iothread = thread(iothread_proc, ioservice.get());
    
      // send
      const string filename = argv[1];
      ifstream file(filename, ifstream::binary);
      if(!file.is_open()) {
        cerr << "could not open " << filename << endl;
        return 2;
      }
      connection->sendFile(file);
    
      // destruct
      connection.reset();
      ioservice->stop();
      if(iothread.joinable())
        iothread.join();
      return 0;
    }
    

    gameserver.h:

    #ifndef _GAMESERVER_H
    #define _GAMESERVER_H
    #include <fstream>
    #include <iostream>
    #include <mutex>
    #include <vector>
    
    #include <boost/asio.hpp>
    #include <boost/bind.hpp>
    #include <boost/endian/conversion.hpp>
    
    namespace GameServerChunkTypes {
      constexpr uint8_t clientFile  = 4;
    }
    
    using namespace boost::asio::ip;
    
    // prototypes
    class GameServerConnection;
    class GameServer;
    
    typedef std::vector<uint8_t> PkgT;
    
    class GameServerConnection
    {
      public:
        // for debugging: capture every byte that was sent and received
        std::ofstream m_debugTcpRecvFile;
        std::ofstream m_debugTcpSendFile;
    
        static constexpr uint32_t s_maxTcpPacketSize = 16384;
    
        // members
        GameServer* m_server;  // This must be nullptr on client side
        tcp::resolver m_tcpResolver;
        tcp::socket m_tcpSocket;
        PkgT m_currentlyReceivedTcpPkg;
    
        // methods
        GameServerConnection(boost::asio::io_service& ios, GameServer* parent);
        virtual ~GameServerConnection();
        bool handleError(const boost::system::error_code& error);  // return if everything is ok
        bool connect(const std::string& address, const uint16_t port);
    
        void sendTcp(std::shared_ptr<PkgT> pkg);
        void handleSentTcp(std::shared_ptr<PkgT> pkg, const boost::system::error_code& error);
        void startReceiveTcpHeader();
        void handleReceivedTcpHeader(std::shared_ptr<PkgT> buf,
                                     const boost::system::error_code& error, const size_t numBytesRead);
        void handleReceivedTcpContent(const boost::system::error_code& error, const size_t numBytesRead);
    
        void sendFile(std::ifstream& file);
        static void pushBlob(const std::vector<uint8_t>& blob, std::vector<PkgT>& pkgs);  // a blob can split over multiple pkgs
    };
    
    /* The GameServer starts with waiting for tcp-connections. Once an incoming tcp-connection has been
    accepted, a GameServerConnection-object will be added to m_connections. All tcp traffic is handled
    via m_connections[i].m_tcpSocket. */
    class GameServer
    {
      protected:
        typedef std::shared_ptr<GameServerConnection> ConnPtr;
    
        tcp::acceptor m_tcpAcceptor;
        std::vector<ConnPtr> m_connections;
        mutable std::recursive_mutex m_connectionsMutex;
    
        PkgT m_blob;
        uint32_t m_blobCur;
        void readSomeBlob(const PkgT& pkg, const size_t pos=0);
      public:
        friend class GameServerConnection;
    
        GameServer(boost::asio::io_service& ios, const uint16_t tcp_port, const bool ipv6=false);
        virtual void kickClient(GameServerConnection* client, const std::string& reason);
        virtual ConnPtr newConnPtr();
        void startAccepting();
        void handleAccepted(ConnPtr conn, const boost::system::error_code& error);
        virtual void handleReceivedTcpPkgServer(GameServerConnection* conn);
    
        void saveBlobIfNecessary();
    };
    
    #endif  // _GAMESERVER_H
    

    gameserver.cpp:

    #include "gameserver.h"
    #include <chrono>
    #include <thread>
    
    using namespace boost::asio::ip;
    
    // GameServerConnection
    GameServerConnection::GameServerConnection(boost::asio::io_service& ios, GameServer* parent)
      : m_server(parent), m_tcpResolver(ios), m_tcpSocket(ios)
    {}
    GameServerConnection::~GameServerConnection()
    {
      try {
        m_tcpSocket.close();
      } catch(...) {}  // don't care about exceptions, socket will be closed anyway
    }
    bool GameServerConnection::handleError(const boost::system::error_code& error)
    {
      if(error.value() == 0)
        return true;
    
      const std::string msg = error.message();
      std::cerr << "There was an error: " << msg << std::endl;
      if(m_server != nullptr)
        m_server->kickClient(this, msg);
      return false;
    }
    bool GameServerConnection::connect(const std::string& address, const uint16_t port)
    {
      assert(m_server == nullptr);
      tcp::resolver::query query(address, std::to_string(port));
      tcp::resolver::iterator endpoint_iterator = m_tcpResolver.resolve(query);
      boost::system::error_code ec;
      m_tcpSocket.connect(*endpoint_iterator, ec);
      if(ec.value() != 0)
        return false;
    
      m_debugTcpSendFile.open("whatClientSent.stream", std::ofstream::binary);
      if(!m_debugTcpSendFile.is_open())
        std::cerr << "Could not open send file\n";
      return true;
    }
    
    void GameServerConnection::sendTcp(std::shared_ptr<PkgT> pkg)
    {
      assert(m_server == nullptr);  // server does not send anything in this example
      m_debugTcpSendFile.write((const char*) pkg->data(), pkg->size());
    
      boost::asio::async_write(m_tcpSocket, boost::asio::buffer(*pkg),
                               boost::bind(&GameServerConnection::handleSentTcp,
                                           this, pkg, boost::asio::placeholders::error()));
    }
    void GameServerConnection::handleSentTcp(std::shared_ptr<PkgT> pkg, const boost::system::error_code& error)
    {
      std::cout << "handleSentTcp() called\n";  // Das wird irgendwie nie aufgerufen.
      pkg.reset();
      if(!handleError(error))
        return;
    }
    void GameServerConnection::startReceiveTcpHeader()
    {
      std::shared_ptr<PkgT> buf(new PkgT{0,0});
      boost::asio::async_read(m_tcpSocket, boost::asio::buffer(*buf),
                              boost::bind(&GameServerConnection::handleReceivedTcpHeader, this, buf,
                                          boost::asio::placeholders::error,
                                          boost::asio::placeholders::bytes_transferred));
    }
    void GameServerConnection::handleReceivedTcpHeader(std::shared_ptr<PkgT> buf,
                                                       const boost::system::error_code& error,
                                                       const size_t numBytesRead)
    {
      if(!handleError(error))
        return;
      assert(buf->size() == 2);
      assert(numBytesRead == 2);
    
      m_debugTcpRecvFile.write((const char*) buf->data(), buf->size());
      m_debugTcpRecvFile.flush();
    
      const uint16_t* size16 = reinterpret_cast<const uint16_t*>(buf->data());
      const uint16_t pkgSize = boost::endian::big_to_native(*size16);
      assert(pkgSize < 32768);
      m_currentlyReceivedTcpPkg.resize(pkgSize);
    
      // receive content
      boost::asio::async_read(m_tcpSocket, boost::asio::buffer(m_currentlyReceivedTcpPkg),
                              boost::bind(&GameServerConnection::handleReceivedTcpContent, this,
                                          boost::asio::placeholders::error,
                                          boost::asio::placeholders::bytes_transferred));
    }
    void GameServerConnection::handleReceivedTcpContent(const boost::system::error_code& error,
                                                        const size_t numBytesRead)
    {
      if(!handleError(error))
        return;
      assert(numBytesRead == m_currentlyReceivedTcpPkg.size());
      assert(m_server != nullptr);  // client does not receive anything in this example
      m_debugTcpRecvFile.write((const char*) m_currentlyReceivedTcpPkg.data(),
                               m_currentlyReceivedTcpPkg.size());
      m_debugTcpRecvFile.flush();
      m_server->handleReceivedTcpPkgServer(this);
    
      // next pkg
      m_currentlyReceivedTcpPkg.clear();
      startReceiveTcpHeader();
    }
    
    void GameServerConnection::sendFile(std::ifstream& file)
    {
      std::vector<PkgT> pkgs;
      pkgs.resize(1);
      pkgs[0].resize(2, 0);  // reserve 2 bytes for size
      pkgs[0].push_back(GameServerChunkTypes::clientFile);
    
      // read from file
      file.seekg(0, file.end);
      const uint32_t fileSize = file.tellg();
      std::vector<uint8_t> buf;
      buf.resize(fileSize);
      file.seekg(0, file.beg);
      file.read((char*) &buf[0], fileSize);
    
      pushBlob(buf, pkgs);
      for(const PkgT& p : pkgs) {
        auto sharedPkg = std::make_shared<PkgT>(p);
        sendTcp(sharedPkg);
        //std::this_thread::sleep_for(std::chrono::milliseconds(1));  // Damit treten keine Verluste auf.
      }
    }
    void GameServerConnection::pushBlob(const std::vector<uint8_t>& blob, std::vector<PkgT>& pkgs)
    {
      assert(pkgs.size() >= 1);
      PkgT& firstPkg = pkgs[pkgs.size() -1];
      const uint32_t blobSize = blob.size();
      {  // write blobSize
        size_t pkgCur = firstPkg.size();
        firstPkg.resize(pkgCur +4, 0);
        uint32_t* sizePtr = reinterpret_cast<uint32_t*>(&firstPkg[pkgCur]);
        *sizePtr = boost::endian::native_to_big(blobSize);
      }
      uint32_t blobCur = 0;
      PkgT* currentPkg = &firstPkg;
      do {
        PkgT pkg;
        if(currentPkg != &firstPkg) {
          pkg.resize(2, 0);
          currentPkg = &pkg;
        }
    
        const uint32_t bytesLeftInCurrentPkg = s_maxTcpPacketSize -currentPkg->size();
        const uint32_t bytesToWriteInCurrentPkg = std::min(blobSize - blobCur, bytesLeftInCurrentPkg);
        currentPkg->insert(currentPkg->end(), blob.begin() +blobCur, blob.begin() +blobCur +bytesToWriteInCurrentPkg);
        blobCur += bytesToWriteInCurrentPkg;
        uint16_t* pkgSize = reinterpret_cast<uint16_t*>(currentPkg->data());
        const uint16_t tmp = currentPkg->size() -2;
        *pkgSize = boost::endian::native_to_big(tmp);
        if(currentPkg != &firstPkg)
          pkgs.push_back(*currentPkg);  // note that firstPkg becomes invalid now
        currentPkg = nullptr;
      } while(blobCur < blobSize);
    }
    
    // GameServer
    GameServer::GameServer(boost::asio::io_service& ios, const uint16_t tcp_port, const bool ipv6)
      : m_tcpAcceptor(ios, tcp::endpoint(ipv6? tcp::v6(): tcp::v4(), tcp_port)), m_blobCur(0)
    {}
    void GameServer::kickClient(GameServerConnection* client, const std::string& reason)
    {
      std::cout << "kicking client (reason: " +reason +")\n";
      std::lock_guard<std::recursive_mutex> lock(m_connectionsMutex);
      auto it = m_connections.begin();
      while(it != m_connections.end()) {
        if(it->get() == client)
          it = m_connections.erase(it);
        else
          ++it;
      }
    
      saveBlobIfNecessary();
    }
    GameServer::ConnPtr GameServer::newConnPtr()
    {
      return ConnPtr(new GameServerConnection(m_tcpAcceptor.get_io_service(), this));
    }
    void GameServer::startAccepting()
    {
      ConnPtr conn = newConnPtr();
      m_tcpAcceptor.async_accept(conn->m_tcpSocket,
                                 boost::bind(&GameServer::handleAccepted, this,
                                             conn, boost::asio::placeholders::error));
    }
    void GameServer::handleAccepted(ConnPtr conn, const boost::system::error_code& error)
    {
      if(error) {
        startAccepting();
        return;
      }
      m_connectionsMutex.lock();
      m_connections.push_back(conn);
      m_connectionsMutex.unlock();
      conn->m_debugTcpRecvFile.open("whatServerReceived.stream", std::ofstream::binary);
      if(!conn->m_debugTcpRecvFile.is_open())
        std::cerr << "Could not open recv file\n";
      conn->startReceiveTcpHeader();
    
      std::cout << "client connected from "
                 + conn->m_tcpSocket.remote_endpoint().address().to_string() << std::endl;
      startAccepting();  // accept other incoming connections
    }
    void GameServer::handleReceivedTcpPkgServer(GameServerConnection* conn)
    {
      const PkgT& pkg = conn->m_currentlyReceivedTcpPkg;
      if(pkg.empty())
        return;
    
      if(m_blobCur < m_blob.size()) {
        readSomeBlob(pkg);
        saveBlobIfNecessary();
        return;
      }
    
      const uint8_t pkgType = pkg[0];  // 1 byte
      switch(pkgType) {
        case GameServerChunkTypes::clientFile: {
          m_blobCur = 0;
          const uint32_t* blobSize = reinterpret_cast<const uint32_t*>(&pkg[1]);  // 4 bytes
          m_blob.resize(boost::endian::big_to_native(*blobSize), 0);
          readSomeBlob(pkg, 5);
          saveBlobIfNecessary();
          break;
        }
        default:
          std::cerr << "network: \treceived unknown tcp packet type\n";
          return;
      }
    }
    
    void GameServer::readSomeBlob(const PkgT& pkg, const size_t pos)
    {
      const uint32_t bytesMissingInBlob = m_blob.size() -m_blobCur;
      const uint32_t bytesLeftInPkg = pkg.size() -pos;
      assert(bytesLeftInPkg <= bytesMissingInBlob);
      const uint32_t bytesToReadNow = std::min(bytesLeftInPkg, bytesMissingInBlob);
      memcpy(&m_blob[m_blobCur], &pkg[pos], bytesToReadNow);
      m_blobCur += bytesToReadNow;
    }
    void GameServer::saveBlobIfNecessary()
    {
      assert(m_blobCur <= m_blob.size());
      if(m_blobCur == m_blob.size() || m_connections.empty()) {
        std::cerr << "Writing blob to file\n";
        std::ofstream f;
        f.open("received.file", std::ofstream::binary);
        f.write((const char*) &m_blob[0], m_blob.size());
        m_blob.clear();
        m_blobCur = 0;
      }
    }
    


  • @lossyTcp sagte in [boost::asio] TCP-Verbindung verlustbehaftet wenn zu schnell gesendet wird:

    async_write

    Dass dein write completion handler nicht aufgerufen wird ist seltsam. Vielleicht hift dir dieses Beispiel weiter:
    https://theboostcpplibraries.com/boost.asio-network-programming



  • Ich tippe darauf, dass auf Seite des Clients die connection zu früh zerstört und das enthaltene Socket geschlossen wird. Die connection muss solange leben, bis alle I/O-Operationen abgeschlossen sind.

    Das könnte man u. a. mit enable_shared_from_this erreichen, indem die Referenzzählung bei den Async-Aufrufen hochgezählt wird (anstelle von this einfach an shared_from_this() binden). In den Async-Completion-Handlern wird die Referenzzählung dann wieder heruntergezählt.

    Alternativ kann man dafür sorgen, dass connection erst dann zerstört wird, wenn io_service::run zurückkehrt und somit keine I/O-Operationen mehr ausstehend sind.



  • Vielen Dank für die Ideen. Folgendes habe ich probiert:

    • Den write completeion handler durch eine globale Funktion ersetzt, sodass ich auf std::bind() verzichten kann. So wird es im vom @RBS2 verlinkten Artikel gemacht.
      Modifikationen der gameserver.cpp:
    void standaloneHandleSentTcp(const boost::system::error_code &ec,
                                 std::size_t bytes_transferred)
    {
      std::cout << "standaloneHandleSentTcp() called\n";  // Wird nie erreicht
      std::cout << "bytes transferred: " << bytes_transferred << std::endl;
      std::cout << "error message: " << ec.message() << std::endl;
    }
    
    // [...]
    
    void GameServerConnection::sendTcp(std::shared_ptr<PkgT> pkg)
    {
      assert(m_server == nullptr);  // server does not send anything in this example
      m_debugTcpSendFile.write((const char*) pkg->data(), pkg->size());
    
      static std::vector<std::shared_ptr<PkgT>> sentPkgs;
      sentPkgs.push_back(pkg);  // prevent destruction of pkg, which is not a parameter of standaloneHandleSentTcp
    
      boost::asio::async_write(m_tcpSocket, boost::asio::buffer(*pkg),
                               standaloneHandleSentTcp);
    //                           boost::bind(&GameServerConnection::handleSentTcp,
    //                                       this, pkg, boost::asio::placeholders::error()));
    }
    

    Der neue Handler wird leider auch nicht aufgerufen.

    • Sichergestellt, dass ioservice->run() terminiert, bevor die connection zerstört wird (client_main.cpp):
    // [...]
      connection->sendFile(file);
    
      this_thread::sleep_for(std::chrono::seconds(10));
    
      // destruct
      std::cout << "destruction begins now\n";
      connection.reset();
    // [...]
    

    Die Meldung "destruction begins now" wird erst deutlich nach "iothread is now terminating." ausgegeben.

    Statische Code-Analyse und Valgrind habe ich übrigens auch auf den Code losgelassen, ohne Ergebnis. Durch Valgrind läuft der Client allerdings wieder so langsam, dass die Übertragung reibungslos klappt. Also eine Art Heisenbug.

    Weitere Ideen?



  • This post is deleted!


  • @wob Bevor du deine Nachricht gelöscht hast, hast du geschrieben, dass das Programm bei dir funktioniert. Daher würde mich interessieren, ob bei dir die Methode GameServerConnection::handleSentTcp() aufgerufen wird. Dies ist ja bei mir selbst dann nicht der Fall, wenn die Übertragung funktioniert.



  • @lossyTcp Hat es nicht, die Dateigröße war zwar identisch, aber ich hatte mich bei der md5 verguckt. Die war unterschiedlich. Daher hatte ich die Nachricht wieder gelöscht. Deine andere Frage kann ich heute Abend beantworten.



  • Dein Code ist unnötig kompliziert, was das Verständnis und das Debugging sehr erschwert.

    Was mir allerdings auffällt ist, dass boost::asio::async_write(..) unmittelbar hintereinander aufgerufen wird. Das ist meiner Meinung nach ein Problem, denn boost::asio::async_write(..) sollte erst wieder neu aufgerufen werden, wenn der entsprechende Completion-Handler aufgerufen wurde. Die entsprechende Stelle in der Doku kann ich leider gerade nicht finden.

    Das erklärt auch, warum es mit der Wartezeit nach sendTcp(..) "funktioniert". Denn dann wird in der Pausenzeit jeweils die Send-Operation abgeschlossen.

    Ausserdem überschreibt auf Serverseite die Funktion saveBlobIfNecessary() das emfangene File, wenn die Verbindung getrennt wird (weil die if-Bedingung erfüllt ist, aber m_blob leer ist).



  • @theta sagte in [boost::asio] TCP-Verbindung verlustbehaftet wenn zu schnell gesendet wird:

    Was mir allerdings auffällt ist, dass boost::asio::async_write(..) unmittelbar hintereinander aufgerufen wird. Das ist meiner Meinung nach ein Problem, denn boost::asio::async_write(..) sollte erst wieder neu aufgerufen werden, wenn der entsprechende Completion-Handler aufgerufen wurde. Die entsprechende Stelle in der Doku kann ich leider gerade nicht finden.

    Hier steht es ja:

    This operation is implemented in terms of zero or more calls to the stream's async_write_some function, and is known as a composed operation. The program must ensure that the stream performs no other write operations (such as async_write, the stream's async_write_some function, or any other composed operations that perform writes) until this operation completes.

    Danke, dass du mich da mit der Nase draufstößt. Den Text habe ich schon zig mal gelesen, aber jedes Mal falsch verstanden. Ich dachte, man muss halt darauf achten, dass man nicht gleichzeitig von einem parallel laufenden Thread Sendeoperationen startet. Aber "operation" != "function call" und das hatte ich übersehen.

    Ich denke, das ist der Knackpunkt. Werde mich wieder melden, wenn ich mein Programm entsprechend geändert habe (vermutlich am Wochenende).



  • Problem gelöst. Folgende Schritte habe ich unternommen (alles auf Seite des Clients):

    • Verhindert, dass mehr als eine Sendeoperation gleichzeitig laufen (gameserver.cpp).
    bool GameServerConnection::sendTcp(std::shared_ptr<PkgT> pkg)
    {
      assert(m_server == nullptr);  // server does not send anything in this example
      {
        std::lock_guard<std::mutex> lock(m_sendMutex);
        if(m_currentlySending)
          return false;
        m_currentlySending = true;
      }
      m_debugTcpSendFile.write((const char*) pkg->data(), pkg->size());
    
      boost::asio::async_write(m_tcpSocket, boost::asio::buffer(*pkg),
                               boost::bind(&GameServerConnection::handleSentTcp,
                                           this, pkg, boost::asio::placeholders::error()));
      return true;
    }
    void GameServerConnection::handleSentTcp(std::shared_ptr<PkgT> pkg, const boost::system::error_code& error)
    {
      {  // unlock sending
        std::lock_guard<std::mutex> lock(m_sendMutex);
        assert(m_currentlySending);
        m_currentlySending = false;
      }
      std::cout << "handleSentTcp() called\n";  // Das wird irgendwie nie aufgerufen.
      pkg.reset();
      if(!handleError(error))
        return;
    }
    
    • Dafür gesorgt, dass ioservice->run() nicht vorzeitig terminiert. Dazu habe ich dem Client einfach gesagt, dass er direkt nach dem Herstellen der Verbindung beginnen soll, asynchron etwas zu empfangen (natürlich wird er nie etwas empfangen).

    Nun endlich empfängt der Server 1:1 das, was der Client sendet. Ich danke euch allen!


Log in to reply