1 /* amok peer management - servers main loop and remote peer stopping */
3 /* Copyright (c) 2006, 2007, 2008, 2009, 2010. The SimGrid Team.
4 * All rights reserved. */
6 /* This program is free software; you can redistribute it and/or modify it
7 * under the terms of the license (GNU LGPL) which comes with this package. */
9 #include "xbt/sysdep.h"
11 #include "amok/peermanagement.h"
13 #include "amok/amok_modinter.h" /* prototype of my module declaration */
14 #include "gras/module.h" /* module mecanism */
16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(amok_pm, amok, "peer management");
20 int amok_pm_moddata_id = -1;
24 } s_amok_pm_moddata_t, *amok_pm_moddata_t;
26 /* Message callbacks */
27 static int amok_pm_cb_kill(gras_msg_cb_ctx_t ctx, void *payload_data)
30 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
35 static int amok_pm_cb_killrpc(gras_msg_cb_ctx_t ctx, void *payload_data)
38 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
40 gras_msg_rpcreturn(30, ctx, NULL);
44 static int amok_pm_cb_get(gras_msg_cb_ctx_t ctx, void *payload)
46 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
47 char *name = *(void **) payload;
48 xbt_dynar_t res = xbt_dict_get(g->groups, name);
50 gras_msg_rpcreturn(30, ctx, &res);
54 static int amok_pm_cb_join(gras_msg_cb_ctx_t ctx, void *payload)
56 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
57 char* group_name = *(char* *) payload;
58 xbt_dynar_t group = xbt_dict_get(g->groups, group_name);
61 gras_socket_t exp = gras_msg_cb_ctx_from(ctx);
62 xbt_peer_t dude = xbt_peer_new(gras_socket_peer_name(exp),
63 gras_socket_peer_port(exp));
65 rank = xbt_dynar_length(group);
66 xbt_dynar_push(group, &dude);
67 VERB3("Contacted by %s:%d. Give it rank #%d", dude->name, dude->port,rank);
69 gras_msg_rpcreturn(10, ctx, &rank);
74 static int amok_pm_cb_leave(gras_msg_cb_ctx_t ctx, void *payload)
76 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
77 char *name = *(void **) payload;
78 xbt_dynar_t group = xbt_dict_get(g->groups, name);
80 gras_socket_t exp = gras_msg_cb_ctx_from(ctx);
81 xbt_peer_t dude = xbt_peer_new(gras_socket_peer_name(exp),
82 gras_socket_peer_port(exp));
87 xbt_dynar_foreach(group, cpt, peer_it) {
88 if (!strcmp(peer_it->name, dude->name) && peer_it->port == dude->port) {
89 xbt_dynar_cursor_rm(group, &cpt);
93 WARN3("Asked to remove %s:%d from group '%s', but not found. Ignoring",
94 dude->name, dude->port, name);
97 gras_msg_rpcreturn(30, ctx, NULL);
101 static int amok_pm_cb_shutdown(gras_msg_cb_ctx_t ctx, void *payload)
103 char *name = *(void **) payload;
104 amok_pm_group_shutdown(name);
106 gras_msg_rpcreturn(30, ctx, NULL);
110 /** \brief Enter the main loop of the program. It won't return until we get a kill message. */
111 void amok_pm_mainloop(double timeOut)
113 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
116 gras_msg_handle(timeOut);
120 /** \brief kill a buddy identified by its peername and port. Note that it is not removed from any group it may belong to. */
121 void amok_pm_kill_hp(char *name, int port)
123 gras_socket_t sock = gras_socket_client(name, port);
125 gras_socket_close(sock);
128 /** \brief kill a buddy to which we have a socket already. Note that it is not removed from any group it may belong to. */
129 void amok_pm_kill(gras_socket_t buddy)
131 gras_msg_send(buddy, "amok_pm_kill", NULL);
134 /** \brief kill syncronously a buddy (do not return before its death). Note that it is not removed from any group it may belong to. */
135 void amok_pm_kill_sync(gras_socket_t buddy)
137 gras_msg_rpccall(buddy, 30, "amok_pm_killrpc", NULL, NULL);
141 /** \brief create a new peermanagement group located on local peer
143 * The dynar elements are of type xbt_peer_t
145 xbt_dynar_t amok_pm_group_new(const char *group_name)
148 xbt_dynar_t res = xbt_dynar_new(sizeof(xbt_peer_t),
149 xbt_peer_free_voidp);
151 xbt_assert0(amok_pm_moddata_id != -1, "Run amok_pm_init first!");
152 g = gras_moddata_by_id(amok_pm_moddata_id);
154 DEBUG1("retrieved groups=%p", g->groups);
156 xbt_dict_set(g->groups, group_name, res, NULL); /*FIXME: leaking xbt_dynar_free_voidp); */
157 VERB1("Group %s created", group_name);
162 /** \brief retrieve all members of the given remote group */
163 xbt_dynar_t amok_pm_group_get(gras_socket_t master, const char *group_name)
167 gras_msg_rpccall(master, 30, "amok_pm_get", &group_name, &res);
171 /** \brief add current peer to the given remote group
173 * Returns the rank of the process in the group.
175 int amok_pm_group_join(gras_socket_t master, const char *group_name)
178 VERB3("Join group '%s' on %s:%d",
179 group_name, gras_socket_peer_name(master),
180 gras_socket_peer_port(master));
181 gras_msg_rpccall(master, 30, "amok_pm_join", &group_name, &rank);
182 VERB4("Joined group '%s' on %s:%d. Got rank %d",
183 group_name, gras_socket_peer_name(master),
184 gras_socket_peer_port(master),
189 /** \brief remove current peer from the given remote group if found
191 * If not found, call is ignored
193 void amok_pm_group_leave(gras_socket_t master, const char *group_name)
195 gras_msg_rpccall(master, 30, "amok_pm_leave", &group_name, NULL);
196 VERB3("Leaved group '%s' on %s:%d",
197 group_name, gras_socket_peer_name(master),
198 gras_socket_peer_port(master));
201 /** \brief stops all members of the given local group */
202 void amok_pm_group_shutdown(const char *group_name)
204 amok_pm_moddata_t g = gras_moddata_by_id(amok_pm_moddata_id);
205 xbt_dynar_t group = xbt_dict_get(g->groups, group_name);
210 xbt_dynar_foreach(group, cpt, peer_it) {
211 amok_pm_kill_hp(peer_it->name, peer_it->port);
214 xbt_dynar_free(&group);
215 xbt_dict_remove(g->groups, group_name);
218 /** \brief stops all members of the given remote group */
219 void amok_pm_group_shutdown_remote(gras_socket_t master,
220 const char *group_name)
222 gras_msg_rpccall(master, 30, "amok_pm_shutdown", &group_name, NULL);
228 * * Module management functions
234 static void _amok_pm_init(void)
236 /* no world-wide globals */
237 /* Datatype and message declarations */
238 gras_datadesc_type_t pm_group_type =
239 gras_datadesc_dynar(gras_datadesc_by_name("xbt_peer_t"),
240 xbt_peer_free_voidp);
242 gras_msgtype_declare("amok_pm_kill", NULL);
243 gras_msgtype_declare_rpc("amok_pm_killrpc", NULL, NULL);
245 gras_msgtype_declare_rpc("amok_pm_get",
246 gras_datadesc_by_name("string"), pm_group_type);
247 gras_msgtype_declare_rpc("amok_pm_join", gras_datadesc_by_name("string"), gras_datadesc_by_name("int"));
248 gras_msgtype_declare_rpc("amok_pm_leave",
249 gras_datadesc_by_name("string"), NULL);
251 gras_msgtype_declare_rpc("amok_pm_shutdown",
252 gras_datadesc_by_name("string"), NULL);
255 static void _amok_pm_join(void *p)
257 /* moddata management */
258 amok_pm_moddata_t mod = (amok_pm_moddata_t) p;
263 mod->groups = xbt_dict_new();
266 gras_cb_register("amok_pm_kill", &amok_pm_cb_kill);
267 gras_cb_register("amok_pm_killrpc", &amok_pm_cb_killrpc);
269 gras_cb_register("amok_pm_get", &amok_pm_cb_get);
270 gras_cb_register("amok_pm_join", &amok_pm_cb_join);
271 gras_cb_register("amok_pm_leave", &amok_pm_cb_leave);
272 gras_cb_register("amok_pm_shutdown", &amok_pm_cb_shutdown);
275 static void _amok_pm_exit(void)
277 /* no world-wide globals */
280 static void _amok_pm_leave(void *p)
283 amok_pm_moddata_t mod = (amok_pm_moddata_t) p;
286 xbt_dict_free(&mod->groups);
289 gras_cb_unregister("amok_pm_kill", &amok_pm_cb_kill);
290 gras_cb_unregister("amok_pm_killrpc", &amok_pm_cb_killrpc);
292 gras_cb_unregister("amok_pm_get", &amok_pm_cb_get);
293 gras_cb_unregister("amok_pm_join", &amok_pm_cb_join);
294 gras_cb_unregister("amok_pm_leave", &amok_pm_cb_leave);
295 gras_cb_unregister("amok_pm_shutdown", &amok_pm_cb_shutdown);
298 void amok_pm_modulecreate()
300 gras_module_add("amok_pm", sizeof(s_amok_pm_moddata_t), &amok_pm_moddata_id,
301 _amok_pm_init, _amok_pm_exit, _amok_pm_join,
309 * * Old module functions (kept for compatibility)
312 /** \brief Initialize the peer management module. Every process must run it before use */
315 gras_module_join("amok_pm");
318 /** \brief Finalize the peer management module. Every process should run it after use */
321 gras_module_leave("amok_pm");