Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Change the prototype of xbt_thread_create(), sorry.
[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,
13                                 "Message reader thread");
14
15 #include "xbt/ex.h"
16 #include "xbt/queue.h"
17 #include "xbt/synchro.h"
18
19 #include "gras/Transport/transport_interface.h" /* gras_select */
20
21 typedef struct s_gras_msg_listener_ {
22   xbt_queue_t incomming_messages;
23   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 */
24   gras_socket_t wakeup_sock_listener_side;
25   gras_socket_t wakeup_sock_master_side;
26   xbt_thread_t listener;
27 } s_gras_msg_listener_t;
28
29 static void listener_function(void *p)
30 {
31   gras_msg_listener_t me = (gras_msg_listener_t) p;
32   s_gras_msg_t msg;
33   xbt_ex_t e;
34   gras_msgtype_t msg_wakeup_listener_t =
35     gras_msgtype_by_name("_wakeup_listener");
36   DEBUG0("I'm the listener");
37   while (1) {
38     DEBUG0("Selecting");
39     msg.expe = gras_trp_select(-1);
40     DEBUG0("Select returned something");
41     gras_msg_recv(msg.expe, &msg);
42     if (msg.type != msg_wakeup_listener_t)
43       xbt_queue_push(me->incomming_messages, &msg);
44     else {
45       char got = *(char *) msg.payl;
46       if (got == '1') {
47         VERB0("Asked to get awake");
48         free(msg.payl);
49       } else {
50         VERB0("Asked to die");
51         //        gras_socket_close(me->wakeup_sock_listener_side);
52         free(msg.payl);
53         return;
54       }
55     }
56     /* empty the list of sockets to trash */
57     TRY {
58       while (1) {
59         int sock;
60         xbt_queue_shift_timed(me->socks_to_close, &sock, 0);
61         if (tcp_close(sock) < 0) {
62           WARN3("error while closing tcp socket %d: %d (%s)\n",
63                 sock, sock_errno, sock_errstr(sock_errno));
64         }
65       }
66     }
67     CATCH(e) {
68       if (e.category != timeout_error)
69         RETHROW;
70       xbt_ex_free(e);
71     }
72   }
73 }
74
75 gras_msg_listener_t gras_msg_listener_launch(xbt_queue_t msg_exchange)
76 {
77   gras_msg_listener_t arg = xbt_new0(s_gras_msg_listener_t, 1);
78   int my_port;
79
80   DEBUG0("Launch listener");
81   arg->incomming_messages = msg_exchange;
82   arg->socks_to_close = xbt_queue_new(0, sizeof(int));
83
84   /* get a free socket for the receiving part of the listener, taking care that it does not get saved as "myport" number */
85   my_port =
86     ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport;
87   arg->wakeup_sock_listener_side =
88     gras_socket_server_range(5000, 6000, -1, 0);
89   ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->myport =
90     my_port;
91   /* Connect the other part of the socket */
92   arg->wakeup_sock_master_side =
93     gras_socket_client(gras_os_myname(),
94                        gras_socket_my_port(arg->wakeup_sock_listener_side));
95
96   /* declare the message used to awake the listener from the master */
97   gras_msgtype_declare("_wakeup_listener", gras_datadesc_by_name("char"));
98
99   /* actually start the thread */
100   arg->listener = xbt_thread_create("listener", listener_function, arg,0/*not joinable*/);
101   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 */
102   return arg;
103 }
104
105 #include "gras/Virtu/virtu_private.h"   /* procdata_t content */
106 void gras_msg_listener_shutdown(gras_msg_listener_t l)
107 {
108   gras_procdata_t *pd = gras_procdata_get();
109   char kill = '0';
110   DEBUG0("Listener quit");
111
112
113   if (pd->listener)
114     gras_msg_send(pd->listener->wakeup_sock_master_side, "_wakeup_listener",
115                   &kill);
116
117   /* FIXME: thread_join is not implemented in SG (remove next conditional when fixed)
118    * But I guess it's not a big deal since we're terminating the thread mainly to
119    * make it free its OS locks on darwin.
120    * darwin is definitly different from the neat & nice SG world */
121   if (gras_if_RL())
122     xbt_thread_join(pd->listener->listener);
123
124   //  gras_socket_close(pd->listener->wakeup_sock_master_side); FIXME: uncommenting this leads to deadlock at terminaison
125   xbt_queue_free(&l->incomming_messages);
126   xbt_queue_free(&l->socks_to_close);
127   xbt_free(l);
128 }
129
130 void gras_msg_listener_awake()
131 {
132   gras_procdata_t *pd;
133   char c = '1';
134
135   DEBUG0("Awaking the listener");
136   pd = gras_procdata_get();
137   if (pd->listener) {
138     gras_msg_send(pd->listener->wakeup_sock_master_side, "_wakeup_listener",
139                   &c);
140   }
141 }
142
143 void gras_msg_listener_close_socket(int sd)
144 {
145   gras_procdata_t *pd = gras_procdata_get();
146   if (pd->listener) {
147     xbt_queue_push(pd->listener->socks_to_close, &sd);
148     gras_msg_listener_awake();
149   } else {
150     /* do it myself */
151     tcp_close(sd);
152   }
153 }