[linux-2.6-block.git] / fs / ocfs2 / cluster / quorum.c

// SPDX-License-Identifier: GPL-2.0-or-later
/* -*- mode: c; c-basic-offset: 8; -*-
 *
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * Copyright (C) 2005 Oracle.  All rights reserved.
 */

/* This quorum hack is only here until we transition to some more rational
 * approach that is driven from userspace.  Honest.  No foolin'.
 *
 * Imagine two nodes lose network connectivity to each other but they're still
 * up and operating in every other way.  Presumably a network timeout indicates
 * that a node is broken and should be recovered.  They can't both recover each
 * other and both carry on without serialising their access to the file system.
 * They need to decide who is authoritative.  Now extend that problem to
 * arbitrary groups of nodes losing connectivity between each other.
 *
 * So we declare that a node which has given up on connecting to a majority
 * of nodes who are still heartbeating will fence itself.
 *
 * There are huge opportunities for races here.  After we give up on a node's
 * connection we need to wait long enough to give heartbeat an opportunity
 * to declare the node as truly dead.  We also need to be careful with the
 * race between when we see a node start heartbeating and when we connect
 * to it.
 *
 * So nodes that are in this transtion put a hold on the quorum decision
 * with a counter.  As they fall out of this transition they drop the count
 * and if they're the last, they fire off the decision.
 */
#include <linux/kernel.h>
#include <linux/workqueue.h>
#include <linux/reboot.h>

#include "heartbeat.h"
#include "nodemanager.h"
#define MLOG_MASK_PREFIX ML_QUORUM
#include "masklog.h"
#include "quorum.h"

static struct o2quo_state {
	spinlock_t		qs_lock;
	struct work_struct	qs_work;
	int			qs_pending;
	int			qs_heartbeating;
	unsigned long		qs_hb_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
	int			qs_connected;
	unsigned long		qs_conn_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
	int			qs_holds;
	unsigned long		qs_hold_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
} o2quo_state;

/* this is horribly heavy-handed.  It should instead flip the file
 * system RO and call some userspace script. */
static void o2quo_fence_self(void)
{
	/* panic spins with interrupts enabled.  with preempt
	 * threads can still schedule, etc, etc */
	o2hb_stop_all_regions();

	switch (o2nm_single_cluster->cl_fence_method) {
	case O2NM_FENCE_PANIC:
		panic("*** ocfs2 is very sorry to be fencing this system by "
		      "panicing ***\n");
		break;
	default:
		WARN_ON(o2nm_single_cluster->cl_fence_method >=
			O2NM_FENCE_METHODS);
		/* fall through */
	case O2NM_FENCE_RESET:
		printk(KERN_ERR "*** ocfs2 is very sorry to be fencing this "
		       "system by restarting ***\n");
		emergency_restart();
		break;
	};
}

/* Indicate that a timeout occurred on a heartbeat region write. The
 * other nodes in the cluster may consider us dead at that time so we
 * want to "fence" ourselves so that we don't scribble on the disk
 * after they think they've recovered us. This can't solve all
 * problems related to writeout after recovery but this hack can at
 * least close some of those gaps. When we have real fencing, this can
 * go away as our node would be fenced externally before other nodes
 * begin recovery. */
void o2quo_disk_timeout(void)
{
	o2quo_fence_self();
}

