Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
6cb3f1e5bd01093657753f8bffe5a792aa3b4276
[simgrid.git] / src / gras / Transport / transport.c
1 /* $Id$ */
2
3 /* transport - low level communication                                      */
4
5 /* Copyright (c) 2004 Martin Quinson. All rights reserved.                  */
6
7 /* This program is free software; you can redistribute it and/or modify it
8  * under the terms of the license (GNU LGPL) which comes with this package. */
9
10 #include "portable.h"
11 #include "gras/Transport/transport_private.h"
12
13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(transport,gras,"Conveying bytes over the network");
14 XBT_LOG_NEW_SUBCATEGORY(raw_trp,transport,"Conveying bytes over the network without formating");
15 static short int _gras_trp_started = 0;
16
17 static xbt_dict_t _gras_trp_plugins;      /* All registered plugins */
18 static void gras_trp_plugin_free(void *p); /* free one of the plugins */
19
20 static void
21 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
22   xbt_error_t errcode;
23
24   gras_trp_plugin_t *plug = xbt_new0(gras_trp_plugin_t, 1);
25   
26   DEBUG1("Create plugin %s",name);
27
28   plug->name=xbt_strdup(name);
29
30   errcode = setup(plug);
31   switch (errcode) {
32   case mismatch_error:
33     /* SG plugin return mismatch when in RL mode (and vice versa) */
34     free(plug->name);
35     free(plug);
36     break;
37
38   case no_error:
39     xbt_dict_set(_gras_trp_plugins,
40                   name, plug, gras_trp_plugin_free);
41     break;
42
43   default:
44     DIE_IMPOSSIBLE;
45   }
46
47 }
48
49 void gras_trp_init(void){
50   if (!_gras_trp_started) {
51      /* make room for all plugins */
52      _gras_trp_plugins=xbt_dict_new();
53
54 #ifdef HAVE_WINSOCK2_H
55      /* initialize the windows mechanism */
56      {  
57         WORD wVersionRequested;
58         WSADATA wsaData;
59         
60         wVersionRequested = MAKEWORD( 2, 0 );
61         xbt_assert0(WSAStartup( wVersionRequested, &wsaData ) == 0,
62                     "Cannot find a usable WinSock DLL");
63         
64         /* Confirm that the WinSock DLL supports 2.0.*/
65         /* Note that if the DLL supports versions greater    */
66         /* than 2.0 in addition to 2.0, it will still return */
67         /* 2.0 in wVersion since that is the version we      */
68         /* requested.                                        */
69         
70         xbt_assert0(LOBYTE( wsaData.wVersion ) == 2 &&
71                     HIBYTE( wsaData.wVersion ) == 0,
72                     "Cannot find a usable WinSock DLL");
73         INFO0("Found and initialized winsock2");
74      }       /* The WinSock DLL is acceptable. Proceed. */
75 #elif HAVE_WINSOCK_H
76      {       WSADATA wsaData;
77         xbt_assert0(WSAStartup( 0x0101, &wsaData ) == 0,
78                     "Cannot find a usable WinSock DLL");
79         INFO0("Found and initialized winsock");
80      }
81 #endif
82    
83      /* Add plugins */
84      gras_trp_plugin_new("tcp", gras_trp_tcp_setup);
85      gras_trp_plugin_new("file",gras_trp_file_setup);
86      gras_trp_plugin_new("sg",gras_trp_sg_setup);
87
88      /* buf is composed, so it must come after the others */
89      gras_trp_plugin_new("buf", gras_trp_buf_setup);
90   }
91    
92   _gras_trp_started++;
93 }
94
95 void
96 gras_trp_exit(void){
97    if (_gras_trp_started == 0) {
98       return;
99    }
100    
101    if ( --_gras_trp_started == 0 ) {
102 #ifdef HAVE_WINSOCK_H
103       if ( WSACleanup() == SOCKET_ERROR ) {
104          if ( WSAGetLastError() == WSAEINPROGRESS ) {
105             WSACancelBlockingCall();
106             WSACleanup();
107          }
108         }
109 #endif
110
111       xbt_dict_free(&_gras_trp_plugins);
112    }
113 }
114
115
116 void gras_trp_plugin_free(void *p) {
117   gras_trp_plugin_t *plug = p;
118
119   if (plug) {
120     if (plug->exit) {
121       plug->exit(plug);
122     } else if (plug->data) {
123       DEBUG1("Plugin %s lacks exit(). Free data anyway.",plug->name);
124       free(plug->data);
125     }
126
127     free(plug->name);
128     free(plug);
129   }
130 }
131
132
133 /**
134  * gras_trp_socket_new:
135  *
136  * Malloc a new socket, and initialize it with defaults
137  */
138 void gras_trp_socket_new(int incoming,
139                          gras_socket_t *dst) {
140
141   gras_socket_t sock=xbt_new0(s_gras_socket_t,1);
142
143   DEBUG1("Create a new socket (%p)", (void*)sock);
144
145   sock->plugin = NULL;
146   sock->sd     = -1;
147   sock->data   = NULL;
148
149   sock->incoming  = incoming ? 1:0;
150   sock->outgoing  = incoming ? 0:1;
151   sock->accepting = incoming ? 1:0;
152
153   sock->port      = -1;
154   sock->peer_port = -1;
155   sock->peer_name = NULL;
156   sock->raw = 0;
157
158   *dst = sock;
159
160   xbt_dynar_push(gras_socketset_get(),dst);
161   XBT_OUT;
162 }
163
164
165 /**
166  * gras_socket_server_ext:
167  *
168  * Opens a server socket and make it ready to be listened to.
169  * In real life, you'll get a TCP socket.
170  */
171 xbt_error_t
172 gras_socket_server_ext(unsigned short port,
173                        
174                        unsigned long int bufSize,
175                        int raw,
176                        
177                        /* OUT */ gras_socket_t *dst) {
178  
179   xbt_error_t errcode;
180   gras_trp_plugin_t *trp;
181   gras_socket_t sock;
182
183   *dst = NULL;
184
185   DEBUG2("Create a server socket from plugin %s on port %d",
186          gras_if_RL() ? "tcp" : "sg",
187          port);
188   TRY(gras_trp_plugin_get_by_name("buf",&trp));
189
190   /* defaults settings */
191   gras_trp_socket_new(1,&sock);
192   sock->plugin= trp;
193   sock->port=port;
194   sock->bufSize = bufSize;
195   sock->raw = raw;
196
197   /* Call plugin socket creation function */
198   DEBUG1("Prepare socket with plugin (fct=%p)",trp->socket_server);
199   errcode = trp->socket_server(trp, sock);
200   DEBUG3("in=%c out=%c accept=%c",
201          sock->incoming?'y':'n', 
202          sock->outgoing?'y':'n',
203          sock->accepting?'y':'n');
204
205   if (errcode != no_error) {
206     free(sock);
207     return errcode;
208   }
209
210   *dst = sock;
211
212   return no_error;
213 }
214    
215 /**
216  * gras_socket_client_ext:
217  *
218  * Opens a client socket to a remote host.
219  * In real life, you'll get a TCP socket.
220  */
221 xbt_error_t
222 gras_socket_client_ext(const char *host,
223                        unsigned short port,
224                        
225                        unsigned long int bufSize,
226                        int raw,
227                        
228                        /* OUT */ gras_socket_t *dst) {
229  
230   xbt_error_t errcode;
231   gras_trp_plugin_t *trp;
232   gras_socket_t sock;
233
234   *dst = NULL;
235
236   TRY(gras_trp_plugin_get_by_name("buf",&trp));
237
238   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
239   /* defaults settings */
240   gras_trp_socket_new(0,&sock);
241   sock->plugin= trp;
242   sock->peer_port = port;
243   sock->peer_name = (char*)strdup(host?host:"localhost");
244   sock->bufSize = bufSize;
245   sock->raw = raw;
246
247   /* plugin-specific */
248   errcode= (*trp->socket_client)(trp, sock);
249   DEBUG3("in=%c out=%c accept=%c",
250          sock->incoming?'y':'n', 
251          sock->outgoing?'y':'n',
252          sock->accepting?'y':'n');
253
254   if (errcode != no_error) {
255     free(sock);
256     return errcode;
257   }
258
259   *dst = sock;
260
261   return no_error;
262 }
263
264 /**
265  * gras_socket_server:
266  *
267  * Opens a server socket and make it ready to be listened to.
268  * In real life, you'll get a TCP socket.
269  */
270 xbt_error_t
271 gras_socket_server(unsigned short port,
272                    /* OUT */ gras_socket_t *dst) {
273    return gras_socket_server_ext(port,32,0,dst);
274 }
275
276 /**
277  * gras_socket_client:
278  *
279  * Opens a client socket to a remote host.
280  * In real life, you'll get a TCP socket.
281  */
282 xbt_error_t
283 gras_socket_client(const char *host,
284                    unsigned short port,
285                    /* OUT */ gras_socket_t *dst) {
286    return gras_socket_client_ext(host,port,32,0,dst);
287 }
288
289
290 void gras_socket_close(gras_socket_t sock) {
291   xbt_dynar_t sockets = gras_socketset_get();
292   gras_socket_t sock_iter;
293   int cursor;
294
295   XBT_IN;
296   /* FIXME: Issue an event when the socket is closed */
297   if (sock) {
298     xbt_dynar_foreach(sockets,cursor,sock_iter) {
299       if (sock == sock_iter) {
300         xbt_dynar_cursor_rm(sockets,&cursor);
301         if (sock->plugin->socket_close) 
302           (* sock->plugin->socket_close)(sock);
303
304         /* free the memory */
305         if (sock->peer_name)
306           free(sock->peer_name);
307         free(sock);
308         XBT_OUT;
309         return;
310       }
311     }
312     WARN0("Ignoring request to free an unknown socket");
313   }
314   XBT_OUT;
315 }
316
317 /**
318  * gras_trp_chunk_send:
319  *
320  * Send a bunch of bytes from on socket
321  */
322 xbt_error_t
323 gras_trp_chunk_send(gras_socket_t sd,
324                     char *data,
325                     long int size) {
326   xbt_assert1(sd->outgoing,
327                "Socket not suited for data send (outgoing=%c)",
328                sd->outgoing?'y':'n');
329   xbt_assert1(sd->plugin->chunk_send,
330                "No function chunk_send on transport plugin %s",
331                sd->plugin->name);
332   return (*sd->plugin->chunk_send)(sd,data,size);
333 }
334 /**
335  * gras_trp_chunk_recv:
336  *
337  * Receive a bunch of bytes from a socket
338  */
339 xbt_error_t 
340 gras_trp_chunk_recv(gras_socket_t sd,
341                     char *data,
342                     long int size) {
343   xbt_assert0(sd->incoming,
344                "Socket not suited for data receive");
345   xbt_assert1(sd->plugin->chunk_recv,
346                "No function chunk_recv on transport plugin %s",
347                sd->plugin->name);
348   return (sd->plugin->chunk_recv)(sd,data,size);
349 }
350
351 /**
352  * gras_trp_flush:
353  *
354  * Make sure all pending communications are done
355  */
356 xbt_error_t 
357 gras_trp_flush(gras_socket_t sd) {
358   return (sd->plugin->flush)(sd);
359 }
360
361 xbt_error_t
362 gras_trp_plugin_get_by_name(const char *name,
363                             gras_trp_plugin_t **dst){
364
365   return xbt_dict_get(_gras_trp_plugins,name,(void**)dst);
366 }
367
368 int   gras_socket_my_port  (gras_socket_t sock) {
369   return sock->port;
370 }
371 int   gras_socket_peer_port(gras_socket_t sock) {
372   return sock->peer_port;
373 }
374 char *gras_socket_peer_name(gras_socket_t sock) {
375   return sock->peer_name;
376 }
377
378 xbt_error_t gras_socket_raw_send(gras_socket_t peer, 
379                                   unsigned int timeout,
380                                   unsigned long int exp_size, 
381                                   unsigned long int msg_size) {
382   xbt_error_t errcode;
383   char *chunk = xbt_malloc(msg_size);
384   int exp_sofar;
385    
386   xbt_assert0(peer->raw,"Asked to send raw data on a regular socket");
387   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
388      CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
389              exp_sofar,exp_size,msg_size,
390              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
391      TRY(gras_trp_chunk_send(peer,chunk,msg_size));
392   }
393   CDEBUG5(raw_trp,"Sent %d of %lu (msg_size=%ld) to %s:%d",
394           exp_sofar,exp_size,msg_size,
395           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
396              
397   free(chunk);
398   return no_error;/* gras_socket_raw_exchange(peer,1,timeout,expSize,msgSize);    */
399 }
400
401 xbt_error_t gras_socket_raw_recv(gras_socket_t peer, 
402                                   unsigned int timeout,
403                                   unsigned long int exp_size, 
404                                   unsigned long int msg_size){
405   
406   xbt_error_t errcode;
407   char *chunk = xbt_malloc(msg_size);
408   int exp_sofar;
409
410   xbt_assert0(peer->raw,"Asked to recveive raw data on a regular socket\n");
411   for (exp_sofar=0; exp_sofar < exp_size; exp_sofar += msg_size) {
412      CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
413              exp_sofar,exp_size,msg_size,
414              gras_socket_peer_name(peer), gras_socket_peer_port(peer));
415      TRY(gras_trp_chunk_recv(peer,chunk,msg_size));
416   }
417   CDEBUG5(raw_trp,"Recvd %d of %lu (msg_size=%ld) from %s:%d",
418           exp_sofar,exp_size,msg_size,
419           gras_socket_peer_name(peer), gras_socket_peer_port(peer));
420
421   free(chunk);
422   return no_error;/* gras_socket_raw_exchange(peer,0,timeout,expSize,msgSize);    */
423 }
424
425 /*
426  * Creating procdata for this module
427  */
428 static void *gras_trp_procdata_new() {
429    gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t,1);
430    
431    res->sockets   = xbt_dynar_new(sizeof(gras_socket_t*), NULL);
432    
433    return (void*)res;
434 }
435
436 /*
437  * Freeing procdata for this module
438  */
439 static void gras_trp_procdata_free(void *data) {
440    gras_trp_procdata_t res = (gras_trp_procdata_t)data;
441    
442    xbt_dynar_free(&( res->sockets ));
443 }
444
445 /*
446  * Module registration
447  */
448 void gras_trp_register() {
449    gras_procdata_add("gras_trp",gras_trp_procdata_new, gras_trp_procdata_free);
450 }
451
452
453 xbt_dynar_t 
454 gras_socketset_get(void) {
455    /* FIXME: KILLME */
456    return ((gras_trp_procdata_t) gras_libdata_get("gras_trp"))->sockets;
457 }