src/gras/Msg/gras_msg_listener.c

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