static void o2quo_make_decision(struct work_struct *work)
{
	int quorum;
	int lowest_hb, lowest_reachable = 0, fence = 0;
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	lowest_hb = find_first_bit(qs->qs_hb_bm, O2NM_MAX_NODES);
	if (lowest_hb != O2NM_MAX_NODES)
		lowest_reachable = test_bit(lowest_hb, qs->qs_conn_bm);

	mlog(0, "heartbeating: %d, connected: %d, "
	     "lowest: %d (%sreachable)\n", qs->qs_heartbeating,
	     qs->qs_connected, lowest_hb, lowest_reachable ? "" : "un");

	if (!test_bit(o2nm_this_node(), qs->qs_hb_bm) ||
	    qs->qs_heartbeating == 1)
		goto out;

	if (qs->qs_heartbeating & 1) {
		/* the odd numbered cluster case is straight forward --
		 * if we can't talk to the majority we're hosed */
		quorum = (qs->qs_heartbeating + 1)/2;
		if (qs->qs_connected < quorum) {
			mlog(ML_ERROR, "fencing this node because it is "
			     "only connected to %u nodes and %u is needed "
			     "to make a quorum out of %u heartbeating nodes\n",
			     qs->qs_connected, quorum,
			     qs->qs_heartbeating);
			fence = 1;
		}
	} else {
		/* the even numbered cluster adds the possibility of each half
		 * of the cluster being able to talk amongst themselves.. in
		 * that case we're hosed if we can't talk to the group that has
		 * the lowest numbered node */
		quorum = qs->qs_heartbeating / 2;
		if (qs->qs_connected < quorum) {
			mlog(ML_ERROR, "fencing this node because it is "
			     "only connected to %u nodes and %u is needed "
			     "to make a quorum out of %u heartbeating nodes\n",
			     qs->qs_connected, quorum,
			     qs->qs_heartbeating);
			fence = 1;
		}
		else if ((qs->qs_connected == quorum) &&
			 !lowest_reachable) {
			mlog(ML_ERROR, "fencing this node because it is "
			     "connected to a half-quorum of %u out of %u "
			     "nodes which doesn't include the lowest active "
			     "node %u\n", quorum, qs->qs_heartbeating,
			     lowest_hb);
			fence = 1;
		}
	}

out:
	if (fence) {
		spin_unlock(&qs->qs_lock);
		o2quo_fence_self();
	} else {
		mlog(ML_NOTICE, "not fencing this node, heartbeating: %d, "
			"connected: %d, lowest: %d (%sreachable)\n",
			qs->qs_heartbeating, qs->qs_connected, lowest_hb,
			lowest_reachable ? "" : "un");
		spin_unlock(&qs->qs_lock);

	}

}

static void o2quo_set_hold(struct o2quo_state *qs, u8 node)
{
	assert_spin_locked(&qs->qs_lock);

	if (!test_and_set_bit(node, qs->qs_hold_bm)) {
		qs->qs_holds++;
		mlog_bug_on_msg(qs->qs_holds == O2NM_MAX_NODES,
			        "node %u\n", node);
		mlog(0, "node %u, %d total\n", node, qs->qs_holds);
	}
}

static void o2quo_clear_hold(struct o2quo_state *qs, u8 node)
{
	assert_spin_locked(&qs->qs_lock);

	if (test_and_clear_bit(node, qs->qs_hold_bm)) {
		mlog(0, "node %u, %d total\n", node, qs->qs_holds - 1);
		if (--qs->qs_holds == 0) {
			if (qs->qs_pending) {
				qs->qs_pending = 0;
				schedule_work(&qs->qs_work);
			}
		}
		mlog_bug_on_msg(qs->qs_holds < 0, "node %u, holds %d\n",
				node, qs->qs_holds);
	}
}

/* as a node comes up we delay the quorum decision until we know the fate of
 * the connection.  the hold will be droped in conn_up or hb_down.  it might be
 * perpetuated by con_err until hb_down.  if we already have a conn, we might
 * be dropping a hold that conn_up got. */
void o2quo_hb_up(u8 node)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	qs->qs_heartbeating++;
	mlog_bug_on_msg(qs->qs_heartbeating == O2NM_MAX_NODES,
		        "node %u\n", node);
	mlog_bug_on_msg(test_bit(node, qs->qs_hb_bm), "node %u\n", node);
	set_bit(node, qs->qs_hb_bm);

	mlog(0, "node %u, %d total\n", node, qs->qs_heartbeating);

	if (!test_bit(node, qs->qs_conn_bm))
		o2quo_set_hold(qs, node);
	else
		o2quo_clear_hold(qs, node);

	spin_unlock(&qs->qs_lock);
}

/* hb going down releases any holds we might have had due to this node from
 * conn_up, conn_err, or hb_up */
void o2quo_hb_down(u8 node)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	qs->qs_heartbeating--;
	mlog_bug_on_msg(qs->qs_heartbeating < 0,
			"node %u, %d heartbeating\n",
			node, qs->qs_heartbeating);
	mlog_bug_on_msg(!test_bit(node, qs->qs_hb_bm), "node %u\n", node);
	clear_bit(node, qs->qs_hb_bm);

	mlog(0, "node %u, %d total\n", node, qs->qs_heartbeating);

	o2quo_clear_hold(qs, node);

	spin_unlock(&qs->qs_lock);
}

