fs/ocfs2/cluster/nodemanager.c

   1 /* -*- mode: c; c-basic-offset: 8; -*-
   2  * vim: noexpandtab sw=8 ts=8 sts=0:
   3  *
   4  * Copyright (C) 2004, 2005 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 "tcp.h"
  27 #include "nodemanager.h"
  28 #include "heartbeat.h"
  29 #include "masklog.h"
  30 #include "sys.h"
  31 #include "ver.h"
  32
  33 /* for now we operate under the assertion that there can be only one
  34  * cluster active at a time.  Changing this will require trickling
  35  * cluster references throughout where nodes are looked up */
  36 struct o2nm_cluster *o2nm_single_cluster = NULL;
  37
  38 char *o2nm_fence_method_desc[O2NM_FENCE_METHODS] = {
  39                 "reset",        /* O2NM_FENCE_RESET */
  40                 "panic",        /* O2NM_FENCE_PANIC */
  41 };
  42
  43 struct o2nm_node *o2nm_get_node_by_num(u8 node_num)
  44 {
  45         struct o2nm_node *node = NULL;
  46
  47         if (node_num >= O2NM_MAX_NODES || o2nm_single_cluster == NULL)
  48                 goto out;
  49
  50         read_lock(&o2nm_single_cluster->cl_nodes_lock);
  51         node = o2nm_single_cluster->cl_nodes[node_num];
  52         if (node)
  53                 config_item_get(&node->nd_item);
  54         read_unlock(&o2nm_single_cluster->cl_nodes_lock);
  55 out:
  56         return node;
  57 }
  58 EXPORT_SYMBOL_GPL(o2nm_get_node_by_num);
  59
  60 int o2nm_configured_node_map(unsigned long *map, unsigned bytes)
  61 {
  62         struct o2nm_cluster *cluster = o2nm_single_cluster;
  63
  64         BUG_ON(bytes < (sizeof(cluster->cl_nodes_bitmap)));
  65
  66         if (cluster == NULL)
  67                 return -EINVAL;
  68
  69         read_lock(&cluster->cl_nodes_lock);
  70         memcpy(map, cluster->cl_nodes_bitmap, sizeof(cluster->cl_nodes_bitmap));
  71         read_unlock(&cluster->cl_nodes_lock);
  72
  73         return 0;
  74 }
  75 EXPORT_SYMBOL_GPL(o2nm_configured_node_map);
  76
  77 static struct o2nm_node *o2nm_node_ip_tree_lookup(struct o2nm_cluster *cluster,
  78                                                   __be32 ip_needle,
  79                                                   struct rb_node ***ret_p,
  80                                                   struct rb_node **ret_parent)
  81 {
  82         struct rb_node **p = &cluster->cl_node_ip_tree.rb_node;
  83         struct rb_node *parent = NULL;
  84         struct o2nm_node *node, *ret = NULL;
  85
  86         while (*p) {
  87                 int cmp;
  88
  89                 parent = *p;
  90                 node = rb_entry(parent, struct o2nm_node, nd_ip_node);
  91
  92                 cmp = memcmp(&ip_needle, &node->nd_ipv4_address,
  93                                 sizeof(ip_needle));
  94                 if (cmp < 0)
  95                         p = &(*p)->rb_left;
  96                 else if (cmp > 0)
  97                         p = &(*p)->rb_right;
  98                 else {
  99                         ret = node;
 100                         break;
 101                 }
 102         }
 103
 104         if (ret_p != NULL)
 105                 *ret_p = p;
 106         if (ret_parent != NULL)
 107                 *ret_parent = parent;
 108
 109         return ret;
 110 }
 111
 112 struct o2nm_node *o2nm_get_node_by_ip(__be32 addr)
 113 {
 114         struct o2nm_node *node = NULL;
 115         struct o2nm_cluster *cluster = o2nm_single_cluster;
 116
 117         if (cluster == NULL)
 118                 goto out;
 119
 120         read_lock(&cluster->cl_nodes_lock);
 121         node = o2nm_node_ip_tree_lookup(cluster, addr, NULL, NULL);
 122         if (node)
 123                 config_item_get(&node->nd_item);
 124         read_unlock(&cluster->cl_nodes_lock);
 125
 126 out:
 127         return node;
 128 }
 129 EXPORT_SYMBOL_GPL(o2nm_get_node_by_ip);
 130
 131 void o2nm_node_put(struct o2nm_node *node)
 132 {
 133         config_item_put(&node->nd_item);
 134 }
 135 EXPORT_SYMBOL_GPL(o2nm_node_put);
 136
 137 void o2nm_node_get(struct o2nm_node *node)
 138 {
 139         config_item_get(&node->nd_item);
 140 }
 141 EXPORT_SYMBOL_GPL(o2nm_node_get);
 142
 143 u8 o2nm_this_node(void)
 144 {
 145         u8 node_num = O2NM_MAX_NODES;
 146
 147         if (o2nm_single_cluster && o2nm_single_cluster->cl_has_local)
 148                 node_num = o2nm_single_cluster->cl_local_node;
 149
 150         return node_num;
 151 }
 152 EXPORT_SYMBOL_GPL(o2nm_this_node);
 153
 154 /* node configfs bits */
 155
 156 static struct o2nm_cluster *to_o2nm_cluster(struct config_item *item)
 157 {
 158         return item ?
 159                 container_of(to_config_group(item), struct o2nm_cluster,
 160                              cl_group)
 161                 : NULL;
 162 }
 163
 164 static struct o2nm_node *to_o2nm_node(struct config_item *item)
 165 {
 166         return item ? container_of(item, struct o2nm_node, nd_item) : NULL;
 167 }
 168
 169 static void o2nm_node_release(struct config_item *item)
 170 {
 171         struct o2nm_node *node = to_o2nm_node(item);
 172         kfree(node);
 173 }
 174
 175 static ssize_t o2nm_node_num_read(struct o2nm_node *node, char *page)
 176 {
 177         return sprintf(page, "%d\n", node->nd_num);
 178 }
 179
 180 static struct o2nm_cluster *to_o2nm_cluster_from_node(struct o2nm_node *node)
 181 {
 182         /* through the first node_set .parent
 183          * mycluster/nodes/mynode == o2nm_cluster->o2nm_node_group->o2nm_node */
 184         return to_o2nm_cluster(node->nd_item.ci_parent->ci_parent);
 185 }
 186
 187 enum {
 188         O2NM_NODE_ATTR_NUM = 0,
 189         O2NM_NODE_ATTR_PORT,
 190         O2NM_NODE_ATTR_ADDRESS,
 191         O2NM_NODE_ATTR_LOCAL,
 192 };
 193
 194 static ssize_t o2nm_node_num_write(struct o2nm_node *node, const char *page,
 195                                    size_t count)
 196 {
 197         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
 198         unsigned long tmp;
 199         char *p = (char *)page;
 200
 201         tmp = simple_strtoul(p, &p, 0);
 202         if (!p || (*p && (*p != '\n')))
 203                 return -EINVAL;
 204
 205         if (tmp >= O2NM_MAX_NODES)
 206                 return -ERANGE;
 207
 208         /* once we're in the cl_nodes tree networking can look us up by
 209          * node number and try to use our address and port attributes
 210          * to connect to this node.. make sure that they've been set
 211          * before writing the node attribute? */
 212         if (!test_bit(O2NM_NODE_ATTR_ADDRESS, &node->nd_set_attributes) ||
 213             !test_bit(O2NM_NODE_ATTR_PORT, &node->nd_set_attributes))
 214                 return -EINVAL; /* XXX */
 215
 216         write_lock(&cluster->cl_nodes_lock);
 217         if (cluster->cl_nodes[tmp])
 218                 p = NULL;
 219         else  {
 220                 cluster->cl_nodes[tmp] = node;
 221                 node->nd_num = tmp;
 222                 set_bit(tmp, cluster->cl_nodes_bitmap);
 223         }
 224         write_unlock(&cluster->cl_nodes_lock);
 225         if (p == NULL)
 226                 return -EEXIST;
 227
 228         return count;
 229 }
 230 static ssize_t o2nm_node_ipv4_port_read(struct o2nm_node *node, char *page)
 231 {
 232         return sprintf(page, "%u\n", ntohs(node->nd_ipv4_port));
 233 }
 234
 235 static ssize_t o2nm_node_ipv4_port_write(struct o2nm_node *node,
 236                                          const char *page, size_t count)
 237 {
 238         unsigned long tmp;
 239         char *p = (char *)page;
 240
 241         tmp = simple_strtoul(p, &p, 0);
 242         if (!p || (*p && (*p != '\n')))
 243                 return -EINVAL;
 244
 245         if (tmp == 0)
 246                 return -EINVAL;
 247         if (tmp >= (u16)-1)
 248                 return -ERANGE;
 249
 250         node->nd_ipv4_port = htons(tmp);
 251
 252         return count;
 253 }
 254
 255 static ssize_t o2nm_node_ipv4_address_read(struct o2nm_node *node, char *page)
 256 {
 257         return sprintf(page, "%pI4\n", &node->nd_ipv4_address);
 258 }
 259
 260 static ssize_t o2nm_node_ipv4_address_write(struct o2nm_node *node,
 261                                             const char *page,
 262                                             size_t count)
 263 {
 264         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
 265         int ret, i;
 266         struct rb_node **p, *parent;
 267         unsigned int octets[4];
 268         __be32 ipv4_addr = 0;
 269
 270         ret = sscanf(page, "%3u.%3u.%3u.%3u", &octets[3], &octets[2],
 271                      &octets[1], &octets[0]);
 272         if (ret != 4)
 273                 return -EINVAL;
 274
 275         for (i = 0; i < ARRAY_SIZE(octets); i++) {
 276                 if (octets[i] > 255)
 277                         return -ERANGE;
 278                 be32_add_cpu(&ipv4_addr, octets[i] << (i * 8));
 279         }
 280
 281         ret = 0;
 282         write_lock(&cluster->cl_nodes_lock);
 283         if (o2nm_node_ip_tree_lookup(cluster, ipv4_addr, &p, &parent))
 284                 ret = -EEXIST;
 285         else {
 286                 rb_link_node(&node->nd_ip_node, parent, p);
 287                 rb_insert_color(&node->nd_ip_node, &cluster->cl_node_ip_tree);
 288         }
 289         write_unlock(&cluster->cl_nodes_lock);
 290         if (ret)
 291                 return ret;
 292
 293         memcpy(&node->nd_ipv4_address, &ipv4_addr, sizeof(ipv4_addr));
 294
 295         return count;
 296 }
 297
 298 static ssize_t o2nm_node_local_read(struct o2nm_node *node, char *page)
 299 {
 300         return sprintf(page, "%d\n", node->nd_local);
 301 }
 302
 303 static ssize_t o2nm_node_local_write(struct o2nm_node *node, const char *page,
 304                                      size_t count)
 305 {
 306         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
 307         unsigned long tmp;
 308         char *p = (char *)page;
 309         ssize_t ret;
 310
 311         tmp = simple_strtoul(p, &p, 0);
 312         if (!p || (*p && (*p != '\n')))
 313                 return -EINVAL;
 314
 315         tmp = !!tmp; /* boolean of whether this node wants to be local */
 316
 317         /* setting local turns on networking rx for now so we require having
 318          * set everything else first */
 319         if (!test_bit(O2NM_NODE_ATTR_ADDRESS, &node->nd_set_attributes) ||
 320             !test_bit(O2NM_NODE_ATTR_NUM, &node->nd_set_attributes) ||
 321             !test_bit(O2NM_NODE_ATTR_PORT, &node->nd_set_attributes))
 322                 return -EINVAL; /* XXX */
 323
 324         /* the only failure case is trying to set a new local node
 325          * when a different one is already set */
 326         if (tmp && tmp == cluster->cl_has_local &&
 327             cluster->cl_local_node != node->nd_num)
 328                 return -EBUSY;
 329
 330         /* bring up the rx thread if we're setting the new local node. */
 331         if (tmp && !cluster->cl_has_local) {
 332                 ret = o2net_start_listening(node);
 333                 if (ret)
 334                         return ret;
 335         }
 336
 337         if (!tmp && cluster->cl_has_local &&
 338             cluster->cl_local_node == node->nd_num) {
 339                 o2net_stop_listening(node);
 340                 cluster->cl_local_node = O2NM_INVALID_NODE_NUM;
 341         }
 342
 343         node->nd_local = tmp;
 344         if (node->nd_local) {
 345                 cluster->cl_has_local = tmp;
 346                 cluster->cl_local_node = node->nd_num;
 347         }
 348
 349         return count;
 350 }
 351
 352 struct o2nm_node_attribute {
 353         struct configfs_attribute attr;
 354         ssize_t (*show)(struct o2nm_node *, char *);
 355         ssize_t (*store)(struct o2nm_node *, const char *, size_t);
 356 };
 357
 358 static struct o2nm_node_attribute o2nm_node_attr_num = {
 359         .attr   = { .ca_owner = THIS_MODULE,
 360                     .ca_name = "num",
 361                     .ca_mode = S_IRUGO | S_IWUSR },
 362         .show   = o2nm_node_num_read,
 363         .store  = o2nm_node_num_write,
 364 };
 365
 366 static struct o2nm_node_attribute o2nm_node_attr_ipv4_port = {
 367         .attr   = { .ca_owner = THIS_MODULE,
 368                     .ca_name = "ipv4_port",
 369                     .ca_mode = S_IRUGO | S_IWUSR },
 370         .show   = o2nm_node_ipv4_port_read,
 371         .store  = o2nm_node_ipv4_port_write,
 372 };
 373
 374 static struct o2nm_node_attribute o2nm_node_attr_ipv4_address = {
 375         .attr   = { .ca_owner = THIS_MODULE,
 376                     .ca_name = "ipv4_address",
 377                     .ca_mode = S_IRUGO | S_IWUSR },
 378         .show   = o2nm_node_ipv4_address_read,
 379         .store  = o2nm_node_ipv4_address_write,
 380 };
 381
 382 static struct o2nm_node_attribute o2nm_node_attr_local = {
 383         .attr   = { .ca_owner = THIS_MODULE,
 384                     .ca_name = "local",
 385                     .ca_mode = S_IRUGO | S_IWUSR },
 386         .show   = o2nm_node_local_read,
 387         .store  = o2nm_node_local_write,
 388 };
 389
 390 static struct configfs_attribute *o2nm_node_attrs[] = {
 391         [O2NM_NODE_ATTR_NUM] = &o2nm_node_attr_num.attr,
 392         [O2NM_NODE_ATTR_PORT] = &o2nm_node_attr_ipv4_port.attr,
 393         [O2NM_NODE_ATTR_ADDRESS] = &o2nm_node_attr_ipv4_address.attr,
 394         [O2NM_NODE_ATTR_LOCAL] = &o2nm_node_attr_local.attr,
 395         NULL,
 396 };
 397
 398 static int o2nm_attr_index(struct configfs_attribute *attr)
 399 {
 400         int i;
 401         for (i = 0; i < ARRAY_SIZE(o2nm_node_attrs); i++) {
 402                 if (attr == o2nm_node_attrs[i])
 403                         return i;
 404         }
 405         BUG();
 406         return 0;
 407 }
 408
 409 static ssize_t o2nm_node_show(struct config_item *item,
 410                               struct configfs_attribute *attr,
 411                               char *page)
 412 {
 413         struct o2nm_node *node = to_o2nm_node(item);
 414         struct o2nm_node_attribute *o2nm_node_attr =
 415                 container_of(attr, struct o2nm_node_attribute, attr);
 416         ssize_t ret = 0;
 417
 418         if (o2nm_node_attr->show)
 419                 ret = o2nm_node_attr->show(node, page);
 420         return ret;
 421 }
 422
 423 static ssize_t o2nm_node_store(struct config_item *item,
 424                                struct configfs_attribute *attr,
 425                                const char *page, size_t count)
 426 {
 427         struct o2nm_node *node = to_o2nm_node(item);
 428         struct o2nm_node_attribute *o2nm_node_attr =
 429                 container_of(attr, struct o2nm_node_attribute, attr);
 430         ssize_t ret;
 431         int attr_index = o2nm_attr_index(attr);
 432
 433         if (o2nm_node_attr->store == NULL) {
 434                 ret = -EINVAL;
 435                 goto out;
 436         }
 437
 438         if (test_bit(attr_index, &node->nd_set_attributes))
 439                 return -EBUSY;
 440
 441         ret = o2nm_node_attr->store(node, page, count);
 442         if (ret < count)
 443                 goto out;
 444
 445         set_bit(attr_index, &node->nd_set_attributes);
 446 out:
 447         return ret;
 448 }
 449
 450 static struct configfs_item_operations o2nm_node_item_ops = {
 451         .release                = o2nm_node_release,
 452         .show_attribute         = o2nm_node_show,
 453         .store_attribute        = o2nm_node_store,
 454 };
 455
 456 static struct config_item_type o2nm_node_type = {
 457         .ct_item_ops    = &o2nm_node_item_ops,
 458         .ct_attrs       = o2nm_node_attrs,
 459         .ct_owner       = THIS_MODULE,
 460 };
 461
 462 /* node set */
 463
 464 struct o2nm_node_group {
 465         struct config_group ns_group;
 466         /* some stuff? */
 467 };
 468
 469 #if 0
 470 static struct o2nm_node_group *to_o2nm_node_group(struct config_group *group)
 471 {
 472         return group ?
 473                 container_of(group, struct o2nm_node_group, ns_group)
 474                 : NULL;
 475 }
 476 #endif
 477
 478 struct o2nm_cluster_attribute {
 479         struct configfs_attribute attr;
 480         ssize_t (*show)(struct o2nm_cluster *, char *);
 481         ssize_t (*store)(struct o2nm_cluster *, const char *, size_t);
 482 };
 483
 484 static ssize_t o2nm_cluster_attr_write(const char *page, ssize_t count,
 485                                        unsigned int *val)
 486 {
 487         unsigned long tmp;
 488         char *p = (char *)page;
 489
 490         tmp = simple_strtoul(p, &p, 0);
 491         if (!p || (*p && (*p != '\n')))
 492                 return -EINVAL;
 493
 494         if (tmp == 0)
 495                 return -EINVAL;
 496         if (tmp >= (u32)-1)
 497                 return -ERANGE;
 498
 499         *val = tmp;
 500
 501         return count;
 502 }
 503
 504 static ssize_t o2nm_cluster_attr_idle_timeout_ms_read(
 505         struct o2nm_cluster *cluster, char *page)
 506 {
 507         return sprintf(page, "%u\n", cluster->cl_idle_timeout_ms);
 508 }
 509
 510 static ssize_t o2nm_cluster_attr_idle_timeout_ms_write(
 511         struct o2nm_cluster *cluster, const char *page, size_t count)
 512 {
 513         ssize_t ret;
 514         unsigned int val;
 515
 516         ret =  o2nm_cluster_attr_write(page, count, &val);
 517
 518         if (ret > 0) {
 519                 if (cluster->cl_idle_timeout_ms != val
 520                         && o2net_num_connected_peers()) {
 521                         mlog(ML_NOTICE,
 522                              "o2net: cannot change idle timeout after "
 523                              "the first peer has agreed to it."
 524                              "  %d connected peers\n",
 525                              o2net_num_connected_peers());
 526                         ret = -EINVAL;
 527                 } else if (val <= cluster->cl_keepalive_delay_ms) {
 528                         mlog(ML_NOTICE, "o2net: idle timeout must be larger "
 529                              "than keepalive delay\n");
 530                         ret = -EINVAL;
 531                 } else {
 532                         cluster->cl_idle_timeout_ms = val;
 533                 }
 534         }
 535
 536         return ret;
 537 }
 538
 539 static ssize_t o2nm_cluster_attr_keepalive_delay_ms_read(
 540         struct o2nm_cluster *cluster, char *page)
 541 {
 542         return sprintf(page, "%u\n", cluster->cl_keepalive_delay_ms);
 543 }
 544
 545 static ssize_t o2nm_cluster_attr_keepalive_delay_ms_write(
 546         struct o2nm_cluster *cluster, const char *page, size_t count)
 547 {
 548         ssize_t ret;
 549         unsigned int val;
 550
 551         ret =  o2nm_cluster_attr_write(page, count, &val);
 552
 553         if (ret > 0) {
 554                 if (cluster->cl_keepalive_delay_ms != val
 555                     && o2net_num_connected_peers()) {
 556                         mlog(ML_NOTICE,
 557                              "o2net: cannot change keepalive delay after"
 558                              " the first peer has agreed to it."
 559                              "  %d connected peers\n",
 560                              o2net_num_connected_peers());
 561                         ret = -EINVAL;
 562                 } else if (val >= cluster->cl_idle_timeout_ms) {
 563                         mlog(ML_NOTICE, "o2net: keepalive delay must be "
 564                              "smaller than idle timeout\n");
 565                         ret = -EINVAL;
 566                 } else {
 567                         cluster->cl_keepalive_delay_ms = val;
 568                 }
 569         }
 570
 571         return ret;
 572 }
 573
 574 static ssize_t o2nm_cluster_attr_reconnect_delay_ms_read(
 575         struct o2nm_cluster *cluster, char *page)
 576 {
 577         return sprintf(page, "%u\n", cluster->cl_reconnect_delay_ms);
 578 }
 579
 580 static ssize_t o2nm_cluster_attr_reconnect_delay_ms_write(
 581         struct o2nm_cluster *cluster, const char *page, size_t count)
 582 {
 583         return o2nm_cluster_attr_write(page, count,
 584                                        &cluster->cl_reconnect_delay_ms);
 585 }
 586
 587 static ssize_t o2nm_cluster_attr_fence_method_read(
 588         struct o2nm_cluster *cluster, char *page)
 589 {
 590         ssize_t ret = 0;
 591
 592         if (cluster)
 593                 ret = sprintf(page, "%s\n",
 594                               o2nm_fence_method_desc[cluster->cl_fence_method]);
 595         return ret;
 596 }
 597
 598 static ssize_t o2nm_cluster_attr_fence_method_write(
 599         struct o2nm_cluster *cluster, const char *page, size_t count)
 600 {
 601         unsigned int i;
 602
 603         if (page[count - 1] != '\n')
 604                 goto bail;
 605
 606         for (i = 0; i < O2NM_FENCE_METHODS; ++i) {
 607                 if (count != strlen(o2nm_fence_method_desc[i]) + 1)
 608                         continue;
 609                 if (strncasecmp(page, o2nm_fence_method_desc[i], count - 1))
 610                         continue;
 611                 if (cluster->cl_fence_method != i) {
 612                         printk(KERN_INFO "ocfs2: Changing fence method to %s\n",
 613                                o2nm_fence_method_desc[i]);
 614                         cluster->cl_fence_method = i;
 615                 }
 616                 return count;
 617         }
 618
 619 bail:
 620         return -EINVAL;
 621 }
 622
 623 static struct o2nm_cluster_attribute o2nm_cluster_attr_idle_timeout_ms = {
 624         .attr   = { .ca_owner = THIS_MODULE,
 625                     .ca_name = "idle_timeout_ms",
 626                     .ca_mode = S_IRUGO | S_IWUSR },
 627         .show   = o2nm_cluster_attr_idle_timeout_ms_read,
 628         .store  = o2nm_cluster_attr_idle_timeout_ms_write,
 629 };
 630
 631 static struct o2nm_cluster_attribute o2nm_cluster_attr_keepalive_delay_ms = {
 632         .attr   = { .ca_owner = THIS_MODULE,
 633                     .ca_name = "keepalive_delay_ms",
 634                     .ca_mode = S_IRUGO | S_IWUSR },
 635         .show   = o2nm_cluster_attr_keepalive_delay_ms_read,
 636         .store  = o2nm_cluster_attr_keepalive_delay_ms_write,
 637 };
 638
 639 static struct o2nm_cluster_attribute o2nm_cluster_attr_reconnect_delay_ms = {
 640         .attr   = { .ca_owner = THIS_MODULE,
 641                     .ca_name = "reconnect_delay_ms",
 642                     .ca_mode = S_IRUGO | S_IWUSR },
 643         .show   = o2nm_cluster_attr_reconnect_delay_ms_read,
 644         .store  = o2nm_cluster_attr_reconnect_delay_ms_write,
 645 };
 646
 647 static struct o2nm_cluster_attribute o2nm_cluster_attr_fence_method = {
 648         .attr   = { .ca_owner = THIS_MODULE,
 649                     .ca_name = "fence_method",
 650                     .ca_mode = S_IRUGO | S_IWUSR },
 651         .show   = o2nm_cluster_attr_fence_method_read,
 652         .store  = o2nm_cluster_attr_fence_method_write,
 653 };
 654
 655 static struct configfs_attribute *o2nm_cluster_attrs[] = {
 656         &o2nm_cluster_attr_idle_timeout_ms.attr,
 657         &o2nm_cluster_attr_keepalive_delay_ms.attr,
 658         &o2nm_cluster_attr_reconnect_delay_ms.attr,
 659         &o2nm_cluster_attr_fence_method.attr,
 660         NULL,
 661 };
 662 static ssize_t o2nm_cluster_show(struct config_item *item,
 663                                  struct configfs_attribute *attr,
 664                                  char *page)
 665 {
 666         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
 667         struct o2nm_cluster_attribute *o2nm_cluster_attr =
 668                 container_of(attr, struct o2nm_cluster_attribute, attr);
 669         ssize_t ret = 0;
 670
 671         if (o2nm_cluster_attr->show)
 672                 ret = o2nm_cluster_attr->show(cluster, page);
 673         return ret;
 674 }
 675
 676 static ssize_t o2nm_cluster_store(struct config_item *item,
 677                                   struct configfs_attribute *attr,
 678                                   const char *page, size_t count)
 679 {
 680         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
 681         struct o2nm_cluster_attribute *o2nm_cluster_attr =
 682                 container_of(attr, struct o2nm_cluster_attribute, attr);
 683         ssize_t ret;
 684
 685         if (o2nm_cluster_attr->store == NULL) {
 686                 ret = -EINVAL;
 687                 goto out;
 688         }
 689
 690         ret = o2nm_cluster_attr->store(cluster, page, count);
 691         if (ret < count)
 692                 goto out;
 693 out:
 694         return ret;
 695 }
 696
 697 static struct config_item *o2nm_node_group_make_item(struct config_group *group,
 698                                                      const char *name)
 699 {
 700         struct o2nm_node *node = NULL;
 701
 702         if (strlen(name) > O2NM_MAX_NAME_LEN)
 703                 return ERR_PTR(-ENAMETOOLONG);
 704
 705         node = kzalloc(sizeof(struct o2nm_node), GFP_KERNEL);
 706         if (node == NULL)
 707                 return ERR_PTR(-ENOMEM);
 708
 709         strcpy(node->nd_name, name); /* use item.ci_namebuf instead? */
 710         config_item_init_type_name(&node->nd_item, name, &o2nm_node_type);
 711         spin_lock_init(&node->nd_lock);
 712
 713         return &node->nd_item;
 714 }
 715
 716 static void o2nm_node_group_drop_item(struct config_group *group,
 717                                       struct config_item *item)
 718 {
 719         struct o2nm_node *node = to_o2nm_node(item);
 720         struct o2nm_cluster *cluster = to_o2nm_cluster(group->cg_item.ci_parent);
 721
 722         o2net_disconnect_node(node);
 723
 724         if (cluster->cl_has_local &&
 725             (cluster->cl_local_node == node->nd_num)) {
 726                 cluster->cl_has_local = 0;
 727                 cluster->cl_local_node = O2NM_INVALID_NODE_NUM;
 728                 o2net_stop_listening(node);
 729         }
 730
 731         /* XXX call into net to stop this node from trading messages */
 732
 733         write_lock(&cluster->cl_nodes_lock);
 734
 735         /* XXX sloppy */
 736         if (node->nd_ipv4_address)
 737                 rb_erase(&node->nd_ip_node, &cluster->cl_node_ip_tree);
 738
 739         /* nd_num might be 0 if the node number hasn't been set.. */
 740         if (cluster->cl_nodes[node->nd_num] == node) {
 741                 cluster->cl_nodes[node->nd_num] = NULL;
 742                 clear_bit(node->nd_num, cluster->cl_nodes_bitmap);
 743         }
 744         write_unlock(&cluster->cl_nodes_lock);
 745
 746         config_item_put(item);
 747 }
 748
 749 static struct configfs_group_operations o2nm_node_group_group_ops = {
 750         .make_item      = o2nm_node_group_make_item,
 751         .drop_item      = o2nm_node_group_drop_item,
 752 };
 753
 754 static struct config_item_type o2nm_node_group_type = {
 755         .ct_group_ops   = &o2nm_node_group_group_ops,
 756         .ct_owner       = THIS_MODULE,
 757 };
 758
 759 /* cluster */
 760
 761 static void o2nm_cluster_release(struct config_item *item)
 762 {
 763         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
 764
 765         kfree(cluster->cl_group.default_groups);
 766         kfree(cluster);
 767 }
 768
 769 static struct configfs_item_operations o2nm_cluster_item_ops = {
 770         .release        = o2nm_cluster_release,
 771         .show_attribute         = o2nm_cluster_show,
 772         .store_attribute        = o2nm_cluster_store,
 773 };
 774
 775 static struct config_item_type o2nm_cluster_type = {
 776         .ct_item_ops    = &o2nm_cluster_item_ops,
 777         .ct_attrs       = o2nm_cluster_attrs,
 778         .ct_owner       = THIS_MODULE,
 779 };
 780
 781 /* cluster set */
 782
 783 struct o2nm_cluster_group {
 784         struct configfs_subsystem cs_subsys;
 785         /* some stuff? */
 786 };
 787
 788 #if 0
 789 static struct o2nm_cluster_group *to_o2nm_cluster_group(struct config_group *group)
 790 {
 791         return group ?
 792                 container_of(to_configfs_subsystem(group), struct o2nm_cluster_group, cs_subsys)
 793                : NULL;
 794 }
 795 #endif
 796
 797 static struct config_group *o2nm_cluster_group_make_group(struct config_group *group,
 798                                                           const char *name)
 799 {
 800         struct o2nm_cluster *cluster = NULL;
 801         struct o2nm_node_group *ns = NULL;
 802         struct config_group *o2hb_group = NULL, *ret = NULL;
 803         void *defs = NULL;
 804
 805         /* this runs under the parent dir's i_mutex; there can be only
 806          * one caller in here at a time */
 807         if (o2nm_single_cluster)
 808                 return ERR_PTR(-ENOSPC);
 809
 810         cluster = kzalloc(sizeof(struct o2nm_cluster), GFP_KERNEL);
 811         ns = kzalloc(sizeof(struct o2nm_node_group), GFP_KERNEL);
 812         defs = kcalloc(3, sizeof(struct config_group *), GFP_KERNEL);
 813         o2hb_group = o2hb_alloc_hb_set();
 814         if (cluster == NULL || ns == NULL || o2hb_group == NULL || defs == NULL)
 815                 goto out;
 816
 817         config_group_init_type_name(&cluster->cl_group, name,
 818                                     &o2nm_cluster_type);
 819         config_group_init_type_name(&ns->ns_group, "node",
 820                                     &o2nm_node_group_type);
 821
 822         cluster->cl_group.default_groups = defs;
 823         cluster->cl_group.default_groups[0] = &ns->ns_group;
 824         cluster->cl_group.default_groups[1] = o2hb_group;
 825         cluster->cl_group.default_groups[2] = NULL;
 826         rwlock_init(&cluster->cl_nodes_lock);
 827         cluster->cl_node_ip_tree = RB_ROOT;
 828         cluster->cl_reconnect_delay_ms = O2NET_RECONNECT_DELAY_MS_DEFAULT;
 829         cluster->cl_idle_timeout_ms    = O2NET_IDLE_TIMEOUT_MS_DEFAULT;
 830         cluster->cl_keepalive_delay_ms = O2NET_KEEPALIVE_DELAY_MS_DEFAULT;
 831         cluster->cl_fence_method       = O2NM_FENCE_RESET;
 832
 833         ret = &cluster->cl_group;
 834         o2nm_single_cluster = cluster;
 835
 836 out:
 837         if (ret == NULL) {
 838                 kfree(cluster);
 839                 kfree(ns);
 840                 o2hb_free_hb_set(o2hb_group);
 841                 kfree(defs);
 842                 ret = ERR_PTR(-ENOMEM);
 843         }
 844
 845         return ret;
 846 }
 847
 848 static void o2nm_cluster_group_drop_item(struct config_group *group, struct config_item *item)
 849 {
 850         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
 851         int i;
 852         struct config_item *killme;
 853
 854         BUG_ON(o2nm_single_cluster != cluster);
 855         o2nm_single_cluster = NULL;
 856
 857         for (i = 0; cluster->cl_group.default_groups[i]; i++) {
 858                 killme = &cluster->cl_group.default_groups[i]->cg_item;
 859                 cluster->cl_group.default_groups[i] = NULL;
 860                 config_item_put(killme);
 861         }
 862
 863         config_item_put(item);
 864 }
 865
 866 static struct configfs_group_operations o2nm_cluster_group_group_ops = {
 867         .make_group     = o2nm_cluster_group_make_group,
 868         .drop_item      = o2nm_cluster_group_drop_item,
 869 };
 870
 871 static struct config_item_type o2nm_cluster_group_type = {
 872         .ct_group_ops   = &o2nm_cluster_group_group_ops,
 873         .ct_owner       = THIS_MODULE,
 874 };
 875
 876 static struct o2nm_cluster_group o2nm_cluster_group = {
 877         .cs_subsys = {
 878                 .su_group = {
 879                         .cg_item = {
 880                                 .ci_namebuf = "cluster",
 881                                 .ci_type = &o2nm_cluster_group_type,
 882                         },
 883                 },
 884         },
 885 };
 886
 887 int o2nm_depend_item(struct config_item *item)
 888 {
 889         return configfs_depend_item(&o2nm_cluster_group.cs_subsys, item);
 890 }
 891
 892 void o2nm_undepend_item(struct config_item *item)
 893 {
 894         configfs_undepend_item(&o2nm_cluster_group.cs_subsys, item);
 895 }
 896
 897 int o2nm_depend_this_node(void)
 898 {
 899         int ret = 0;
 900         struct o2nm_node *local_node;
 901
 902         local_node = o2nm_get_node_by_num(o2nm_this_node());
 903         if (!local_node) {
 904                 ret = -EINVAL;
 905                 goto out;
 906         }
 907
 908         ret = o2nm_depend_item(&local_node->nd_item);
 909         o2nm_node_put(local_node);
 910
 911 out:
 912         return ret;
 913 }
 914
 915 void o2nm_undepend_this_node(void)
 916 {
 917         struct o2nm_node *local_node;
 918
 919         local_node = o2nm_get_node_by_num(o2nm_this_node());
 920         BUG_ON(!local_node);
 921
 922         o2nm_undepend_item(&local_node->nd_item);
 923         o2nm_node_put(local_node);
 924 }
 925
 926
 927 static void __exit exit_o2nm(void)
 928 {
 929         /* XXX sync with hb callbacks and shut down hb? */
 930         o2net_unregister_hb_callbacks();
 931         configfs_unregister_subsystem(&o2nm_cluster_group.cs_subsys);
 932         o2cb_sys_shutdown();
 933
 934         o2net_exit();
 935         o2hb_exit();
 936 }
 937
 938 static int __init init_o2nm(void)
 939 {
 940         int ret = -1;
 941
 942         cluster_print_version();
 943
 944         ret = o2hb_init();
 945         if (ret)
 946                 goto out;
 947
 948         ret = o2net_init();
 949         if (ret)
 950                 goto out_o2hb;
 951
 952         ret = o2net_register_hb_callbacks();
 953         if (ret)
 954                 goto out_o2net;
 955
 956         config_group_init(&o2nm_cluster_group.cs_subsys.su_group);
 957         mutex_init(&o2nm_cluster_group.cs_subsys.su_mutex);
 958         ret = configfs_register_subsystem(&o2nm_cluster_group.cs_subsys);
 959         if (ret) {
 960                 printk(KERN_ERR "nodemanager: Registration returned %d\n", ret);
 961                 goto out_callbacks;
 962         }
 963
 964         ret = o2cb_sys_init();
 965         if (!ret)
 966                 goto out;
 967
 968         configfs_unregister_subsystem(&o2nm_cluster_group.cs_subsys);
 969 out_callbacks:
 970         o2net_unregister_hb_callbacks();
 971 out_o2net:
 972         o2net_exit();
 973 out_o2hb:
 974         o2hb_exit();
 975 out:
 976         return ret;
 977 }
 978
 979 MODULE_AUTHOR("Oracle");
 980 MODULE_LICENSE("GPL");
 981
 982 module_init(init_o2nm)
 983 module_exit(exit_o2nm)