Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
new buffered plugin. Use either tcp or sg to do the grunt work
[simgrid.git] / src / gras / Transport / transport_plugin_buf.c
1 /* $Id$ */
2
3 /* buf trp (transport) - buffered transport using the TCP one            */
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 <netinet/in.h>   /* htonl/ntohl */
12 #include <stdlib.h>
13 #include <string.h>       /* memset */
14
15 #include "gras_private.h"
16 #include "transport_private.h"
17
18 GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(trp_buf,transport);
19
20 /***
21  *** Prototypes 
22  ***/
23 gras_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self,
24                                         const char *host,
25                                         unsigned short port,
26                                         /* OUT */ gras_socket_t *sock);
27 gras_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self,
28                                         unsigned short port,
29                                         /* OUT */ gras_socket_t *sock);
30 gras_error_t gras_trp_buf_socket_accept(gras_socket_t  *sock,
31                                         gras_socket_t **dst);
32
33 void         gras_trp_buf_socket_close(gras_socket_t *sd);
34   
35 gras_error_t gras_trp_buf_chunk_send(gras_socket_t *sd,
36                                      const char *data,
37                                      long int size);
38
39 gras_error_t gras_trp_buf_chunk_recv(gras_socket_t *sd,
40                                      char *data,
41                                      long int size);
42 gras_error_t gras_trp_buf_flush(gras_socket_t  *sock);
43
44
45 /***
46  *** Specific plugin part
47  ***/
48
49 typedef struct {
50   gras_trp_plugin_t *super;
51 } gras_trp_buf_plug_data_t;
52
53 /***
54  *** Specific socket part
55  ***/
56
57 typedef struct {
58   uint32_t size;
59   char *data;
60   int pos; /* for receive; not exchanged over the net */
61 } gras_trp_buf_t;
62
63 struct gras_trp_bufdata_{
64   gras_trp_buf_t in;
65   gras_trp_buf_t out;
66   int buffsize;
67 };
68
69 gras_error_t gras_trp_buf_init_sock(gras_socket_t *sock) {
70   gras_trp_bufdata_t *data=malloc(sizeof(gras_trp_bufdata_t));
71   
72   GRAS_IN;
73   if (!data)
74     RAISE_MALLOC;
75   data->in.size  = 0;
76   data->in.data  = NULL;
77   data->in.pos   = 0; /* useless, indeed, since size==pos */
78   data->out.size = 0;
79   data->out.data = NULL;
80   data->out.pos  = 0;
81   //  data->buffsize = 32 * 1024 - 4; /* default socket buffsize (32k) - headers */ 
82   data->buffsize = 1000 * 1024 ; /* 1Mb */ 
83   sock->bufdata = data;
84   return no_error;
85 }
86
87
88 /***
89  *** Code
90  ***/
91 gras_error_t
92 gras_trp_buf_setup(gras_trp_plugin_t *plug) {
93   gras_error_t errcode;
94   gras_trp_buf_plug_data_t *data =malloc(sizeof(gras_trp_buf_plug_data_t));
95   if (!data)
96     RAISE_MALLOC;
97
98   GRAS_IN;
99   TRY(gras_trp_plugin_get_by_name(gras_if_RL() ? "tcp" : "sg",
100                                   &(data->super)));
101   DEBUG1("Derivate a buffer plugin from %s",gras_if_RL() ? "tcp" : "sg");
102
103   plug->socket_client = gras_trp_buf_socket_client;
104   plug->socket_server = gras_trp_buf_socket_server;
105   plug->socket_accept = gras_trp_buf_socket_accept;
106   plug->socket_close  = gras_trp_buf_socket_close;
107
108   plug->chunk_send    = gras_trp_buf_chunk_send;
109   plug->chunk_recv    = gras_trp_buf_chunk_recv;
110
111   plug->flush         = gras_trp_buf_flush;
112
113   plug->data = (void*)data;
114   plug->exit = NULL;
115   
116   return no_error;
117 }
118
119 gras_error_t gras_trp_buf_socket_client(gras_trp_plugin_t *self,
120                                         const char *host,
121                                         unsigned short port,
122                                         /* OUT */ gras_socket_t *sock){
123   gras_error_t errcode;
124   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)self->data)->super;
125
126   GRAS_IN;
127   TRY(super->socket_client(super,host,port,sock));
128   sock->plugin = self;
129   TRY(gras_trp_buf_init_sock(sock));
130     
131   return no_error;
132 }
133
134 /**
135  * gras_trp_buf_socket_server:
136  *
137  * Open a socket used to receive messages.
138  */
139 gras_error_t gras_trp_buf_socket_server(gras_trp_plugin_t *self,
140                                         unsigned short port,
141                                         /* OUT */ gras_socket_t *sock){
142   gras_error_t errcode;
143   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)self->data)->super;
144
145   GRAS_IN;
146   TRY(super->socket_server(super,port,sock));
147   sock->plugin = self;
148   TRY(gras_trp_buf_init_sock(sock));
149   return no_error;
150 }
151
152 gras_error_t
153 gras_trp_buf_socket_accept(gras_socket_t  *sock,
154                            gras_socket_t **dst) {
155   gras_error_t errcode;
156   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
157       
158   GRAS_IN;
159   TRY(super->socket_accept(sock,dst));
160   (*dst)->plugin = sock->plugin;
161   TRY(gras_trp_buf_init_sock(*dst));
162   return no_error;
163 }
164
165 void gras_trp_buf_socket_close(gras_socket_t *sock){
166   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
167   gras_trp_bufdata_t *data=sock->bufdata;
168
169   GRAS_IN;
170   if (data->in.size || data->out.size)
171     gras_trp_buf_flush(sock);
172   if (data->in.data)
173     free(data->in.data);
174   if (data->out.data)
175     free(data->out.data);
176   free(data);
177
178   return super->socket_close(sock);
179 }
180
181 /**
182  * gras_trp_buf_chunk_send:
183  *
184  * Send data on a TCP socket
185  */
186 gras_error_t 
187 gras_trp_buf_chunk_send(gras_socket_t *sock,
188                         const char *chunk,
189                         long int size) {
190
191   gras_error_t errcode;
192   gras_trp_bufdata_t *data=(gras_trp_bufdata_t*)sock->bufdata;
193   int chunk_pos=0;
194
195   GRAS_IN;
196   /* Let underneath plugin check for direction, we work even in duplex */
197   gras_assert0(size >= 0, "Cannot send a negative amount of data");
198
199   if (!data->out.data && !(data->out.data = malloc(data->buffsize)))
200     RAISE_MALLOC;
201
202   while (chunk_pos < size) {
203     /* size of the chunck to receive in that shot */
204     long int thissize = min(size-chunk_pos,data->buffsize - data->out.size);
205     DEBUG5("Set the chars %d..%ld into the buffer (size=%ld, ctn='%.*s')",
206            (int)data->out.size,
207            ((int)data->out.size) + thissize -1,
208            size, chunk_pos, chunk);
209
210     memcpy(data->out.data + data->out.size, chunk + chunk_pos, thissize);
211
212     data->out.size += thissize;
213     chunk_pos      += thissize;
214     DEBUG5("New pos = %d; Still to send = %ld of %ld; ctn sofar='%.*s'",
215            data->out.size,size-chunk_pos,size,(int)chunk_pos,chunk);
216
217     if (data->out.size == data->buffsize) /* out of space. Flush it */
218       TRY(gras_trp_buf_flush(sock));
219   }
220
221   GRAS_OUT;
222   return no_error;
223 }
224
225 /**
226  * gras_trp_buf_chunk_recv:
227  *
228  * Receive data on a TCP socket.
229  */
230 gras_error_t 
231 gras_trp_buf_chunk_recv(gras_socket_t *sock,
232                         char *chunk,
233                         long int size) {
234
235   gras_error_t errcode;
236   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
237   gras_trp_bufdata_t *data=sock->bufdata;
238   long int chunck_pos = 0;
239
240   /* Let underneath plugin check for direction, we work even in duplex */
241   gras_assert0(sock, "Cannot recv on an NULL socket");
242   gras_assert0(size >= 0, "Cannot receive a negative amount of data");
243   
244   GRAS_IN;
245   if (!data->in.data && !(data->in.data = malloc(data->buffsize)))
246     RAISE_MALLOC;
247
248   while (chunck_pos < size) {
249     /* size of the chunck to receive in that shot */
250     long int thissize;
251
252     if (data->in.size == data->in.pos) { /* out of data. Get more */
253       uint32_t nextsize;
254       DEBUG0("Recv the size");
255       TRY(super->chunk_recv(sock,(char*)&nextsize, 4));
256       data->in.size = ntohl(nextsize);
257
258       VERB1("Recv the chunk (size=%d)",data->in.size);
259       TRY(super->chunk_recv(sock, data->in.data, data->in.size));
260       data->in.pos=0;
261     }
262     
263     thissize = min(size-chunck_pos ,  data->in.size - data->in.pos);
264     DEBUG2("Get the chars %d..%ld out of the buffer",
265            data->in.pos,
266            data->in.pos + thissize - 1);
267     memcpy(chunk+chunck_pos, data->in.data + data->in.pos, thissize);
268
269     data->in.pos += thissize;
270     chunck_pos   += thissize;
271     DEBUG5("New pos = %d; Still to receive = %ld of %ld. Ctn so far='%.*s'",
272            data->in.pos,size - chunck_pos,size,(int)chunck_pos,chunk);
273   }
274
275   GRAS_OUT;
276   return no_error;
277 }
278
279 /**
280  * gras_trp_buf_flush:
281  *
282  * Make sure the data is sent
283  */
284 gras_error_t 
285 gras_trp_buf_flush(gras_socket_t *sock) {
286   gras_error_t errcode;
287   uint32_t size;
288   gras_trp_plugin_t *super=((gras_trp_buf_plug_data_t*)sock->plugin->data)->super;
289   gras_trp_bufdata_t *data=sock->bufdata;
290
291   GRAS_IN;
292   size = htonl(data->out.size);
293   DEBUG1("Send the size (=%d)",data->out.size);
294   TRY(super->chunk_send(sock,(char*) &size, 4));
295
296   DEBUG1("Send the chunk (size=%d)",data->out.size);
297   TRY(super->chunk_send(sock, data->out.data, data->out.size));
298   VERB1("Chunk sent (size=%d)",data->out.size);
299   data->out.size = 0;
300   return no_error;
301 }