heartbeat.c

Go to the documentation of this file.

/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * Copyright (C) 2004, 2005, 2012 Oracle.  All rights reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 */

#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/configfs.h>

#include "heartbeat.h"
#include "tcp.h"
#include "nodemanager.h"

#include "masklog.h"

/*
 * The first heartbeat pass had one global thread that would serialize all hb
 * callback calls.  This global serializing sem should only be removed once
 * we've made sure that all callees can deal with being called concurrently
 * from multiple hb region threads.
 */
static DECLARE_RWSEM(r2hb_callback_sem);

/*
 * multiple hb threads are watching multiple regions.  A node is live
 * whenever any of the threads sees activity from the node in its region.
 */
static DEFINE_SPINLOCK(r2hb_live_lock);
static unsigned long r2hb_live_node_bitmap[BITS_TO_LONGS(R2NM_MAX_NODES)];

static struct r2hb_callback {
    struct list_head list;
} r2hb_callbacks[R2HB_NUM_CB];

enum r2hb_heartbeat_modes {
    R2HB_HEARTBEAT_LOCAL        = 0,
    R2HB_HEARTBEAT_GLOBAL,
    R2HB_HEARTBEAT_NUM_MODES,
};

char *r2hb_heartbeat_mode_desc[R2HB_HEARTBEAT_NUM_MODES] = {
        "local",    /* R2HB_HEARTBEAT_LOCAL */
        "global",   /* R2HB_HEARTBEAT_GLOBAL */
};

unsigned int r2hb_dead_threshold = R2HB_DEFAULT_DEAD_THRESHOLD;
unsigned int r2hb_heartbeat_mode = R2HB_HEARTBEAT_LOCAL;

/* Only sets a new threshold if there are no active regions.
 *
 * No locking or otherwise interesting code is required for reading
 * r2hb_dead_threshold as it can't change once regions are active and
 * it's not interesting to anyone until then anyway. */
static void r2hb_dead_threshold_set(unsigned int threshold)
{
    if (threshold > R2HB_MIN_DEAD_THRESHOLD) {
        spin_lock(&r2hb_live_lock);
        r2hb_dead_threshold = threshold;
        spin_unlock(&r2hb_live_lock);
    }
}

static int r2hb_global_hearbeat_mode_set(unsigned int hb_mode)
{
    int ret = -1;

    if (hb_mode < R2HB_HEARTBEAT_NUM_MODES) {
        spin_lock(&r2hb_live_lock);
        r2hb_heartbeat_mode = hb_mode;
        ret = 0;
        spin_unlock(&r2hb_live_lock);
    }

    return ret;
}

void r2hb_exit(void)
{
}

int r2hb_init(void)
{
    int i;

    for (i = 0; i < ARRAY_SIZE(r2hb_callbacks); i++)
        INIT_LIST_HEAD(&r2hb_callbacks[i].list);

    memset(r2hb_live_node_bitmap, 0, sizeof(r2hb_live_node_bitmap));

    return 0;
}

/* if we're already in a callback then we're already serialized by the sem */
static void r2hb_fill_node_map_from_callback(unsigned long *map,
                         unsigned bytes)
{
    BUG_ON(bytes < (BITS_TO_LONGS(R2NM_MAX_NODES) * sizeof(unsigned long)));

    memcpy(map, &r2hb_live_node_bitmap, bytes);
}

/*
 * get a map of all nodes that are heartbeating in any regions
 */
void r2hb_fill_node_map(unsigned long *map, unsigned bytes)
{
    /* callers want to serialize this map and callbacks so that they
     * can trust that they don't miss nodes coming to the party */
    down_read(&r2hb_callback_sem);
    spin_lock(&r2hb_live_lock);
    r2hb_fill_node_map_from_callback(map, bytes);
    spin_unlock(&r2hb_live_lock);
    up_read(&r2hb_callback_sem);
}
EXPORT_SYMBOL_GPL(r2hb_fill_node_map);

/*
 * heartbeat configfs bits.  The heartbeat set is a default set under
 * the cluster set in nodemanager.c.
 */

/* heartbeat set */

struct r2hb_hb_group {
    struct config_group hs_group;
    /* some stuff? */
};

static struct r2hb_hb_group *to_r2hb_hb_group(struct config_group *group)
{
    return group ?
        container_of(group, struct r2hb_hb_group, hs_group)
        : NULL;
}

static struct config_item r2hb_config_item;

static struct config_item *r2hb_hb_group_make_item(struct config_group *group,
                              const char *name)
{
    int ret;

    if (strlen(name) > R2HB_MAX_REGION_NAME_LEN) {
        ret = -ENAMETOOLONG;
        goto free;
    }

    config_item_put(&r2hb_config_item);

    return &r2hb_config_item;
free:
    return ERR_PTR(ret);
}

