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/Virtu/virtu_interface.h"
  18 //#include "gras/DataDesc/datadesc_interface.h"
  19 #include "gras/Transport/transport_interface.h" /* gras_select */
  20 //#include "portable.h" /* execinfo when available to propagate exceptions */
  21
  22
  23 typedef struct s_gras_msg_listener_ {
  24   xbt_queue_t incomming_messages;
  25   xbt_thread_t listener;
  26 } s_gras_msg_listener_t;
  27
  28 static void listener_function(void *p) {
  29   gras_msg_listener_t me = (gras_msg_listener_t)p;
  30   s_gras_msg_t msg;
  31         xbt_ex_t e;
  32         int found =0;
  33   while (1) {
  34                 TRY {
  35     msg.expe = gras_trp_select(0.5);
  36                 found =1;
  37                 }
  38                 CATCH(e) {
  39                 //      gras_os_sleep(0.01);
  40
  41                 }
  42                 if (found) {
  43                         gras_msg_recv(msg.expe, &msg);
  44                         xbt_queue_push(me->incomming_messages, &msg);
  45                         found =0;
  46                 }
  47   }
  48 }
  49
  50 gras_msg_listener_t
  51 gras_msg_listener_launch(xbt_queue_t msg_exchange){
  52   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t,1);
  53
  54   arg->incomming_messages = msg_exchange;
  55
  56   arg->listener =  xbt_thread_create(listener_function,arg);
  57   return arg;
  58 }
  59
  60 void gras_msg_listener_shutdown(gras_msg_listener_t l) {
  61   xbt_thread_cancel(l->listener);
  62   xbt_queue_free(&l->incomming_messages);
  63   xbt_free(l);
  64 }