sockets > Problème de communication interprocess

Bonjour!

Je cherche à faire communiquer (de manière optimale) plusieurs processus entre eux via des sockets. Il y a une application serveur qui instancie une socket d'écoute et quatre applications clientes qui intègrent chacune une socket de service pour envoyer des requêtes au serveur.

> J'ai un code fonctionnel qui réalise cette fonctionnalité à peu près correctement mais je suis confronté à des problèmes lorsque plusieurs clients émettent au même moment des requêtes vers le serveur. Ce dernier les reçoit mais les données reçus sont ou tronquées, ou même mélangées.

Voici un peu plus bas le code intégré dans chaque processus pour la gestion des sockets.
Fonctions dédiées pour création de la socket d'écoute et pour les sockets de service. ça repose sur 2 tableaux de type SOCKET* pour gérer plusieurs sockets...

Le code que j'ai écrit pour la lecture et pour l'écriture des messages sur les sockets n'est pas bien conçu...

La fonction "read_datas" pour récupérer à la demande les informations reçues dans un buffer.
La fonction "send_datas" pour envoyer des données sur la socket. (Je crois qu'il manque un contrôle ici pour voir si la socket est prête pour envoyer des données... et le mécanisme qui permet de ré-envoyer les données en cas de problème :mrgreen:)
La fonction "Request" qui utilise 'send_datas' et qui attend une sorte d'accusé de réception pour s'assurer que la demande a bien été acheminée. (Ici, ensemble fragile et bourrin car tant qu'aucune chaine de retour n'a été récupérée, on fait une pause et on rescrute la file de réception jusqu'à recevoir l'accusé, donc possibilité de boucle infinie si la communication ne s'effectue pas à un moment donné) :aie:

Le code :

Code:

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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
 
//#ifdef _WIN32			// of course, socket.h and socket.cpp are Windows only
#include "stdafx.h"
//#endif
 
#include <stdlib.h>
#include <stdio.h>
#include "sockets.h"
 
 
 
#include <winsock.h>
 
 
/// GLOBAL VARS section
SOCKET *service_socket = NULL;
SOCKET *listen_socket = NULL;
int ListenAllocated = 0;
int ClientAllocated = 0;
 
 
////////////
/// you may have problems while using this function, so set the debug constante to true is so
#define DEBUG_SOCKET_API		true
////////////
 
////////////////////////////////////////////////////////////////////////////////////////////
/////////       Your server process MUST ALLOCATE BOTH server and client Socket (s)     ////////////
/////////       Your client process ALLOCATE ONLY client Socket(s)                                      ////////////
////////////////////////////////////////////////////////////////////////////////////////////
 
 
 
int Alloc_ServiceSocks(int nb)
{
	service_socket = (SOCKET*) malloc(sizeof(SOCKET)*nb);
	if (!service_socket)
		return 0;
 
	ClientAllocated = nb;
	return 1;
}
 
 
int Alloc_ListenSocks(int nb)
{
	listen_socket = (SOCKET*) malloc(sizeof(SOCKET)*nb);
	if (!listen_socket)
		return 0;
 
	ListenAllocated = nb;
	return 1;
}
 
 
void close_Service_socket(int SockID)
{
	int err;
	if ( err = closesocket(service_socket[SockID]) != 0)			/* ferme le socket gerant la connexion cliente */
		printf("\nERROR while closing service_socket[%d]: %d", SockID, err);
}
 
 
void close_Listen_socket(int SockID)
{
	int err;
	if ( err = closesocket(listen_socket[SockID]) != 0)         // ferme le socket gerant la connexion serveur 
		printf("\nERROR while closing listen_socket[%d]: %d", SockID, err);
}
 
 
void Release_ServiceSocks()
{
	free(service_socket);
}
 
void Release_ListenSocks()
{
	free(listen_socket);
}
 
 
 
 
int InitWinsock(void) 
{ 
	WORD wVersionRequested;  
	WSADATA WSAData;              /* structure WSADATA définie dans winsock.h */ 
	int err; 
 
	wVersionRequested = MAKEWORD(1,1);      /* 1.1 version voulut de Winsock */      
	err = WSAStartup(wVersionRequested, &WSAData); /* appel de notre fonction */          
 
	/* controle d'erreur */ 
	if(err == 0)
	{
		if (DEBUG_SOCKET_API)
			fprintf(stdout, "\n[*] Winsock INIT ok"); 
	}
	else
	{ 
		fprintf(stderr, "\nWinsock INIT failed, error code : %d", GetLastError()); 
		return(-1); 
	} 
 
	return(0);
}
 
 
 
 
int Create_ListenSocket(u_short nPort, int SockID) 
{
 
	if (SockID > ListenAllocated-1)
	{
		fprintf(stderr, "\nUnable to create server socket! first call AllocPairedSocks()");
		return 0;
	}	
 
	struct sockaddr_in sin;          /* déclaration de la structure sockaddr_in */ 
 
	memset(&sin, 0x0, sizeof(struct sockaddr_in)); 
	sin.sin_addr.s_addr		= htonl(INADDR_ANY);		/* définit l'adresse du server */ 
	sin.sin_family			= AF_INET;					/* famille du socket */ 
	sin.sin_port			= htons(nPort);				/* port sur lequel va etre assigner le server (1337 pour l'exemple)*/ 
 
	listen_socket[SockID] = socket(AF_INET, SOCK_STREAM, IPPROTO_IP); /* appel de la fonction socket */ 
 
	if(listen_socket[SockID] != INVALID_SOCKET)  
	{
		if (DEBUG_SOCKET_API)
			fprintf(stdout, "\n[*] server Socket[%d] successfuly created.", SockID); 
	}	
	else
	{ 
		fprintf(stderr, "\nError while creating server socket[%d], error code: %d", SockID, WSAGetLastError()); 
		return(0); 
	}
 
	if(bind(listen_socket[SockID], (sockaddr *)&sin, sizeof(struct sockaddr_in)) != SOCKET_ERROR) /* associe l'adresse local au socket */ 
	{	
		if (DEBUG_SOCKET_API)
			fprintf(stdout, "\n[*] Association IP Adress - socket[%d] success.", SockID); 
	}	
	else 
	{ 
		fprintf(stderr, "\nError while associating IP Adress with server socket[%d], error code: %d", SockID, WSAGetLastError()); 
		return(0); 
	} 
 
	// setup listen_socket[SockID] as NON BLOCKING
	// under UNIX OS, see fcntl...
	u_long *sopt_value = new u_long(1);
	*sopt_value = 1;
	ioctlsocket(listen_socket[SockID], FIONBIO, sopt_value);		// 1 to set FIONBIO enabled, otherwise set it to 0
 
 
	return(1); /* retourne le socket */ 
} 
 
 
 
 
 
//int Server(SOCKET listen_socket)
int Server(int SockID)
{ 
	struct sockaddr_in adresse; 
	int adresse_size = 0; 
 
	memset(&adresse, 0, sizeof(struct sockaddr_in)); 
 
	/* passe en mode écoute le serveur */ 
	if(listen(listen_socket[SockID], 3) == 0)
	{
		if (DEBUG_SOCKET_API)
			fprintf(stdout, "\n[*] Server listening on Socket[%d].", SockID); 
	}
	else 
	{ 
		fprintf(stderr, "\nServer listening operation failed on Socket[%d], error code: %d", SockID, WSAGetLastError()); 
		return(-1); 
	}     
 
	if (DEBUG_SOCKET_API)
		fprintf(stderr, "\nWaiting for client connexion (Socket[%d]), code: %d   ", SockID, WSAGetLastError());
 
	while(1)
	{ 
		adresse_size = sizeof(adresse); 
		/* accept les connexions clientes */ 
		service_socket[SockID] = accept(listen_socket[SockID], (struct sockaddr *)&adresse, &adresse_size);  
		//printf("\n attente du client...");
 
		if(service_socket[SockID] == INVALID_SOCKET) 
		{
			if (DEBUG_SOCKET_API)
				fprintf(stdout, ".");
			Sleep(500);
			//return(-1); 
		}
		else	// si ok...
		{
			if (DEBUG_SOCKET_API)
				fprintf(stdout, "\n[*] Client successfuly connected on Socket[%d]", SockID);
			break;
		}
	} 
 
    return(0); 
} 
 
 
 
int Create_ServiceSocket(const char* ipadd, u_short nPort, int SockID) 
{ 
	if (SockID > ClientAllocated-1)
	{
		fprintf(stderr, "\nUnable to create Service(client) socket! first call AllocPairedSocks()");
		return 0;
	}	
 
	struct sockaddr_in sin;          /* déclaration de la structure sockaddr_in */ 
 
	memset(&sin, 0x0, sizeof(struct sockaddr_in)); 
	sin.sin_addr.s_addr		= inet_addr(ipadd);			/* définit l'adresse IP du server */
	sin.sin_family			= AF_INET;					/* famille du socket */ 
	sin.sin_port			= htons(nPort);				/* port sur lequel va etre assigner le server (1337 pour l'exemple)*/ 
 
	service_socket[SockID] = socket(AF_INET, SOCK_STREAM, IPPROTO_IP); /* appel de la fonction socket */ 
 
	if(service_socket[SockID] != INVALID_SOCKET)
	{
		if (DEBUG_SOCKET_API)
			fprintf(stdout, "\n[*] Client Socket[%d] successfuly created.", SockID); 
	}
	else
	{ 
		fprintf(stderr, "\nError while creating client socket[%d], error code: %d", SockID, WSAGetLastError()); 
		return(0); 
	}
 
	if (DEBUG_SOCKET_API)
		fprintf(stdout, "\nConnection to server ");
 
	if (!connect(service_socket[SockID], (sockaddr *)&sin, sizeof(struct sockaddr_in)) )
	{
		if (DEBUG_SOCKET_API)
			fprintf(stdout, " OK!");
	}
	else
	{
		return(0);
	}
 
	// setup service_socket[SockID] as NON BLOCKING
	// under UNIX OS, see fcntl...
 
	u_long *sopt_value = new u_long(1);
	*sopt_value = 1;
	ioctlsocket(service_socket[SockID], FIONBIO, sopt_value);		// 1 to set FIONBIO enabled, otherwise set it to 0
 
	return(1); /* retourne le socket */ 
} 
 
 
 
 
 
 
 
fd_set socks;        	/* Socket file descriptors we want to wake up for, using select() */
struct timeval timeout;  /* Timeout for select */
int readsocks;	     			/* Number of sockets ready for reading */
int retval;
 
 
int read_datas(char* buffer, int SockID) 
{ 
	int ret = 0;
	timeout.tv_sec = 0;
	timeout.tv_usec = 50000;
 
	FD_ZERO(&socks); 
	FD_SET(service_socket[SockID], &socks); 
 
 
	readsocks = select((int)service_socket[SockID]+1, &socks, (fd_set *) 0, (fd_set *) 0, &timeout);
 
	if (readsocks < 0)
	{
				printf("\nSelect() Error!\n");
				exit(EXIT_FAILURE);
	}
	if (readsocks == 0)	/* Nothing to read, just show that we're alive */
	{
			//printf("Nothing to read...\n");
	}
	else ret = recv(service_socket[SockID], buffer, MESSAGE_SIZE, 0);
 
 
	if (ret)
		return 0;
	else
		return -1;
} 
 
 
 
 
// avoided unuseful Param1 (socket_type)
//unsigned int send_datas(char* message, int SockID)
unsigned int send_datas(const char* message, int SockID)
{
	unsigned int ret;
 
	if (SockID > ClientAllocated-1)
	{
		printf("\ninvalid SockID argument in envoi_donnees() call!");
		return -1;
	}
 
	ret = send(service_socket[SockID], message, strlen(message), 0);	// 0: connexion normale
 
 
	if(ret == SOCKET_ERROR)
	{
		fprintf(stdout, "\nError while sending datas!");
		fprintf(stdout, "\ndatas=%s  on socket %d", message, SockID);
	}
 
	return ret;
}
 
 
 
 
 
/////////////////////////////////////////////////////////////////////////////
// this function to be sure that the message has been well received /////////
// the paired process must respond something (at least something like "ok" //
/////////////////////////////////////////////////////////////////////////////
 
void Request(int SockID, char *req, char* ret)
{
	//printf("\nenvoi de l'instruction: %s", req);
	memset(ret, '\0', MESSAGE_SIZE);
	send_datas(req, SockID);
	do
	{	
		Sleep(5);
		read_datas(ret, SockID);
	}
	while(!strcmp(ret, ""));	// loop while we haven't the answer (it shouldn't be longer!)
}

Tout ça n'est pas très clair pour moi, l'ensemble doit se comporter de manière très réactive. Je conçois une application pour la robotique qui doit "se rapprocher" d'un système temps réel.
Je pense qu'il existe d'autres moyens de réaliser ceci: :lol:

Citation:

send_datas(req, SockID);
do
{
Sleep(5);
read_datas(ret, SockID);
}
while(!strcmp(ret, "")); // loop while we haven't the answer (it shouldn't be longer!)

J'espère que vous comprendrez mon besoin.
Vos remarques et idées sont les bienvenues et je vous remercie pour l'intérêt que vous portez à ce topic.

Bonjour,
Je n'ai pas encore trouvé solution au problème, mais j'ai vu que beaucoup d'entre vous parliez de la librairie Boost dans cette rubrique.
Je vais voir ça de plus près.

Je vais peut être poster le code qui utilise les fonctions citées auparavant :P

Code:

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
	while (application)
	{
		// En gros je vérifie périodiquement qu'est ce qu'il y a à lire sur la socket de service...
		read_datas(buffer, MOTORCTRL_SOCKID);
 
		if (strcmp(buffer, "")) // S'il y a quelque chose à lire...
		{
 
			//////// 1) je place le message fraichement reçu dans une pile FIFO (premier message arrivé = premier traité)
			//
			total_items = explode(superArray, (const char*)buffer);
 
			for (item = 0; item < total_items; item++)
			{
				sizeof_stack = stack_write(thestack, (const char*) superArray[item]);
				if (sizeof_stack == 0)
				{
					printf("\n[MOTORCTRL] > WARNING! stack is full!!!");					
					tracelog("\n[MOTORCTRL] > WARNING! stack is full!!!");
				}
			}
 
			// j'efface le dernier message reçu dans le tampon
			memset(buffer, '\0', MESSAGE_SIZE);
 
			for (int b=0; b<100; b++)
				memset(superArray[b], '\0', 100);
 
 
			///////// 2) Maintenant je vide ma pile en traitant chacun des messages reçus
			while (stack_read_as_FIFO(thestack, buffer))
			{
				curr_msg = "";
				curr_msg.insert(0, buffer);
 
				GetEventAct(curr_msg, buffer);
 
				memset(buffer, '\0', MESSAGE_SIZE);
			}
 
		}
 
		// J'endors mon process ici un labs de temps pour pas bouffer tout le CPU
		#ifdef _WIN32
		Sleep(400);	// ms for Win...
		#else
		sleep(4);
		#endif
 
	}

Salut,
Je n'ai pas regardé en détail mais :
1/ Ton code est du C, ici c'est du C++. Faut choisir ;)
2/ En particulier, Boost est une bibliothèque C++
3/
Code:

1 2 3 u_long *sopt_value = new u_long(1); *sopt_value = 1; ioctlsocket(listen_socket[SockID], FIONBIO, sopt_value);
:alerte: Fuite mémoire :alerte:
Code:

1 2 u_long sopt_value = 1; ioctlsocket(listen_socket[SockID], FIONBIO, &sopt_value);
4/ On ne voit pas si tu utilises un ou plusieurs buffer pour la réception selon la socket

Salut 3DArchi,

Merci pour le retour.
Je vais corriger la fuite mémoire de suite.

1/ Je resterai bien sur du C, au moins je comprends à peu près ce que je fais :)

4/ Pour ce qui est des sockets, chacune d'elles stocke ce qu'elle reçoit dans un tampon, qui lui copie son contenu dans une pile avant d'être effacé.
J'ai pensé à ce mécanisme pour le cas où je ne souhaiterais pas exécuter tout de suite les instructions reçues.

Donc à chaque socket sa pile d'instruction...

Petite illustration de traitement pour une des sockets:

Citation:

effacer pile

Tant que (Application = vrai)

effacer buffer

buffer <- Lire données socket

si (longueur buffer > 0)
ajouter à pile FIFO chaque instruction reçue dans buffer
// buffer contient la série d'instructions séparées par des ';'
fin de si

// Ici je choisis de retirer (ou non) une instruction de ma pile et d'exécuter
// la fonction qui va bien...
// ça donne approximativement:

si (longueur pile > 0)
tmpInstruction <- dépiler instruction (FIFO)

si (tmpInstruction = "motor.forward")
avance()
sinon si (tmpInstruction = "motor.backward")
recule()
...
...
sinon affiche "instruction non reconnue: " suivi de la chaine tmpInstruction

fin de si

Fin Tant Que

La pile étant dimensionnée en conséquent, je n'ai pas rencontré de débordement (cas où l'on reçoit plus d'instructions qu'on a le temps de les traiter)

Il m'arrive de voir s'afficher à l'écran par exemple instruction non reconnue: "motor.fo"
Cette instruction est bien dépilée mais pas reconnue car tronquée :cry: et donc elle passe à la trappe.
J'observe ce phénomène quand plusieurs messages transitent simultanément. Ce qui a pour effet de faire tôt ou tard figer l'ensemble...