src/gras/Msg/gras_msg_listener.c

   1 /* Thread in charge of listening the network and queuing incoming messages  */
   2
   3 /* Copyright (c) 2007, 2008, 2009, 2010. The SimGrid Team.
   4  * All rights reserved.                                                     */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include "gras/Msg/msg_private.h"
  10 #include "gras/Transport/transport_private.h"
  11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg_read, gras_msg,
  12                                 "Message reader thread");
  13
  14 #include "xbt/ex.h"
  15 #include "xbt/queue.h"
  16 #include "xbt/synchro.h"
  17
  18 #include "gras/Transport/transport_interface.h" /* gras_select */
  19
  20 typedef struct s_gras_msg_listener_ {
  21   xbt_queue_t incomming_messages; /* messages received from the wire and still to be used by master */
  22   xbt_queue_t socks_to_close;   /* let the listener close the sockets, since it may be selecting on them. Darwin don't like this trick */
  23   gras_socket_t wakeup_sock_listener_side;
  24   gras_socket_t wakeup_sock_master_side;
  25   xbt_thread_t listener;
  26 } s_gras_msg_listener_t;
  27
  28 static void listener_function(void *p)
  29 {
  30   gras_msg_listener_t me = (gras_msg_listener_t) p;
  31   gras_msg_t msg;
  32   xbt_ex_t e;
  33   gras_msgtype_t msg_wakeup_listener_t =
  34     gras_msgtype_by_name("_wakeup_listener");
  35   DEBUG0("I'm the listener");
  36   while (1) {
  37     msg = gras_msg_recv_any();
  38     if (msg->type != msg_wakeup_listener_t)
  39       xbt_queue_push(me->incomming_messages, msg);
  40     else {
  41       char got = *(char *) msg->payl;
  42       if (got == '1') {
  43         VERB0("Asked to get awake");
  44         free(msg->payl);
  45         free(msg);
  46       } else {
  47         VERB0("Asked to die");
  48         //gras_socket_close(me->wakeup_sock_listener_side);
  49         free(msg->payl);
  50         free(msg);
  51         return;
  52       }
  53     }
  54     /* empty the list of sockets to trash */
  55     TRY {
  56       while (1) {
  57         int sock;
  58         xbt_queue_shift_timed(me->socks_to_close, &sock, 0);
  59         if (tcp_close(sock) < 0) {
  60 #ifdef _WIN32
  61           WARN2("error while closing tcp socket %d: %d (%s)\n", sock, sock_errno);
  62 #else
  63           WARN3("error while closing tcp socket %d: %d (%s)\n",
  64                 sock, sock_errno, sock_errstr(sock_errno));
  65 #endif
  66         }
  67       }
  68     }
  69     CATCH(e) {
  70       if (e.category != timeout_error)
  71         RETHROW;
  72       xbt_ex_free(e);
  73     }
  74   }
  75 }
  76
  77 gras_msg_listener_t gras_msg_listener_launch(xbt_queue_t msg_exchange)
  78 {
  79   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t, 1);
  80   int my_port;
  81
  82   DEBUG0("Launch listener");
  83   arg->incomming_messages = msg_exchange;
  84   arg->socks_to_close = xbt_queue_new(0, sizeof(int));
  85
  86   /* get a free socket for the receiving part of the listener, taking care that it does not get saved as "myport" number */
  87   my_port =
  88     ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport;
  89   arg->wakeup_sock_listener_side =
  90     gras_socket_server_range(5000, 6000, -1, 0);
  91   ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport =
  92     my_port;
  93   /* Connect the other part of the socket */
  94   arg->wakeup_sock_master_side =
  95     gras_socket_client(gras_os_myname(),
  96                        gras_socket_my_port(arg->wakeup_sock_listener_side));
  97
  98   /* declare the message used to awake the listener from the master */
  99   gras_msgtype_declare("_wakeup_listener", gras_datadesc_by_name("char"));
 100
 101   /* actually start the thread */
 102   arg->listener = xbt_thread_create("listener", listener_function, arg,1/*joinable*/);
 103   gras_os_sleep(0);             /* TODO: useless? give the listener a chance to initialize even if the main is empty and we cancel it right afterward */
 104   return arg;
 105 }
 106
 107 #include "gras/Virtu/virtu_private.h"   /* procdata_t content */
 108 void gras_msg_listener_shutdown()
 109 {
 110   gras_procdata_t *pd = gras_procdata_get();
 111   char kill = '0';
 112   DEBUG0("Listener quit");
 113
 114   if (pd->listener)
 115     gras_msg_send(pd->listener->wakeup_sock_master_side, "_wakeup_listener",
 116           &kill);
 117
 118   xbt_thread_join(pd->listener->listener);
 119
 120   //  gras_socket_close(pd->listener->wakeup_sock_master_side); FIXME: uncommenting this leads to deadlock at terminaison
 121   xbt_queue_free(&pd->listener->incomming_messages);
 122   xbt_queue_free(&pd->listener->socks_to_close);
 123   xbt_free(pd->listener);
 124 }
 125
 126 void gras_msg_listener_awake()
 127 {
 128   gras_procdata_t *pd;
 129   char c = '1';
 130
 131   DEBUG0("Awaking the listener");
 132   pd = gras_procdata_get();
 133   if (pd->listener) {
 134     gras_msg_send(pd->listener->wakeup_sock_master_side, "_wakeup_listener",
 135                   &c);
 136   }
 137 }
 138
 139 void gras_msg_listener_close_socket(int sd)
 140 {
 141   gras_procdata_t *pd = gras_procdata_get();
 142   if (pd->listener) {
 143     xbt_queue_push(pd->listener->socks_to_close, &sd);
 144     gras_msg_listener_awake();
 145   } else {
 146     /* do it myself */
 147     tcp_close(sd);
 148   }
 149 }