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