/* this tells us that we've decided that the node is still heartbeating
 * even though we've lost it's conn.  it must only be called after conn_err
 * and indicates that we must now make a quorum decision in the future,
 * though we might be doing so after waiting for holds to drain.  Here
 * we'll be dropping the hold from conn_err. */
void o2quo_hb_still_up(u8 node)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	mlog(0, "node %u\n", node);

	qs->qs_pending = 1;
	o2quo_clear_hold(qs, node);

	spin_unlock(&qs->qs_lock);
}

/* This is analogous to hb_up.  as a node's connection comes up we delay the
 * quorum decision until we see it heartbeating.  the hold will be droped in
 * hb_up or hb_down.  it might be perpetuated by con_err until hb_down.  if
 * it's already heartbeating we might be dropping a hold that conn_up got.
 * */
void o2quo_conn_up(u8 node)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	qs->qs_connected++;
	mlog_bug_on_msg(qs->qs_connected == O2NM_MAX_NODES,
		        "node %u\n", node);
	mlog_bug_on_msg(test_bit(node, qs->qs_conn_bm), "node %u\n", node);
	set_bit(node, qs->qs_conn_bm);

	mlog(0, "node %u, %d total\n", node, qs->qs_connected);

	if (!test_bit(node, qs->qs_hb_bm))
		o2quo_set_hold(qs, node);
	else
		o2quo_clear_hold(qs, node);

	spin_unlock(&qs->qs_lock);
}

/* we've decided that we won't ever be connecting to the node again.  if it's
 * still heartbeating we grab a hold that will delay decisions until either the
 * node stops heartbeating from hb_down or the caller decides that the node is
 * still up and calls still_up */
void o2quo_conn_err(u8 node)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock(&qs->qs_lock);

	if (test_bit(node, qs->qs_conn_bm)) {
		qs->qs_connected--;
		mlog_bug_on_msg(qs->qs_connected < 0,
				"node %u, connected %d\n",
				node, qs->qs_connected);

		clear_bit(node, qs->qs_conn_bm);

		if (test_bit(node, qs->qs_hb_bm))
			o2quo_set_hold(qs, node);
	}

	mlog(0, "node %u, %d total\n", node, qs->qs_connected);


	spin_unlock(&qs->qs_lock);
}

void o2quo_init(void)
{
	struct o2quo_state *qs = &o2quo_state;

	spin_lock_init(&qs->qs_lock);
	INIT_WORK(&qs->qs_work, o2quo_make_decision);
}

void o2quo_exit(void)
{
	struct o2quo_state *qs = &o2quo_state;

	flush_work(&qs->qs_work);
}
Commit	Line	Data
328970de	1	// SPDX-License-Identifier: GPL-2.0-or-later
98211489 ZB	2	/* -- mode: c; c-basic-offset: 8; --
	3	*
	4	* vim: noexpandtab sw=8 ts=8 sts=0:
	5	*
	6	* Copyright (C) 2005 Oracle. All rights reserved.
98211489 ZB	7	*/
	8
	9	/* This quorum hack is only here until we transition to some more rational
	10	* approach that is driven from userspace. Honest. No foolin'.
	11	*
	12	* Imagine two nodes lose network connectivity to each other but they're still
	13	* up and operating in every other way. Presumably a network timeout indicates
	14	* that a node is broken and should be recovered. They can't both recover each
	15	* other and both carry on without serialising their access to the file system.
	16	* They need to decide who is authoritative. Now extend that problem to
	17	* arbitrary groups of nodes losing connectivity between each other.
	18	*
	19	* So we declare that a node which has given up on connecting to a majority
	20	* of nodes who are still heartbeating will fence itself.
	21	*
	22	* There are huge opportunities for races here. After we give up on a node's
	23	* connection we need to wait long enough to give heartbeat an opportunity
	24	* to declare the node as truly dead. We also need to be careful with the
	25	* race between when we see a node start heartbeating and when we connect
	26	* to it.
	27	*
	28	* So nodes that are in this transtion put a hold on the quorum decision
	29	* with a counter. As they fall out of this transition they drop the count
	30	* and if they're the last, they fire off the decision.
	31	*/
	32	#include <linux/kernel.h>
