Merge git://git.kernel.org/pub/scm/linux/kernel/git/wim/linux-2.6-watchdog
[linux-2.6] / 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/sysctl.h>
25 #include <linux/configfs.h>
26
27 #include "endian.h"
28 #include "tcp.h"
29 #include "nodemanager.h"
30 #include "heartbeat.h"
31 #include "masklog.h"
32 #include "sys.h"
33 #include "ver.h"
34
35 /* for now we operate under the assertion that there can be only one
36  * cluster active at a time.  Changing this will require trickling
37  * cluster references throughout where nodes are looked up */
38 static struct o2nm_cluster *o2nm_single_cluster = NULL;
39
40 #define OCFS2_MAX_HB_CTL_PATH 256
41 static char ocfs2_hb_ctl_path[OCFS2_MAX_HB_CTL_PATH] = "/sbin/ocfs2_hb_ctl";
42
43 static ctl_table ocfs2_nm_table[] = {
44         {
45                 .ctl_name       = 1,
46                 .procname       = "hb_ctl_path",
47                 .data           = ocfs2_hb_ctl_path,
48                 .maxlen         = OCFS2_MAX_HB_CTL_PATH,
49                 .mode           = 0644,
50                 .proc_handler   = &proc_dostring,
51                 .strategy       = &sysctl_string,
52         },
53         { .ctl_name = 0 }
54 };
55
56 static ctl_table ocfs2_mod_table[] = {
57         {
58                 .ctl_name       = KERN_OCFS2_NM,
59                 .procname       = "nm",
60                 .data           = NULL,
61                 .maxlen         = 0,
62                 .mode           = 0555,
63                 .child          = ocfs2_nm_table
64         },
65         { .ctl_name = 0}
66 };
67
68 static ctl_table ocfs2_kern_table[] = {
69         {
70                 .ctl_name       = KERN_OCFS2,
71                 .procname       = "ocfs2",
72                 .data           = NULL,
73                 .maxlen         = 0,
74                 .mode           = 0555,
75                 .child          = ocfs2_mod_table
76         },
77         { .ctl_name = 0}
78 };
79
80 static ctl_table ocfs2_root_table[] = {
81         {
82                 .ctl_name       = CTL_FS,
83                 .procname       = "fs",
84                 .data           = NULL,
85                 .maxlen         = 0,
86                 .mode           = 0555,
87                 .child          = ocfs2_kern_table
88         },
89         { .ctl_name = 0 }
90 };
91
92 static struct ctl_table_header *ocfs2_table_header = NULL;
93
94 const char *o2nm_get_hb_ctl_path(void)
95 {
96         return ocfs2_hb_ctl_path;
97 }
98 EXPORT_SYMBOL_GPL(o2nm_get_hb_ctl_path);
99
100 struct o2nm_cluster {
101         struct config_group     cl_group;
102         unsigned                cl_has_local:1;
103         u8                      cl_local_node;
104         rwlock_t                cl_nodes_lock;
105         struct o2nm_node        *cl_nodes[O2NM_MAX_NODES];
106         struct rb_root          cl_node_ip_tree;
107         /* this bitmap is part of a hack for disk bitmap.. will go eventually. - zab */
108         unsigned long   cl_nodes_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
109 };
110
111 struct o2nm_node *o2nm_get_node_by_num(u8 node_num)
112 {
113         struct o2nm_node *node = NULL;
114
115         if (node_num >= O2NM_MAX_NODES || o2nm_single_cluster == NULL)
116                 goto out;
117
118         read_lock(&o2nm_single_cluster->cl_nodes_lock);
119         node = o2nm_single_cluster->cl_nodes[node_num];
120         if (node)
121                 config_item_get(&node->nd_item);
122         read_unlock(&o2nm_single_cluster->cl_nodes_lock);
123 out:
124         return node;
125 }
126 EXPORT_SYMBOL_GPL(o2nm_get_node_by_num);
127
128 int o2nm_configured_node_map(unsigned long *map, unsigned bytes)
129 {
130         struct o2nm_cluster *cluster = o2nm_single_cluster;
131
132         BUG_ON(bytes < (sizeof(cluster->cl_nodes_bitmap)));
133
134         if (cluster == NULL)
135                 return -EINVAL;
136
137         read_lock(&cluster->cl_nodes_lock);
138         memcpy(map, cluster->cl_nodes_bitmap, sizeof(cluster->cl_nodes_bitmap));
139         read_unlock(&cluster->cl_nodes_lock);
140
141         return 0;
142 }
143 EXPORT_SYMBOL_GPL(o2nm_configured_node_map);
144
145 static struct o2nm_node *o2nm_node_ip_tree_lookup(struct o2nm_cluster *cluster,
146                                                   __be32 ip_needle,
147                                                   struct rb_node ***ret_p,
148                                                   struct rb_node **ret_parent)
149 {
150         struct rb_node **p = &cluster->cl_node_ip_tree.rb_node;
151         struct rb_node *parent = NULL;
152         struct o2nm_node *node, *ret = NULL;
153
154         while (*p) {
155                 int cmp;
156
157                 parent = *p;
158                 node = rb_entry(parent, struct o2nm_node, nd_ip_node);
159
160                 cmp = memcmp(&ip_needle, &node->nd_ipv4_address,
161                                 sizeof(ip_needle));
162                 if (cmp < 0)
163                         p = &(*p)->rb_left;
164                 else if (cmp > 0)
165                         p = &(*p)->rb_right;
166                 else {
167                         ret = node;
168                         break;
169                 }
170         }
171
172         if (ret_p != NULL)
173                 *ret_p = p;
174         if (ret_parent != NULL)
175                 *ret_parent = parent;
176
177         return ret;
178 }
179
180 struct o2nm_node *o2nm_get_node_by_ip(__be32 addr)
181 {
182         struct o2nm_node *node = NULL;
183         struct o2nm_cluster *cluster = o2nm_single_cluster;
184
185         if (cluster == NULL)
186                 goto out;
187
188         read_lock(&cluster->cl_nodes_lock);
189         node = o2nm_node_ip_tree_lookup(cluster, addr, NULL, NULL);
190         if (node)
191                 config_item_get(&node->nd_item);
192         read_unlock(&cluster->cl_nodes_lock);
193
194 out:
195         return node;
196 }
197 EXPORT_SYMBOL_GPL(o2nm_get_node_by_ip);
198
199 void o2nm_node_put(struct o2nm_node *node)
200 {
201         config_item_put(&node->nd_item);
202 }
203 EXPORT_SYMBOL_GPL(o2nm_node_put);
204
205 void o2nm_node_get(struct o2nm_node *node)
206 {
207         config_item_get(&node->nd_item);
208 }
209 EXPORT_SYMBOL_GPL(o2nm_node_get);
210
211 u8 o2nm_this_node(void)
212 {
213         u8 node_num = O2NM_MAX_NODES;
214
215         if (o2nm_single_cluster && o2nm_single_cluster->cl_has_local)
216                 node_num = o2nm_single_cluster->cl_local_node;
217
218         return node_num;
219 }
220 EXPORT_SYMBOL_GPL(o2nm_this_node);
221
222 /* node configfs bits */
223
224 static struct o2nm_cluster *to_o2nm_cluster(struct config_item *item)
225 {
226         return item ?
227                 container_of(to_config_group(item), struct o2nm_cluster,
228                              cl_group)
229                 : NULL;
230 }
231
232 static struct o2nm_node *to_o2nm_node(struct config_item *item)
233 {
234         return item ? container_of(item, struct o2nm_node, nd_item) : NULL;
235 }
236
237 static void o2nm_node_release(struct config_item *item)
238 {
239         struct o2nm_node *node = to_o2nm_node(item);
240         kfree(node);
241 }
242
243 static ssize_t o2nm_node_num_read(struct o2nm_node *node, char *page)
244 {
245         return sprintf(page, "%d\n", node->nd_num);
246 }
247
248 static struct o2nm_cluster *to_o2nm_cluster_from_node(struct o2nm_node *node)
249 {
250         /* through the first node_set .parent
251          * mycluster/nodes/mynode == o2nm_cluster->o2nm_node_group->o2nm_node */
252         return to_o2nm_cluster(node->nd_item.ci_parent->ci_parent);
253 }
254
255 enum {
256         O2NM_NODE_ATTR_NUM = 0,
257         O2NM_NODE_ATTR_PORT,
258         O2NM_NODE_ATTR_ADDRESS,
259         O2NM_NODE_ATTR_LOCAL,
260 };
261
262 static ssize_t o2nm_node_num_write(struct o2nm_node *node, const char *page,
263                                    size_t count)
264 {
265         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
266         unsigned long tmp;
267         char *p = (char *)page;
268
269         tmp = simple_strtoul(p, &p, 0);
270         if (!p || (*p && (*p != '\n')))
271                 return -EINVAL;
272
273         if (tmp >= O2NM_MAX_NODES)
274                 return -ERANGE;
275
276         /* once we're in the cl_nodes tree networking can look us up by
277          * node number and try to use our address and port attributes
278          * to connect to this node.. make sure that they've been set
279          * before writing the node attribute? */
280         if (!test_bit(O2NM_NODE_ATTR_ADDRESS, &node->nd_set_attributes) ||
281             !test_bit(O2NM_NODE_ATTR_PORT, &node->nd_set_attributes))
282                 return -EINVAL; /* XXX */
283
284         write_lock(&cluster->cl_nodes_lock);
285         if (cluster->cl_nodes[tmp])
286                 p = NULL;
287         else  {
288                 cluster->cl_nodes[tmp] = node;
289                 node->nd_num = tmp;
290                 set_bit(tmp, cluster->cl_nodes_bitmap);
291         }
292         write_unlock(&cluster->cl_nodes_lock);
293         if (p == NULL)
294                 return -EEXIST;
295
296         return count;
297 }
298 static ssize_t o2nm_node_ipv4_port_read(struct o2nm_node *node, char *page)
299 {
300         return sprintf(page, "%u\n", ntohs(node->nd_ipv4_port));
301 }
302
303 static ssize_t o2nm_node_ipv4_port_write(struct o2nm_node *node,
304                                          const char *page, size_t count)
305 {
306         unsigned long tmp;
307         char *p = (char *)page;
308
309         tmp = simple_strtoul(p, &p, 0);
310         if (!p || (*p && (*p != '\n')))
311                 return -EINVAL;
312
313         if (tmp == 0)
314                 return -EINVAL;
315         if (tmp >= (u16)-1)
316                 return -ERANGE;
317
318         node->nd_ipv4_port = htons(tmp);
319
320         return count;
321 }
322
323 static ssize_t o2nm_node_ipv4_address_read(struct o2nm_node *node, char *page)
324 {
325         return sprintf(page, "%u.%u.%u.%u\n", NIPQUAD(node->nd_ipv4_address));
326 }
327
328 static ssize_t o2nm_node_ipv4_address_write(struct o2nm_node *node,
329                                             const char *page,
330                                             size_t count)
331 {
332         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
333         int ret, i;
334         struct rb_node **p, *parent;
335         unsigned int octets[4];
336         __be32 ipv4_addr = 0;
337
338         ret = sscanf(page, "%3u.%3u.%3u.%3u", &octets[3], &octets[2],
339                      &octets[1], &octets[0]);
340         if (ret != 4)
341                 return -EINVAL;
342
343         for (i = 0; i < ARRAY_SIZE(octets); i++) {
344                 if (octets[i] > 255)
345                         return -ERANGE;
346                 be32_add_cpu(&ipv4_addr, octets[i] << (i * 8));
347         }
348
349         ret = 0;
350         write_lock(&cluster->cl_nodes_lock);
351         if (o2nm_node_ip_tree_lookup(cluster, ipv4_addr, &p, &parent))
352                 ret = -EEXIST;
353         else {
354                 rb_link_node(&node->nd_ip_node, parent, p);
355                 rb_insert_color(&node->nd_ip_node, &cluster->cl_node_ip_tree);
356         }
357         write_unlock(&cluster->cl_nodes_lock);
358         if (ret)
359                 return ret;
360
361         memcpy(&node->nd_ipv4_address, &ipv4_addr, sizeof(ipv4_addr));
362
363         return count;
364 }
365
366 static ssize_t o2nm_node_local_read(struct o2nm_node *node, char *page)
367 {
368         return sprintf(page, "%d\n", node->nd_local);
369 }
370
371 static ssize_t o2nm_node_local_write(struct o2nm_node *node, const char *page,
372                                      size_t count)
373 {
374         struct o2nm_cluster *cluster = to_o2nm_cluster_from_node(node);
375         unsigned long tmp;
376         char *p = (char *)page;
377         ssize_t ret;
378
379         tmp = simple_strtoul(p, &p, 0);
380         if (!p || (*p && (*p != '\n')))
381                 return -EINVAL;
382
383         tmp = !!tmp; /* boolean of whether this node wants to be local */
384
385         /* setting local turns on networking rx for now so we require having
386          * set everything else first */
387         if (!test_bit(O2NM_NODE_ATTR_ADDRESS, &node->nd_set_attributes) ||
388             !test_bit(O2NM_NODE_ATTR_NUM, &node->nd_set_attributes) ||
389             !test_bit(O2NM_NODE_ATTR_PORT, &node->nd_set_attributes))
390                 return -EINVAL; /* XXX */
391
392         /* the only failure case is trying to set a new local node
393          * when a different one is already set */
394         if (tmp && tmp == cluster->cl_has_local &&
395             cluster->cl_local_node != node->nd_num)
396                 return -EBUSY;
397
398         /* bring up the rx thread if we're setting the new local node. */
399         if (tmp && !cluster->cl_has_local) {
400                 ret = o2net_start_listening(node);
401                 if (ret)
402                         return ret;
403         }
404
405         if (!tmp && cluster->cl_has_local &&
406             cluster->cl_local_node == node->nd_num) {
407                 o2net_stop_listening(node);
408                 cluster->cl_local_node = O2NM_INVALID_NODE_NUM;
409         }
410
411         node->nd_local = tmp;
412         if (node->nd_local) {
413                 cluster->cl_has_local = tmp;
414                 cluster->cl_local_node = node->nd_num;
415         }
416
417         return count;
418 }
419
420 struct o2nm_node_attribute {
421         struct configfs_attribute attr;
422         ssize_t (*show)(struct o2nm_node *, char *);
423         ssize_t (*store)(struct o2nm_node *, const char *, size_t);
424 };
425
426 static struct o2nm_node_attribute o2nm_node_attr_num = {
427         .attr   = { .ca_owner = THIS_MODULE,
428                     .ca_name = "num",
429                     .ca_mode = S_IRUGO | S_IWUSR },
430         .show   = o2nm_node_num_read,
431         .store  = o2nm_node_num_write,
432 };
433
434 static struct o2nm_node_attribute o2nm_node_attr_ipv4_port = {
435         .attr   = { .ca_owner = THIS_MODULE,
436                     .ca_name = "ipv4_port",
437                     .ca_mode = S_IRUGO | S_IWUSR },
438         .show   = o2nm_node_ipv4_port_read,
439         .store  = o2nm_node_ipv4_port_write,
440 };
441
442 static struct o2nm_node_attribute o2nm_node_attr_ipv4_address = {
443         .attr   = { .ca_owner = THIS_MODULE,
444                     .ca_name = "ipv4_address",
445                     .ca_mode = S_IRUGO | S_IWUSR },
446         .show   = o2nm_node_ipv4_address_read,
447         .store  = o2nm_node_ipv4_address_write,
448 };
449
450 static struct o2nm_node_attribute o2nm_node_attr_local = {
451         .attr   = { .ca_owner = THIS_MODULE,
452                     .ca_name = "local",
453                     .ca_mode = S_IRUGO | S_IWUSR },
454         .show   = o2nm_node_local_read,
455         .store  = o2nm_node_local_write,
456 };
457
458 static struct configfs_attribute *o2nm_node_attrs[] = {
459         [O2NM_NODE_ATTR_NUM] = &o2nm_node_attr_num.attr,
460         [O2NM_NODE_ATTR_PORT] = &o2nm_node_attr_ipv4_port.attr,
461         [O2NM_NODE_ATTR_ADDRESS] = &o2nm_node_attr_ipv4_address.attr,
462         [O2NM_NODE_ATTR_LOCAL] = &o2nm_node_attr_local.attr,
463         NULL,
464 };
465
466 static int o2nm_attr_index(struct configfs_attribute *attr)
467 {
468         int i;
469         for (i = 0; i < ARRAY_SIZE(o2nm_node_attrs); i++) {
470                 if (attr == o2nm_node_attrs[i])
471                         return i;
472         }
473         BUG();
474         return 0;
475 }
476
477 static ssize_t o2nm_node_show(struct config_item *item,
478                               struct configfs_attribute *attr,
479                               char *page)
480 {
481         struct o2nm_node *node = to_o2nm_node(item);
482         struct o2nm_node_attribute *o2nm_node_attr =
483                 container_of(attr, struct o2nm_node_attribute, attr);
484         ssize_t ret = 0;
485
486         if (o2nm_node_attr->show)
487                 ret = o2nm_node_attr->show(node, page);
488         return ret;
489 }
490
491 static ssize_t o2nm_node_store(struct config_item *item,
492                                struct configfs_attribute *attr,
493                                const char *page, size_t count)
494 {
495         struct o2nm_node *node = to_o2nm_node(item);
496         struct o2nm_node_attribute *o2nm_node_attr =
497                 container_of(attr, struct o2nm_node_attribute, attr);
498         ssize_t ret;
499         int attr_index = o2nm_attr_index(attr);
500
501         if (o2nm_node_attr->store == NULL) {
502                 ret = -EINVAL;
503                 goto out;
504         }
505
506         if (test_bit(attr_index, &node->nd_set_attributes))
507                 return -EBUSY;
508
509         ret = o2nm_node_attr->store(node, page, count);
510         if (ret < count)
511                 goto out;
512
513         set_bit(attr_index, &node->nd_set_attributes);
514 out:
515         return ret;
516 }
517
518 static struct configfs_item_operations o2nm_node_item_ops = {
519         .release                = o2nm_node_release,
520         .show_attribute         = o2nm_node_show,
521         .store_attribute        = o2nm_node_store,
522 };
523
524 static struct config_item_type o2nm_node_type = {
525         .ct_item_ops    = &o2nm_node_item_ops,
526         .ct_attrs       = o2nm_node_attrs,
527         .ct_owner       = THIS_MODULE,
528 };
529
530 /* node set */
531
532 struct o2nm_node_group {
533         struct config_group ns_group;
534         /* some stuff? */
535 };
536
537 #if 0
538 static struct o2nm_node_group *to_o2nm_node_group(struct config_group *group)
539 {
540         return group ?
541                 container_of(group, struct o2nm_node_group, ns_group)
542                 : NULL;
543 }
544 #endif
545
546 static struct config_item *o2nm_node_group_make_item(struct config_group *group,
547                                                      const char *name)
548 {
549         struct o2nm_node *node = NULL;
550         struct config_item *ret = NULL;
551
552         if (strlen(name) > O2NM_MAX_NAME_LEN)
553                 goto out; /* ENAMETOOLONG */
554
555         node = kcalloc(1, sizeof(struct o2nm_node), GFP_KERNEL);
556         if (node == NULL)
557                 goto out; /* ENOMEM */
558
559         strcpy(node->nd_name, name); /* use item.ci_namebuf instead? */
560         config_item_init_type_name(&node->nd_item, name, &o2nm_node_type);
561         spin_lock_init(&node->nd_lock);
562
563         ret = &node->nd_item;
564
565 out:
566         if (ret == NULL)
567                 kfree(node);
568
569         return ret;
570 }
571
572 static void o2nm_node_group_drop_item(struct config_group *group,
573                                       struct config_item *item)
574 {
575         struct o2nm_node *node = to_o2nm_node(item);
576         struct o2nm_cluster *cluster = to_o2nm_cluster(group->cg_item.ci_parent);
577
578         o2net_disconnect_node(node);
579
580         if (cluster->cl_has_local &&
581             (cluster->cl_local_node == node->nd_num)) {
582                 cluster->cl_has_local = 0;
583                 cluster->cl_local_node = O2NM_INVALID_NODE_NUM;
584                 o2net_stop_listening(node);
585         }
586
587         /* XXX call into net to stop this node from trading messages */
588
589         write_lock(&cluster->cl_nodes_lock);
590
591         /* XXX sloppy */
592         if (node->nd_ipv4_address)
593                 rb_erase(&node->nd_ip_node, &cluster->cl_node_ip_tree);
594
595         /* nd_num might be 0 if the node number hasn't been set.. */
596         if (cluster->cl_nodes[node->nd_num] == node) {
597                 cluster->cl_nodes[node->nd_num] = NULL;
598                 clear_bit(node->nd_num, cluster->cl_nodes_bitmap);
599         }
600         write_unlock(&cluster->cl_nodes_lock);
601
602         config_item_put(item);
603 }
604
605 static struct configfs_group_operations o2nm_node_group_group_ops = {
606         .make_item      = o2nm_node_group_make_item,
607         .drop_item      = o2nm_node_group_drop_item,
608 };
609
610 static struct config_item_type o2nm_node_group_type = {
611         .ct_group_ops   = &o2nm_node_group_group_ops,
612         .ct_owner       = THIS_MODULE,
613 };
614
615 /* cluster */
616
617 static void o2nm_cluster_release(struct config_item *item)
618 {
619         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
620
621         kfree(cluster->cl_group.default_groups);
622         kfree(cluster);
623 }
624
625 static struct configfs_item_operations o2nm_cluster_item_ops = {
626         .release        = o2nm_cluster_release,
627 };
628
629 static struct config_item_type o2nm_cluster_type = {
630         .ct_item_ops    = &o2nm_cluster_item_ops,
631         .ct_owner       = THIS_MODULE,
632 };
633
634 /* cluster set */
635
636 struct o2nm_cluster_group {
637         struct configfs_subsystem cs_subsys;
638         /* some stuff? */
639 };
640
641 #if 0
642 static struct o2nm_cluster_group *to_o2nm_cluster_group(struct config_group *group)
643 {
644         return group ?
645                 container_of(to_configfs_subsystem(group), struct o2nm_cluster_group, cs_subsys)
646                : NULL;
647 }
648 #endif
649
650 static struct config_group *o2nm_cluster_group_make_group(struct config_group *group,
651                                                           const char *name)
652 {
653         struct o2nm_cluster *cluster = NULL;
654         struct o2nm_node_group *ns = NULL;
655         struct config_group *o2hb_group = NULL, *ret = NULL;
656         void *defs = NULL;
657
658         /* this runs under the parent dir's i_mutex; there can be only
659          * one caller in here at a time */
660         if (o2nm_single_cluster)
661                 goto out; /* ENOSPC */
662
663         cluster = kcalloc(1, sizeof(struct o2nm_cluster), GFP_KERNEL);
664         ns = kcalloc(1, sizeof(struct o2nm_node_group), GFP_KERNEL);
665         defs = kcalloc(3, sizeof(struct config_group *), GFP_KERNEL);
666         o2hb_group = o2hb_alloc_hb_set();
667         if (cluster == NULL || ns == NULL || o2hb_group == NULL || defs == NULL)
668                 goto out;
669
670         config_group_init_type_name(&cluster->cl_group, name,
671                                     &o2nm_cluster_type);
672         config_group_init_type_name(&ns->ns_group, "node",
673                                     &o2nm_node_group_type);
674
675         cluster->cl_group.default_groups = defs;
676         cluster->cl_group.default_groups[0] = &ns->ns_group;
677         cluster->cl_group.default_groups[1] = o2hb_group;
678         cluster->cl_group.default_groups[2] = NULL;
679         rwlock_init(&cluster->cl_nodes_lock);
680         cluster->cl_node_ip_tree = RB_ROOT;
681
682         ret = &cluster->cl_group;
683         o2nm_single_cluster = cluster;
684
685 out:
686         if (ret == NULL) {
687                 kfree(cluster);
688                 kfree(ns);
689                 o2hb_free_hb_set(o2hb_group);
690                 kfree(defs);
691         }
692
693         return ret;
694 }
695
696 static void o2nm_cluster_group_drop_item(struct config_group *group, struct config_item *item)
697 {
698         struct o2nm_cluster *cluster = to_o2nm_cluster(item);
699         int i;
700         struct config_item *killme;
701
702         BUG_ON(o2nm_single_cluster != cluster);
703         o2nm_single_cluster = NULL;
704
705         for (i = 0; cluster->cl_group.default_groups[i]; i++) {
706                 killme = &cluster->cl_group.default_groups[i]->cg_item;
707                 cluster->cl_group.default_groups[i] = NULL;
708                 config_item_put(killme);
709         }
710
711         config_item_put(item);
712 }
713
714 static struct configfs_group_operations o2nm_cluster_group_group_ops = {
715         .make_group     = o2nm_cluster_group_make_group,
716         .drop_item      = o2nm_cluster_group_drop_item,
717 };
718
719 static struct config_item_type o2nm_cluster_group_type = {
720         .ct_group_ops   = &o2nm_cluster_group_group_ops,
721         .ct_owner       = THIS_MODULE,
722 };
723
724 static struct o2nm_cluster_group o2nm_cluster_group = {
725         .cs_subsys = {
726                 .su_group = {
727                         .cg_item = {
728                                 .ci_namebuf = "cluster",
729                                 .ci_type = &o2nm_cluster_group_type,
730                         },
731                 },
732         },
733 };
734
735 static void __exit exit_o2nm(void)
736 {
737         if (ocfs2_table_header)
738                 unregister_sysctl_table(ocfs2_table_header);
739
740         /* XXX sync with hb callbacks and shut down hb? */
741         o2net_unregister_hb_callbacks();
742         configfs_unregister_subsystem(&o2nm_cluster_group.cs_subsys);
743         o2cb_sys_shutdown();
744
745         o2net_exit();
746 }
747
748 static int __init init_o2nm(void)
749 {
750         int ret = -1;
751
752         cluster_print_version();
753
754         o2hb_init();
755         o2net_init();
756
757         ocfs2_table_header = register_sysctl_table(ocfs2_root_table, 0);
758         if (!ocfs2_table_header) {
759                 printk(KERN_ERR "nodemanager: unable to register sysctl\n");
760                 ret = -ENOMEM; /* or something. */
761                 goto out_o2net;
762         }
763
764         ret = o2net_register_hb_callbacks();
765         if (ret)
766                 goto out_sysctl;
767
768         config_group_init(&o2nm_cluster_group.cs_subsys.su_group);
769         init_MUTEX(&o2nm_cluster_group.cs_subsys.su_sem);
770         ret = configfs_register_subsystem(&o2nm_cluster_group.cs_subsys);
771         if (ret) {
772                 printk(KERN_ERR "nodemanager: Registration returned %d\n", ret);
773                 goto out_callbacks;
774         }
775
776         ret = o2cb_sys_init();
777         if (!ret)
778                 goto out;
779
780         configfs_unregister_subsystem(&o2nm_cluster_group.cs_subsys);
781 out_callbacks:
782         o2net_unregister_hb_callbacks();
783 out_sysctl:
784         unregister_sysctl_table(ocfs2_table_header);
785 out_o2net:
786         o2net_exit();
787 out:
788         return ret;
789 }
790
791 MODULE_AUTHOR("Oracle");
792 MODULE_LICENSE("GPL");
793
794 module_init(init_o2nm)
795 module_exit(exit_o2nm)