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
  32   while (1) {
  33     msg.expe = gras_trp_select(-1);
  34     gras_msg_recv(msg.expe, &msg);
  35     xbt_queue_push(me->incomming_messages, &msg);
  36   }
  37 }
  38
  39 gras_msg_listener_t
  40 gras_msg_listener_launch(xbt_queue_t msg_exchange){
  41   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t,1);
  42
  43   arg->incomming_messages = msg_exchange;
  44
  45   arg->listener =  xbt_thread_create(listener_function,&arg);
  46   return arg;
  47 }
  48
  49 void gras_msg_listener_shutdown(gras_msg_listener_t l) {
  50   xbt_thread_cancel(l->listener);
  51   xbt_queue_free(&l->incomming_messages);
  52   xbt_free(l);
  53 }