98211489	33	#include <linux/workqueue.h>
bebe6f12	34	#include <linux/reboot.h>
98211489 ZB	35
	36	#include "heartbeat.h"
	37	#include "nodemanager.h"
	38	#define MLOG_MASK_PREFIX ML_QUORUM
	39	#include "masklog.h"
	40	#include "quorum.h"
	41
	42	static struct o2quo_state {
	43	spinlock_t qs_lock;
	44	struct work_struct qs_work;
	45	int qs_pending;
	46	int qs_heartbeating;
	47	unsigned long qs_hb_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
	48	int qs_connected;
	49	unsigned long qs_conn_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
	50	int qs_holds;
	51	unsigned long qs_hold_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
	52	} o2quo_state;
	53
	54	/* this is horribly heavy-handed. It should instead flip the file
	55	* system RO and call some userspace script. */
	56	static void o2quo_fence_self(void)
	57	{
	58	/* panic spins with interrupts enabled. with preempt
	59	* threads can still schedule, etc, etc */
	60	o2hb_stop_all_regions();
bebe6f12	61
f6656d26 SM	62	switch (o2nm_single_cluster->cl_fence_method) {
	63	case O2NM_FENCE_PANIC:
	64	panic("*** ocfs2 is very sorry to be fencing this system by "
	65	"panicing ***\n");
	66	break;
	67	default:
	68	WARN_ON(o2nm_single_cluster->cl_fence_method >=
	69	O2NM_FENCE_METHODS);
0a4c9265	70	/* fall through */
f6656d26 SM	71	case O2NM_FENCE_RESET:
	72	printk(KERN_ERR "*** ocfs2 is very sorry to be fencing this "
	73	"system by restarting ***\n");
	74	emergency_restart();
	75	break;
	76	};
98211489 ZB	77	}
98211489 ZB	78
e926d8a1	79	/* Indicate that a timeout occurred on a heartbeat region write. The
98211489 ZB	80	* other nodes in the cluster may consider us dead at that time so we
	81	* want to "fence" ourselves so that we don't scribble on the disk
	82	* after they think they've recovered us. This can't solve all
	83	* problems related to writeout after recovery but this hack can at
	84	* least close some of those gaps. When we have real fencing, this can
	85	* go away as our node would be fenced externally before other nodes
	86	* begin recovery. */
	87	void o2quo_disk_timeout(void)
	88	{
	89	o2quo_fence_self();
	90	}
	91
