src/gras/Transport/transport_plugin_sg.c

   1 /* file trp (transport) - send/receive a bunch of bytes in SG realm         */
   2
   3 /* Note that this is only used to debug other parts of GRAS since message   */
   4 /*  exchange in SG realm is implemented directly without mimicing real life */
   5 /*  This would be terribly unefficient.                                     */
   6
   7 /* Copyright (c) 2004, 2005, 2006, 2007, 2009, 2010. The SimGrid Team.
   8  * All rights reserved.                                                     */
   9
  10 /* This program is free software; you can redistribute it and/or modify it
  11  * under the terms of the license (GNU LGPL) which comes with this package. */
  12
  13 #include "xbt/ex.h"
  14
  15 #include "simgrid/simix.h"
  16 #include "gras/Msg/msg_private.h"
  17 #include "gras/Transport/transport_private.h"
  18 #include "gras/Virtu/virtu_sg.h"
  19 #include "xbt/xbt_socket_private.h" /* FIXME */
  20
  21 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_trp_sg, gras_trp,
  22                                 "SimGrid pseudo-transport");
  23
  24 /***
  25  *** Prototypes
  26  ***/
  27
  28 /* retrieve the port record associated to a numerical port on an host */
  29 static gras_sg_portrec_t find_port(gras_hostdata_t * hd, int port);
  30
  31 void gras_trp_sg_socket_client(xbt_trp_plugin_t self,
  32                                const char*host,
  33                                int port,
  34                                /* OUT */ xbt_socket_t sock);
  35 void gras_trp_sg_socket_server(xbt_trp_plugin_t self,
  36                                int port,
  37                                /* OUT */ xbt_socket_t sock);
  38 void gras_trp_sg_socket_close(xbt_socket_t sd);
  39
  40 void gras_trp_sg_chunk_send_raw(xbt_socket_t sd,
  41                                 const char *data, unsigned long int size);
  42 void gras_trp_sg_chunk_send(xbt_socket_t sd,
  43                             const char *data,
  44                             unsigned long int size, int stable_ignored);
  45
  46 int gras_trp_sg_chunk_recv(xbt_socket_t sd,
  47                            char *data, unsigned long int size);
  48
  49 /***
  50  *** Specific plugin part
  51  ***/
  52 typedef struct {
  53   int placeholder;              /* nothing plugin specific so far */
  54 } gras_trp_sg_plug_data_t;
  55
  56
  57 /***
  58  *** Code
  59  ***/
  60 static gras_sg_portrec_t find_port(gras_hostdata_t * hd, int port)
  61 {
  62   unsigned int cpt;
  63   gras_sg_portrec_t pr;
  64
  65   xbt_assert(hd, "Please run gras_process_init on each process");
  66
  67   xbt_dynar_foreach(hd->ports, cpt, pr) {
  68     if (pr->port == port)
  69       return pr;
  70   }
  71   return NULL;
  72 }
  73
  74 /***
  75  *** Info about who's speaking
  76  ***/
  77 static int gras_trp_sg_my_port(xbt_socket_t s)
  78 {
  79   gras_trp_sg_sock_data_t sockdata = xbt_socket_get_data(s);
  80   if (gras_socket_im_the_server(s))
  81     return sockdata->server_port;
  82   else
  83     return sockdata->client_port;
  84 }
  85 static int gras_trp_sg_peer_port(xbt_socket_t s)
  86 {
  87   gras_trp_sg_sock_data_t sockdata = xbt_socket_get_data(s);
  88   if (gras_socket_im_the_server(s))
  89     return sockdata->client_port;
  90   else
  91     return sockdata->server_port;
  92 }
  93
  94 static const char* gras_trp_sg_peer_name(xbt_socket_t s)
  95 {
  96   gras_trp_sg_sock_data_t sockdata = xbt_socket_get_data(s);
  97   if (gras_socket_im_the_server(s))
  98     return SIMIX_host_get_name(simcall_process_get_host(sockdata->client));
  99   else {
 100     return SIMIX_host_get_name(simcall_process_get_host(sockdata->server));
 101   }
 102 }
 103 static const char* gras_trp_sg_peer_proc(xbt_socket_t s) {
 104   THROW_UNIMPLEMENTED;
 105 }
 106 static void gras_trp_sg_peer_proc_set(xbt_socket_t s,char *name) {
 107   THROW_UNIMPLEMENTED;
 108 }
 109
 110 void gras_trp_sg_setup(xbt_trp_plugin_t plug)
 111 {
 112   plug->my_port = gras_trp_sg_my_port;
 113   plug->peer_port = gras_trp_sg_peer_port;
 114   plug->peer_name = gras_trp_sg_peer_name;
 115   plug->peer_proc = gras_trp_sg_peer_proc;
 116   plug->peer_proc_set = gras_trp_sg_peer_proc_set;
 117
 118   gras_trp_sg_plug_data_t *data = xbt_new(gras_trp_sg_plug_data_t, 1);
 119
 120   plug->data = data;
 121
 122   plug->socket_client = gras_trp_sg_socket_client;
 123   plug->socket_server = gras_trp_sg_socket_server;
 124   plug->socket_close = gras_trp_sg_socket_close;
 125
 126   plug->raw_send = gras_trp_sg_chunk_send_raw;
 127   plug->send = gras_trp_sg_chunk_send;
 128   plug->raw_recv = plug->recv = gras_trp_sg_chunk_recv;
 129
 130   plug->flush = NULL;           /* nothing cached */
 131 }
 132
 133 void gras_trp_sg_socket_client(xbt_trp_plugin_t self,
 134                                const char*host,
 135                                int port,
 136                                /* OUT */ xbt_socket_t sock)
 137 {
 138
 139   smx_host_t peer;
 140   gras_hostdata_t *hd;
 141   gras_trp_sg_sock_data_t data;
 142   gras_sg_portrec_t pr;
 143
 144   /* make sure this socket will reach someone */
 145   if (!(peer = simcall_host_get_by_name(host)))
 146     THROWF(mismatch_error, 0,
 147            "Can't connect to %s: no such host.\n", host);
 148
 149   if (!(hd = (gras_hostdata_t *) simcall_host_get_data(peer)))
 150     THROWF(mismatch_error, 0,
 151            "can't connect to %s: no process on this host",
 152            host);
 153
 154   pr = find_port(hd, port);
 155
 156   if (pr == NULL) {
 157     THROWF(mismatch_error, 0,
 158            "can't connect to %s:%d, no process listen on this port",
 159            host, port);
 160   }
 161
 162   /* Ensure that the listener is expecting the kind of stuff we want to send */
 163   if (pr->meas && !sock->meas) {
 164     THROWF(mismatch_error, 0,
 165            "can't connect to %s:%d in regular mode, the process listen "
 166            "in measurement mode on this port", host,
 167            port);
 168   }
 169   if (!pr->meas && sock->meas) {
 170     THROWF(mismatch_error, 0,
 171            "can't connect to %s:%d in measurement mode, the process listen "
 172            "in regular mode on this port", host,
 173            port);
 174   }
 175
 176   /* create simulation data of the socket */
 177   data = xbt_new0(s_gras_trp_sg_sock_data_t, 1);
 178   data->client = SIMIX_process_self();
 179   data->server = pr->server;
 180   data->server_port = port;
 181   data->client_port = gras_os_myport();
 182
 183   /* initialize synchronization stuff on the socket */
 184   data->rdv_server = pr->rdv;
 185   data->rdv_client = simcall_rdv_create(NULL);
 186   data->comm_recv = simcall_comm_irecv(data->rdv_client, &data->msg, NULL, NULL, NULL);
 187
 188   /* connect that simulation data to the socket */
 189   sock->data = data;
 190   sock->incoming = 1;
 191
 192   XBT_DEBUG("%s (PID %d) connects in %s mode to %s:%d (rdv_ser:%p, rdv_cli:%p, comm:%p)",
 193          simcall_process_get_name(SIMIX_process_self()), gras_os_getpid(),
 194          sock->meas ? "meas" : "regular", host, port,
 195          data->rdv_server,data->rdv_client,data->comm_recv);
 196 }
 197
 198 void gras_trp_sg_socket_server(xbt_trp_plugin_t self, int port, xbt_socket_t sock)
 199 {
 200
 201   gras_hostdata_t *hd =
 202       (gras_hostdata_t *) SIMIX_host_self_get_data();
 203   gras_sg_portrec_t pr;
 204   gras_trp_sg_sock_data_t data;
 205
 206   xbt_assert(hd, "Please run gras_process_init on each process");
 207
 208   sock->accepting = 1;
 209
 210   /* Check whether a server is already listening on that port or not */
 211   pr = find_port(hd, port);
 212
 213   if (pr)
 214     THROWF(mismatch_error, 0,
 215            "can't listen on address %s:%d: port already in use.",
 216            SIMIX_host_self_get_name(), port);
 217
 218   /* This port is free, let's take it */
 219   pr = xbt_new(s_gras_sg_portrec_t, 1);
 220   pr->port = port;
 221   pr->meas = sock->meas;
 222   pr->server = SIMIX_process_self();
 223   xbt_dynar_push(hd->ports, &pr);
 224   pr->rdv = simcall_rdv_create(NULL);
 225
 226   /* Create the socket */
 227   data = xbt_new0(s_gras_trp_sg_sock_data_t, 1);
 228   data->server = SIMIX_process_self();
 229   data->server_port = port;
 230   data->client = NULL;
 231   data->rdv_server = pr->rdv;
 232   data->rdv_client = NULL;
 233   data->comm_recv = simcall_comm_irecv(pr->rdv, &data->msg, NULL, NULL, NULL);
 234
 235   sock->data = data;
 236
 237   XBT_VERB
 238       ("'%s' (%d) ears on %s:%d%s (%p; data:%p); Here rdv: %p; Remote rdv: %p; Comm %p",
 239        simcall_process_get_name(SIMIX_process_self()), gras_os_getpid(),
 240        SIMIX_host_self_get_name(), port,
 241        sock->meas ? " (mode meas)" : "", sock, data,
 242        (data->server ==
 243         SIMIX_process_self())? data->rdv_server : data->rdv_client,
 244        (data->server ==
 245         SIMIX_process_self())? data->rdv_client : data->rdv_server,
 246        data->comm_recv);
 247
 248 }
 249
 250 void gras_trp_sg_socket_close(xbt_socket_t sock)
 251 {
 252   gras_hostdata_t *hd =
 253       (gras_hostdata_t *) SIMIX_host_self_get_data();
 254   unsigned int cpt;
 255   gras_sg_portrec_t pr;
 256
 257   XBT_IN(" (sock=%p)", sock);
 258
 259   if (!sock)
 260     return;
 261
 262   xbt_assert(hd, "Please run gras_process_init on each process");
 263
 264   gras_trp_sg_sock_data_t sockdata = sock->data;
 265
 266   if (sock->incoming && !sock->outgoing && sockdata->server_port >= 0) {
 267     /* server mode socket. Unregister it from 'OS' tables */
 268     xbt_dynar_foreach(hd->ports, cpt, pr) {
 269       XBT_DEBUG("Check pr %u of %lu", cpt, xbt_dynar_length(hd->ports));
 270       if (pr->port == sockdata->server_port) {
 271         xbt_dynar_cursor_rm(hd->ports, &cpt);
 272         XBT_OUT();
 273         return;
 274       }
 275     }
 276     XBT_WARN
 277         ("socket_close called on the unknown incoming socket %p (port=%d)",
 278          sock, sockdata->server_port);
 279   }
 280   /* FIXME: kill the rdv point if receiver side */
 281   free(sock->data);
 282   XBT_OUT();
 283 }
 284
 285 typedef struct {
 286   int size;
 287   void *data;
 288 } sg_task_data_t;
 289
 290 void gras_trp_sg_chunk_send(xbt_socket_t sock,
 291                             const char *data,
 292                             unsigned long int size, int stable_ignored)
 293 {
 294   gras_trp_sg_chunk_send_raw(sock, data, size);
 295 }
 296
 297 void gras_trp_sg_chunk_send_raw(xbt_socket_t sock,
 298                                 const char *data, unsigned long int size)
 299 {
 300 #ifdef KILLME
 301   char name[256];
 302   static unsigned int count = 0;
 303
 304   smx_action_t act;             /* simix action */
 305   gras_trp_procdata_t trp_remote_proc;
 306   gras_msg_procdata_t msg_remote_proc;
 307   gras_msg_t msg;               /* message to send */
 308
 309   //gras_trp_sg_sock_data_t sock_data = (gras_trp_sg_sock_data_t) sock->data;
 310   xbt_assert(sock->meas,
 311               "SG chunk exchange shouldn't be used on non-measurement sockets");
 312
 313
 314   /* creates simix action and waits its ends, waits in the sender host
 315      condition */
 316   /*
 317   if (XBT_LOG_ISENABLED(gras_trp_sg, xbt_log_priority_debug)) {
 318     smx_process_t remote_dude =
 319         (sock_data->server ==
 320          SIMIX_process_self())? (sock_data->client) : (sock_data->server);
 321     smx_host_t remote_host = simcall_process_get_host(remote_dude);
 322   }
 323   */
 324   //SIMIX_network_send(sock_data->rdv,size,1,-1,NULL,0,NULL,NULL);
 325 #endif
 326   THROW_UNIMPLEMENTED;
 327 }
 328
 329 int gras_trp_sg_chunk_recv(xbt_socket_t sock,
 330                            char *data, unsigned long int size)
 331 {
 332   //gras_trp_sg_sock_data_t *sock_data =
 333   //    (gras_trp_sg_sock_data_t *) sock->data;
 334
 335   //simcall_comm_recv(sock_data->rdv,-1,NULL,0,NULL);
 336   THROW_UNIMPLEMENTED;
 337 #ifdef KILLME
 338   gras_trp_sg_sock_data_t *remote_sock_data;
 339   xbt_socket_t remote_socket = NULL;
 340   gras_msg_t msg_got;
 341   gras_msg_procdata_t msg_procdata =
 342       (gras_msg_procdata_t) gras_libdata_by_name("gras_msg");
 343   gras_trp_procdata_t trp_proc =
 344       (gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id);
 345
 346   xbt_assert(sock->meas,
 347               "SG chunk exchange shouldn't be used on non-measurement sockets");
 348   xbt_queue_shift_timed(trp_proc->meas_selectable_sockets,
 349                         &remote_socket, 60);
 350
 351   if (remote_socket == NULL) {
 352     THROWF(timeout_error, 0, "Timeout");
 353   }
 354
 355   remote_sock_data = (gras_trp_sg_sock_data_t *) remote_socket->data;
 356   msg_got = xbt_fifo_shift(msg_procdata->msg_to_receive_queue_meas);
 357
 358   sock_data = (gras_trp_sg_sock_data_t *) sock->data;
 359
 360   /* ok, I'm here, you can continue the communication */
 361   simcall_cond_signal(remote_sock_data->cond);
 362
 363   simcall_mutex_lock(remote_sock_data->mutex);
 364   /* wait for communication end */
 365   simcall_cond_wait(remote_sock_data->cond, remote_sock_data->mutex);
 366
 367   if (msg_got->payl_size != size)
 368     THROWF(mismatch_error, 0,
 369            "Got %d bytes when %ld where expected (in %s->%s:%d)",
 370            msg_got->payl_size, size,
 371            simcall_host_get_name(sock_data->to_host),
 372            SIMIX_host_self_get_name(), sock->peer_port);
 373
 374   if (data)
 375     memcpy(data, msg_got->payl, size);
 376
 377   xbt_free(msg_got->payl);
 378
 379   xbt_free(msg_got);
 380   simcall_mutex_unlock(remote_sock_data->mutex);
 381 #endif
 382   return 0;
 383 }