Comment empêcher un serveur basé sur ASIO de se terminer ?

Question

J'ai lu quelques tutoriels sur Boost ASIO. Jusqu'à présent, j'ai compris que l'ensemble de l'envoi et de la réception est une boucle qui ne peut être itérée qu'une seule fois. Jetez un coup d'œil au code simple suivant :

client.cpp :

#include <boost/asio.hpp>
#include <boost/array.hpp>
#include <iostream>
#include <string>

boost::asio::io_service io_service;
boost::asio::ip::tcp::resolver resolver(io_service);
boost::asio::ip::tcp::socket sock(io_service);
boost::array<char, 4096> buffer;

void read_handler(const boost::system::error_code &ec, std::size_t bytes_transferred)
{
  if (!ec)
  {
    std::cout << std::string(buffer.data(), bytes_transferred) << std::endl;
    sock.async_read_some(boost::asio::buffer(buffer), read_handler);
  }
}

void connect_handler(const boost::system::error_code &ec)
{
  if (!ec)
  {        
    sock.async_read_some(boost::asio::buffer(buffer), read_handler);
  }
}

void resolve_handler(const boost::system::error_code &ec, boost::asio::ip::tcp::resolver::iterator it)
{
  if (!ec)
  {
    sock.async_connect(*it, connect_handler);
  }
}

int main()
{
  boost::asio::ip::tcp::resolver::query query("localhost", "2013");
  resolver.async_resolve(query, resolve_handler);
  io_service.run();
}

le programme resolves une adresse, connects au serveur et reads les données, et enfin ends lorsqu'il n'y a pas de données. Ma question : Comment puis-je continuer cette boucle ? Je veux dire, comment puis-je maintenir cette connexion entre un client et un serveur pendant toute la durée de vie de mon application afin que le serveur envoie des données chaque fois qu'il a quelque chose à envoyer ? J'ai essayé de briser ce cercle mais tout semble piégé à l'intérieur. io_service.run() La même question se pose dans le cas de mon serveur :

server.cpp :

#include <boost/asio.hpp>
#include <string>

boost::asio::io_service io_service;
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), 2013);
boost::asio::ip::tcp::acceptor acceptor(io_service, endpoint);
boost::asio::ip::tcp::socket sock(io_service);
std::string data = "Hello, world!";

void write_handler(const boost::system::error_code &ec, std::size_t bytes_transferred)
{
}

void accept_handler(const boost::system::error_code &ec)
{
  if (!ec)
  {
    boost::asio::async_write(sock, boost::asio::buffer(data), write_handler);
  }
}

int main()
{
  acceptor.listen();
  acceptor.async_accept(sock, accept_handler);
  io_service.run();
}

Ce n'est qu'un exemple. Dans une application réelle, j'aimerais garder le socket ouvert et le réutiliser pour d'autres échanges de données (en lecture et en écriture). Comment puis-je le faire ?

J'apprécie vos aimables commentaires. Si vous avez des références à des solutions faciles pour résoudre ce problème, j'apprécierais que vous les mentionniez. Je vous remercie.

Mise à jour (exemple de code serveur)

Sur la base de la réponse donnée ci-dessous (mise à jour 2), j'ai écrit le code du serveur. Veuillez noter que le code est simplifié (capable de compiler et d'exécuter cependant). Notez également que le io_service ne reviendra jamais car il est toujours en attente d'une nouvelle connexion. _Et c'est ainsi que le io_service.run ne revient jamais et fonctionne pour toujours_ . chaque fois que vous voulez que io_service.run revienne, il suffit de faire en sorte que l'accepteur n'accepte plus rien. s'il vous plaît, faites-le d'une des nombreuses façons dont je ne me souviens pas actuellement (sérieusement, comment faire cela de manière propre ? :) ).

profiter :

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <string>
#include <iostream>
#include <vector>
#include <time.h>
boost::asio::io_service io_service;
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), 2013);
boost::asio::ip::tcp::acceptor acceptor(io_service, endpoint);
//boost::asio::ip::tcp::socket sock(io_service);
std::string data = "Hello, world!";

class Observer;
std::vector<Observer*> observers;

class Observer
{
public:
    Observer(boost::asio::ip::tcp::socket *socket_):socket_obs(socket_){}
    void notify(std::string data)
    {
        std::cout << "notify called data[" << data << "]" << std::endl;
        boost::asio::async_write(*socket_obs, boost::asio::buffer(data) , boost::bind(&Observer::write_handler, this,boost::asio::placeholders::error));
    }
    void write_handler(const boost::system::error_code &ec)
    {
         if (!ec) //no error: done, just wait for the next notification
             return;
         socket_obs->close(); //client will get error and exit its read_handler
         observers.erase(std::find(observers.begin(), observers.end(),this));
         std::cout << "Observer::write_handler  returns as nothing was written" << std::endl;
    }
private:
        boost::asio::ip::tcp::socket *socket_obs;
};

class server
{
public:
     void CreatSocketAndAccept()
     {
          socket_ = new boost::asio::ip::tcp::socket(io_service);
          observers.push_back(new Observer(socket_));
          acceptor.async_accept(*socket_,boost::bind(&server::handle_accept, this,boost::asio::placeholders::error));
     }
      server(boost::asio::io_service& io_service)
      {
          acceptor.listen();
          CreatSocketAndAccept();
      }

