src/gras/Transport/transport_plugin_buf.c

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