Discussion :

[Chatanga]

Bonjour,

Je butte actuellement sur un problème de lecture/écriture concurrente avec un boost::asio:ip::tcp::iostream. L'écriture sur une instance de cette classe est bloquante tant qu'une lecture (bloquante elle aussi) est en cours.

Si j'ai bien compris, c'est une limitation du iostream encapsulant le streambuf TCP. La sentry associée veille à ce que le streambuf soit accédé de manière exclusive. Ca me choque un peu dans la mesure où les zones get & set d'un streambuf ne constituent pas forcément une même fifo (et même si c'était le cas on pourrait imaginer un comportement non bloquant) mais, si je ne me suis pas mélangé les pinceaux, c'est pourtant bien ce qui se passe.

Il est néanmoins possible de lire et d'écrire simultanéement sur une socket boost (le streambuf TCP gérant bien deux buffers décorélés) alors pourquoi ce comportement de la part d'un boost::asio:ip::tcp::iostream ? Il y aurait-il un moyen de construire un iostream permettant lecture et écriture simultanées ?

[Alp]

Salut,

1/ En protégeant avec un mutex tu ne peux pas lire et écrire (chacun son tour) comme tu le souhaites ?

2/ async_read et async_write ne sont pas bloquant, il me semble, donc tu dois pouvoir creuser dans cette voie-là.

[Chatanga]

1/ En protégeant avec un mutex tu ne peux pas lire et écrire (chacun son tour) comme tu le souhaites ?

Malheureusement la lecture est bloquante. En fait, j'ai un thread qui lit en boucle des blocs de tailles variables (les appels étant bloquants tant que le réseau n'a pas fournit assez de données).

L'écriture se fait de temps en temps dans un autre thread mais a donc toujours lieu, sauf coup de chance, durant un appel bloquant en lecture du thread précédent.

2/ async_read et async_write ne sont pas bloquant, il me semble, donc tu dois pouvoir creuser dans cette voie-là.

Ca m'obligerait à gérer moi-même l'aspect bloquant (dont j'ai besoin, c'est un stream après-tout). Mais pourquoi pas, ça peut être plus clean que la solution ci-dessous :

la seule solution que j'ai réussie à trouver depuis peu est de créer un istream et un ostream avec chacun leur propre streambuf mais qui pointent sur un même socket en interne (ce streambuf est une classe à moi). La seule raison qui m'oblige à avoir deux streambuf distincts est d'éviter le lock (c'est le streambuf qui sert de clef).

Ca fontionne mais il faut encore que je vois si c'est vraiment complétement sécurisé d'un point de vue multithreads (et puis ça reste un peu lourd d'avoir ces duplications).

[Alp]

Et si un thread écrit dans le socket pendant qu'un autre thread lit le socket... quelle maitrise as-tu sur le comportement ainsi induit ?

[Chatanga]

Et si un thread écrit dans le socket pendant qu'un autre thread lit le socket... quelle maitrise as-tu sur le comportement ainsi induit ?

En fait, mon streambuf est réentrant donc, si le asio::ip::tcp::socket est thread-safe, mon streambuf l'est aussi. La question devient donc : peut-on lire et écrire simultanément dans un tel socket sans problème ?

En pratique, ça marche mais, dans l'absolu, ça reste à prouver. Personnellement, je n'utilise asio que depuis très peu de temps et je ne peux rien affirmer. En particulier, je ne sais pas si les strands sont juste là pour faciliter le multi-threads ou sont en fait nécessaire pour en faire.

[Alp]

A priori écriture/lecture asynchrone, mutexes, ...

[Chatanga]

Hummmm... Vraiment ?

Qu'un socket boost ne soit pas safe si on fait plusieurs lectures ou plusieurs écritures simultanées, je suis d'accord, mais ce n'est pas mon cas (et si c'était le cas l'utilisation d'un i/ostream réglerait le problème). La lecture est toujours faite par le même thread, de même que pour l'écriture. Si c'est un socket POSIX utilisé sous la moquette par boost, je vois mal comment il pourrait se prendre les pieds dans son propre tapis.

Dans le cas contraire, je vois aussi mal ce que je pourrais mutexer. Et puis ça voudrait dire qu'il est impossible de garantir une écriture immédiate dans une socket sans parler du fait que boost pourrait perdre du temps à me notifier une lecture avant d'effectuer une écriture...

Il ne me reste plus qu'à inspecter plus en détail les sybillines sources de boost pour démêler le vrai du faux

[Alp]

D'un autre côté, dis-toi que dans un environnement MT et/ou en utilisant l'IO asynchrone, tu ne pourras pas contrôler les écritures/lectures. Tu délègues justement tout ça à Asio.

Bon, maintenant, la question qui tue : peux-tu expliquer ce que ton application doit faire et ce que représente ce fameux socket qui doit pouvoir être écrit et lu à n'importe quel moment.

[Chatanga]

Bon, maintenant, la question qui tue : peux-tu expliquer ce que ton application doit faire et ce que représente ce fameux socket qui doit pouvoir être écrit et lu à n'importe quel moment.

Oh mais c'est tout bête. L'application en question est une sorte de messagerie. Il y a un serveur qui a pour tâche de router les messages entre les différents clients connectés. Par exemple, le client A balance un message au client B en l'envoyant au serveur et en lui indiquant que c'est pour B.

Côté client, il y a une API toute simple qui offre la possibilité d'envoyer des messages et d'en recevoir. Cette partie réception est assurée par un thread qui écoute la connexion et extrait les messages qui arrivent pour les stocker dans une file où l'utilisateur peut venir les récupérer.

Pour le client, émission et réception sont donc décorrélées et ne s'enchaînent pas.

[Alp]

