Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Introduce the flush, use the buffered transport
[simgrid.git] / src / gras / Transport / transport.c
1 /* $Id$ */
2
3 /* transport - low level communication                                      */
4
5 /* Authors: Martin Quinson                                                  */
6 /* Copyright (C) 2004 Martin Quinson.                                       */
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 <time.h>       /* time() */
12
13 #include "Transport/transport_private.h"
14
15 GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(transport,GRAS);
16
17 static gras_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 gras_trp_socket_free(void *s); /* free one socket */
21
22 gras_error_t
23 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup);
24
25 gras_error_t
26 gras_trp_plugin_new(const char *name, gras_trp_setup_t setup) {
27   gras_error_t errcode;
28
29   gras_trp_plugin_t *plug = malloc(sizeof(gras_trp_plugin_t));
30   
31   DEBUG1("Create plugin %s",name);
32
33   if (!plug) 
34     RAISE_MALLOC;
35
36   memset(plug,0,sizeof(gras_trp_plugin_t));
37
38   plug->name=strdup(name);
39   if (!plug->name)
40     RAISE_MALLOC;
41
42   errcode = setup(plug);
43   switch (errcode) {
44   case mismatch_error:
45     /* SG plugin return mismatch when in RL mode (and vice versa) */
46     free(plug->name);
47     free(plug);
48     break;
49
50   case no_error:
51     TRY(gras_dict_set(_gras_trp_plugins, 
52                       name, plug, gras_trp_plugin_free));
53     break;
54
55   default:
56     free(plug);
57     return errcode;
58   }
59   return no_error;
60 }
61
62 gras_error_t 
63 gras_trp_init(void){
64   gras_error_t errcode;
65   
66   /* make room for all plugins */
67   TRY(gras_dict_new(&_gras_trp_plugins));
68
69   /* Add them */
70   TRY(gras_trp_plugin_new("tcp", gras_trp_tcp_setup));
71   TRY(gras_trp_plugin_new("file",gras_trp_file_setup));
72   TRY(gras_trp_plugin_new("sg",gras_trp_sg_setup));
73
74   /* buf is composed, so it must come after the others */
75   TRY(gras_trp_plugin_new("buf", gras_trp_buf_setup));
76
77   return no_error;
78 }
79
80 void
81 gras_trp_exit(void){
82   gras_dict_free(&_gras_trp_plugins);
83 }
84
85
86 void gras_trp_plugin_free(void *p) {
87   gras_trp_plugin_t *plug = p;
88
89   if (plug) {
90     if (plug->exit) {
91       plug->exit(plug);
92     } else if (plug->data) {
93       DEBUG1("Plugin %s lacks exit(). Free data anyway.",plug->name);
94       free(plug->data);
95     }
96
97     free(plug->name);
98     free(plug);
99   }
100 }
101
102
103 /**
104  * gras_trp_socket_new:
105  *
106  * Malloc a new socket, and initialize it with defaults
107  */
108 gras_error_t gras_trp_socket_new(int incoming,
109                                  gras_socket_t **dst) {
110
111   gras_socket_t *sock;
112
113   if (! (sock=malloc(sizeof(gras_socket_t))) )
114     RAISE_MALLOC;
115   DEBUG1("Create a new socket (%p)", sock);
116
117   sock->plugin = NULL;
118   sock->sd     = -1;
119   sock->data   = NULL;
120
121   sock->incoming  = incoming ? 1:0;
122   sock->outgoing  = incoming ? 0:1;
123   sock->accepting = incoming ? 1:0;
124
125   sock->port      = -1;
126   sock->peer_port = -1;
127   sock->peer_name = NULL;
128   sock->raw = 0;
129
130   *dst = sock;
131
132   return gras_dynar_push(gras_socketset_get(),dst);
133 }
134
135
136 /**
137  * gras_socket_server:
138  *
139  * Opens a server socket and make it ready to be listened to.
140  * In real life, you'll get a TCP socket.
141  */
142 gras_error_t
143 gras_socket_server(unsigned short port,
144                    /* OUT */ gras_socket_t **dst) {
145  
146   gras_error_t errcode;
147   gras_trp_plugin_t *trp;
148   gras_socket_t *sock;
149
150   *dst = NULL;
151
152   DEBUG1("Create a server socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
153   TRY(gras_trp_plugin_get_by_name("buf",&trp));
154
155   /* defaults settings */
156   TRY(gras_trp_socket_new(1,&sock));
157   sock->plugin= trp;
158   sock->port=port;
159
160   /* Call plugin socket creation function */
161   errcode = trp->socket_server(trp, port, sock);
162   DEBUG3("in=%c out=%c accept=%c",
163          sock->incoming?'y':'n', 
164          sock->outgoing?'y':'n',
165          sock->accepting?'y':'n');
166
167   if (errcode != no_error) {
168     free(sock);
169     return errcode;
170   }
171
172   *dst = sock;
173
174   return no_error;
175 }
176
177 /**
178  * gras_socket_client:
179  *
180  * Opens a client socket to a remote host.
181  * In real life, you'll get a TCP socket.
182  */
183 gras_error_t
184 gras_socket_client(const char *host,
185                    unsigned short port,
186                    /* OUT */ gras_socket_t **dst) {
187  
188   gras_error_t errcode;
189   gras_trp_plugin_t *trp;
190   gras_socket_t *sock;
191
192   *dst = NULL;
193
194   TRY(gras_trp_plugin_get_by_name("buf",&trp));
195
196   DEBUG1("Create a client socket from plugin %s",gras_if_RL() ? "tcp" : "sg");
197   /* defaults settings */
198   TRY(gras_trp_socket_new(0,&sock));
199   sock->plugin= trp;
200   sock->peer_port = port;
201   sock->peer_name = strdup(host?host:"localhost");
202
203   /* plugin-specific */
204   errcode= (*trp->socket_client)(trp, 
205                                  host ? host : "localhost", port,
206                                  sock);
207   DEBUG3("in=%c out=%c accept=%c",
208          sock->incoming?'y':'n', 
209          sock->outgoing?'y':'n',
210          sock->accepting?'y':'n');
211
212   if (errcode != no_error) {
213     free(sock);
214     return errcode;
215   }
216
217   *dst = sock;
218
219   return no_error;
220 }
221
222 void gras_socket_close(gras_socket_t *sock) {
223   gras_dynar_t *sockets = gras_socketset_get();
224   gras_socket_t *sock_iter;
225   int cursor;
226
227   /* FIXME: Issue an event when the socket is closed */
228   if (sock) {
229     gras_dynar_foreach(sockets,cursor,sock_iter) {
230       if (sock == sock_iter) {
231         gras_dynar_cursor_rm(sockets,&cursor);
232         if ( sock->plugin->socket_close) 
233           (* sock->plugin->socket_close)(sock);
234
235         /* free the memory */
236         if (sock->peer_name)
237           free(sock->peer_name);
238         free(sock);
239         return;
240       }
241     }
242     WARN0("Ignoring request to free an unknown socket");
243   }
244 }
245
246 /**
247  * gras_trp_chunk_send:
248  *
249  * Send a bunch of bytes from on socket
250  */
251 gras_error_t
252 gras_trp_chunk_send(gras_socket_t *sd,
253                     char *data,
254                     long int size) {
255   gras_assert1(sd->outgoing,
256                "Socket not suited for data send (outgoing=%c)",
257                sd->outgoing?'y':'n');
258   gras_assert1(sd->plugin->chunk_send,
259                "No function chunk_send on transport plugin %s",
260                sd->plugin->name);
261   return (*sd->plugin->chunk_send)(sd,data,size);
262 }
263 /**
264  * gras_trp_chunk_recv:
265  *
266  * Receive a bunch of bytes from a socket
267  */
268 gras_error_t 
269 gras_trp_chunk_recv(gras_socket_t *sd,
270                     char *data,
271                     long int size) {
272   gras_assert0(sd->incoming,
273                "Socket not suited for data receive");
274   gras_assert1(sd->plugin->chunk_recv,
275                "No function chunk_recv on transport plugin %s",
276                sd->plugin->name);
277   return (sd->plugin->chunk_recv)(sd,data,size);
278 }
279
280 /**
281  * gras_trp_flush:
282  *
283  * Make sure all pending communications are done
284  */
285 gras_error_t 
286 gras_trp_flush(gras_socket_t *sd) {
287   return (sd->plugin->flush)(sd);
288 }
289
290 gras_error_t
291 gras_trp_plugin_get_by_name(const char *name,
292                             gras_trp_plugin_t **dst){
293
294   return gras_dict_get(_gras_trp_plugins,name,(void**)dst);
295 }
296
297 int   gras_socket_my_port  (gras_socket_t *sock) {
298   return sock->port;
299 }
300 int   gras_socket_peer_port(gras_socket_t *sock) {
301   return sock->peer_port;
302 }
303 char *gras_socket_peer_name(gras_socket_t *sock) {
304   return sock->peer_name;
305 }