      void handle_accept(const boost::system::error_code& e)
      {
          CreatSocketAndAccept();
      }
private:
  boost::asio::ip::tcp::socket *socket_;
};

class Agent
{
public:
    void update(std::string data)
    {
        if(!observers.empty())
        {
//          std::cout << "calling notify data[" << data << "]" << std::endl;
            observers[0]->notify(data);
        }
    }

};
Agent agent;
void AgentSim()
{
    int i = 0;
    sleep(10);//wait for me to start client
    while(i++ < 10)
    {
        std::ostringstream out("");
        out << data << i ;
//      std::cout << "calling update data[" << out.str() << "]" << std::endl;
        agent.update(out.str());
        sleep(1);
    }
}
void run()
{
    io_service.run();
    std::cout << "io_service returned" << std::endl;
}
int main()
{
    server server_(io_service);

    boost::thread thread_1(AgentSim);
    boost::thread thread_2(run);
    thread_2.join();
    thread_1.join();
}

Answer 1

Vous pouvez simplifier la logique des programmes basés sur l'asynchronisme de la manière suivante : chaque fonction qui appelle une fonction async_X fournit un gestionnaire. C'est un peu comme les transitions entre les états d'une machine d'état, où les gestionnaires sont les états et les appels asynchrones sont les transitions entre les états. Le simple fait de quitter un gestionnaire sans appeler une fonction async_* équivaut à une transition vers un état final. Tout ce que le programme "fait" (envoyer des données, recevoir des données, connecter des sockets, etc.) se produit pendant les transitions.

Si vous voyez les choses de cette manière, votre client ressemble à ceci (uniquement le "bon chemin", c'est-à-dire sans erreurs) :

<start>         --(resolve)----> resolve_handler
resolve_handler --(connect)----> connect_handler
connect_handler --(read data)--> read_handler
read_handler    --(read data)--> read_handler

Votre serveur ressemble à ceci :

<start>         --(accept)-----> accept handler
accept_handler  --(write data)-> write_handler
write_handler   ---------------> <end>

Puisque votre write_handler ne fait rien, il effectue une transition vers l'état final, ce qui signifie que ioservice::run renvoie. La question est maintenant de savoir ce que vous voulez faire une fois que les données ont été écrites sur la socket. En fonction de cela, vous devrez définir une transition correspondante, c'est-à-dire un appel asynchrone qui fait ce que vous voulez faire.

Mise à jour : D'après votre commentaire, je vois que vous voulez attendre que les prochaines données soient prêtes, c'est-à-dire le prochain tic-tac. Les transitions se présentent alors comme suit :

write_handler   --(wait for tick/data)--> dataready
dataready       --(write data)----------> write_handler

Vous voyez, cela introduit un nouvel état (handler), que j'ai appelé dataready On pourrait aussi bien l'appeler tick_handler ou autre. La transition vers le write_handler est facile :

void dataready()
{
  // get the new data...
  async_write(sock, buffer(data), write_handler);
}

La transition entre le write_handler peut être un simple async_wait sur un minuteur. Si les données proviennent de l'extérieur et que vous ne savez pas exactement quand elles seront prêtes, attendez un certain temps, vérifiez si les données sont là, et si ce n'est pas le cas, attendez encore un peu :

write_handler    --(wait some time)--> checkForData
checkForData:no  --(wait some time)--> checkForData
checkForData:yes --(write data)------> write_handler

ou, en (pseudo)code :

void write_handler(const error_code &ec, size_t bytes_transferred)
{
  //...
  async_wait(ticklenght, checkForData);
}

void checkForData(/*insert wait handler signature here*/)
{
  if (dataIsReady())
  {
    async_write(sock, buffer(data), write_handler);
  }
  else
  {
    async_wait(shortTime, checkForData):
  }
}

Mise à jour 2 : D'après votre commentaire, vous avez déjà un agent qui s'occupe de la gestion du temps d'une manière ou d'une autre (en appelant la mise à jour chaque fois que c'est nécessaire). Voici comment je résoudrais ce problème :

1. L'agent dispose d'une liste d'observateurs qui reçoivent une notification lorsqu'il y a de nouvelles données lors d'un appel de mise à jour.
2. Chaque observateur gère une connexion client (socket).
3. Le serveur se contente d'attendre les connexions en cours, de créer des observateurs à partir de ces connexions et de les enregistrer auprès de l'agent.

Je ne suis pas très au fait de la syntaxe exacte de l'ASIO, et il s'agira donc d'un pseudo-code un peu lourd :

Serveur :

void Server::accept_handler()
{
    obs = new Observer(socket);
    agent.register(obs);
    new socket; //observer takes care of the old one
    async_accept(..., accept_handler);
}

Agent :

void Agent::update()
{
    if (newDataAvailable())
    {
        for (auto& obs : observers)
        {
             obs->notify(data);
        }
    }
}

Observateur :

void Observer::notify(data)
{
    async_write(sock, data, write_handler);
}

void Observer::write_handler(error_code ec, ...)
{
     if (!ec) //no error: done, just wait for the next notification
         return;
     //on error: close the connection and unregister
     agent.unregister(this);
     socket.close(); //client will get error and exit its read_handler
}