Alors, autant à utiliser Boost, je mettrais un timer qui disons toutes les demi-secondes émettrait un signal qui lui déclencherait la création de thread de lecture de la pile de messages reçus et ensuite les transmet pour l'affichage.
Pour l'écriture, tu émets le signal quand l'utilisateur valide son message et à ça tu connectes un simple thread d'envoi de message.

Donc j'utiliserais en plus d'Asio : Bind, Signal&Function, Thread

Pour le parsing y'a ce qu'il faut dans Boost également.

[loufoque]

Un iostream, c'est bufferisé etc.
Ce n'est pas la même chose qu'un file descriptor.

[Chatanga]

Mieux valant tard que jamais, voici un complément d'information sur mon problème que j'ai récemment eu le temps d'investiguer plus en détail. L'extrait de code qui suit illustre l'utilisation du tcp::iostream et de ma solution. Une solution qui n'est d'ailleurs pas idéale puisque a priori non thread-safe.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
void Client::connect( const string & address, int port )
{
	askedToHalt_ = false;
	boost::thread( boost::bind< int >( boost::ref( *this ), address, port, false ) );
}
 
void Client::disconnect()
{
	askedToHalt_ = true;
	threads_.join_all();
}
 
int Client::operator()( const string & host, int port, bool useWorkaround )
{
	askedToHalt_ = false;
 
	if ( useWorkaround )
	{
		auto_ptr< tcp::socket > pSocket = openSocket( host, port );
 
		SocketBuf in( *pSocket );
		SocketBuf out( *pSocket );
		istream input( &in );
		ostream output( &out );
 
		/*
		La classe tcp::socket n'est pas annoncée comme thread-safe si
		partagé dans la documentation Boost. Je ne connais cependant
		pas assez std::streambuf pour affirmer que ce soit un problème.
 
                        +-------------+
                        | tcp::socket |
                        +-------------+
                          ^         ^
                          |         |
                +-----------+     +-----------+
                | SocketBuf |     | SocketBuf |
                +-----------+     +-----------+
                      ^                 ^
                      |                 |
                 +---------+       +---------+
                 | istream |       | ostream |
                 +---------+       +---------+
 
                  thread #1         thread #2
 
		*/
 
		threads_.create_thread( boost::bind< int >( boost::ref( *this ), &input ) );
		threads_.create_thread( boost::bind< int >( boost::ref( *this ), &output ) );
		threads_.join_all();
 
		pSocket->close();
	}
	else
	{
		stringstream portAsString;
		portAsString << port;
 
		tcp::iostream inAndOut( host, portAsString.str() );
 
		threads_.create_thread( boost::bind< int >( boost::ref( *this ), static_cast< istream * >( &inAndOut ) ) );
		threads_.create_thread( boost::bind< int >( boost::ref( *this ), static_cast< ostream * >( &inAndOut ) ) );
		threads_.join_all();
 
		inAndOut.close();
	}
 
	return 0;
}
 
int Client::operator()( istream * in )
{
	while ( !askedToHalt_ )
	{
		enum { bufferSize = 4 };
		char buffer[ bufferSize + 1 ];
		memset( buffer, 0, bufferSize + 1 );
		in->read( buffer, bufferSize ); // Sans le workaround, l'écriture est bloquée durant l'attente en lecture.
 
		canPrintOut_.lock();
		cout << "Recu : '" << buffer << "'" << endl;
		canPrintOut_.unlock();
	}
	return 0;
}
 
int Client::operator()( ostream * out )
{
	char c = 'a';
	while ( !askedToHalt_ )
	{
		pause( 2 );
 
		canPrintOut_.lock();
		cout << "Envoye : '" << c << "'" << endl;
		canPrintOut_.unlock();
 
		*out << c << flush;
		if ( ++c > 'z' ) c = 'a';
	}
	return 0;
}

Je peux fournir mon petit programme de test pour ceux qui seraient intéressés mais l'extrait qui précède devrait suffire à voire le problème je pense. Dans la foulée, je balance aussi le code de SocketBuf mais ce dernier n'a rien de spécifique, c'est surtout le fait que j'en utilise deux dans ma solution qui est significatif. Le problème c'est que mes deux instances pointent sur le même tcp::socket qui n'est pas thread-safe dans un contexte MT (cf. la doc Boost).

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
SocketBuf::SocketBuf( tcp::socket & socket ) : socket_( socket )
{
	setg( get_buffer_.begin(), get_buffer_.begin() + depot_max, get_buffer_.begin() + depot_max );
	setp( 0, 0 );
}
 
int SocketBuf::underflow()
{
	if ( gptr() == egptr() )
	{
		boost::system::error_code codeErreur;
		size_t bytes_transferred = socket_.receive( buffer( buffer( get_buffer_ ) + depot_max ), 0, codeErreur );
		if ( codeErreur )
			return traits_type::eof();
 
		setg( get_buffer_.begin(), get_buffer_.begin() + depot_max, get_buffer_.begin() + depot_max + bytes_transferred );
		return traits_type::to_int_type( *gptr() );
	}
	else
	{
		return traits_type::eof();
	}
}
 
int SocketBuf::overflow( int c )
{
	if ( traits_type::eq_int_type( c, traits_type::eof() ) )
	{
		return traits_type::not_eof( c );
	}
	else
	{
		boost::system::error_code codeErreur;
		char_type ch = traits_type::to_char_type( c );
		socket_.send( buffer( &ch, sizeof( char_type ) ), 0, codeErreur );
		if ( codeErreur )
			return traits_type::eof();
		return c;
	}
}
 
int SocketBuf::sync()
{
	return overflow( traits_type::eof() );
}

En conclusion, j'utilise actuellement ma solution en priant que l'implémentation du std::streambuf dont hérite mon SocketBuf accède indirectement de manière thread-safe au tcp::socket...