static void r2hb_hb_group_drop_item(struct config_group *group,
                       struct config_item *item)
{
    if (r2hb_global_heartbeat_active()) {
        pr_notice("ramster: Heartbeat %s on region %s (%s)\n",
            "stopped/aborted", config_item_name(item),
            "no region");
    }

    config_item_put(item);
}

struct r2hb_hb_group_attribute {
    struct configfs_attribute attr;
    ssize_t (*show)(struct r2hb_hb_group *, char *);
    ssize_t (*store)(struct r2hb_hb_group *, const char *, size_t);
};

static ssize_t r2hb_hb_group_show(struct config_item *item,
                     struct configfs_attribute *attr,
                     char *page)
{
    struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
    struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
        container_of(attr, struct r2hb_hb_group_attribute, attr);
    ssize_t ret = 0;

    if (r2hb_hb_group_attr->show)
        ret = r2hb_hb_group_attr->show(reg, page);
    return ret;
}

static ssize_t r2hb_hb_group_store(struct config_item *item,
                      struct configfs_attribute *attr,
                      const char *page, size_t count)
{
    struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
    struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
        container_of(attr, struct r2hb_hb_group_attribute, attr);
    ssize_t ret = -EINVAL;

    if (r2hb_hb_group_attr->store)
        ret = r2hb_hb_group_attr->store(reg, page, count);
    return ret;
}

static ssize_t r2hb_hb_group_threshold_show(struct r2hb_hb_group *group,
                             char *page)
{
    return sprintf(page, "%u\n", r2hb_dead_threshold);
}

static ssize_t r2hb_hb_group_threshold_store(struct r2hb_hb_group *group,
                            const char *page,
                            size_t count)
{
    unsigned long tmp;
    char *p = (char *)page;
    int err;

    err = kstrtoul(p, 10, &tmp);
    if (err)
        return err;

    /* this will validate ranges for us. */
    r2hb_dead_threshold_set((unsigned int) tmp);

    return count;
}

static
ssize_t r2hb_hb_group_mode_show(struct r2hb_hb_group *group,
                       char *page)
{
    return sprintf(page, "%s\n",
               r2hb_heartbeat_mode_desc[r2hb_heartbeat_mode]);
}

static
ssize_t r2hb_hb_group_mode_store(struct r2hb_hb_group *group,
                    const char *page, size_t count)
{
    unsigned int i;
    int ret;
    size_t len;

    len = (page[count - 1] == '\n') ? count - 1 : count;
    if (!len)
        return -EINVAL;

    for (i = 0; i < R2HB_HEARTBEAT_NUM_MODES; ++i) {
        if (strnicmp(page, r2hb_heartbeat_mode_desc[i], len))
            continue;

        ret = r2hb_global_hearbeat_mode_set(i);
        if (!ret)
            pr_notice("ramster: Heartbeat mode set to %s\n",
                   r2hb_heartbeat_mode_desc[i]);
        return count;
    }

    return -EINVAL;

}

static struct r2hb_hb_group_attribute r2hb_hb_group_attr_threshold = {
    .attr   = { .ca_owner = THIS_MODULE,
            .ca_name = "dead_threshold",
            .ca_mode = S_IRUGO | S_IWUSR },
    .show   = r2hb_hb_group_threshold_show,
    .store  = r2hb_hb_group_threshold_store,
};

static struct r2hb_hb_group_attribute r2hb_hb_group_attr_mode = {
    .attr   = { .ca_owner = THIS_MODULE,
        .ca_name = "mode",
        .ca_mode = S_IRUGO | S_IWUSR },
    .show   = r2hb_hb_group_mode_show,
    .store  = r2hb_hb_group_mode_store,
};

static struct configfs_attribute *r2hb_hb_group_attrs[] = {
    &r2hb_hb_group_attr_threshold.attr,
    &r2hb_hb_group_attr_mode.attr,
    NULL,
};

static struct configfs_item_operations r2hb_hearbeat_group_item_ops = {
    .show_attribute     = r2hb_hb_group_show,
    .store_attribute    = r2hb_hb_group_store,
};

static struct configfs_group_operations r2hb_hb_group_group_ops = {
    .make_item  = r2hb_hb_group_make_item,
    .drop_item  = r2hb_hb_group_drop_item,
};

static struct config_item_type r2hb_hb_group_type = {
    .ct_group_ops   = &r2hb_hb_group_group_ops,
    .ct_item_ops    = &r2hb_hearbeat_group_item_ops,
    .ct_attrs   = r2hb_hb_group_attrs,
    .ct_owner   = THIS_MODULE,
};

