Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Gras listener thread of each process do select(-1) instead of while(1) {select(0.5)}
[simgrid.git] / 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 #include "gras/Transport/transport_private.h"
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_msg_read,gras_msg,"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;
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   gras_msg_listener_t me = (gras_msg_listener_t)p;
29   s_gras_msg_t msg;
30   gras_msgtype_t msg_wakeup_listener_t = gras_msgtype_by_name("_wakeup_listener");
31   DEBUG0("I'm the listener");
32   while (1) {
33     DEBUG0("Selecting");
34     msg.expe = gras_trp_select(-1);
35     DEBUG0("Select returned something");
36     gras_msg_recv(msg.expe, &msg);
37     if (msg.type!=msg_wakeup_listener_t)
38         xbt_queue_push(me->incomming_messages, &msg);
39     else  {
40         DEBUG0("Asked to get awake");
41     }
42   }
43 }
44
45 gras_msg_listener_t
46 gras_msg_listener_launch(xbt_queue_t msg_exchange){
47   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t,1);
48   int my_port;
49
50   DEBUG0("Launch listener");
51   arg->incomming_messages = msg_exchange;
52
53   /* get a free socket for the receiving part of the listener, taking care that it does not get saved as "myport" number */
54   my_port = ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport;
55   arg->wakeup_sock_listener_side=gras_socket_server_range(5000,6000,-1,0);
56   ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport = my_port;
57   /* Connect the other part of the socket */
58   arg->wakeup_sock_master_side=gras_socket_client(gras_os_myname(),gras_socket_my_port(arg->wakeup_sock_listener_side));
59
60   /* declare the message used to awake the listener from the master */
61   gras_msgtype_declare("_wakeup_listener",gras_datadesc_by_name("char"));
62
63   /* actually start the thread */
64   arg->listener =  xbt_thread_create("listener",listener_function,arg);
65   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 */
66   return arg;
67 }
68
69 void gras_msg_listener_shutdown(gras_msg_listener_t l) {
70   DEBUG0("Listener quit");
71   xbt_thread_cancel(l->listener);
72   xbt_queue_free(&l->incomming_messages);
73   xbt_free(l);
74 }
75
76 #include "gras/Virtu/virtu_private.h" /* procdata_t content */
77 void gras_msg_listener_awake(){
78         gras_procdata_t *pd;
79         char c = '1';
80
81         DEBUG0("Awaking the listener");
82         pd = gras_procdata_get();
83         if (pd->listener) {
84                 gras_msg_send(pd->listener->wakeup_sock_master_side,"_wakeup_listener",&c);
85         }
86 }