Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Fix (c) dates; make clear this is deprecated code
[simgrid.git] / src / gras / SG / gras_sg.c
1 /* $Id$ */
2
3 /* gras_sg - legacy implementation of GRAS on top of the SimGrid simulator  */
4 /* This file should be KILLED whenever the raw sockets work in the new gras */
5
6 /* Copyright (c) 2003 Martin Quinson. All rights reserved.                  */
7
8 /* This program is free software; you can redistribute it and/or modify it
9  * under the terms of the license (GNU LGPL) which comes with this package. */
10
11 #include <stdio.h>
12 #include <string.h>
13
14 #include "gras_sg.h"
15
16 XBT_LOG_DEFAULT_CATEGORY(GRAS);
17
18 /* **************************************************************************
19  * Openning/Maintaining/Closing connexions (private functions for both raw
20  * and regular sockets)
21  * **************************************************************************/
22 xbt_error_t
23 _gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
24                       int raw, unsigned int bufSize, /* OUT */ gras_sock_t **sock) {
25
26   gras_hostdata_t *hd=hd=(gras_hostdata_t *)MSG_host_get_data(MSG_host_self());
27   grasProcessData_t *pd=(grasProcessData_t *)MSG_process_get_data(MSG_process_self());
28   int port,i;
29   const char *host=MSG_host_get_name(MSG_host_self());
30
31   xbt_assert0(hd,"HostData=NULL !! Please run grasInit on each process\n");
32   xbt_assert0(pd,"ProcessData=NULL !! Please run grasInit on each process\n");
33
34   for (port=startingPort ; port <= endingPort ; port++) {
35     for (i=0; i<hd->portLen && hd->port[i] != port; i++);
36     if (i<hd->portLen && hd->port[i] == port)
37       continue;
38
39     /* port not used so far. Do it */
40     if (i == hd->portLen) {
41       /* need to enlarge the tables */
42       if (hd->portLen++) {
43         hd->port2chan=(int*)realloc(hd->port2chan,hd->portLen*sizeof(int));
44         hd->port     =(int*)realloc(hd->port     ,hd->portLen*sizeof(int));
45         hd->raw      =(int*)realloc(hd->raw      ,hd->portLen*sizeof(int));
46       } else {
47         hd->port2chan=(int*)malloc(hd->portLen*sizeof(int));
48         hd->port     =(int*)malloc(hd->portLen*sizeof(int));
49         hd->raw      =(int*)malloc(hd->portLen*sizeof(int));
50       }
51       if (!hd->port2chan || !hd->port || !hd->raw) {
52         fprintf(stderr,"GRAS: PANIC: A malloc error did lose all ports attribution on this host\n");
53         hd->portLen = 0;
54         return malloc_error;
55       }
56     }
57     hd->port2chan[ i ]=raw ? pd->rawChan : pd->chan;
58     hd->port[ i ]=port;
59     hd->raw[ i ]=raw;
60
61     /* Create the socket */
62     if (!(*sock=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
63       fprintf(stderr,"GRAS: openServerSocket: out of memory\n");
64       return malloc_error;
65     }    
66     
67     (*sock)->server_sock  = 1;
68     (*sock)->raw_sock     = raw;
69     (*sock)->from_PID     = -1;
70     (*sock)->to_PID       = MSG_process_self_PID();
71     (*sock)->to_host      = MSG_host_self();
72     (*sock)->to_port      = port;  
73     (*sock)->to_chan      = pd->chan;
74
75     /*
76     fprintf(stderr,"GRAS: '%s' (%d) ears on %s:%d%s (%p).\n",
77             MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
78             host,port,raw? " (mode RAW)":"",*sock);
79     */
80     return no_error;
81   }
82   /* if we go out of the previous for loop, that's that we didn't find any
83      suited port number */
84
85   fprintf(stderr,
86           "GRAS: can't find an empty port between %d and %d to open a socket on host %s\n.",
87           startingPort,endingPort,host);
88   return mismatch_error;
89 }
90
91 xbt_error_t
92 _gras_sock_client_open(const char *host, short port, int raw, unsigned int bufSize,
93                      /* OUT */ gras_sock_t **sock) {
94   m_host_t peer;
95   gras_hostdata_t *hd;
96   int i;
97
98   /* make sure this socket will reach someone */
99   if (!(peer=MSG_get_host_by_name(host))) {
100       fprintf(stderr,"GRAS: can't connect to %s: no such host.\n",host);
101       return mismatch_error;
102   }
103   if (!(hd=(gras_hostdata_t *)MSG_host_get_data(peer))) {
104       fprintf(stderr,"GRAS: can't connect to %s: no process on this host.\n",host);
105       return mismatch_error;
106   }
107   for (i=0; i<hd->portLen && port != hd->port[i]; i++);
108   if (i == hd->portLen) {
109     fprintf(stderr,"GRAS: can't connect to %s:%d, no process listen on this port.\n",host,port);
110     return mismatch_error;
111   } 
112
113   if (hd->raw[i] && !raw) {
114     fprintf(stderr,"GRAS: can't connect to %s:%d in regular mode, the process listen in raw mode on this port.\n",host,port);
115     return mismatch_error;
116   }
117   if (!hd->raw[i] && raw) {
118     fprintf(stderr,"GRAS: can't connect to %s:%d in raw mode, the process listen in regular mode on this port.\n",host,port);
119     return mismatch_error;
120   }
121     
122
123   /* Create the socket */
124   if (!(*sock=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
125       fprintf(stderr,"GRAS: openClientSocket: out of memory\n");
126       return malloc_error;
127   }    
128
129   (*sock)->server_sock  = 0;
130   (*sock)->raw_sock     = raw;
131   (*sock)->from_PID     = MSG_process_self_PID();
132   (*sock)->to_PID       = hd->proc[ hd->port2chan[i] ];
133   (*sock)->to_host      = peer;
134   (*sock)->to_port      = port;  
135   (*sock)->to_chan      = hd->port2chan[i];
136
137   /*
138   fprintf(stderr,"GRAS: %s (PID %d) connects in %s mode to %s:%d (to_PID=%d).\n",
139           MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
140           raw?"RAW":"regular",host,port,(*sock)->to_PID);
141   */
142   return no_error;
143 }
144
145 xbt_error_t _gras_sock_close(int raw, gras_sock_t *sd) {
146   gras_hostdata_t *hd=(gras_hostdata_t *)MSG_host_get_data(MSG_host_self());
147   int i;
148
149   xbt_assert0(hd,"HostData=NULL !! Please run grasInit on each process\n");
150
151   if (!sd) return no_error;
152   if (raw && !sd->raw_sock) {
153       fprintf(stderr,"GRAS: gras_rawsock_close: Was passed a regular socket. Please use gras_sock_close()\n");
154   }
155   if (!raw && sd->raw_sock) {
156       fprintf(stderr,"GRAS: grasSockClose: Was passed a raw socket. Please use gras_rawsock_close()\n");
157   }
158   if (sd->server_sock) {
159     /* server mode socket. Un register it from 'OS' tables */
160     for (i=0; 
161          i<hd->portLen && sd->to_port != hd->port[i]; 
162          i++);
163
164     if (i==hd->portLen) {
165       fprintf(stderr,"GRAS: closeSocket: The host does not know this server socket.\n");
166     } else {
167       memmove(&(hd->port[i]),      &(hd->port[i+1]),      (hd->portLen -i -1) * sizeof(int));
168       memmove(&(hd->raw[i]),       &(hd->raw[i+1]),       (hd->portLen -i -1) * sizeof(int));
169       memmove(&(hd->port2chan[i]), &(hd->port2chan[i+1]), (hd->portLen -i -1) * sizeof(int));
170       hd->portLen--;
171     }
172   } 
173   free(sd);
174   return no_error;
175 }
176
177 /* **************************************************************************
178  * Creating/Using regular sockets
179  * **************************************************************************/
180 xbt_error_t
181 gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
182                      /* OUT */ gras_sock_t **sock) {
183   return _gras_sock_server_open(startingPort,endingPort,0,0,sock);
184 }
185
186 xbt_error_t
187 gras_sock_client_open(const char *host, short port,
188                      /* OUT */ gras_sock_t **sock) {
189   return _gras_sock_client_open(host,port,0,0,sock);
190 }
191
192 xbt_error_t gras_sock_close(gras_sock_t *sd) {
193   return _gras_sock_close(0,sd);
194 }
195
196 unsigned short
197 gras_sock_get_my_port(gras_sock_t *sd) {
198   if (!sd || !sd->server_sock) return -1;
199   return sd->to_port;
200 }
201
202 unsigned short
203 gras_sock_get_peer_port(gras_sock_t *sd) {
204   if (!sd || sd->server_sock) return -1;
205   return sd->to_port;
206 }
207
208 char *
209 gras_sock_get_peer_name(gras_sock_t *sd) {
210   m_process_t proc;
211
212   if (!sd) return NULL;
213   if ((proc=MSG_process_from_PID(sd->to_PID))) {
214     return (char*) MSG_host_get_name(MSG_process_get_host(proc));
215   } else {
216     fprintf(stderr,"GRAS: try to access hostname of unknown process %d\n",sd->to_PID);
217     return (char*) "(dead or unknown host)";
218   }
219 }
220 /* **************************************************************************
221  * Creating/Using raw sockets
222  * **************************************************************************/
223 xbt_error_t gras_rawsock_server_open(unsigned short startingPort, unsigned short endingPort,
224                                   unsigned int bufSize, gras_rawsock_t **sock) {
225   return _gras_sock_server_open(startingPort,endingPort,1,bufSize,(gras_sock_t**)sock);
226 }
227
228 xbt_error_t gras_rawsock_client_open(const char *host, short port, 
229                                   unsigned int bufSize, gras_rawsock_t **sock) {
230   return _gras_sock_client_open(host,port,1,bufSize,(gras_sock_t**)sock);
231 }
232
233 xbt_error_t gras_rawsock_close(gras_rawsock_t *sd) {
234   return _gras_sock_close(1,(gras_sock_t*)sd);
235 }
236
237 unsigned short
238 gras_rawsock_get_peer_port(gras_rawsock_t *sd) {
239   if (!sd || !sd->server_sock) return -1;
240   return sd->to_port;
241 }
242
243 xbt_error_t
244 gras_rawsock_send(gras_rawsock_t *sock, unsigned int expSize, unsigned int msgSize) {
245   unsigned int bytesTotal;
246   static unsigned int count=0;
247   m_task_t task=NULL;
248   char name[256];
249
250   xbt_assert0(sock->raw_sock,"Asked to send raw data on a regular socket\n");
251
252   for(bytesTotal = 0;
253       bytesTotal < expSize;
254       bytesTotal += msgSize) {
255     
256     sprintf(name,"Raw data[%d]",count++);
257
258     task=MSG_task_create(name,0,((double)msgSize)/(1024.0*1024.0),NULL);
259     /*
260     fprintf(stderr, "%f:%s: gras_rawsock_send(%f %s -> %s) BEGIN\n",
261             gras_time(),
262             MSG_process_get_name(MSG_process_self()),
263             ((double)msgSize)/(1024.0*1024.0),
264             MSG_host_get_name( MSG_host_self()),
265             MSG_host_get_name( sock->to_host));
266     */
267     if (MSG_task_put(task, sock->to_host,sock->to_chan) != MSG_OK) {
268       fprintf(stderr,"GRAS: msgSend: Problem during the MSG_task_put()\n");
269       return unknown_error;
270     }
271     /*fprintf(stderr, "%f:%s: gras_rawsock_send(%f -> %s) END\n",
272             gras_time(),
273             MSG_process_get_name(MSG_process_self()),
274             ((double)msgSize)/(1024.0*1024.0),
275             MSG_host_get_name( sock->to_host));*/
276   }
277   return no_error;
278 }
279
280 xbt_error_t
281 gras_rawsock_recv(gras_rawsock_t *sock, unsigned int expSize, unsigned int msgSize,
282                 unsigned int timeout) {
283   grasProcessData_t *pd=(grasProcessData_t *)MSG_process_get_data(MSG_process_self());
284   unsigned int bytesTotal;
285   m_task_t task=NULL;
286   double startTime;
287
288   xbt_assert0(sock->raw_sock,"Asked to receive raw data on a regular socket\n");
289
290   startTime=gras_time();
291   for(bytesTotal = 0;
292       bytesTotal < expSize;
293       bytesTotal += msgSize) {
294     
295     task=NULL;
296     /*
297     fprintf(stderr, "%f:%s: gras_rawsock_recv() BEGIN\n",
298             gras_time(),
299             MSG_process_get_name(MSG_process_self()));
300     */
301     do {
302       if (MSG_task_Iprobe((m_channel_t) pd->rawChan)) { 
303         if (MSG_task_get(&task, (m_channel_t) pd->rawChan) != MSG_OK) {
304           fprintf(stderr,"GRAS: Error in MSG_task_get()\n");
305           return unknown_error;
306         }
307         if (MSG_task_destroy(task) != MSG_OK) {
308           fprintf(stderr,"GRAS: Error in MSG_task_destroy()\n");
309           return unknown_error;
310         }
311         /*
312         fprintf(stderr, "%f:%s: gras_rawsock_recv() END\n",
313                 gras_time(),
314                 MSG_process_get_name(MSG_process_self()));
315         */
316         break;
317       } else { 
318         MSG_process_sleep(0.0001);
319       }
320     } while (gras_time() - startTime < timeout);
321
322     if (gras_time() - startTime > timeout)
323       return timeout_error;
324   }
325   return no_error;
326 }
327
328
329 /* **************************************************************************
330  * Actually exchanging messages
331  * **************************************************************************/
332
333 xbt_error_t
334 grasMsgRecv(gras_msg_t **msg,
335             double timeOut) {
336
337   double startTime=gras_time();
338   grasProcessData_t *pd=grasProcessDataGet();
339   m_task_t task=NULL;
340
341   do {
342     if (MSG_task_Iprobe((m_channel_t) pd->chan)) {
343       if (MSG_task_get(&task, (m_channel_t) pd->chan) != MSG_OK) {
344         fprintf(stderr,"GRAS: Error in MSG_task_get()\n");
345         return unknown_error;
346       }
347       
348       *msg=(gras_msg_t*)MSG_task_get_data(task);
349       /*
350         { 
351         int i,j;
352         gras_msg_t *__dm_=*msg;
353         
354         fprintf(stderr, "grasMsgRecv(%s) = %d seq (",
355         __dm_->entry->name, __dm_->entry->seqCount );
356         
357         for (i=0; i<__dm_->entry->seqCount; i++) {
358         fprintf(stderr,"%d elem {",__dm_->dataCount[i]);
359         for (j=0; j<__dm_->dataCount[i]; j++) { 
360         fprintf(stderr,"%p; ",__dm_->data[i]);
361         }
362         fprintf(stderr,"},");
363         }
364         fprintf(stderr, ")\n");
365         }
366       */
367
368       if (MSG_task_destroy(task) != MSG_OK) {
369         fprintf(stderr,"GRAS: Error in MSG_task_destroy()\n");
370         return unknown_error;
371       }
372       return no_error;
373
374     } else {
375       MSG_process_sleep(1);
376     }
377   } while (gras_time()-startTime < timeOut || MSG_task_Iprobe((m_channel_t) pd->chan));
378   return timeout_error;
379 }
380
381 xbt_error_t
382 gras_msg_send(gras_sock_t *sd,
383             gras_msg_t *_msg,
384             e_xbt_free_directive_t freeDirective) {
385   
386   grasProcessData_t *pd=grasProcessDataGet();
387   m_task_t task;
388   static int count=0;
389   char name[256];
390   gras_msg_t *msg;
391   gras_sock_t *answer;
392
393   /* arg validity checks */
394   xbt_assert0(msg,"Trying to send NULL message");
395   xbt_assert0(sd ,"Trying to send over a NULL socket");
396   
397   if (!(answer=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
398     RAISE_MALLOC;
399   }
400   answer->server_sock=0;
401   answer->raw_sock   =0;
402   answer->from_PID   = sd->to_PID;
403   answer->to_PID     = MSG_process_self_PID();
404   answer->to_host    = MSG_host_self();
405   answer->to_port    = 0;
406   answer->to_chan    = pd->chan;
407     
408   sprintf(name,"%s[%d]",_msg->entry->name,count++);
409   /* if freeDirective == free_never, we have to build a copy of the message */
410   if (freeDirective == free_never) {
411     msg=gras_msg_copy(_msg);
412   } else {
413     msg=_msg;
414   }
415   msg->sock = answer;
416
417   /*
418   fprintf(stderr,"Send %s with answer(%p)=",msg->entry->name,msg->sock);
419   fprintf(stderr,"(server=%d,raw=%d,fromPID=%d,toPID=%d,toHost=%p,toPort=%d,toChan=%d)\n",
420           msg->sock->server_sock,msg->sock->raw_sock,msg->sock->from_PID,
421           msg->sock->to_PID,msg->sock->to_host,msg->sock->to_port,msg->sock->to_chan);
422   fprintf(stderr,"Send over %p=(server=%d,raw=%d,fromPID=%d,toPID=%d,toHost=%p,toPort=%d,toChan=%d)\n",
423           sd,sd->server_sock,sd->raw_sock,sd->from_PID,sd->to_PID,sd->to_host,sd->to_port,sd->to_chan);
424   */
425
426   /*
427   { 
428     int i,j;
429     gras_msg_t *__dm_=msg;
430
431     fprintf(stderr, "gras_msg_send(%s) = %d seq (",
432             __dm_->entry->name, __dm_->entry->seqCount );
433
434     for (i=0; i<__dm_->entry->seqCount; i++) {
435       fprintf(stderr,"%d elem {",__dm_->dataCount[i]);
436       for (j=0; j<__dm_->dataCount[i]; j++) { 
437         fprintf(stderr,"%p; ",__dm_->data[i]);
438       }
439       fprintf(stderr,"},");
440     }
441     fprintf(stderr, ")\n");
442   }
443   */
444
445   /* Send it */
446   task=MSG_task_create(name,0,((double)msg->header->dataSize)/(1024.0*1024.0),msg);
447   if (MSG_task_put(task, sd->to_host,sd->to_chan) != MSG_OK) {
448     fprintf(stderr,"GRAS: msgSend: Problem during the MSG_task_put()\n");
449     return unknown_error;
450   }
451   return no_error;
452 }
453
454 gras_sock_t *gras_sock_new(void) {
455   return malloc(sizeof(gras_sock_t));
456 }
457
458 void grasSockFree(gras_sock_t *s) {
459   if (s) free (s);
460 }
461
462 /* **************************************************************************
463  * Process data
464  * **************************************************************************/
465 grasProcessData_t *grasProcessDataGet() {
466   return (grasProcessData_t *)MSG_process_get_data(MSG_process_self());
467 }
468
469 /* **************************************************************************
470  * Wrappers over OS functions
471  * **************************************************************************/
472 double gras_time() {
473   return MSG_getClock();
474 }
475
476 void gras_sleep(unsigned long sec, unsigned long usec) {
477   MSG_process_sleep((double)sec + ((double)usec)/1000000);
478 }
479