drivers/staging/ramster/cluster/heartbeat.c

   1 /* -*- mode: c; c-basic-offset: 8; -*-
   2  * vim: noexpandtab sw=8 ts=8 sts=0:
   3  *
   4  * Copyright (C) 2004, 2005, 2012 Oracle.  All rights reserved.
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public
  17  * License along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA.
  20  */
  21
  22 #include <linux/kernel.h>
  23 #include <linux/module.h>
  24 #include <linux/configfs.h>
  25
  26 #include "heartbeat.h"
  27 #include "tcp.h"
  28 #include "nodemanager.h"
  29
  30 #include "masklog.h"
  31
  32 /*
  33  * The first heartbeat pass had one global thread that would serialize all hb
  34  * callback calls.  This global serializing sem should only be removed once
  35  * we've made sure that all callees can deal with being called concurrently
  36  * from multiple hb region threads.
  37  */
  38 static DECLARE_RWSEM(r2hb_callback_sem);
  39
  40 /*
  41  * multiple hb threads are watching multiple regions.  A node is live
  42  * whenever any of the threads sees activity from the node in its region.
  43  */
  44 static DEFINE_SPINLOCK(r2hb_live_lock);
  45 static unsigned long r2hb_live_node_bitmap[BITS_TO_LONGS(R2NM_MAX_NODES)];
  46
  47 static struct r2hb_callback {
  48         struct list_head list;
  49 } r2hb_callbacks[R2HB_NUM_CB];
  50
  51 enum r2hb_heartbeat_modes {
  52         R2HB_HEARTBEAT_LOCAL            = 0,
  53         R2HB_HEARTBEAT_GLOBAL,
  54         R2HB_HEARTBEAT_NUM_MODES,
  55 };
  56
  57 char *r2hb_heartbeat_mode_desc[R2HB_HEARTBEAT_NUM_MODES] = {
  58                 "local",        /* R2HB_HEARTBEAT_LOCAL */
  59                 "global",       /* R2HB_HEARTBEAT_GLOBAL */
  60 };
  61
  62 unsigned int r2hb_dead_threshold = R2HB_DEFAULT_DEAD_THRESHOLD;
  63 unsigned int r2hb_heartbeat_mode = R2HB_HEARTBEAT_LOCAL;
  64
  65 /* Only sets a new threshold if there are no active regions.
  66  *
  67  * No locking or otherwise interesting code is required for reading
  68  * r2hb_dead_threshold as it can't change once regions are active and
  69  * it's not interesting to anyone until then anyway. */
  70 static void r2hb_dead_threshold_set(unsigned int threshold)
  71 {
  72         if (threshold > R2HB_MIN_DEAD_THRESHOLD) {
  73                 spin_lock(&r2hb_live_lock);
  74                 r2hb_dead_threshold = threshold;
  75                 spin_unlock(&r2hb_live_lock);
  76         }
  77 }
  78
  79 static int r2hb_global_hearbeat_mode_set(unsigned int hb_mode)
  80 {
  81         int ret = -1;
  82
  83         if (hb_mode < R2HB_HEARTBEAT_NUM_MODES) {
  84                 spin_lock(&r2hb_live_lock);
  85                 r2hb_heartbeat_mode = hb_mode;
  86                 ret = 0;
  87                 spin_unlock(&r2hb_live_lock);
  88         }
  89
  90         return ret;
  91 }
  92
  93 void r2hb_exit(void)
  94 {
  95 }
  96
  97 int r2hb_init(void)
  98 {
  99         int i;
 100
 101         for (i = 0; i < ARRAY_SIZE(r2hb_callbacks); i++)
 102                 INIT_LIST_HEAD(&r2hb_callbacks[i].list);
 103
 104         memset(r2hb_live_node_bitmap, 0, sizeof(r2hb_live_node_bitmap));
 105
 106         return 0;
 107 }
 108
 109 /* if we're already in a callback then we're already serialized by the sem */
 110 static void r2hb_fill_node_map_from_callback(unsigned long *map,
 111                                              unsigned bytes)
 112 {
 113         BUG_ON(bytes < (BITS_TO_LONGS(R2NM_MAX_NODES) * sizeof(unsigned long)));
 114
 115         memcpy(map, &r2hb_live_node_bitmap, bytes);
 116 }
 117
 118 /*
 119  * get a map of all nodes that are heartbeating in any regions
 120  */
 121 void r2hb_fill_node_map(unsigned long *map, unsigned bytes)
 122 {
 123         /* callers want to serialize this map and callbacks so that they
 124          * can trust that they don't miss nodes coming to the party */
 125         down_read(&r2hb_callback_sem);
 126         spin_lock(&r2hb_live_lock);
 127         r2hb_fill_node_map_from_callback(map, bytes);
 128         spin_unlock(&r2hb_live_lock);
 129         up_read(&r2hb_callback_sem);
 130 }
 131 EXPORT_SYMBOL_GPL(r2hb_fill_node_map);
 132
 133 /*
 134  * heartbeat configfs bits.  The heartbeat set is a default set under
 135  * the cluster set in nodemanager.c.
 136  */
 137
 138 /* heartbeat set */
 139
 140 struct r2hb_hb_group {
 141         struct config_group hs_group;
 142         /* some stuff? */
 143 };
 144
 145 static struct r2hb_hb_group *to_r2hb_hb_group(struct config_group *group)
 146 {
 147         return group ?
 148                 container_of(group, struct r2hb_hb_group, hs_group)
 149                 : NULL;
 150 }
 151
 152 static struct config_item r2hb_config_item;
 153
 154 static struct config_item *r2hb_hb_group_make_item(struct config_group *group,
 155                                                           const char *name)
 156 {
 157         int ret;
 158
 159         if (strlen(name) > R2HB_MAX_REGION_NAME_LEN) {
 160                 ret = -ENAMETOOLONG;
 161                 goto free;
 162         }
 163
 164         config_item_put(&r2hb_config_item);
 165
 166         return &r2hb_config_item;
 167 free:
 168         return ERR_PTR(ret);
 169 }
 170
 171 static void r2hb_hb_group_drop_item(struct config_group *group,
 172                                            struct config_item *item)
 173 {
 174         if (r2hb_global_heartbeat_active()) {
 175                 printk(KERN_NOTICE "ramster: Heartbeat %s "
 176                         "on region %s (%s)\n",
 177                         "stopped/aborted", config_item_name(item),
 178                         "no region");
 179         }
 180
 181         config_item_put(item);
 182 }
 183
 184 struct r2hb_hb_group_attribute {
 185         struct configfs_attribute attr;
 186         ssize_t (*show)(struct r2hb_hb_group *, char *);
 187         ssize_t (*store)(struct r2hb_hb_group *, const char *, size_t);
 188 };
 189
 190 static ssize_t r2hb_hb_group_show(struct config_item *item,
 191                                          struct configfs_attribute *attr,
 192                                          char *page)
 193 {
 194         struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
 195         struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
 196                 container_of(attr, struct r2hb_hb_group_attribute, attr);
 197         ssize_t ret = 0;
 198
 199         if (r2hb_hb_group_attr->show)
 200                 ret = r2hb_hb_group_attr->show(reg, page);
 201         return ret;
 202 }
 203
 204 static ssize_t r2hb_hb_group_store(struct config_item *item,
 205                                           struct configfs_attribute *attr,
 206                                           const char *page, size_t count)
 207 {
 208         struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
 209         struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
 210                 container_of(attr, struct r2hb_hb_group_attribute, attr);
 211         ssize_t ret = -EINVAL;
 212
 213         if (r2hb_hb_group_attr->store)
 214                 ret = r2hb_hb_group_attr->store(reg, page, count);
 215         return ret;
 216 }
 217
 218 static ssize_t r2hb_hb_group_threshold_show(struct r2hb_hb_group *group,
 219                                                      char *page)
 220 {
 221         return sprintf(page, "%u\n", r2hb_dead_threshold);
 222 }
 223
 224 static ssize_t r2hb_hb_group_threshold_store(struct r2hb_hb_group *group,
 225                                                     const char *page,
 226                                                     size_t count)
 227 {
 228         unsigned long tmp;
 229         char *p = (char *)page;
 230         int err;
 231
 232         err = kstrtoul(p, 10, &tmp);
 233         if (err)
 234                 return err;
 235
 236         /* this will validate ranges for us. */
 237         r2hb_dead_threshold_set((unsigned int) tmp);
 238
 239         return count;
 240 }
 241
 242 static
 243 ssize_t r2hb_hb_group_mode_show(struct r2hb_hb_group *group,
 244                                        char *page)
 245 {
 246         return sprintf(page, "%s\n",
 247                        r2hb_heartbeat_mode_desc[r2hb_heartbeat_mode]);
 248 }
 249
 250 static
 251 ssize_t r2hb_hb_group_mode_store(struct r2hb_hb_group *group,
 252                                         const char *page, size_t count)
 253 {
 254         unsigned int i;
 255         int ret;
 256         size_t len;
 257
 258         len = (page[count - 1] == '\n') ? count - 1 : count;
 259         if (!len)
 260                 return -EINVAL;
 261
 262         for (i = 0; i < R2HB_HEARTBEAT_NUM_MODES; ++i) {
 263                 if (strnicmp(page, r2hb_heartbeat_mode_desc[i], len))
 264                         continue;
 265
 266                 ret = r2hb_global_hearbeat_mode_set(i);
 267                 if (!ret)
 268                         printk(KERN_NOTICE "ramster: Heartbeat mode "
 269                                                 "set to %s\n",
 270                                r2hb_heartbeat_mode_desc[i]);
 271                 return count;
 272         }
 273
 274         return -EINVAL;
 275
 276 }
 277
 278 static struct r2hb_hb_group_attribute r2hb_hb_group_attr_threshold = {
 279         .attr   = { .ca_owner = THIS_MODULE,
 280                     .ca_name = "dead_threshold",
 281                     .ca_mode = S_IRUGO | S_IWUSR },
 282         .show   = r2hb_hb_group_threshold_show,
 283         .store  = r2hb_hb_group_threshold_store,
 284 };
 285
 286 static struct r2hb_hb_group_attribute r2hb_hb_group_attr_mode = {
 287         .attr   = { .ca_owner = THIS_MODULE,
 288                 .ca_name = "mode",
 289                 .ca_mode = S_IRUGO | S_IWUSR },
 290         .show   = r2hb_hb_group_mode_show,
 291         .store  = r2hb_hb_group_mode_store,
 292 };
 293
 294 static struct configfs_attribute *r2hb_hb_group_attrs[] = {
 295         &r2hb_hb_group_attr_threshold.attr,
 296         &r2hb_hb_group_attr_mode.attr,
 297         NULL,
 298 };
 299
 300 static struct configfs_item_operations r2hb_hearbeat_group_item_ops = {
 301         .show_attribute         = r2hb_hb_group_show,
 302         .store_attribute        = r2hb_hb_group_store,
 303 };
 304
 305 static struct configfs_group_operations r2hb_hb_group_group_ops = {
 306         .make_item      = r2hb_hb_group_make_item,
 307         .drop_item      = r2hb_hb_group_drop_item,
 308 };
 309
 310 static struct config_item_type r2hb_hb_group_type = {
 311         .ct_group_ops   = &r2hb_hb_group_group_ops,
 312         .ct_item_ops    = &r2hb_hearbeat_group_item_ops,
 313         .ct_attrs       = r2hb_hb_group_attrs,
 314         .ct_owner       = THIS_MODULE,
 315 };
 316
 317 /* this is just here to avoid touching group in heartbeat.h which the
 318  * entire damn world #includes */
 319 struct config_group *r2hb_alloc_hb_set(void)
 320 {
 321         struct r2hb_hb_group *hs = NULL;
 322         struct config_group *ret = NULL;
 323
 324         hs = kzalloc(sizeof(struct r2hb_hb_group), GFP_KERNEL);
 325         if (hs == NULL)
 326                 goto out;
 327
 328         config_group_init_type_name(&hs->hs_group, "heartbeat",
 329                                     &r2hb_hb_group_type);
 330
 331         ret = &hs->hs_group;
 332 out:
 333         if (ret == NULL)
 334                 kfree(hs);
 335         return ret;
 336 }
 337
 338 void r2hb_free_hb_set(struct config_group *group)
 339 {
 340         struct r2hb_hb_group *hs = to_r2hb_hb_group(group);
 341         kfree(hs);
 342 }
 343
 344 /* hb callback registration and issuing */
 345
 346 static struct r2hb_callback *hbcall_from_type(enum r2hb_callback_type type)
 347 {
 348         if (type == R2HB_NUM_CB)
 349                 return ERR_PTR(-EINVAL);
 350
 351         return &r2hb_callbacks[type];
 352 }
 353
 354 void r2hb_setup_callback(struct r2hb_callback_func *hc,
 355                          enum r2hb_callback_type type,
 356                          r2hb_cb_func *func,
 357                          void *data,
 358                          int priority)
 359 {
 360         INIT_LIST_HEAD(&hc->hc_item);
 361         hc->hc_func = func;
 362         hc->hc_data = data;
 363         hc->hc_priority = priority;
 364         hc->hc_type = type;
 365         hc->hc_magic = R2HB_CB_MAGIC;
 366 }
 367 EXPORT_SYMBOL_GPL(r2hb_setup_callback);
 368
 369 int r2hb_register_callback(const char *region_uuid,
 370                            struct r2hb_callback_func *hc)
 371 {
 372         struct r2hb_callback_func *tmp;
 373         struct list_head *iter;
 374         struct r2hb_callback *hbcall;
 375         int ret;
 376
 377         BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);
 378         BUG_ON(!list_empty(&hc->hc_item));
 379
 380         hbcall = hbcall_from_type(hc->hc_type);
 381         if (IS_ERR(hbcall)) {
 382                 ret = PTR_ERR(hbcall);
 383                 goto out;
 384         }
 385
 386         down_write(&r2hb_callback_sem);
 387
 388         list_for_each(iter, &hbcall->list) {
 389                 tmp = list_entry(iter, struct r2hb_callback_func, hc_item);
 390                 if (hc->hc_priority < tmp->hc_priority) {
 391                         list_add_tail(&hc->hc_item, iter);
 392                         break;
 393                 }
 394         }
 395         if (list_empty(&hc->hc_item))
 396                 list_add_tail(&hc->hc_item, &hbcall->list);
 397
 398         up_write(&r2hb_callback_sem);
 399         ret = 0;
 400 out:
 401         mlog(ML_CLUSTER, "returning %d on behalf of %p for funcs %p\n",
 402              ret, __builtin_return_address(0), hc);
 403         return ret;
 404 }
 405 EXPORT_SYMBOL_GPL(r2hb_register_callback);
 406
 407 void r2hb_unregister_callback(const char *region_uuid,
 408                               struct r2hb_callback_func *hc)
 409 {
 410         BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);
 411
 412         mlog(ML_CLUSTER, "on behalf of %p for funcs %p\n",
 413              __builtin_return_address(0), hc);
 414
 415         /* XXX Can this happen _with_ a region reference? */
 416         if (list_empty(&hc->hc_item))
 417                 return;
 418
 419         down_write(&r2hb_callback_sem);
 420
 421         list_del_init(&hc->hc_item);
 422
 423         up_write(&r2hb_callback_sem);
 424 }
 425 EXPORT_SYMBOL_GPL(r2hb_unregister_callback);
 426
 427 int r2hb_check_node_heartbeating_from_callback(u8 node_num)
 428 {
 429         unsigned long testing_map[BITS_TO_LONGS(R2NM_MAX_NODES)];
 430
 431         r2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
 432         if (!test_bit(node_num, testing_map)) {
 433                 mlog(ML_HEARTBEAT,
 434                      "node (%u) does not have heartbeating enabled.\n",
 435                      node_num);
 436                 return 0;
 437         }
 438
 439         return 1;
 440 }
 441 EXPORT_SYMBOL_GPL(r2hb_check_node_heartbeating_from_callback);
 442
 443 void r2hb_stop_all_regions(void)
 444 {
 445 }
 446 EXPORT_SYMBOL_GPL(r2hb_stop_all_regions);
 447
 448 /*
 449  * this is just a hack until we get the plumbing which flips file systems
 450  * read only and drops the hb ref instead of killing the node dead.
 451  */
 452 int r2hb_global_heartbeat_active(void)
 453 {
 454         return (r2hb_heartbeat_mode == R2HB_HEARTBEAT_GLOBAL);
 455 }
 456 EXPORT_SYMBOL(r2hb_global_heartbeat_active);
 457
 458 /* added for RAMster */
 459 void r2hb_manual_set_node_heartbeating(int node_num)
 460 {
 461         if (node_num < R2NM_MAX_NODES)
 462                 set_bit(node_num, r2hb_live_node_bitmap);
 463 }
 464 EXPORT_SYMBOL(r2hb_manual_set_node_heartbeating);