/* this is just here to avoid touching group in heartbeat.h which the
 * entire damn world #includes */
struct config_group *r2hb_alloc_hb_set(void)
{
    struct r2hb_hb_group *hs = NULL;
    struct config_group *ret = NULL;

    hs = kzalloc(sizeof(struct r2hb_hb_group), GFP_KERNEL);
    if (hs == NULL)
        goto out;

    config_group_init_type_name(&hs->hs_group, "heartbeat",
                    &r2hb_hb_group_type);

    ret = &hs->hs_group;
out:
    if (ret == NULL)
        kfree(hs);
    return ret;
}

void r2hb_free_hb_set(struct config_group *group)
{
    struct r2hb_hb_group *hs = to_r2hb_hb_group(group);
    kfree(hs);
}

/* hb callback registration and issuing */

static struct r2hb_callback *hbcall_from_type(enum r2hb_callback_type type)
{
    if (type == R2HB_NUM_CB)
        return ERR_PTR(-EINVAL);

    return &r2hb_callbacks[type];
}

void r2hb_setup_callback(struct r2hb_callback_func *hc,
             enum r2hb_callback_type type,
             r2hb_cb_func *func,
             void *data,
             int priority)
{
    INIT_LIST_HEAD(&hc->hc_item);
    hc->hc_func = func;
    hc->hc_data = data;
    hc->hc_priority = priority;
    hc->hc_type = type;
    hc->hc_magic = R2HB_CB_MAGIC;
}
EXPORT_SYMBOL_GPL(r2hb_setup_callback);

int r2hb_register_callback(const char *region_uuid,
               struct r2hb_callback_func *hc)
{
    struct r2hb_callback_func *tmp;
    struct list_head *iter;
    struct r2hb_callback *hbcall;
    int ret;

    BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);
    BUG_ON(!list_empty(&hc->hc_item));

    hbcall = hbcall_from_type(hc->hc_type);
    if (IS_ERR(hbcall)) {
        ret = PTR_ERR(hbcall);
        goto out;
    }

    down_write(&r2hb_callback_sem);

    list_for_each(iter, &hbcall->list) {
        tmp = list_entry(iter, struct r2hb_callback_func, hc_item);
        if (hc->hc_priority < tmp->hc_priority) {
            list_add_tail(&hc->hc_item, iter);
            break;
        }
    }
    if (list_empty(&hc->hc_item))
        list_add_tail(&hc->hc_item, &hbcall->list);

    up_write(&r2hb_callback_sem);
    ret = 0;
out:
    mlog(ML_CLUSTER, "returning %d on behalf of %p for funcs %p\n",
         ret, __builtin_return_address(0), hc);
    return ret;
}
EXPORT_SYMBOL_GPL(r2hb_register_callback);

void r2hb_unregister_callback(const char *region_uuid,
                  struct r2hb_callback_func *hc)
{
    BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);

    mlog(ML_CLUSTER, "on behalf of %p for funcs %p\n",
         __builtin_return_address(0), hc);

    /* XXX Can this happen _with_ a region reference? */
    if (list_empty(&hc->hc_item))
        return;

    down_write(&r2hb_callback_sem);

    list_del_init(&hc->hc_item);

    up_write(&r2hb_callback_sem);
}
EXPORT_SYMBOL_GPL(r2hb_unregister_callback);

int r2hb_check_node_heartbeating_from_callback(u8 node_num)
{
    unsigned long testing_map[BITS_TO_LONGS(R2NM_MAX_NODES)];

    r2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
    if (!test_bit(node_num, testing_map)) {
        mlog(ML_HEARTBEAT,
             "node (%u) does not have heartbeating enabled.\n",
             node_num);
        return 0;
    }

    return 1;
}
EXPORT_SYMBOL_GPL(r2hb_check_node_heartbeating_from_callback);

void r2hb_stop_all_regions(void)
{
}
EXPORT_SYMBOL_GPL(r2hb_stop_all_regions);

/*
 * this is just a hack until we get the plumbing which flips file systems
 * read only and drops the hb ref instead of killing the node dead.
 */
int r2hb_global_heartbeat_active(void)
{
    return (r2hb_heartbeat_mode == R2HB_HEARTBEAT_GLOBAL);
}
EXPORT_SYMBOL(r2hb_global_heartbeat_active);

/* added for RAMster */
void r2hb_manual_set_node_heartbeating(int node_num)
{
    if (node_num < R2NM_MAX_NODES)
        set_bit(node_num, r2hb_live_node_bitmap);
}
EXPORT_SYMBOL(r2hb_manual_set_node_heartbeating);