Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
46638a3a6a40039e83d1997b549648bf6ccc45c8
[simgrid.git] / src / gras / Transport / transport_plugin_tcp.c
1 /* $Id$ */
2
3 /* tcp trp (transport) - send/receive a bunch of bytes from a tcp socket    */
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 <unistd.h>       /* close() pipe() read() write() */
12 #include <signal.h>       /* close() pipe() read() write() */
13 #include <netinet/in.h>   /* sometimes required for #include <arpa/inet.h> */
14 #include <netinet/tcp.h>  /* TCP_NODELAY */
15 #include <arpa/inet.h>    /* inet_ntoa() */
16 #include <netdb.h>        /* getprotobyname() */
17 #include <sys/time.h>     /* struct timeval */
18 #include <errno.h>        /* errno */
19 #include <sys/wait.h>     /* waitpid() */
20 #include <sys/socket.h>   /* getpeername() socket() */
21 #include <stdlib.h>
22 #include <string.h>       /* memset */
23
24 #include "gras_private.h"
25 #include "transport_private.h"
26
27 GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(trp_tcp,transport);
28
29 /***
30  *** Prototypes 
31  ***/
32 gras_error_t gras_trp_tcp_socket_client(gras_trp_plugin_t *self,
33                                         const char *host,
34                                         unsigned short port,
35                                         /* OUT */ gras_socket_t *sock);
36 gras_error_t gras_trp_tcp_socket_server(gras_trp_plugin_t *self,
37                                         unsigned short port,
38                                         /* OUT */ gras_socket_t *sock);
39 gras_error_t gras_trp_tcp_socket_accept(gras_socket_t  *sock,
40                                         gras_socket_t **dst);
41
42 void         gras_trp_tcp_socket_close(gras_socket_t *sd);
43   
44 gras_error_t gras_trp_tcp_chunk_send(gras_socket_t *sd,
45                                      char *data,
46                                      long int size);
47
48 gras_error_t gras_trp_tcp_chunk_recv(gras_socket_t *sd,
49                                      char *data,
50                                      long int size);
51
52 void gras_trp_tcp_exit(gras_trp_plugin_t *plug);
53
54
55 static int TcpProtoNumber(void);
56 /***
57  *** Specific plugin part
58  ***/
59
60 typedef struct {
61   fd_set msg_socks;
62   fd_set raw_socks;
63 } gras_trp_tcp_plug_data_t;
64
65 /***
66  *** Specific socket part
67  ***/
68
69 typedef struct {
70   int buffsize;
71 } gras_trp_tcp_sock_data_t;
72
73
74 /***
75  *** Code
76  ***/
77 gras_error_t
78 gras_trp_tcp_setup(gras_trp_plugin_t *plug) {
79
80   gras_trp_tcp_plug_data_t *data = malloc(sizeof(gras_trp_tcp_plug_data_t));
81   if (!data)
82     RAISE_MALLOC;
83
84   FD_ZERO(&(data->msg_socks));
85   FD_ZERO(&(data->raw_socks));
86
87   plug->socket_client = gras_trp_tcp_socket_client;
88   plug->socket_server = gras_trp_tcp_socket_server;
89   plug->socket_accept = gras_trp_tcp_socket_accept;
90   plug->socket_close  = gras_trp_tcp_socket_close;
91
92   plug->chunk_send    = gras_trp_tcp_chunk_send;
93   plug->chunk_recv    = gras_trp_tcp_chunk_recv;
94
95   plug->data = (void*)data;
96   plug->exit = gras_trp_tcp_exit;
97
98   return no_error;
99 }
100
101 void gras_trp_tcp_exit(gras_trp_plugin_t *plug) {
102   DEBUG1("Exit plugin TCP (free %p)", plug->data);
103   free(plug->data);
104 }
105
106 gras_error_t gras_trp_tcp_socket_client(gras_trp_plugin_t *self,
107                                         const char *host,
108                                         unsigned short port,
109                                         /* OUT */ gras_socket_t *sock){
110   
111   struct sockaddr_in addr;
112   struct hostent *he;
113   struct in_addr *haddr;
114
115   sock->incoming = 1; /* TCP sockets are duplex'ed */
116
117   sock->sd = socket (AF_INET, SOCK_STREAM, 0);
118   
119   if (sock->sd < 0) {
120     RAISE1(system_error,
121            "Failed to create socket: %s",
122            strerror (errno));
123   }
124   
125   he = gethostbyname (host);
126   if (he == NULL) {
127     RAISE2(system_error,
128            "Failed to lookup hostname %s: %s",
129            host, strerror (errno));
130   }
131   
132   haddr = ((struct in_addr *) (he->h_addr_list)[0]);
133   
134   memset(&addr, 0, sizeof(struct sockaddr_in));
135   memcpy (&addr.sin_addr, haddr, sizeof(struct in_addr));
136   addr.sin_family = AF_INET;
137   addr.sin_port = htons (port);
138
139   if (connect (sock->sd, (struct sockaddr*) &addr, sizeof (addr)) < 0) {
140     close(sock->sd);
141     RAISE3(system_error,
142            "Failed to connect socket to %s:%d (%s)",
143            host, port, strerror (errno));
144   }
145   
146   return no_error;
147 }
148
149 /**
150  * gras_trp_tcp_socket_server:
151  *
152  * Open a socket used to receive messages.
153  */
154 gras_error_t gras_trp_tcp_socket_server(gras_trp_plugin_t *self,
155                                         unsigned short port,
156                                         /* OUT */ gras_socket_t *sock){
157 //  int size = bufSize * 1024;
158   int on = 1;
159   struct sockaddr_in server;
160
161   gras_trp_tcp_plug_data_t *tcp=(gras_trp_tcp_plug_data_t*)self->data;
162  
163   sock->outgoing  = 1; /* TCP => duplex mode */
164
165   server.sin_port = htons((u_short)port);
166   server.sin_addr.s_addr = INADDR_ANY;
167   server.sin_family = AF_INET;
168   if((sock->sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
169     RAISE1(system_error,"socket allocation failed: %s", strerror(errno));
170   }
171
172   (void)setsockopt(sock->sd, SOL_SOCKET, SO_REUSEADDR, 
173                    (char *)&on, sizeof(on));
174    /*
175   (void)setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *)&size, sizeof(size));
176   (void)setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *)&size, sizeof(size));
177     */
178   if (bind(sock->sd, (struct sockaddr *)&server, sizeof(server)) == -1) {
179     close(sock->sd);
180     RAISE2(system_error,"Cannot bind to port %d: %s",port, strerror(errno));
181   }
182
183   if (listen(sock->sd, 5) < 0) {
184     close(sock->sd);
185     RAISE2(system_error,"Cannot listen to port %d: %s",port,strerror(errno));
186   }
187
188   if (sock->raw)
189     FD_SET(sock->sd, &(tcp->raw_socks));
190   else
191     FD_SET(sock->sd, &(tcp->msg_socks));
192
193   DEBUG2("Openned a server socket on port %d (sock %d)",port,sock->sd);
194   
195   return no_error;
196 }
197
198 gras_error_t
199 gras_trp_tcp_socket_accept(gras_socket_t  *sock,
200                            gras_socket_t **dst) {
201   gras_socket_t *res;
202   gras_error_t errcode;
203   
204   struct sockaddr_in peer_in;
205   socklen_t peer_in_len = sizeof(peer_in);
206
207   int sd;
208   int tmp_errno;
209                         
210   TRY(gras_trp_socket_new(1,&res));
211
212   sd = accept(sock->sd, (struct sockaddr *)&peer_in, &peer_in_len);
213   tmp_errno = errno;
214
215   if(sd == -1) {
216     gras_socket_close(sock);
217     RAISE1(system_error,
218            "Accept failed (%s). Droping server socket.", strerror(tmp_errno));
219   } else {
220     int i = 1;
221     socklen_t s = sizeof(int);
222   
223     if (setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char *)&i, s) 
224         || setsockopt(sd, TcpProtoNumber(), TCP_NODELAY, (char *)&i, s)) {
225       WARN0("setsockopt failed, cannot condition the accepted socket");
226     }
227  
228     /* FIXME: bufSize removed until we can have optionsets 
229        i = ((gras_trp_tcp_sock_specific_t*)sock->specific)->buffsize;
230        if (setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *)&i, s)
231        || setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *)&i, s)) {
232        WARNING0("setsockopt failed, cannot set buffsize");      
233        }
234     */
235      
236     res->plugin    = sock->plugin;
237     res->incoming  = sock->incoming;
238     res->outgoing  = sock->outgoing;
239     res->accepting = 0;
240     res->sd        = sd;
241     res->port      = -1;
242     res->peer_port = peer_in.sin_port;
243
244     /* FIXME: Lock to protect inet_ntoa */
245     if (((struct sockaddr *)&peer_in)->sa_family != AF_INET) {
246       res->peer_name = strdup("unknown");
247     } else {
248       struct in_addr addrAsInAddr;
249       char *tmp;
250  
251       addrAsInAddr.s_addr = peer_in.sin_addr.s_addr;
252       
253       tmp = inet_ntoa(addrAsInAddr);
254       if (tmp != NULL) {
255         res->peer_name = strdup(tmp);
256       } else {
257         res->peer_name = strdup("unknown");
258       }
259     }
260
261     VERB3("accepted socket %d to %s:%d", sd, res->peer_name,res->peer_port);
262     
263     *dst = res;
264
265     return no_error;
266   }
267 }
268
269 void gras_trp_tcp_socket_close(gras_socket_t *sock){
270   gras_trp_tcp_plug_data_t *tcp;
271   
272   if (!sock) return; /* close only once */
273   tcp=sock->plugin->data;
274
275   DEBUG1("close tcp connection %d", sock->sd);
276
277   /* FIXME: no pipe in GRAS so far  
278   if(!FD_ISSET(sd, &connectedPipes)) {
279     if(shutdown(sd, 2) < 0) {
280       GetNWSLock(&lock);
281       tmp_errno = errno;
282       ReleaseNWSLock(&lock);
283       
284       / * The other side may have beaten us to the reset. * /
285       if ((tmp_errno!=ENOTCONN) && (tmp_errno!=ECONNRESET)) {
286         WARN1("CloseSocket: shutdown error %d\n", tmp_errno);
287       }
288     }
289   } */
290
291   /* forget about the socket */
292   if (sock->raw)
293     FD_CLR(sock->sd, &(tcp->raw_socks));
294   else
295     FD_CLR(sock->sd, &(tcp->msg_socks));
296
297   /* close the socket */
298   if(close(sock->sd) < 0) {
299     WARN3("error while closing tcp socket %d: %d (%s)\n", 
300              sock->sd, errno, strerror(errno));
301   }
302 }
303
304 /**
305  * gras_trp_tcp_chunk_send:
306  *
307  * Send data on a TCP socket
308  */
309 gras_error_t 
310 gras_trp_tcp_chunk_send(gras_socket_t *sock,
311                         char *data,
312                         long int size) {
313   
314   /* TCP sockets are in duplex mode, don't check direction */
315   gras_assert0(size >= 0, "Cannot send a negative amount of data");
316
317   while (size) {
318     int status = 0;
319     
320     status = write(sock->sd, data, (size_t)size);
321     DEBUG3("write(%d, %p, %ld);", sock->sd, data, size);
322     
323     if (status == -1) {
324       RAISE4(system_error,"write(%d,%p,%ld) failed: %s",
325              sock->sd, data, size,
326              strerror(errno));
327     }
328     
329     if (status) {
330       size  -= status;
331       data  += status;
332     } else {
333       RAISE0(system_error,"file descriptor closed");
334     }
335   }
336
337   return no_error;
338 }
339 /**
340  * gras_trp_tcp_chunk_recv:
341  *
342  * Receive data on a TCP socket.
343  */
344 gras_error_t 
345 gras_trp_tcp_chunk_recv(gras_socket_t *sock,
346                         char *data,
347                         long int size) {
348
349   /* TCP sockets are in duplex mode, don't check direction */
350   gras_assert0(sock, "Cannot recv on an NULL socket");
351   gras_assert0(size >= 0, "Cannot receive a negative amount of data");
352   
353   while (size) {
354     int status = 0;
355     
356     status = read(sock->sd, data, (size_t)size);
357     DEBUG3("read(%d, %p, %ld);", sock->sd, data, size);
358     
359     if (status == -1) {
360       RAISE4(system_error,"read(%d,%p,%d) failed: %s",
361              sock->sd, data, (int)size,
362              strerror(errno));
363     }
364     
365     if (status) {
366       size  -= status;
367       data  += status;
368     } else {
369       RAISE0(system_error,"file descriptor closed");
370     }
371   }
372   
373   return no_error;
374 }
375
376
377 /*
378  * Returns the tcp protocol number from the network protocol data base.
379  *
380  * getprotobyname() is not thread safe. We need to lock it.
381  */
382 static int TcpProtoNumber(void) {
383   struct protoent *fetchedEntry;
384   static int returnValue = 0;
385   
386   if(returnValue == 0) {
387     fetchedEntry = getprotobyname("tcp");
388     gras_assert0(fetchedEntry, "getprotobyname(tcp) gave NULL");
389     returnValue = fetchedEntry->p_proto;
390   }
391   
392   return returnValue;
393 }