boost :: asio :: sécurité des threads socket

Après réétudier le code pour async_write je suis maintenant convaincue que toute opération d'écriture est thread-safe si et seulement si la taille du paquet est plus petit que

default_max_transfer_size = 65536;

Ce qui se passe, c'est que dès qu'un async_write est appelé un async_write_some est appelé dans le même thread. Tous les threads dans la piscine d'appeler une certaine forme de io_service::course sur l'appel de async_write_some pour que l'opération d'écriture jusqu'à la fin.

Ces async_write_some appels et interleave si il doit être appelé plus d'une fois (les paquets sont de plus de 65536).

ASIO n'est pas de file d'attente de l'écriture d'un socket que vous attendez, une finition après l'autre. Afin d'assurer à la fois fil et interleave sûr écrit considérer le morceau de code suivant:

    void my_connection::async_serialized_write(
boost::shared_ptr<transmission> outpacket) {
m_tx_mutex.lock();
bool in_progress = !m_pending_transmissions.empty();
m_pending_transmissions.push(outpacket);
if (!in_progress) {
if (m_pending_transmissions.front()->scatter_buffers.size() > 0) {
boost::asio::async_write(m_socket,
m_pending_transmissions.front()->scatter_buffers,
boost::asio::transfer_all(),
boost::bind(&my_connection::handle_async_serialized_write,
shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
} else { //Send single buffer
boost::asio::async_write(m_socket,
boost::asio::buffer(
m_pending_transmissions.front()->buffer_references.front(),                          m_pending_transmissions.front()->num_bytes_left),
boost::asio::transfer_all(),
boost::bind(
&my_connection::handle_async_serialized_write,
shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
}
m_tx_mutex.unlock();
}
void my_connection::handle_async_serialized_write(
const boost::system::error_code& e, size_t bytes_transferred) {
if (!e) {
boost::shared_ptr<transmission> transmission;
m_tx_mutex.lock();
transmission = m_pending_transmissions.front();
m_pending_transmissions.pop();
if (!m_pending_transmissions.empty()) {
if (m_pending_transmissions.front()->scatter_buffers.size() > 0) {
boost::asio::async_write(m_socket,
m_pending_transmissions.front()->scatter_buffers,
boost::asio::transfer_exactly(
m_pending_transmissions.front()->num_bytes_left),
boost::bind(
&chreosis_connection::handle_async_serialized_write,
shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
} else { //Send single buffer
boost::asio::async_write(m_socket,
boost::asio::buffer(
m_pending_transmissions.front()->buffer_references.front(),
m_pending_transmissions.front()->num_bytes_left),
boost::asio::transfer_all(),
boost::bind(
&my_connection::handle_async_serialized_write,
shared_from_this(),
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
}
m_tx_mutex.unlock();
transmission->handler(e, bytes_transferred, transmission);
} else {
MYLOG_ERROR(
m_connection_oid.toString() << " " << "handle_async_serialized_write: " << e.message());
stop(connection_stop_reasons::stop_async_handler_error);
}
}

Ce fait, fondamentalement, une file d'attente pour l'envoi d'un paquet à la fois. async_write n'est appelé qu'après la première écriture réussit puis appelle le gestionnaire d'origine pour le premier à écrire.

Il aurait été plus facile si asio fait écrire les files d'attente automatique par prise d'eau.

La clé de la compréhension de l'ASIO est de réaliser que l'achèvement des gestionnaires d'exécuter uniquement dans le contexte d'un thread qui a appelé io_service.run() n'importe quel thread appelle la méthode asynchrone. Si vous n'avez appelé io_service.run() dans un thread alors tous d'achèvement des gestionnaires d'exécuter en série dans le contexte de ce thread. Si vous avez appelé io_service.run() en plus d'un thread à l'achèvement des gestionnaires s'exécute dans le contexte de l'un de ces fils. Vous pouvez considérer cela comme un pool de threads où les fils dans la piscine sont ces fils qui ont appelé io_service.run() sur le même io_service objet.

Si vous avez plusieurs threads appeler io_service.run() ensuite, vous pouvez forcer l'achèvement des gestionnaires d'être sérialisé en les mettant dans un strand.

Pour répondre à la dernière partie de votre question, vous devriez appeler boost::async_write(). Cela permettra d'envoi de l'opération d'écriture sur un thread qui a appelé io_service.run() et invoqué le gestionnaire d'achèvement lorsque l'écriture s'est fait. Si vous avez besoin de sérialiser cette opération, alors c'est un peu plus compliqué et vous devriez lire la documentation sur les brins ici.

Considérons d'abord que le socket est un cours d'eau et n'est pas en interne de protection contre simultanément en lecture et/ou écriture. Il y a trois différentes considérations.

1. Une exécution simultanée des fonctions qui accèdent à la même prise.
2. Une exécution simultanée des délégués qui entourent le même socket.
3. Exécution entrelacée de délégués qui écrivent dans le même socket.

La exemple de discussion est asynchrone, mais ne sont pas simultanées. Le io_service est exécuter à partir d'un seul fil, ce qui rend tous les chats des opérations de la clientèle non-simultané. En d'autres termes, il évite tous ces problèmes. Même les async_write doit en interne complet de l'envoi de toutes les parties d'un message avant tout autre travail peut se poursuivre, en évitant le problème de l'entrelacement.

Gestionnaires sont utilisées que par un thread qui appelle une surcharge de run(), run_one(), poll() ou poll_one() pour le io_service.

En publiant des travaux pour le thread unique io_service les autres threads peuvent en toute sécurité d'éviter à la fois de concurrence et de blocage par la file d'attente de travail dans le io_service. Cependant, si votre scénario s'oppose à vous de tous les travaux de mise en mémoire tampon pour un socket donné, les choses deviennent plus compliquées. Vous pouvez avoir besoin de bloquer la communication par socket (mais pas de fils), par opposition à la file d'attente de travail indefinately. Aussi, le travail de la file d'attente peut être très difficile à gérer, car il est entièrement opaque.

Si votre io_service exécute plus d'un fil, vous pouvez facilement éviter les problèmes ci-dessus, mais vous ne pouvez invoquer la lecture ou de l'écriture de l'gestionnaires d'autres lectures ou d'écritures (et au démarrage). Cette séquences de tous les accès à la socket, tout en restant non-bloquant. La sécurité vient du fait que le modèle est à l'aide d'un thread à un moment donné. Mais la publication de travailler à partir d'un thread indépendant est problématique, même si vous n'avez pas l'esprit de mise en mémoire tampon.

Un brin est un pilote asio de la classe que les postes de travail à un io_service d'une manière qui garantit la non-simultanées de l'invocation. Toutefois, l'utilisation d'un brin d'invoquer async_read et/ou async_write résout uniquement le premier de ces trois problèmes. Ces fonctions en interne poste de travail à la io_service de la douille. Si ce service est en cours d'exécution de plusieurs threads les travaux peuvent être exécutés simultanément.

Alors, comment voulez-vous, pour un socket donné, en toute sécurité invoquer async_read et/ou async_write en même temps?

1. Avec les appelants, le premier problème peut être résolu avec un mutex ou un brin, à l'aide de l'ancien si vous ne voulez pas de tampon, le travail et le dernier, si vous ne. Cela protège le socket pendant les appels de fonction, mais ne fait rien pour les autres problèmes.

2. Le deuxième problème semble plus difficile, car il est difficile de voir ce qui se passe à l'intérieur du code d'exécution asynchrone des deux fonctions. L'async les fonctions à la fois de poste de travail à la io_service de la douille.

De la poussée de la prise de la source:

/**
* This constructor creates a stream socket without opening it. The socket
* needs to be opened and then connected or accepted before data can be sent
* or received on it.
*
* @param io_service The io_service object that the stream socket will use to
* dispatch handlers for any asynchronous operations performed on the socket.
*/
explicit basic_stream_socket(boost::asio::io_service& io_service)
: basic_socket<Protocol, StreamSocketService>(io_service)
{
}

Et de la io_service::run()

/**
* The run() function blocks until all work has finished and there are no
* more handlers to be dispatched, or until the io_service has been stopped.
*
* Multiple threads may call the run() function to set up a pool of threads
* from which the io_service may execute handlers. All threads that are
* waiting in the pool are equivalent and the io_service may choose any one
* of them to invoke a handler.
*
* ...
*/
BOOST_ASIO_DECL std::size_t run();

Donc si vous donnez un socket plusieurs threads, il n'a pas le choix d'utiliser plusieurs threads - en dépit de ne pas thread-safe. La seule façon d'éviter ce problème (à part le remplacement de la douille de mise en œuvre) est de donner à la prise d'un thread pour travailler avec. Pour une même prise, c'est ce que vous voulez de toute façon (donc ne pas la peine de courir à écrire un remplacement).

1. Le troisième problème peut être résolu en utilisant une (autre) mutex est verrouillé avant de la async_write, passé dans le gestionnaire d'achèvement et déverrouillé à ce point. Cela permettra d'éviter toute appelant à partir de début à écrire jusqu'à ce que toutes les parties de la précédente écrire sont complètes.

Noter que le async_write postes de travail à une file d'attente - c'est comment il est capable de revenir, presque immédiatement. Si vous mettez trop de travail, vous pourriez avoir à traiter avec un certain nombre de conséquences. Malgré l'utilisation d'un seul io_service fil pour le socket, vous pouvez avoir n'importe quel nombre de threads de travail via l'affichage simultané ou non des appels simultanés à async_write.

D'autre part, async_read est simple. Il n'y a pas d'entrelacement de problème, et vous n'en boucle à partir du gestionnaire de l'appel précédent. Vous pouvez ou ne voulez pas envoyer les résultats du travail d'un autre thread ou une file d'attente, mais si vous effectuez sur le gestionnaire d'achèvement fil que vous êtes tout simplement le blocage de toutes les lectures et les écritures sur votre single-threaded socket.

Mise à JOUR

J'ai fait un peu plus de creuser dans la mise en œuvre de l'implémentation sous-jacente de la socket stream (pour une plate-forme). Il semble être le cas que le support de façon constante exécute plate-forme de prise d'appels sur l'invocation de fil, pas le délégué posté le io_service. En d'autres termes, malgré le fait que async_read et async_write semblent revenir immédiatement, ils le font en fait exécuter toutes les opérations de socket avant de retourner. Seuls les gestionnaires sont affichés à l'io_service. Ce n'est ni documentées ni exposés par le exaple code que j'ai examinés, mais en supposant qu'il est garanti comportement, il a un impact significatif sur le deuxième problème ci-dessus.

En supposant que le travail posté à l'io_service n'intègre pas les opérations de socket, il n'est pas nécessaire de limiter la io_service à un seul thread. Il permet toutefois d'insister sur l'importance de se prémunir contre une exécution simultanée de l'async fonctions. Ainsi, par exemple, si l'on suit l'exemple de discussion, mais plutôt en ajoute un autre thread pour le io_service, il devient un problème. Avec async les appels de fonction en cours d'exécution dans gestionnaires de fonction, vous avez simultanées de l'exécution de la fonction. Cela nécessiterait un mutex, ou tous les async les appels de fonction pour être réutilisé pour l'exécution sur un brin.

Mise à JOUR 2

À l'égard de la troisième problème (interleaving), si la taille des données dépasse 65536 octets, le travail est divisé interne à async_write et envoyé dans les pièces. Mais il est essentiel de comprendre que, si il n'y a plus qu'à un fil dans la io_service, des morceaux de travail autres que le premier sera affecté à différents threads. Tout cela se produit à l'interne dans le async_write fonction avant l'achèvement de votre gestionnaire est appelé. La mise en œuvre crée son propre intermédiaire d'achèvement gestionnaires et les utilise pour l'exécution de tous, mais la première opération de socket.

Cela signifie que toute la garde autour de l'async_write appel (mutex ou verticale) sera pas protéger le support si il y a plusieurs io_service fils et plus de 64 ko de données à la poste (par défaut, cela peut varier). Par conséquent, dans ce cas, l'entrelacement de la garde est nécessaire non seulement pour interleave sécurité, mais aussi de la sécurité des threads de la douille. J'ai vérifié tout cela dans un débogueur.

LE MUTEX OPTION

La async_read et async_write fonctions en interne, l'utilisation de la io_service afin d'obtenir des fils sur lesquels post-achèvement des gestionnaires, le blocage jusqu'à ce que les threads sont disponibles. Cela les rend dangereux pour la garde avec des mutex serrures. Lorsqu'un mutex est utilisé pour la garde de ces fonctions, un blocage se produire lorsque les threads contre la serrure, mourir de faim les io_service. Étant donné qu'il n'y a pas d'autre moyen de garde async_write lors de l'envoi d' > 64 ko avec une multithread io_service, il se verrouille dans un seul thread dans ce scénario, ce qui, évidemment, se résout la question de la simultanéité.

Cela dépend de si vous accédez à même socket objet à partir de plusieurs threads. Disons que vous avez deux threads s'exécutant même io_service::run() fonction.

Si par exemple vous faire de la lecture et de l'écriture en même temps ou peut-être effectuer d'annuler l'opération
d'autres fil. Ensuite, il n'est pas sûr.

Toutefois, si votre protocole ne fait qu'une seule opération à la fois.

1. Si un seul thread exécute la io_service exécuté, alors il n'y a pas de problème. Si vous voulez exécuter quelque chose sur la prise de la autre thread, vous pouvez appeler io_service::post() avec
   gestionnaire de cette opération sur le socket de sorte qu'il serait exécuté dans le même thread.
2. Si vous avez plusieurs threads d'exécution io_service::run et vous essayez d'effectuer les opérations simultanément, disons - annuler et de l'opération de lecture, alors vous devriez utiliser des brins. Il y a un tutoriel pour ce en Boost.Asio de la documentation.