nodemanager.c

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