c4028958	92	static void o2quo_make_decision(struct work_struct *work)
98211489 ZB	93	{
	94	int quorum;
	95	int lowest_hb, lowest_reachable = 0, fence = 0;
	96	struct o2quo_state *qs = &o2quo_state;
	97
	98	spin_lock(&qs->qs_lock);
	99
	100	lowest_hb = find_first_bit(qs->qs_hb_bm, O2NM_MAX_NODES);
	101	if (lowest_hb != O2NM_MAX_NODES)
	102	lowest_reachable = test_bit(lowest_hb, qs->qs_conn_bm);
	103
	104	mlog(0, "heartbeating: %d, connected: %d, "
	105	"lowest: %d (%sreachable)\n", qs->qs_heartbeating,
	106	qs->qs_connected, lowest_hb, lowest_reachable ? "" : "un");
	107
	108	if (!test_bit(o2nm_this_node(), qs->qs_hb_bm) \|\|
	109	qs->qs_heartbeating == 1)
	110	goto out;
	111
	112	if (qs->qs_heartbeating & 1) {
	113	/* the odd numbered cluster case is straight forward --
	114	* if we can't talk to the majority we're hosed */
	115	quorum = (qs->qs_heartbeating + 1)/2;
	116	if (qs->qs_connected < quorum) {
	117	mlog(ML_ERROR, "fencing this node because it is "
	118	"only connected to %u nodes and %u is needed "
	119	"to make a quorum out of %u heartbeating nodes\n",
	120	qs->qs_connected, quorum,
	121	qs->qs_heartbeating);
	122	fence = 1;
	123	}
	124	} else {
	125	/* the even numbered cluster adds the possibility of each half
	126	* of the cluster being able to talk amongst themselves.. in
	127	* that case we're hosed if we can't talk to the group that has
	128	* the lowest numbered node */
	129	quorum = qs->qs_heartbeating / 2;
	130	if (qs->qs_connected < quorum) {
	131	mlog(ML_ERROR, "fencing this node because it is "
	132	"only connected to %u nodes and %u is needed "
	133	"to make a quorum out of %u heartbeating nodes\n",
	134	qs->qs_connected, quorum,
	135	qs->qs_heartbeating);
	136	fence = 1;
	137	}
	138	else if ((qs->qs_connected == quorum) &&
	139	!lowest_reachable) {
	140	mlog(ML_ERROR, "fencing this node because it is "
	141	"connected to a half-quorum of %u out of %u "
	142	"nodes which doesn't include the lowest active "
	143	"node %u\n", quorum, qs->qs_heartbeating,
	144	lowest_hb);
	145	fence = 1;
	146	}
	147	}
	148
	149	out:
8c7b638c JB	150	if (fence) {
8c7b638c JB	151	spin_unlock(&qs->qs_lock);
98211489	152	o2quo_fence_self();
8c7b638c JB	153	} else {
	154	mlog(ML_NOTICE, "not fencing this node, heartbeating: %d, "
	155	"connected: %d, lowest: %d (%sreachable)\n",
	156	qs->qs_heartbeating, qs->qs_connected, lowest_hb,
	157	lowest_reachable ? "" : "un");
	158	spin_unlock(&qs->qs_lock);
	159
	160	}
	161
98211489 ZB	162	}
	163
	164	static void o2quo_set_hold(struct o2quo_state *qs, u8 node)
	165	{
	166	assert_spin_locked(&qs->qs_lock);
	167
	168	if (!test_and_set_bit(node, qs->qs_hold_bm)) {
	169	qs->qs_holds++;
	170	mlog_bug_on_msg(qs->qs_holds == O2NM_MAX_NODES,
	171	"node %u\n", node);
	172	mlog(0, "node %u, %d total\n", node, qs->qs_holds);
	173	}
	174	}
	175
	176	static void o2quo_clear_hold(struct o2quo_state *qs, u8 node)
	177	{
	178	assert_spin_locked(&qs->qs_lock);
	179
	180	if (test_and_clear_bit(node, qs->qs_hold_bm)) {
	181	mlog(0, "node %u, %d total\n", node, qs->qs_holds - 1);
	182	if (--qs->qs_holds == 0) {
	183	if (qs->qs_pending) {
	184	qs->qs_pending = 0;
	185	schedule_work(&qs->qs_work);
	186	}
	187	}
	188	mlog_bug_on_msg(qs->qs_holds < 0, "node %u, holds %d\n",
	189	node, qs->qs_holds);
	190	}
	191	}
	192
	193	/* as a node comes up we delay the quorum decision until we know the fate of
	194	* the connection. the hold will be droped in conn_up or hb_down. it might be
	195	* perpetuated by con_err until hb_down. if we already have a conn, we might
	196	* be dropping a hold that conn_up got. */
	197	void o2quo_hb_up(u8 node)
	198	{
	199	struct o2quo_state *qs = &o2quo_state;
	200
	201	spin_lock(&qs->qs_lock);
	202
	203	qs->qs_heartbeating++;
	204	mlog_bug_on_msg(qs->qs_heartbeating == O2NM_MAX_NODES,
	205	"node %u\n", node);
	206	mlog_bug_on_msg(test_bit(node, qs->qs_hb_bm), "node %u\n", node);
	207	set_bit(node, qs->qs_hb_bm);
	208
	209	mlog(0, "node %u, %d total\n", node, qs->qs_heartbeating);
	210
	211	if (!test_bit(node, qs->qs_conn_bm))
	212	o2quo_set_hold(qs, node);
	213	else
	214	o2quo_clear_hold(qs, node);
	215
	216	spin_unlock(&qs->qs_lock);
	217	}
	218
	219	/* hb going down releases any holds we might have had due to this node from
	220	* conn_up, conn_err, or hb_up */
	221	void o2quo_hb_down(u8 node)
	222	{
	223	struct o2quo_state *qs = &o2quo_state;
	224
	225	spin_lock(&qs->qs_lock);
226
227	qs->qs_heartbeating--;
228	mlog_bug_on_msg(qs->qs_heartbeating < 0,
229	"node %u, %d heartbeating\n",
230	node, qs->qs_heartbeating);
231	mlog_bug_on_msg(!test_bit(node, qs->qs_hb_bm), "node %u\n", node);
232	clear_bit(node, qs->qs_hb_bm);
233
234	mlog(0, "node %u, %d total\n", node, qs->qs_heartbeating);
235
236	o2quo_clear_hold(qs, node);
237
238	spin_unlock(&qs->qs_lock);
239	}
240
241	/* this tells us that we've decided that the node is still heartbeating
242	* even though we've lost it's conn. it must only be called after conn_err
243	* and indicates that we must now make a quorum decision in the future,
244	* though we might be doing so after waiting for holds to drain. Here
245	* we'll be dropping the hold from conn_err. */
246	void o2quo_hb_still_up(u8 node)
247	{
248	struct o2quo_state *qs = &o2quo_state;
249
250	spin_lock(&qs->qs_lock);
251
252	mlog(0, "node %u\n", node);
253
254	qs->qs_pending = 1;
255	o2quo_clear_hold(qs, node);
256
257	spin_unlock(&qs->qs_lock);
258	}
259
25985edc	260	/* This is analogous to hb_up. as a node's connection comes up we delay the
98211489 ZB	261	* quorum decision until we see it heartbeating. the hold will be droped in
98211489 ZB	262	* hb_up or hb_down. it might be perpetuated by con_err until hb_down. if
b4d8ed4f	263	* it's already heartbeating we might be dropping a hold that conn_up got.
98211489 ZB	264	* */
	265	void o2quo_conn_up(u8 node)
	266	{
	267	struct o2quo_state *qs = &o2quo_state;
	268
	269	spin_lock(&qs->qs_lock);
	270
	271	qs->qs_connected++;
	272	mlog_bug_on_msg(qs->qs_connected == O2NM_MAX_NODES,
	273	"node %u\n", node);
	274	mlog_bug_on_msg(test_bit(node, qs->qs_conn_bm), "node %u\n", node);
	275	set_bit(node, qs->qs_conn_bm);
	276
	277	mlog(0, "node %u, %d total\n", node, qs->qs_connected);
	278
	279	if (!test_bit(node, qs->qs_hb_bm))
	280	o2quo_set_hold(qs, node);
	281	else
	282	o2quo_clear_hold(qs, node);
	283
	284	spin_unlock(&qs->qs_lock);
	285	}
	286
	287	/* we've decided that we won't ever be connecting to the node again. if it's
	288	* still heartbeating we grab a hold that will delay decisions until either the
	289	* node stops heartbeating from hb_down or the caller decides that the node is
	290	* still up and calls still_up */
	291	void o2quo_conn_err(u8 node)
	292	{
	293	struct o2quo_state *qs = &o2quo_state;
	294
	295	spin_lock(&qs->qs_lock);
	296
	297	if (test_bit(node, qs->qs_conn_bm)) {
	298	qs->qs_connected--;
	299	mlog_bug_on_msg(qs->qs_connected < 0,
	300	"node %u, connected %d\n",
	301	node, qs->qs_connected);
	302
	303	clear_bit(node, qs->qs_conn_bm);
fc2af28b YZ	304
	305	if (test_bit(node, qs->qs_hb_bm))
	306	o2quo_set_hold(qs, node);
98211489 ZB	307	}
	308
	309	mlog(0, "node %u, %d total\n", node, qs->qs_connected);
	310
98211489 ZB	311
	312	spin_unlock(&qs->qs_lock);
	313	}
	314
	315	void o2quo_init(void)
	316	{
	317	struct o2quo_state *qs = &o2quo_state;
	318
	319	spin_lock_init(&qs->qs_lock);
c4028958	320	INIT_WORK(&qs->qs_work, o2quo_make_decision);
98211489 ZB	321	}
	322
	323	void o2quo_exit(void)
	324	{
9b00a818 TH	325	struct o2quo_state *qs = &o2quo_state;
9b00a818 TH	326
43829731	327	flush_work(&qs->qs_work);
98211489	328	}