fs/ocfs2/stackglue.c

   1 /* -*- mode: c; c-basic-offset: 8; -*-
   2  * vim: noexpandtab sw=8 ts=8 sts=0:
   3  *
   4  * stackglue.c
   5  *
   6  * Code which implements an OCFS2 specific interface to underlying
   7  * cluster stacks.
   8  *
   9  * Copyright (C) 2007 Oracle.  All rights reserved.
  10  *
  11  * This program is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU General Public
  13  * License as published by the Free Software Foundation, version 2.
  14  *
  15  * This program is distributed in the hope that it will be useful,
  16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18  * General Public License for more details.
  19  */
  20
  21 #include <linux/list.h>
  22 #include <linux/spinlock.h>
  23 #include <linux/module.h>
  24 #include <linux/slab.h>
  25 #include <linux/kmod.h>
  26 #include <linux/fs.h>
  27 #include <linux/kobject.h>
  28 #include <linux/sysfs.h>
  29 #include <linux/sysctl.h>
  30
  31 #include "ocfs2_fs.h"
  32
  33 #include "stackglue.h"
  34
  35 #define OCFS2_STACK_PLUGIN_O2CB         "o2cb"
  36 #define OCFS2_STACK_PLUGIN_USER         "user"
  37
  38 static struct ocfs2_locking_protocol *lproto;
  39 static DEFINE_SPINLOCK(ocfs2_stack_lock);
  40 static LIST_HEAD(ocfs2_stack_list);
  41 static char cluster_stack_name[OCFS2_STACK_LABEL_LEN + 1];
  42
  43 /*
  44  * The stack currently in use.  If not null, active_stack->sp_count > 0,
  45  * the module is pinned, and the locking protocol cannot be changed.
  46  */
  47 static struct ocfs2_stack_plugin *active_stack;
  48
  49 static struct ocfs2_stack_plugin *ocfs2_stack_lookup(const char *name)
  50 {
  51         struct ocfs2_stack_plugin *p;
  52
  53         assert_spin_locked(&ocfs2_stack_lock);
  54
  55         list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
  56                 if (!strcmp(p->sp_name, name))
  57                         return p;
  58         }
  59
  60         return NULL;
  61 }
  62
  63 static int ocfs2_stack_driver_request(const char *stack_name,
  64                                       const char *plugin_name)
  65 {
  66         int rc;
  67         struct ocfs2_stack_plugin *p;
  68
  69         spin_lock(&ocfs2_stack_lock);
  70
  71         /*
  72          * If the stack passed by the filesystem isn't the selected one,
  73          * we can't continue.
  74          */
  75         if (strcmp(stack_name, cluster_stack_name)) {
  76                 rc = -EBUSY;
  77                 goto out;
  78         }
  79
  80         if (active_stack) {
  81                 /*
  82                  * If the active stack isn't the one we want, it cannot
  83                  * be selected right now.
  84                  */
  85                 if (!strcmp(active_stack->sp_name, plugin_name))
  86                         rc = 0;
  87                 else
  88                         rc = -EBUSY;
  89                 goto out;
  90         }
  91
  92         p = ocfs2_stack_lookup(plugin_name);
  93         if (!p || !try_module_get(p->sp_owner)) {
  94                 rc = -ENOENT;
  95                 goto out;
  96         }
  97
  98         /* Ok, the stack is pinned */
  99         p->sp_count++;
 100         active_stack = p;
 101
 102         rc = 0;
 103
 104 out:
 105         spin_unlock(&ocfs2_stack_lock);
 106         return rc;
 107 }
 108
 109 /*
 110  * This function looks up the appropriate stack and makes it active.  If
 111  * there is no stack, it tries to load it.  It will fail if the stack still
 112  * cannot be found.  It will also fail if a different stack is in use.
 113  */
 114 static int ocfs2_stack_driver_get(const char *stack_name)
 115 {
 116         int rc;
 117         char *plugin_name = OCFS2_STACK_PLUGIN_O2CB;
 118
 119         /*
 120          * Classic stack does not pass in a stack name.  This is
 121          * compatible with older tools as well.
 122          */
 123         if (!stack_name || !*stack_name)
 124                 stack_name = OCFS2_STACK_PLUGIN_O2CB;
 125
 126         if (strlen(stack_name) != OCFS2_STACK_LABEL_LEN) {
 127                 printk(KERN_ERR
 128                        "ocfs2 passed an invalid cluster stack label: \"%s\"\n",
 129                        stack_name);
 130                 return -EINVAL;
 131         }
 132
 133         /* Anything that isn't the classic stack is a user stack */
 134         if (strcmp(stack_name, OCFS2_STACK_PLUGIN_O2CB))
 135                 plugin_name = OCFS2_STACK_PLUGIN_USER;
 136
 137         rc = ocfs2_stack_driver_request(stack_name, plugin_name);
 138         if (rc == -ENOENT) {
 139                 request_module("ocfs2_stack_%s", plugin_name);
 140                 rc = ocfs2_stack_driver_request(stack_name, plugin_name);
 141         }
 142
 143         if (rc == -ENOENT) {
 144                 printk(KERN_ERR
 145                        "ocfs2: Cluster stack driver \"%s\" cannot be found\n",
 146                        plugin_name);
 147         } else if (rc == -EBUSY) {
 148                 printk(KERN_ERR
 149                        "ocfs2: A different cluster stack is in use\n");
 150         }
 151
 152         return rc;
 153 }
 154
 155 static void ocfs2_stack_driver_put(void)
 156 {
 157         spin_lock(&ocfs2_stack_lock);
 158         BUG_ON(active_stack == NULL);
 159         BUG_ON(active_stack->sp_count == 0);
 160
 161         active_stack->sp_count--;
 162         if (!active_stack->sp_count) {
 163                 module_put(active_stack->sp_owner);
 164                 active_stack = NULL;
 165         }
 166         spin_unlock(&ocfs2_stack_lock);
 167 }
 168
 169 int ocfs2_stack_glue_register(struct ocfs2_stack_plugin *plugin)
 170 {
 171         int rc;
 172
 173         spin_lock(&ocfs2_stack_lock);
 174         if (!ocfs2_stack_lookup(plugin->sp_name)) {
 175                 plugin->sp_count = 0;
 176                 plugin->sp_proto = lproto;
 177                 list_add(&plugin->sp_list, &ocfs2_stack_list);
 178                 printk(KERN_INFO "ocfs2: Registered cluster interface %s\n",
 179                        plugin->sp_name);
 180                 rc = 0;
 181         } else {
 182                 printk(KERN_ERR "ocfs2: Stack \"%s\" already registered\n",
 183                        plugin->sp_name);
 184                 rc = -EEXIST;
 185         }
 186         spin_unlock(&ocfs2_stack_lock);
 187
 188         return rc;
 189 }
 190 EXPORT_SYMBOL_GPL(ocfs2_stack_glue_register);
 191
 192 void ocfs2_stack_glue_unregister(struct ocfs2_stack_plugin *plugin)
 193 {
 194         struct ocfs2_stack_plugin *p;
 195
 196         spin_lock(&ocfs2_stack_lock);
 197         p = ocfs2_stack_lookup(plugin->sp_name);
 198         if (p) {
 199                 BUG_ON(p != plugin);
 200                 BUG_ON(plugin == active_stack);
 201                 BUG_ON(plugin->sp_count != 0);
 202                 list_del_init(&plugin->sp_list);
 203                 printk(KERN_INFO "ocfs2: Unregistered cluster interface %s\n",
 204                        plugin->sp_name);
 205         } else {
 206                 printk(KERN_ERR "Stack \"%s\" is not registered\n",
 207                        plugin->sp_name);
 208         }
 209         spin_unlock(&ocfs2_stack_lock);
 210 }
 211 EXPORT_SYMBOL_GPL(ocfs2_stack_glue_unregister);
 212
 213 void ocfs2_stack_glue_set_locking_protocol(struct ocfs2_locking_protocol *proto)
 214 {
 215         struct ocfs2_stack_plugin *p;
 216
 217         BUG_ON(proto == NULL);
 218
 219         spin_lock(&ocfs2_stack_lock);
 220         BUG_ON(active_stack != NULL);
 221
 222         lproto = proto;
 223         list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
 224                 p->sp_proto = lproto;
 225         }
 226
 227         spin_unlock(&ocfs2_stack_lock);
 228 }
 229 EXPORT_SYMBOL_GPL(ocfs2_stack_glue_set_locking_protocol);
 230
 231
 232 /*
 233  * The ocfs2_dlm_lock() and ocfs2_dlm_unlock() functions take
 234  * "struct ocfs2_lock_res *astarg" instead of "void *astarg" because the
 235  * underlying stack plugins need to pilfer the lksb off of the lock_res.
 236  * If some other structure needs to be passed as an astarg, the plugins
 237  * will need to be given a different avenue to the lksb.
 238  */
 239 int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
 240                    int mode,
 241                    union ocfs2_dlm_lksb *lksb,
 242                    u32 flags,
 243                    void *name,
 244                    unsigned int namelen,
 245                    struct ocfs2_lock_res *astarg)
 246 {
 247         BUG_ON(lproto == NULL);
 248
 249         return active_stack->sp_ops->dlm_lock(conn, mode, lksb, flags,
 250                                               name, namelen, astarg);
 251 }
 252 EXPORT_SYMBOL_GPL(ocfs2_dlm_lock);
 253
 254 int ocfs2_dlm_unlock(struct ocfs2_cluster_connection *conn,
 255                      union ocfs2_dlm_lksb *lksb,
 256                      u32 flags,
 257                      struct ocfs2_lock_res *astarg)
 258 {
 259         BUG_ON(lproto == NULL);
 260
 261         return active_stack->sp_ops->dlm_unlock(conn, lksb, flags, astarg);
 262 }
 263 EXPORT_SYMBOL_GPL(ocfs2_dlm_unlock);
 264
 265 int ocfs2_dlm_lock_status(union ocfs2_dlm_lksb *lksb)
 266 {
 267         return active_stack->sp_ops->lock_status(lksb);
 268 }
 269 EXPORT_SYMBOL_GPL(ocfs2_dlm_lock_status);
 270
 271 /*
 272  * Why don't we cast to ocfs2_meta_lvb?  The "clean" answer is that we
 273  * don't cast at the glue level.  The real answer is that the header
 274  * ordering is nigh impossible.
 275  */
 276 void *ocfs2_dlm_lvb(union ocfs2_dlm_lksb *lksb)
 277 {
 278         return active_stack->sp_ops->lock_lvb(lksb);
 279 }
 280 EXPORT_SYMBOL_GPL(ocfs2_dlm_lvb);
 281
 282 void ocfs2_dlm_dump_lksb(union ocfs2_dlm_lksb *lksb)
 283 {
 284         active_stack->sp_ops->dump_lksb(lksb);
 285 }
 286 EXPORT_SYMBOL_GPL(ocfs2_dlm_dump_lksb);
 287
 288 int ocfs2_cluster_connect(const char *stack_name,
 289                           const char *group,
 290                           int grouplen,
 291                           void (*recovery_handler)(int node_num,
 292                                                    void *recovery_data),
 293                           void *recovery_data,
 294                           struct ocfs2_cluster_connection **conn)
 295 {
 296         int rc = 0;
 297         struct ocfs2_cluster_connection *new_conn;
 298
 299         BUG_ON(group == NULL);
 300         BUG_ON(conn == NULL);
 301         BUG_ON(recovery_handler == NULL);
 302
 303         if (grouplen > GROUP_NAME_MAX) {
 304                 rc = -EINVAL;
 305                 goto out;
 306         }
 307
 308         new_conn = kzalloc(sizeof(struct ocfs2_cluster_connection),
 309                            GFP_KERNEL);
 310         if (!new_conn) {
 311                 rc = -ENOMEM;
 312                 goto out;
 313         }
 314
 315         memcpy(new_conn->cc_name, group, grouplen);
 316         new_conn->cc_namelen = grouplen;
 317         new_conn->cc_recovery_handler = recovery_handler;
 318         new_conn->cc_recovery_data = recovery_data;
 319
 320         /* Start the new connection at our maximum compatibility level */
 321         new_conn->cc_version = lproto->lp_max_version;
 322
 323         /* This will pin the stack driver if successful */
 324         rc = ocfs2_stack_driver_get(stack_name);
 325         if (rc)
 326                 goto out_free;
 327
 328         rc = active_stack->sp_ops->connect(new_conn);
 329         if (rc) {
 330                 ocfs2_stack_driver_put();
 331                 goto out_free;
 332         }
 333
 334         *conn = new_conn;
 335
 336 out_free:
 337         if (rc)
 338                 kfree(new_conn);
 339
 340 out:
 341         return rc;
 342 }
 343 EXPORT_SYMBOL_GPL(ocfs2_cluster_connect);
 344
 345 /* If hangup_pending is 0, the stack driver will be dropped */
 346 int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn,
 347                              int hangup_pending)
 348 {
 349         int ret;
 350
 351         BUG_ON(conn == NULL);
 352
 353         ret = active_stack->sp_ops->disconnect(conn, hangup_pending);
 354
 355         /* XXX Should we free it anyway? */
 356         if (!ret) {
 357                 kfree(conn);
 358                 if (!hangup_pending)
 359                         ocfs2_stack_driver_put();
 360         }
 361
 362         return ret;
 363 }
 364 EXPORT_SYMBOL_GPL(ocfs2_cluster_disconnect);
 365
 366 void ocfs2_cluster_hangup(const char *group, int grouplen)
 367 {
 368         BUG_ON(group == NULL);
 369         BUG_ON(group[grouplen] != '\0');
 370
 371         if (active_stack->sp_ops->hangup)
 372                 active_stack->sp_ops->hangup(group, grouplen);
 373
 374         /* cluster_disconnect() was called with hangup_pending==1 */
 375         ocfs2_stack_driver_put();
 376 }
 377 EXPORT_SYMBOL_GPL(ocfs2_cluster_hangup);
 378
 379 int ocfs2_cluster_this_node(unsigned int *node)
 380 {
 381         return active_stack->sp_ops->this_node(node);
 382 }
 383 EXPORT_SYMBOL_GPL(ocfs2_cluster_this_node);
 384
 385
 386 /*
 387  * Sysfs bits
 388  */
 389
 390 static ssize_t ocfs2_max_locking_protocol_show(struct kobject *kobj,
 391                                                struct kobj_attribute *attr,
 392                                                char *buf)
 393 {
 394         ssize_t ret = 0;
 395
 396         spin_lock(&ocfs2_stack_lock);
 397         if (lproto)
 398                 ret = snprintf(buf, PAGE_SIZE, "%u.%u\n",
 399                                lproto->lp_max_version.pv_major,
 400                                lproto->lp_max_version.pv_minor);
 401         spin_unlock(&ocfs2_stack_lock);
 402
 403         return ret;
 404 }
 405
 406 static struct kobj_attribute ocfs2_attr_max_locking_protocol =
 407         __ATTR(max_locking_protocol, S_IFREG | S_IRUGO,
 408                ocfs2_max_locking_protocol_show, NULL);
 409
 410 static ssize_t ocfs2_loaded_cluster_plugins_show(struct kobject *kobj,
 411                                                  struct kobj_attribute *attr,
 412                                                  char *buf)
 413 {
 414         ssize_t ret = 0, total = 0, remain = PAGE_SIZE;
 415         struct ocfs2_stack_plugin *p;
 416
 417         spin_lock(&ocfs2_stack_lock);
 418         list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
 419                 ret = snprintf(buf, remain, "%s\n",
 420                                p->sp_name);
 421                 if (ret < 0) {
 422                         total = ret;
 423                         break;
 424                 }
 425                 if (ret == remain) {
 426                         /* snprintf() didn't fit */
 427                         total = -E2BIG;
 428                         break;
 429                 }
 430                 total += ret;
 431                 remain -= ret;
 432         }
 433         spin_unlock(&ocfs2_stack_lock);
 434
 435         return total;
 436 }
 437
 438 static struct kobj_attribute ocfs2_attr_loaded_cluster_plugins =
 439         __ATTR(loaded_cluster_plugins, S_IFREG | S_IRUGO,
 440                ocfs2_loaded_cluster_plugins_show, NULL);
 441
 442 static ssize_t ocfs2_active_cluster_plugin_show(struct kobject *kobj,
 443                                                 struct kobj_attribute *attr,
 444                                                 char *buf)
 445 {
 446         ssize_t ret = 0;
 447
 448         spin_lock(&ocfs2_stack_lock);
 449         if (active_stack) {
 450                 ret = snprintf(buf, PAGE_SIZE, "%s\n",
 451                                active_stack->sp_name);
 452                 if (ret == PAGE_SIZE)
 453                         ret = -E2BIG;
 454         }
 455         spin_unlock(&ocfs2_stack_lock);
 456
 457         return ret;
 458 }
 459
 460 static struct kobj_attribute ocfs2_attr_active_cluster_plugin =
 461         __ATTR(active_cluster_plugin, S_IFREG | S_IRUGO,
 462                ocfs2_active_cluster_plugin_show, NULL);
 463
 464 static ssize_t ocfs2_cluster_stack_show(struct kobject *kobj,
 465                                         struct kobj_attribute *attr,
 466                                         char *buf)
 467 {
 468         ssize_t ret;
 469         spin_lock(&ocfs2_stack_lock);
 470         ret = snprintf(buf, PAGE_SIZE, "%s\n", cluster_stack_name);
 471         spin_unlock(&ocfs2_stack_lock);
 472
 473         return ret;
 474 }
 475
 476 static ssize_t ocfs2_cluster_stack_store(struct kobject *kobj,
 477                                          struct kobj_attribute *attr,
 478                                          const char *buf, size_t count)
 479 {
 480         size_t len = count;
 481         ssize_t ret;
 482
 483         if (len == 0)
 484                 return len;
 485
 486         if (buf[len - 1] == '\n')
 487                 len--;
 488
 489         if ((len != OCFS2_STACK_LABEL_LEN) ||
 490             (strnlen(buf, len) != len))
 491                 return -EINVAL;
 492
 493         spin_lock(&ocfs2_stack_lock);
 494         if (active_stack) {
 495                 if (!strncmp(buf, cluster_stack_name, len))
 496                         ret = count;
 497                 else
 498                         ret = -EBUSY;
 499         } else {
 500                 memcpy(cluster_stack_name, buf, len);
 501                 ret = count;
 502         }
 503         spin_unlock(&ocfs2_stack_lock);
 504
 505         return ret;
 506 }
 507
 508
 509 static struct kobj_attribute ocfs2_attr_cluster_stack =
 510         __ATTR(cluster_stack, S_IFREG | S_IRUGO | S_IWUSR,
 511                ocfs2_cluster_stack_show,
 512                ocfs2_cluster_stack_store);
 513
 514 static struct attribute *ocfs2_attrs[] = {
 515         &ocfs2_attr_max_locking_protocol.attr,
 516         &ocfs2_attr_loaded_cluster_plugins.attr,
 517         &ocfs2_attr_active_cluster_plugin.attr,
 518         &ocfs2_attr_cluster_stack.attr,
 519         NULL,
 520 };
 521
 522 static struct attribute_group ocfs2_attr_group = {
 523         .attrs = ocfs2_attrs,
 524 };
 525
 526 static struct kset *ocfs2_kset;
 527
 528 static void ocfs2_sysfs_exit(void)
 529 {
 530         kset_unregister(ocfs2_kset);
 531 }
 532
 533 static int ocfs2_sysfs_init(void)
 534 {
 535         int ret;
 536
 537         ocfs2_kset = kset_create_and_add("ocfs2", NULL, fs_kobj);
 538         if (!ocfs2_kset)
 539                 return -ENOMEM;
 540
 541         ret = sysfs_create_group(&ocfs2_kset->kobj, &ocfs2_attr_group);
 542         if (ret)
 543                 goto error;
 544
 545         return 0;
 546
 547 error:
 548         kset_unregister(ocfs2_kset);
 549         return ret;
 550 }
 551
 552 /*
 553  * Sysctl bits
 554  *
 555  * The sysctl lives at /proc/sys/fs/ocfs2/nm/hb_ctl_path.  The 'nm' doesn't
 556  * make as much sense in a multiple cluster stack world, but it's safer
 557  * and easier to preserve the name.
 558  */
 559
 560 #define FS_OCFS2_NM             1
 561
 562 #define OCFS2_MAX_HB_CTL_PATH 256
 563 static char ocfs2_hb_ctl_path[OCFS2_MAX_HB_CTL_PATH] = "/sbin/ocfs2_hb_ctl";
 564
 565 static ctl_table ocfs2_nm_table[] = {
 566         {
 567                 .ctl_name       = 1,
 568                 .procname       = "hb_ctl_path",
 569                 .data           = ocfs2_hb_ctl_path,
 570                 .maxlen         = OCFS2_MAX_HB_CTL_PATH,
 571                 .mode           = 0644,
 572                 .proc_handler   = &proc_dostring,
 573                 .strategy       = &sysctl_string,
 574         },
 575         { .ctl_name = 0 }
 576 };
 577
 578 static ctl_table ocfs2_mod_table[] = {
 579         {
 580                 .ctl_name       = FS_OCFS2_NM,
 581                 .procname       = "nm",
 582                 .data           = NULL,
 583                 .maxlen         = 0,
 584                 .mode           = 0555,
 585                 .child          = ocfs2_nm_table
 586         },
 587         { .ctl_name = 0}
 588 };
 589
 590 static ctl_table ocfs2_kern_table[] = {
 591         {
 592                 .ctl_name       = FS_OCFS2,
 593                 .procname       = "ocfs2",
 594                 .data           = NULL,
 595                 .maxlen         = 0,
 596                 .mode           = 0555,
 597                 .child          = ocfs2_mod_table
 598         },
 599         { .ctl_name = 0}
 600 };
 601
 602 static ctl_table ocfs2_root_table[] = {
 603         {
 604                 .ctl_name       = CTL_FS,
 605                 .procname       = "fs",
 606                 .data           = NULL,
 607                 .maxlen         = 0,
 608                 .mode           = 0555,
 609                 .child          = ocfs2_kern_table
 610         },
 611         { .ctl_name = 0 }
 612 };
 613
 614 static struct ctl_table_header *ocfs2_table_header = NULL;
 615
 616 const char *ocfs2_get_hb_ctl_path(void)
 617 {
 618         return ocfs2_hb_ctl_path;
 619 }
 620 EXPORT_SYMBOL_GPL(ocfs2_get_hb_ctl_path);
 621
 622
 623 /*
 624  * Initialization
 625  */
 626
 627 static int __init ocfs2_stack_glue_init(void)
 628 {
 629         strcpy(cluster_stack_name, OCFS2_STACK_PLUGIN_O2CB);
 630
 631         ocfs2_table_header = register_sysctl_table(ocfs2_root_table);
 632         if (!ocfs2_table_header) {
 633                 printk(KERN_ERR
 634                        "ocfs2 stack glue: unable to register sysctl\n");
 635                 return -ENOMEM; /* or something. */
 636         }
 637
 638         return ocfs2_sysfs_init();
 639 }
 640
 641 static void __exit ocfs2_stack_glue_exit(void)
 642 {
 643         lproto = NULL;
 644         ocfs2_sysfs_exit();
 645         if (ocfs2_table_header)
 646                 unregister_sysctl_table(ocfs2_table_header);
 647 }
 648
 649 MODULE_AUTHOR("Oracle");
 650 MODULE_DESCRIPTION("ocfs2 cluter stack glue layer");
 651 MODULE_LICENSE("GPL");
 652 module_init(ocfs2_stack_glue_init);
 653 module_exit(ocfs2_stack_glue_exit);