src/gras/Msg/gras_msg_listener.c

   1 /* $Id$ */
   2
   3 /* Thread in charge of listening the network and queuing incoming messages  */
   4
   5 /* Copyright (c) 2007 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 "gras/Msg/msg_private.h"
  11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg_read,gras_msg,"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_thread_t listener;
  22 } s_gras_msg_listener_t;
  23
  24 static void listener_function(void *p) {
  25   gras_msg_listener_t me = (gras_msg_listener_t)p;
  26   s_gras_msg_t msg;
  27   xbt_ex_t e;
  28   int found =0;
  29   DEBUG0("I'm the listener");
  30   while (1) {
  31     TRY {
  32       DEBUG0("Select for .5 sec");
  33       msg.expe = gras_trp_select(0.5);
  34       found =1;
  35       DEBUG0("Select returned something");
  36     }
  37     CATCH(e) {
  38       xbt_ex_free(e);
  39     }
  40     if (found) {
  41       gras_msg_recv(msg.expe, &msg);
  42       xbt_queue_push(me->incomming_messages, &msg);
  43       found =0;
  44     }
  45   }
  46 }
  47
  48 gras_msg_listener_t
  49 gras_msg_listener_launch(xbt_queue_t msg_exchange){
  50   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t,1);
  51
  52   DEBUG0("Launch listener");
  53   arg->incomming_messages = msg_exchange;
  54
  55   arg->listener =  xbt_thread_create("listener",listener_function,arg);
  56   gras_os_sleep(0); /* give the listener a chance to initialize even if the main is empty and we cancel it right afterward */
  57   return arg;
  58 }
  59
  60 void gras_msg_listener_shutdown(gras_msg_listener_t l) {
  61   DEBUG0("Listener quit");
  62   xbt_thread_cancel(l->listener);
  63   xbt_queue_free(&l->incomming_messages);
  64   xbt_free(l);
  65 }