src/amok/HostManagement/hostmanagement.c

   1 /* $Id$ */
   2
   3 /* amok host management - servers main loop and remote host stopping        */
   4
   5 /* Copyright (c) 2006 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 "xbt/sysdep.h"
  11 #include "xbt/host.h"
  12 #include "amok/hostmanagement.h"
  13 #include "gras/Virtu/virtu_interface.h" /* libdata */
  14
  15 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(amok_hm,amok,"Host management");
  16
  17
  18 /* libdata management */
  19 static int amok_hm_libdata_id;
  20 typedef struct {
  21   int done;
  22   xbt_dict_t groups;
  23 } s_amok_hm_libdata_t, *amok_hm_libdata_t;
  24
  25 static void *amok_hm_libdata_new() {
  26   amok_hm_libdata_t res=xbt_new(s_amok_hm_libdata_t,1);
  27   res->done = 0;
  28   res->groups = xbt_dict_new();
  29   return res;
  30 }
  31 static void amok_hm_libdata_free(void *d) {
  32   amok_hm_libdata_t data=(amok_hm_libdata_t)d;
  33   xbt_dict_free(&data->groups);
  34   free (data);
  35 }
  36
  37 /* Message callbacks */
  38 static int amok_hm_cb_kill(gras_msg_cb_ctx_t ctx,
  39                            void             *payload_data) {
  40
  41   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
  42   g->done = 1;
  43   return 1;
  44 }
  45 static int amok_hm_cb_killrpc(gras_msg_cb_ctx_t ctx,
  46                               void             *payload_data) {
  47
  48   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
  49   g->done = 1;
  50   gras_msg_rpcreturn(30,ctx,NULL);
  51   return 1;
  52 }
  53
  54 static int amok_hm_cb_get(gras_msg_cb_ctx_t ctx, void *payload) {
  55   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
  56   char *name = *(void**)payload;
  57   xbt_dynar_t res = xbt_dict_get(g->groups, name);
  58
  59   gras_msg_rpcreturn(30, ctx, &res);
  60   return 1;
  61 }
  62 static int amok_hm_cb_join(gras_msg_cb_ctx_t ctx, void *payload) {
  63   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
  64   char *name = *(void**)payload;
  65   xbt_dynar_t group = xbt_dict_get(g->groups, name);
  66
  67   gras_socket_t exp = gras_msg_cb_ctx_from(ctx);
  68   xbt_host_t dude = xbt_host_new(gras_socket_peer_name(exp),
  69                                  gras_socket_peer_port(exp));
  70
  71   xbt_dynar_push(group,&dude);
  72
  73   gras_msg_rpcreturn(30, ctx, NULL);
  74   return 1;
  75 }
  76 static int amok_hm_cb_leave(gras_msg_cb_ctx_t ctx, void *payload) {
  77   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
  78   char *name = *(void**)payload;
  79   xbt_dynar_t group = xbt_dict_get(g->groups, name);
  80
  81   gras_socket_t exp = gras_msg_cb_ctx_from(ctx);
  82   xbt_host_t dude = xbt_host_new(gras_socket_peer_name(exp),
  83                                  gras_socket_peer_port(exp));
  84
  85   int cpt;
  86   xbt_host_t host_it;
  87
  88   xbt_dynar_foreach(group, cpt, host_it) {
  89     if (!strcmp(host_it->name, dude->name) &&
  90         host_it->port == dude->port) {
  91       xbt_dynar_cursor_rm (group,&cpt);
  92       goto end;
  93     }
  94   }
  95   WARN3("Asked to remove %s:%d from group '%s', but not found. Ignoring",
  96         dude->name,dude->port, name);
  97
  98  end:
  99   gras_msg_rpcreturn(30, ctx, NULL);
 100   return 1;
 101 }
 102
 103 static int amok_hm_cb_shutdown(gras_msg_cb_ctx_t ctx, void *payload) {
 104   char *name = *(void**)payload;
 105   amok_hm_group_shutdown_local(name);
 106
 107   gras_msg_rpcreturn(30, ctx, NULL);
 108   return 1;
 109 }
 110
 111
 112 /* Initialization stuff */
 113 static short amok_hm_used = 0;
 114
 115 /** \brief Initialize the host management module. Every process must run it before use */
 116 void amok_hm_init() {
 117   /* pure INIT part */
 118   if (! amok_hm_used) {
 119
 120     /* dependencies */
 121     amok_base_init();
 122
 123     /* module data on each process */
 124     amok_hm_libdata_id = gras_procdata_add("amok_hm",
 125                                            amok_hm_libdata_new,
 126                                            amok_hm_libdata_free);
 127
 128     /* Datatype and message declarations */
 129     gras_msgtype_declare("amok_hm_kill",NULL);
 130     gras_msgtype_declare_rpc("amok_hm_kill",NULL,NULL);
 131
 132     gras_msgtype_declare_rpc("amok_hm_get",
 133                              gras_datadesc_by_name("string"),
 134                              gras_datadesc_by_name("xbt_dynar_t"));
 135     gras_msgtype_declare_rpc("amok_hm_join",
 136                              gras_datadesc_by_name("string"),
 137                              NULL);
 138     gras_msgtype_declare_rpc("amok_hm_leave",
 139                              gras_datadesc_by_name("string"),
 140                              NULL);
 141
 142     gras_msgtype_declare_rpc("amok_hm_shutdown",
 143                              gras_datadesc_by_name("string"),
 144                              NULL);
 145   }
 146   amok_hm_used++;
 147
 148   /* JOIN part */
 149   gras_cb_register(gras_msgtype_by_name("amok_hm_kill"),
 150                    &amok_hm_cb_kill);
 151   gras_cb_register(gras_msgtype_by_name("amok_hm_killrpc"),
 152                    &amok_hm_cb_killrpc);
 153
 154   gras_cb_register(gras_msgtype_by_name("amok_hm_get"),
 155                    &amok_hm_cb_get);
 156   gras_cb_register(gras_msgtype_by_name("amok_hm_join"),
 157                    &amok_hm_cb_join);
 158   gras_cb_register(gras_msgtype_by_name("amok_hm_leave"),
 159                    &amok_hm_cb_leave);
 160   gras_cb_register(gras_msgtype_by_name("amok_hm_shutdown"),
 161                    &amok_hm_cb_shutdown);
 162 }
 163
 164 /** \brief Finalize the host management module. Every process should run it after use */
 165 void amok_hm_exit() {
 166   /* pure EXIT part */
 167   amok_hm_used--;
 168
 169   /* LEAVE part */
 170   gras_cb_unregister(gras_msgtype_by_name("amok_hm_kill"),
 171                      &amok_hm_cb_kill);
 172   gras_cb_unregister(gras_msgtype_by_name("amok_hm_killrpc"),
 173                      &amok_hm_cb_killrpc);
 174
 175   gras_cb_unregister(gras_msgtype_by_name("amok_hm_get"),
 176                      &amok_hm_cb_get);
 177   gras_cb_unregister(gras_msgtype_by_name("amok_hm_join"),
 178                      &amok_hm_cb_join);
 179   gras_cb_unregister(gras_msgtype_by_name("amok_hm_leave"),
 180                      &amok_hm_cb_leave);
 181
 182   gras_cb_unregister(gras_msgtype_by_name("amok_hm_shutdown"),
 183                      &amok_hm_cb_shutdown);
 184 }
 185
 186
 187 /** \brief Enter the main loop of the program. It won't return until we get a kill message. */
 188 void amok_hm_mainloop(double timeOut) {
 189   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
 190
 191   while (!g->done) {
 192     gras_msg_handle(timeOut);
 193   }
 194 }
 195
 196 /** \brief kill a buddy identified by its hostname and port */
 197 void amok_hm_kill_hp(char *name,int port) {
 198   gras_socket_t sock=gras_socket_client(name,port);
 199   amok_hm_kill(sock);
 200   gras_socket_close(sock);
 201 }
 202
 203 /** \brief kill a buddy to which we have a socket already */
 204 void amok_hm_kill(gras_socket_t buddy) {
 205   gras_msg_send(buddy,gras_msgtype_by_name("amok_hm_kill"),NULL);
 206 }
 207
 208 /** \brief kill syncronously a buddy (do not return before its death) */
 209 void amok_hm_kill_sync(gras_socket_t buddy) {
 210   gras_msg_rpccall(buddy,30,gras_msgtype_by_name("amok_hm_killrpc"),NULL,NULL);
 211 }
 212
 213
 214 /** \brief create a new hostmanagement group located on local host
 215  *
 216  * The dynar elements are of type xbt_host_t
 217  */
 218 xbt_dynar_t amok_hm_group_new(char *group_name) {
 219   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
 220   xbt_dynar_t res = xbt_dynar_new(sizeof(xbt_host_t),
 221                                   &xbt_host_free_voidp);
 222
 223   xbt_dict_set(g->groups,group_name,res,&xbt_dynar_free_voidp);
 224
 225   return res;
 226 }
 227 /** \brief retrieve all members of the given remote group */
 228 xbt_dynar_t amok_hm_group_get(gras_socket_t master, char *group_name) {
 229   xbt_dynar_t res;
 230
 231   gras_msg_rpccall(master,30,gras_msgtype_by_name("amok_hm_get"),
 232                    &group_name,&res);
 233   return res;
 234 }
 235
 236 /** \brief add current host to the given remote group */
 237 void        amok_hm_group_join(gras_socket_t master, char *group_name) {
 238   gras_msg_rpccall(master,30,gras_msgtype_by_name("amok_hm_join"),
 239                    &group_name,NULL);
 240 }
 241 /** \brief remove current host from the given remote group if found
 242  *
 243  * If not found, call is ignored
 244  */
 245 void        amok_hm_group_leave(gras_socket_t master, char *group_name) {
 246   gras_msg_rpccall(master,30,gras_msgtype_by_name("amok_hm_leave"),
 247                    &group_name,NULL);
 248 }
 249
 250 /** \brief stops all members of the given local group */
 251 void amok_hm_group_shutdown_local(char *group_name) {
 252   amok_hm_libdata_t g=gras_libdata_by_id(amok_hm_libdata_id);
 253   xbt_dynar_t group = xbt_dict_get(g->groups, group_name);
 254
 255   int cpt;
 256   xbt_host_t host_it;
 257
 258   xbt_dynar_foreach(group, cpt, host_it) {
 259     amok_hm_kill_hp(host_it->name, host_it->port);
 260   }
 261
 262   xbt_dict_remove(g->groups,group_name);
 263 }
 264 /** \brief stops all members of the given remote group */
 265 void amok_hm_group_shutdown_remote(gras_socket_t master, char *group_name){
 266   gras_msg_rpccall(master,30,gras_msgtype_by_name("amok_hm_shutdown"),
 267                    &group_name,NULL);
 268 }