xref: /linux/fs/ocfs2/cluster/heartbeat.c (revision aaa44952bbd1d4db14a4d676bf9595bb5db7e7b0)
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* -*- mode: c; c-basic-offset: 8; -*-
3  * vim: noexpandtab sw=8 ts=8 sts=0:
4  *
5  * Copyright (C) 2004, 2005 Oracle.  All rights reserved.
6  */
7 
8 #include <linux/kernel.h>
9 #include <linux/sched.h>
10 #include <linux/jiffies.h>
11 #include <linux/module.h>
12 #include <linux/fs.h>
13 #include <linux/bio.h>
14 #include <linux/blkdev.h>
15 #include <linux/delay.h>
16 #include <linux/file.h>
17 #include <linux/kthread.h>
18 #include <linux/configfs.h>
19 #include <linux/random.h>
20 #include <linux/crc32.h>
21 #include <linux/time.h>
22 #include <linux/debugfs.h>
23 #include <linux/slab.h>
24 #include <linux/bitmap.h>
25 #include <linux/ktime.h>
26 #include "heartbeat.h"
27 #include "tcp.h"
28 #include "nodemanager.h"
29 #include "quorum.h"
30 
31 #include "masklog.h"
32 
33 
34 /*
35  * The first heartbeat pass had one global thread that would serialize all hb
36  * callback calls.  This global serializing sem should only be removed once
37  * we've made sure that all callees can deal with being called concurrently
38  * from multiple hb region threads.
39  */
40 static DECLARE_RWSEM(o2hb_callback_sem);
41 
42 /*
43  * multiple hb threads are watching multiple regions.  A node is live
44  * whenever any of the threads sees activity from the node in its region.
45  */
46 static DEFINE_SPINLOCK(o2hb_live_lock);
47 static struct list_head o2hb_live_slots[O2NM_MAX_NODES];
48 static unsigned long o2hb_live_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
49 static LIST_HEAD(o2hb_node_events);
50 static DECLARE_WAIT_QUEUE_HEAD(o2hb_steady_queue);
51 
52 /*
53  * In global heartbeat, we maintain a series of region bitmaps.
54  * 	- o2hb_region_bitmap allows us to limit the region number to max region.
55  * 	- o2hb_live_region_bitmap tracks live regions (seen steady iterations).
56  * 	- o2hb_quorum_region_bitmap tracks live regions that have seen all nodes
57  * 		heartbeat on it.
58  * 	- o2hb_failed_region_bitmap tracks the regions that have seen io timeouts.
59  */
60 static unsigned long o2hb_region_bitmap[BITS_TO_LONGS(O2NM_MAX_REGIONS)];
61 static unsigned long o2hb_live_region_bitmap[BITS_TO_LONGS(O2NM_MAX_REGIONS)];
62 static unsigned long o2hb_quorum_region_bitmap[BITS_TO_LONGS(O2NM_MAX_REGIONS)];
63 static unsigned long o2hb_failed_region_bitmap[BITS_TO_LONGS(O2NM_MAX_REGIONS)];
64 
65 #define O2HB_DB_TYPE_LIVENODES		0
66 #define O2HB_DB_TYPE_LIVEREGIONS	1
67 #define O2HB_DB_TYPE_QUORUMREGIONS	2
68 #define O2HB_DB_TYPE_FAILEDREGIONS	3
69 #define O2HB_DB_TYPE_REGION_LIVENODES	4
70 #define O2HB_DB_TYPE_REGION_NUMBER	5
71 #define O2HB_DB_TYPE_REGION_ELAPSED_TIME	6
72 #define O2HB_DB_TYPE_REGION_PINNED	7
73 struct o2hb_debug_buf {
74 	int db_type;
75 	int db_size;
76 	int db_len;
77 	void *db_data;
78 };
79 
80 static struct o2hb_debug_buf *o2hb_db_livenodes;
81 static struct o2hb_debug_buf *o2hb_db_liveregions;
82 static struct o2hb_debug_buf *o2hb_db_quorumregions;
83 static struct o2hb_debug_buf *o2hb_db_failedregions;
84 
85 #define O2HB_DEBUG_DIR			"o2hb"
86 #define O2HB_DEBUG_LIVENODES		"livenodes"
87 #define O2HB_DEBUG_LIVEREGIONS		"live_regions"
88 #define O2HB_DEBUG_QUORUMREGIONS	"quorum_regions"
89 #define O2HB_DEBUG_FAILEDREGIONS	"failed_regions"
90 #define O2HB_DEBUG_REGION_NUMBER	"num"
91 #define O2HB_DEBUG_REGION_ELAPSED_TIME	"elapsed_time_in_ms"
92 #define O2HB_DEBUG_REGION_PINNED	"pinned"
93 
94 static struct dentry *o2hb_debug_dir;
95 
96 static LIST_HEAD(o2hb_all_regions);
97 
98 static struct o2hb_callback {
99 	struct list_head list;
100 } o2hb_callbacks[O2HB_NUM_CB];
101 
102 static struct o2hb_callback *hbcall_from_type(enum o2hb_callback_type type);
103 
104 enum o2hb_heartbeat_modes {
105 	O2HB_HEARTBEAT_LOCAL		= 0,
106 	O2HB_HEARTBEAT_GLOBAL,
107 	O2HB_HEARTBEAT_NUM_MODES,
108 };
109 
110 static const char *o2hb_heartbeat_mode_desc[O2HB_HEARTBEAT_NUM_MODES] = {
111 	"local",	/* O2HB_HEARTBEAT_LOCAL */
112 	"global",	/* O2HB_HEARTBEAT_GLOBAL */
113 };
114 
115 unsigned int o2hb_dead_threshold = O2HB_DEFAULT_DEAD_THRESHOLD;
116 static unsigned int o2hb_heartbeat_mode = O2HB_HEARTBEAT_LOCAL;
117 
118 /*
119  * o2hb_dependent_users tracks the number of registered callbacks that depend
120  * on heartbeat. o2net and o2dlm are two entities that register this callback.
121  * However only o2dlm depends on the heartbeat. It does not want the heartbeat
122  * to stop while a dlm domain is still active.
123  */
124 static unsigned int o2hb_dependent_users;
125 
126 /*
127  * In global heartbeat mode, all regions are pinned if there are one or more
128  * dependent users and the quorum region count is <= O2HB_PIN_CUT_OFF. All
129  * regions are unpinned if the region count exceeds the cut off or the number
130  * of dependent users falls to zero.
131  */
132 #define O2HB_PIN_CUT_OFF		3
133 
134 /*
135  * In local heartbeat mode, we assume the dlm domain name to be the same as
136  * region uuid. This is true for domains created for the file system but not
137  * necessarily true for userdlm domains. This is a known limitation.
138  *
139  * In global heartbeat mode, we pin/unpin all o2hb regions. This solution
140  * works for both file system and userdlm domains.
141  */
142 static int o2hb_region_pin(const char *region_uuid);
143 static void o2hb_region_unpin(const char *region_uuid);
144 
145 /* Only sets a new threshold if there are no active regions.
146  *
147  * No locking or otherwise interesting code is required for reading
148  * o2hb_dead_threshold as it can't change once regions are active and
149  * it's not interesting to anyone until then anyway. */
150 static void o2hb_dead_threshold_set(unsigned int threshold)
151 {
152 	if (threshold > O2HB_MIN_DEAD_THRESHOLD) {
153 		spin_lock(&o2hb_live_lock);
154 		if (list_empty(&o2hb_all_regions))
155 			o2hb_dead_threshold = threshold;
156 		spin_unlock(&o2hb_live_lock);
157 	}
158 }
159 
160 static int o2hb_global_heartbeat_mode_set(unsigned int hb_mode)
161 {
162 	int ret = -1;
163 
164 	if (hb_mode < O2HB_HEARTBEAT_NUM_MODES) {
165 		spin_lock(&o2hb_live_lock);
166 		if (list_empty(&o2hb_all_regions)) {
167 			o2hb_heartbeat_mode = hb_mode;
168 			ret = 0;
169 		}
170 		spin_unlock(&o2hb_live_lock);
171 	}
172 
173 	return ret;
174 }
175 
176 struct o2hb_node_event {
177 	struct list_head        hn_item;
178 	enum o2hb_callback_type hn_event_type;
179 	struct o2nm_node        *hn_node;
180 	int                     hn_node_num;
181 };
182 
183 struct o2hb_disk_slot {
184 	struct o2hb_disk_heartbeat_block *ds_raw_block;
185 	u8			ds_node_num;
186 	u64			ds_last_time;
187 	u64			ds_last_generation;
188 	u16			ds_equal_samples;
189 	u16			ds_changed_samples;
190 	struct list_head	ds_live_item;
191 };
192 
193 /* each thread owns a region.. when we're asked to tear down the region
194  * we ask the thread to stop, who cleans up the region */
195 struct o2hb_region {
196 	struct config_item	hr_item;
197 
198 	struct list_head	hr_all_item;
199 	unsigned		hr_unclean_stop:1,
200 				hr_aborted_start:1,
201 				hr_item_pinned:1,
202 				hr_item_dropped:1,
203 				hr_node_deleted:1;
204 
205 	/* protected by the hr_callback_sem */
206 	struct task_struct 	*hr_task;
207 
208 	unsigned int		hr_blocks;
209 	unsigned long long	hr_start_block;
210 
211 	unsigned int		hr_block_bits;
212 	unsigned int		hr_block_bytes;
213 
214 	unsigned int		hr_slots_per_page;
215 	unsigned int		hr_num_pages;
216 
217 	struct page             **hr_slot_data;
218 	struct block_device	*hr_bdev;
219 	struct o2hb_disk_slot	*hr_slots;
220 
221 	/* live node map of this region */
222 	unsigned long		hr_live_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
223 	unsigned int		hr_region_num;
224 
225 	struct dentry		*hr_debug_dir;
226 	struct o2hb_debug_buf	*hr_db_livenodes;
227 	struct o2hb_debug_buf	*hr_db_regnum;
228 	struct o2hb_debug_buf	*hr_db_elapsed_time;
229 	struct o2hb_debug_buf	*hr_db_pinned;
230 
231 	/* let the person setting up hb wait for it to return until it
232 	 * has reached a 'steady' state.  This will be fixed when we have
233 	 * a more complete api that doesn't lead to this sort of fragility. */
234 	atomic_t		hr_steady_iterations;
235 
236 	/* terminate o2hb thread if it does not reach steady state
237 	 * (hr_steady_iterations == 0) within hr_unsteady_iterations */
238 	atomic_t		hr_unsteady_iterations;
239 
240 	char			hr_dev_name[BDEVNAME_SIZE];
241 
242 	unsigned int		hr_timeout_ms;
243 
244 	/* randomized as the region goes up and down so that a node
245 	 * recognizes a node going up and down in one iteration */
246 	u64			hr_generation;
247 
248 	struct delayed_work	hr_write_timeout_work;
249 	unsigned long		hr_last_timeout_start;
250 
251 	/* negotiate timer, used to negotiate extending hb timeout. */
252 	struct delayed_work	hr_nego_timeout_work;
253 	unsigned long		hr_nego_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
254 
255 	/* Used during o2hb_check_slot to hold a copy of the block
256 	 * being checked because we temporarily have to zero out the
257 	 * crc field. */
258 	struct o2hb_disk_heartbeat_block *hr_tmp_block;
259 
260 	/* Message key for negotiate timeout message. */
261 	unsigned int		hr_key;
262 	struct list_head	hr_handler_list;
263 
264 	/* last hb status, 0 for success, other value for error. */
265 	int			hr_last_hb_status;
266 };
267 
268 struct o2hb_bio_wait_ctxt {
269 	atomic_t          wc_num_reqs;
270 	struct completion wc_io_complete;
271 	int               wc_error;
272 };
273 
274 #define O2HB_NEGO_TIMEOUT_MS (O2HB_MAX_WRITE_TIMEOUT_MS/2)
275 
276 enum {
277 	O2HB_NEGO_TIMEOUT_MSG = 1,
278 	O2HB_NEGO_APPROVE_MSG = 2,
279 };
280 
281 struct o2hb_nego_msg {
282 	u8 node_num;
283 };
284 
285 static void o2hb_write_timeout(struct work_struct *work)
286 {
287 	int failed, quorum;
288 	struct o2hb_region *reg =
289 		container_of(work, struct o2hb_region,
290 			     hr_write_timeout_work.work);
291 
292 	mlog(ML_ERROR, "Heartbeat write timeout to device %s after %u "
293 	     "milliseconds\n", reg->hr_dev_name,
294 	     jiffies_to_msecs(jiffies - reg->hr_last_timeout_start));
295 
296 	if (o2hb_global_heartbeat_active()) {
297 		spin_lock(&o2hb_live_lock);
298 		if (test_bit(reg->hr_region_num, o2hb_quorum_region_bitmap))
299 			set_bit(reg->hr_region_num, o2hb_failed_region_bitmap);
300 		failed = bitmap_weight(o2hb_failed_region_bitmap,
301 					O2NM_MAX_REGIONS);
302 		quorum = bitmap_weight(o2hb_quorum_region_bitmap,
303 					O2NM_MAX_REGIONS);
304 		spin_unlock(&o2hb_live_lock);
305 
306 		mlog(ML_HEARTBEAT, "Number of regions %d, failed regions %d\n",
307 		     quorum, failed);
308 
309 		/*
310 		 * Fence if the number of failed regions >= half the number
311 		 * of  quorum regions
312 		 */
313 		if ((failed << 1) < quorum)
314 			return;
315 	}
316 
317 	o2quo_disk_timeout();
318 }
319 
320 static void o2hb_arm_timeout(struct o2hb_region *reg)
321 {
322 	/* Arm writeout only after thread reaches steady state */
323 	if (atomic_read(&reg->hr_steady_iterations) != 0)
324 		return;
325 
326 	mlog(ML_HEARTBEAT, "Queue write timeout for %u ms\n",
327 	     O2HB_MAX_WRITE_TIMEOUT_MS);
328 
329 	if (o2hb_global_heartbeat_active()) {
330 		spin_lock(&o2hb_live_lock);
331 		clear_bit(reg->hr_region_num, o2hb_failed_region_bitmap);
332 		spin_unlock(&o2hb_live_lock);
333 	}
334 	cancel_delayed_work(&reg->hr_write_timeout_work);
335 	schedule_delayed_work(&reg->hr_write_timeout_work,
336 			      msecs_to_jiffies(O2HB_MAX_WRITE_TIMEOUT_MS));
337 
338 	cancel_delayed_work(&reg->hr_nego_timeout_work);
339 	/* negotiate timeout must be less than write timeout. */
340 	schedule_delayed_work(&reg->hr_nego_timeout_work,
341 			      msecs_to_jiffies(O2HB_NEGO_TIMEOUT_MS));
342 	memset(reg->hr_nego_node_bitmap, 0, sizeof(reg->hr_nego_node_bitmap));
343 }
344 
345 static void o2hb_disarm_timeout(struct o2hb_region *reg)
346 {
347 	cancel_delayed_work_sync(&reg->hr_write_timeout_work);
348 	cancel_delayed_work_sync(&reg->hr_nego_timeout_work);
349 }
350 
351 static int o2hb_send_nego_msg(int key, int type, u8 target)
352 {
353 	struct o2hb_nego_msg msg;
354 	int status, ret;
355 
356 	msg.node_num = o2nm_this_node();
357 again:
358 	ret = o2net_send_message(type, key, &msg, sizeof(msg),
359 			target, &status);
360 
361 	if (ret == -EAGAIN || ret == -ENOMEM) {
362 		msleep(100);
363 		goto again;
364 	}
365 
366 	return ret;
367 }
368 
369 static void o2hb_nego_timeout(struct work_struct *work)
370 {
371 	unsigned long live_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
372 	int master_node, i, ret;
373 	struct o2hb_region *reg;
374 
375 	reg = container_of(work, struct o2hb_region, hr_nego_timeout_work.work);
376 	/* don't negotiate timeout if last hb failed since it is very
377 	 * possible io failed. Should let write timeout fence self.
378 	 */
379 	if (reg->hr_last_hb_status)
380 		return;
381 
382 	o2hb_fill_node_map(live_node_bitmap, sizeof(live_node_bitmap));
383 	/* lowest node as master node to make negotiate decision. */
384 	master_node = find_next_bit(live_node_bitmap, O2NM_MAX_NODES, 0);
385 
386 	if (master_node == o2nm_this_node()) {
387 		if (!test_bit(master_node, reg->hr_nego_node_bitmap)) {
388 			printk(KERN_NOTICE "o2hb: node %d hb write hung for %ds on region %s (%s).\n",
389 				o2nm_this_node(), O2HB_NEGO_TIMEOUT_MS/1000,
390 				config_item_name(&reg->hr_item), reg->hr_dev_name);
391 			set_bit(master_node, reg->hr_nego_node_bitmap);
392 		}
393 		if (memcmp(reg->hr_nego_node_bitmap, live_node_bitmap,
394 				sizeof(reg->hr_nego_node_bitmap))) {
395 			/* check negotiate bitmap every second to do timeout
396 			 * approve decision.
397 			 */
398 			schedule_delayed_work(&reg->hr_nego_timeout_work,
399 				msecs_to_jiffies(1000));
400 
401 			return;
402 		}
403 
404 		printk(KERN_NOTICE "o2hb: all nodes hb write hung, maybe region %s (%s) is down.\n",
405 			config_item_name(&reg->hr_item), reg->hr_dev_name);
406 		/* approve negotiate timeout request. */
407 		o2hb_arm_timeout(reg);
408 
409 		i = -1;
410 		while ((i = find_next_bit(live_node_bitmap,
411 				O2NM_MAX_NODES, i + 1)) < O2NM_MAX_NODES) {
412 			if (i == master_node)
413 				continue;
414 
415 			mlog(ML_HEARTBEAT, "send NEGO_APPROVE msg to node %d\n", i);
416 			ret = o2hb_send_nego_msg(reg->hr_key,
417 					O2HB_NEGO_APPROVE_MSG, i);
418 			if (ret)
419 				mlog(ML_ERROR, "send NEGO_APPROVE msg to node %d fail %d\n",
420 					i, ret);
421 		}
422 	} else {
423 		/* negotiate timeout with master node. */
424 		printk(KERN_NOTICE "o2hb: node %d hb write hung for %ds on region %s (%s), negotiate timeout with node %d.\n",
425 			o2nm_this_node(), O2HB_NEGO_TIMEOUT_MS/1000, config_item_name(&reg->hr_item),
426 			reg->hr_dev_name, master_node);
427 		ret = o2hb_send_nego_msg(reg->hr_key, O2HB_NEGO_TIMEOUT_MSG,
428 				master_node);
429 		if (ret)
430 			mlog(ML_ERROR, "send NEGO_TIMEOUT msg to node %d fail %d\n",
431 				master_node, ret);
432 	}
433 }
434 
435 static int o2hb_nego_timeout_handler(struct o2net_msg *msg, u32 len, void *data,
436 				void **ret_data)
437 {
438 	struct o2hb_region *reg = data;
439 	struct o2hb_nego_msg *nego_msg;
440 
441 	nego_msg = (struct o2hb_nego_msg *)msg->buf;
442 	printk(KERN_NOTICE "o2hb: receive negotiate timeout message from node %d on region %s (%s).\n",
443 		nego_msg->node_num, config_item_name(&reg->hr_item), reg->hr_dev_name);
444 	if (nego_msg->node_num < O2NM_MAX_NODES)
445 		set_bit(nego_msg->node_num, reg->hr_nego_node_bitmap);
446 	else
447 		mlog(ML_ERROR, "got nego timeout message from bad node.\n");
448 
449 	return 0;
450 }
451 
452 static int o2hb_nego_approve_handler(struct o2net_msg *msg, u32 len, void *data,
453 				void **ret_data)
454 {
455 	struct o2hb_region *reg = data;
456 
457 	printk(KERN_NOTICE "o2hb: negotiate timeout approved by master node on region %s (%s).\n",
458 		config_item_name(&reg->hr_item), reg->hr_dev_name);
459 	o2hb_arm_timeout(reg);
460 	return 0;
461 }
462 
463 static inline void o2hb_bio_wait_init(struct o2hb_bio_wait_ctxt *wc)
464 {
465 	atomic_set(&wc->wc_num_reqs, 1);
466 	init_completion(&wc->wc_io_complete);
467 	wc->wc_error = 0;
468 }
469 
470 /* Used in error paths too */
471 static inline void o2hb_bio_wait_dec(struct o2hb_bio_wait_ctxt *wc,
472 				     unsigned int num)
473 {
474 	/* sadly atomic_sub_and_test() isn't available on all platforms.  The
475 	 * good news is that the fast path only completes one at a time */
476 	while(num--) {
477 		if (atomic_dec_and_test(&wc->wc_num_reqs)) {
478 			BUG_ON(num > 0);
479 			complete(&wc->wc_io_complete);
480 		}
481 	}
482 }
483 
484 static void o2hb_wait_on_io(struct o2hb_bio_wait_ctxt *wc)
485 {
486 	o2hb_bio_wait_dec(wc, 1);
487 	wait_for_completion(&wc->wc_io_complete);
488 }
489 
490 static void o2hb_bio_end_io(struct bio *bio)
491 {
492 	struct o2hb_bio_wait_ctxt *wc = bio->bi_private;
493 
494 	if (bio->bi_status) {
495 		mlog(ML_ERROR, "IO Error %d\n", bio->bi_status);
496 		wc->wc_error = blk_status_to_errno(bio->bi_status);
497 	}
498 
499 	o2hb_bio_wait_dec(wc, 1);
500 	bio_put(bio);
501 }
502 
503 /* Setup a Bio to cover I/O against num_slots slots starting at
504  * start_slot. */
505 static struct bio *o2hb_setup_one_bio(struct o2hb_region *reg,
506 				      struct o2hb_bio_wait_ctxt *wc,
507 				      unsigned int *current_slot,
508 				      unsigned int max_slots, int op,
509 				      int op_flags)
510 {
511 	int len, current_page;
512 	unsigned int vec_len, vec_start;
513 	unsigned int bits = reg->hr_block_bits;
514 	unsigned int spp = reg->hr_slots_per_page;
515 	unsigned int cs = *current_slot;
516 	struct bio *bio;
517 	struct page *page;
518 
519 	/* Testing has shown this allocation to take long enough under
520 	 * GFP_KERNEL that the local node can get fenced. It would be
521 	 * nicest if we could pre-allocate these bios and avoid this
522 	 * all together. */
523 	bio = bio_alloc(GFP_ATOMIC, 16);
524 	if (!bio) {
525 		mlog(ML_ERROR, "Could not alloc slots BIO!\n");
526 		bio = ERR_PTR(-ENOMEM);
527 		goto bail;
528 	}
529 
530 	/* Must put everything in 512 byte sectors for the bio... */
531 	bio->bi_iter.bi_sector = (reg->hr_start_block + cs) << (bits - 9);
532 	bio_set_dev(bio, reg->hr_bdev);
533 	bio->bi_private = wc;
534 	bio->bi_end_io = o2hb_bio_end_io;
535 	bio_set_op_attrs(bio, op, op_flags);
536 
537 	vec_start = (cs << bits) % PAGE_SIZE;
538 	while(cs < max_slots) {
539 		current_page = cs / spp;
540 		page = reg->hr_slot_data[current_page];
541 
542 		vec_len = min(PAGE_SIZE - vec_start,
543 			      (max_slots-cs) * (PAGE_SIZE/spp) );
544 
545 		mlog(ML_HB_BIO, "page %d, vec_len = %u, vec_start = %u\n",
546 		     current_page, vec_len, vec_start);
547 
548 		len = bio_add_page(bio, page, vec_len, vec_start);
549 		if (len != vec_len) break;
550 
551 		cs += vec_len / (PAGE_SIZE/spp);
552 		vec_start = 0;
553 	}
554 
555 bail:
556 	*current_slot = cs;
557 	return bio;
558 }
559 
560 static int o2hb_read_slots(struct o2hb_region *reg,
561 			   unsigned int begin_slot,
562 			   unsigned int max_slots)
563 {
564 	unsigned int current_slot = begin_slot;
565 	int status;
566 	struct o2hb_bio_wait_ctxt wc;
567 	struct bio *bio;
568 
569 	o2hb_bio_wait_init(&wc);
570 
571 	while(current_slot < max_slots) {
572 		bio = o2hb_setup_one_bio(reg, &wc, &current_slot, max_slots,
573 					 REQ_OP_READ, 0);
574 		if (IS_ERR(bio)) {
575 			status = PTR_ERR(bio);
576 			mlog_errno(status);
577 			goto bail_and_wait;
578 		}
579 
580 		atomic_inc(&wc.wc_num_reqs);
581 		submit_bio(bio);
582 	}
583 
584 	status = 0;
585 
586 bail_and_wait:
587 	o2hb_wait_on_io(&wc);
588 	if (wc.wc_error && !status)
589 		status = wc.wc_error;
590 
591 	return status;
592 }
593 
594 static int o2hb_issue_node_write(struct o2hb_region *reg,
595 				 struct o2hb_bio_wait_ctxt *write_wc)
596 {
597 	int status;
598 	unsigned int slot;
599 	struct bio *bio;
600 
601 	o2hb_bio_wait_init(write_wc);
602 
603 	slot = o2nm_this_node();
604 
605 	bio = o2hb_setup_one_bio(reg, write_wc, &slot, slot+1, REQ_OP_WRITE,
606 				 REQ_SYNC);
607 	if (IS_ERR(bio)) {
608 		status = PTR_ERR(bio);
609 		mlog_errno(status);
610 		goto bail;
611 	}
612 
613 	atomic_inc(&write_wc->wc_num_reqs);
614 	submit_bio(bio);
615 
616 	status = 0;
617 bail:
618 	return status;
619 }
620 
621 static u32 o2hb_compute_block_crc_le(struct o2hb_region *reg,
622 				     struct o2hb_disk_heartbeat_block *hb_block)
623 {
624 	__le32 old_cksum;
625 	u32 ret;
626 
627 	/* We want to compute the block crc with a 0 value in the
628 	 * hb_cksum field. Save it off here and replace after the
629 	 * crc. */
630 	old_cksum = hb_block->hb_cksum;
631 	hb_block->hb_cksum = 0;
632 
633 	ret = crc32_le(0, (unsigned char *) hb_block, reg->hr_block_bytes);
634 
635 	hb_block->hb_cksum = old_cksum;
636 
637 	return ret;
638 }
639 
640 static void o2hb_dump_slot(struct o2hb_disk_heartbeat_block *hb_block)
641 {
642 	mlog(ML_ERROR, "Dump slot information: seq = 0x%llx, node = %u, "
643 	     "cksum = 0x%x, generation 0x%llx\n",
644 	     (long long)le64_to_cpu(hb_block->hb_seq),
645 	     hb_block->hb_node, le32_to_cpu(hb_block->hb_cksum),
646 	     (long long)le64_to_cpu(hb_block->hb_generation));
647 }
648 
649 static int o2hb_verify_crc(struct o2hb_region *reg,
650 			   struct o2hb_disk_heartbeat_block *hb_block)
651 {
652 	u32 read, computed;
653 
654 	read = le32_to_cpu(hb_block->hb_cksum);
655 	computed = o2hb_compute_block_crc_le(reg, hb_block);
656 
657 	return read == computed;
658 }
659 
660 /*
661  * Compare the slot data with what we wrote in the last iteration.
662  * If the match fails, print an appropriate error message. This is to
663  * detect errors like... another node hearting on the same slot,
664  * flaky device that is losing writes, etc.
665  * Returns 1 if check succeeds, 0 otherwise.
666  */
667 static int o2hb_check_own_slot(struct o2hb_region *reg)
668 {
669 	struct o2hb_disk_slot *slot;
670 	struct o2hb_disk_heartbeat_block *hb_block;
671 	char *errstr;
672 
673 	slot = &reg->hr_slots[o2nm_this_node()];
674 	/* Don't check on our 1st timestamp */
675 	if (!slot->ds_last_time)
676 		return 0;
677 
678 	hb_block = slot->ds_raw_block;
679 	if (le64_to_cpu(hb_block->hb_seq) == slot->ds_last_time &&
680 	    le64_to_cpu(hb_block->hb_generation) == slot->ds_last_generation &&
681 	    hb_block->hb_node == slot->ds_node_num)
682 		return 1;
683 
684 #define ERRSTR1		"Another node is heartbeating on device"
685 #define ERRSTR2		"Heartbeat generation mismatch on device"
686 #define ERRSTR3		"Heartbeat sequence mismatch on device"
687 
688 	if (hb_block->hb_node != slot->ds_node_num)
689 		errstr = ERRSTR1;
690 	else if (le64_to_cpu(hb_block->hb_generation) !=
691 		 slot->ds_last_generation)
692 		errstr = ERRSTR2;
693 	else
694 		errstr = ERRSTR3;
695 
696 	mlog(ML_ERROR, "%s (%s): expected(%u:0x%llx, 0x%llx), "
697 	     "ondisk(%u:0x%llx, 0x%llx)\n", errstr, reg->hr_dev_name,
698 	     slot->ds_node_num, (unsigned long long)slot->ds_last_generation,
699 	     (unsigned long long)slot->ds_last_time, hb_block->hb_node,
700 	     (unsigned long long)le64_to_cpu(hb_block->hb_generation),
701 	     (unsigned long long)le64_to_cpu(hb_block->hb_seq));
702 
703 	return 0;
704 }
705 
706 static inline void o2hb_prepare_block(struct o2hb_region *reg,
707 				      u64 generation)
708 {
709 	int node_num;
710 	u64 cputime;
711 	struct o2hb_disk_slot *slot;
712 	struct o2hb_disk_heartbeat_block *hb_block;
713 
714 	node_num = o2nm_this_node();
715 	slot = &reg->hr_slots[node_num];
716 
717 	hb_block = (struct o2hb_disk_heartbeat_block *)slot->ds_raw_block;
718 	memset(hb_block, 0, reg->hr_block_bytes);
719 	/* TODO: time stuff */
720 	cputime = ktime_get_real_seconds();
721 	if (!cputime)
722 		cputime = 1;
723 
724 	hb_block->hb_seq = cpu_to_le64(cputime);
725 	hb_block->hb_node = node_num;
726 	hb_block->hb_generation = cpu_to_le64(generation);
727 	hb_block->hb_dead_ms = cpu_to_le32(o2hb_dead_threshold * O2HB_REGION_TIMEOUT_MS);
728 
729 	/* This step must always happen last! */
730 	hb_block->hb_cksum = cpu_to_le32(o2hb_compute_block_crc_le(reg,
731 								   hb_block));
732 
733 	mlog(ML_HB_BIO, "our node generation = 0x%llx, cksum = 0x%x\n",
734 	     (long long)generation,
735 	     le32_to_cpu(hb_block->hb_cksum));
736 }
737 
738 static void o2hb_fire_callbacks(struct o2hb_callback *hbcall,
739 				struct o2nm_node *node,
740 				int idx)
741 {
742 	struct o2hb_callback_func *f;
743 
744 	list_for_each_entry(f, &hbcall->list, hc_item) {
745 		mlog(ML_HEARTBEAT, "calling funcs %p\n", f);
746 		(f->hc_func)(node, idx, f->hc_data);
747 	}
748 }
749 
750 /* Will run the list in order until we process the passed event */
751 static void o2hb_run_event_list(struct o2hb_node_event *queued_event)
752 {
753 	struct o2hb_callback *hbcall;
754 	struct o2hb_node_event *event;
755 
756 	/* Holding callback sem assures we don't alter the callback
757 	 * lists when doing this, and serializes ourselves with other
758 	 * processes wanting callbacks. */
759 	down_write(&o2hb_callback_sem);
760 
761 	spin_lock(&o2hb_live_lock);
762 	while (!list_empty(&o2hb_node_events)
763 	       && !list_empty(&queued_event->hn_item)) {
764 		event = list_entry(o2hb_node_events.next,
765 				   struct o2hb_node_event,
766 				   hn_item);
767 		list_del_init(&event->hn_item);
768 		spin_unlock(&o2hb_live_lock);
769 
770 		mlog(ML_HEARTBEAT, "Node %s event for %d\n",
771 		     event->hn_event_type == O2HB_NODE_UP_CB ? "UP" : "DOWN",
772 		     event->hn_node_num);
773 
774 		hbcall = hbcall_from_type(event->hn_event_type);
775 
776 		/* We should *never* have gotten on to the list with a
777 		 * bad type... This isn't something that we should try
778 		 * to recover from. */
779 		BUG_ON(IS_ERR(hbcall));
780 
781 		o2hb_fire_callbacks(hbcall, event->hn_node, event->hn_node_num);
782 
783 		spin_lock(&o2hb_live_lock);
784 	}
785 	spin_unlock(&o2hb_live_lock);
786 
787 	up_write(&o2hb_callback_sem);
788 }
789 
790 static void o2hb_queue_node_event(struct o2hb_node_event *event,
791 				  enum o2hb_callback_type type,
792 				  struct o2nm_node *node,
793 				  int node_num)
794 {
795 	assert_spin_locked(&o2hb_live_lock);
796 
797 	BUG_ON((!node) && (type != O2HB_NODE_DOWN_CB));
798 
799 	event->hn_event_type = type;
800 	event->hn_node = node;
801 	event->hn_node_num = node_num;
802 
803 	mlog(ML_HEARTBEAT, "Queue node %s event for node %d\n",
804 	     type == O2HB_NODE_UP_CB ? "UP" : "DOWN", node_num);
805 
806 	list_add_tail(&event->hn_item, &o2hb_node_events);
807 }
808 
809 static void o2hb_shutdown_slot(struct o2hb_disk_slot *slot)
810 {
811 	struct o2hb_node_event event =
812 		{ .hn_item = LIST_HEAD_INIT(event.hn_item), };
813 	struct o2nm_node *node;
814 	int queued = 0;
815 
816 	node = o2nm_get_node_by_num(slot->ds_node_num);
817 	if (!node)
818 		return;
819 
820 	spin_lock(&o2hb_live_lock);
821 	if (!list_empty(&slot->ds_live_item)) {
822 		mlog(ML_HEARTBEAT, "Shutdown, node %d leaves region\n",
823 		     slot->ds_node_num);
824 
825 		list_del_init(&slot->ds_live_item);
826 
827 		if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
828 			clear_bit(slot->ds_node_num, o2hb_live_node_bitmap);
829 
830 			o2hb_queue_node_event(&event, O2HB_NODE_DOWN_CB, node,
831 					      slot->ds_node_num);
832 			queued = 1;
833 		}
834 	}
835 	spin_unlock(&o2hb_live_lock);
836 
837 	if (queued)
838 		o2hb_run_event_list(&event);
839 
840 	o2nm_node_put(node);
841 }
842 
843 static void o2hb_set_quorum_device(struct o2hb_region *reg)
844 {
845 	if (!o2hb_global_heartbeat_active())
846 		return;
847 
848 	/* Prevent race with o2hb_heartbeat_group_drop_item() */
849 	if (kthread_should_stop())
850 		return;
851 
852 	/* Tag region as quorum only after thread reaches steady state */
853 	if (atomic_read(&reg->hr_steady_iterations) != 0)
854 		return;
855 
856 	spin_lock(&o2hb_live_lock);
857 
858 	if (test_bit(reg->hr_region_num, o2hb_quorum_region_bitmap))
859 		goto unlock;
860 
861 	/*
862 	 * A region can be added to the quorum only when it sees all
863 	 * live nodes heartbeat on it. In other words, the region has been
864 	 * added to all nodes.
865 	 */
866 	if (memcmp(reg->hr_live_node_bitmap, o2hb_live_node_bitmap,
867 		   sizeof(o2hb_live_node_bitmap)))
868 		goto unlock;
869 
870 	printk(KERN_NOTICE "o2hb: Region %s (%s) is now a quorum device\n",
871 	       config_item_name(&reg->hr_item), reg->hr_dev_name);
872 
873 	set_bit(reg->hr_region_num, o2hb_quorum_region_bitmap);
874 
875 	/*
876 	 * If global heartbeat active, unpin all regions if the
877 	 * region count > CUT_OFF
878 	 */
879 	if (bitmap_weight(o2hb_quorum_region_bitmap,
880 			   O2NM_MAX_REGIONS) > O2HB_PIN_CUT_OFF)
881 		o2hb_region_unpin(NULL);
882 unlock:
883 	spin_unlock(&o2hb_live_lock);
884 }
885 
886 static int o2hb_check_slot(struct o2hb_region *reg,
887 			   struct o2hb_disk_slot *slot)
888 {
889 	int changed = 0, gen_changed = 0;
890 	struct o2hb_node_event event =
891 		{ .hn_item = LIST_HEAD_INIT(event.hn_item), };
892 	struct o2nm_node *node;
893 	struct o2hb_disk_heartbeat_block *hb_block = reg->hr_tmp_block;
894 	u64 cputime;
895 	unsigned int dead_ms = o2hb_dead_threshold * O2HB_REGION_TIMEOUT_MS;
896 	unsigned int slot_dead_ms;
897 	int tmp;
898 	int queued = 0;
899 
900 	memcpy(hb_block, slot->ds_raw_block, reg->hr_block_bytes);
901 
902 	/*
903 	 * If a node is no longer configured but is still in the livemap, we
904 	 * may need to clear that bit from the livemap.
905 	 */
906 	node = o2nm_get_node_by_num(slot->ds_node_num);
907 	if (!node) {
908 		spin_lock(&o2hb_live_lock);
909 		tmp = test_bit(slot->ds_node_num, o2hb_live_node_bitmap);
910 		spin_unlock(&o2hb_live_lock);
911 		if (!tmp)
912 			return 0;
913 	}
914 
915 	if (!o2hb_verify_crc(reg, hb_block)) {
916 		/* all paths from here will drop o2hb_live_lock for
917 		 * us. */
918 		spin_lock(&o2hb_live_lock);
919 
920 		/* Don't print an error on the console in this case -
921 		 * a freshly formatted heartbeat area will not have a
922 		 * crc set on it. */
923 		if (list_empty(&slot->ds_live_item))
924 			goto out;
925 
926 		/* The node is live but pushed out a bad crc. We
927 		 * consider it a transient miss but don't populate any
928 		 * other values as they may be junk. */
929 		mlog(ML_ERROR, "Node %d has written a bad crc to %s\n",
930 		     slot->ds_node_num, reg->hr_dev_name);
931 		o2hb_dump_slot(hb_block);
932 
933 		slot->ds_equal_samples++;
934 		goto fire_callbacks;
935 	}
936 
937 	/* we don't care if these wrap.. the state transitions below
938 	 * clear at the right places */
939 	cputime = le64_to_cpu(hb_block->hb_seq);
940 	if (slot->ds_last_time != cputime)
941 		slot->ds_changed_samples++;
942 	else
943 		slot->ds_equal_samples++;
944 	slot->ds_last_time = cputime;
945 
946 	/* The node changed heartbeat generations. We assume this to
947 	 * mean it dropped off but came back before we timed out. We
948 	 * want to consider it down for the time being but don't want
949 	 * to lose any changed_samples state we might build up to
950 	 * considering it live again. */
951 	if (slot->ds_last_generation != le64_to_cpu(hb_block->hb_generation)) {
952 		gen_changed = 1;
953 		slot->ds_equal_samples = 0;
954 		mlog(ML_HEARTBEAT, "Node %d changed generation (0x%llx "
955 		     "to 0x%llx)\n", slot->ds_node_num,
956 		     (long long)slot->ds_last_generation,
957 		     (long long)le64_to_cpu(hb_block->hb_generation));
958 	}
959 
960 	slot->ds_last_generation = le64_to_cpu(hb_block->hb_generation);
961 
962 	mlog(ML_HEARTBEAT, "Slot %d gen 0x%llx cksum 0x%x "
963 	     "seq %llu last %llu changed %u equal %u\n",
964 	     slot->ds_node_num, (long long)slot->ds_last_generation,
965 	     le32_to_cpu(hb_block->hb_cksum),
966 	     (unsigned long long)le64_to_cpu(hb_block->hb_seq),
967 	     (unsigned long long)slot->ds_last_time, slot->ds_changed_samples,
968 	     slot->ds_equal_samples);
969 
970 	spin_lock(&o2hb_live_lock);
971 
972 fire_callbacks:
973 	/* dead nodes only come to life after some number of
974 	 * changes at any time during their dead time */
975 	if (list_empty(&slot->ds_live_item) &&
976 	    slot->ds_changed_samples >= O2HB_LIVE_THRESHOLD) {
977 		mlog(ML_HEARTBEAT, "Node %d (id 0x%llx) joined my region\n",
978 		     slot->ds_node_num, (long long)slot->ds_last_generation);
979 
980 		set_bit(slot->ds_node_num, reg->hr_live_node_bitmap);
981 
982 		/* first on the list generates a callback */
983 		if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
984 			mlog(ML_HEARTBEAT, "o2hb: Add node %d to live nodes "
985 			     "bitmap\n", slot->ds_node_num);
986 			set_bit(slot->ds_node_num, o2hb_live_node_bitmap);
987 
988 			o2hb_queue_node_event(&event, O2HB_NODE_UP_CB, node,
989 					      slot->ds_node_num);
990 
991 			changed = 1;
992 			queued = 1;
993 		}
994 
995 		list_add_tail(&slot->ds_live_item,
996 			      &o2hb_live_slots[slot->ds_node_num]);
997 
998 		slot->ds_equal_samples = 0;
999 
1000 		/* We want to be sure that all nodes agree on the
1001 		 * number of milliseconds before a node will be
1002 		 * considered dead. The self-fencing timeout is
1003 		 * computed from this value, and a discrepancy might
1004 		 * result in heartbeat calling a node dead when it
1005 		 * hasn't self-fenced yet. */
1006 		slot_dead_ms = le32_to_cpu(hb_block->hb_dead_ms);
1007 		if (slot_dead_ms && slot_dead_ms != dead_ms) {
1008 			/* TODO: Perhaps we can fail the region here. */
1009 			mlog(ML_ERROR, "Node %d on device %s has a dead count "
1010 			     "of %u ms, but our count is %u ms.\n"
1011 			     "Please double check your configuration values "
1012 			     "for 'O2CB_HEARTBEAT_THRESHOLD'\n",
1013 			     slot->ds_node_num, reg->hr_dev_name, slot_dead_ms,
1014 			     dead_ms);
1015 		}
1016 		goto out;
1017 	}
1018 
1019 	/* if the list is dead, we're done.. */
1020 	if (list_empty(&slot->ds_live_item))
1021 		goto out;
1022 
1023 	/* live nodes only go dead after enough consequtive missed
1024 	 * samples..  reset the missed counter whenever we see
1025 	 * activity */
1026 	if (slot->ds_equal_samples >= o2hb_dead_threshold || gen_changed) {
1027 		mlog(ML_HEARTBEAT, "Node %d left my region\n",
1028 		     slot->ds_node_num);
1029 
1030 		clear_bit(slot->ds_node_num, reg->hr_live_node_bitmap);
1031 
1032 		/* last off the live_slot generates a callback */
1033 		list_del_init(&slot->ds_live_item);
1034 		if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
1035 			mlog(ML_HEARTBEAT, "o2hb: Remove node %d from live "
1036 			     "nodes bitmap\n", slot->ds_node_num);
1037 			clear_bit(slot->ds_node_num, o2hb_live_node_bitmap);
1038 
1039 			/* node can be null */
1040 			o2hb_queue_node_event(&event, O2HB_NODE_DOWN_CB,
1041 					      node, slot->ds_node_num);
1042 
1043 			changed = 1;
1044 			queued = 1;
1045 		}
1046 
1047 		/* We don't clear this because the node is still
1048 		 * actually writing new blocks. */
1049 		if (!gen_changed)
1050 			slot->ds_changed_samples = 0;
1051 		goto out;
1052 	}
1053 	if (slot->ds_changed_samples) {
1054 		slot->ds_changed_samples = 0;
1055 		slot->ds_equal_samples = 0;
1056 	}
1057 out:
1058 	spin_unlock(&o2hb_live_lock);
1059 
1060 	if (queued)
1061 		o2hb_run_event_list(&event);
1062 
1063 	if (node)
1064 		o2nm_node_put(node);
1065 	return changed;
1066 }
1067 
1068 static int o2hb_highest_node(unsigned long *nodes, int numbits)
1069 {
1070 	return find_last_bit(nodes, numbits);
1071 }
1072 
1073 static int o2hb_lowest_node(unsigned long *nodes, int numbits)
1074 {
1075 	return find_first_bit(nodes, numbits);
1076 }
1077 
1078 static int o2hb_do_disk_heartbeat(struct o2hb_region *reg)
1079 {
1080 	int i, ret, highest_node, lowest_node;
1081 	int membership_change = 0, own_slot_ok = 0;
1082 	unsigned long configured_nodes[BITS_TO_LONGS(O2NM_MAX_NODES)];
1083 	unsigned long live_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
1084 	struct o2hb_bio_wait_ctxt write_wc;
1085 
1086 	ret = o2nm_configured_node_map(configured_nodes,
1087 				       sizeof(configured_nodes));
1088 	if (ret) {
1089 		mlog_errno(ret);
1090 		goto bail;
1091 	}
1092 
1093 	/*
1094 	 * If a node is not configured but is in the livemap, we still need
1095 	 * to read the slot so as to be able to remove it from the livemap.
1096 	 */
1097 	o2hb_fill_node_map(live_node_bitmap, sizeof(live_node_bitmap));
1098 	i = -1;
1099 	while ((i = find_next_bit(live_node_bitmap,
1100 				  O2NM_MAX_NODES, i + 1)) < O2NM_MAX_NODES) {
1101 		set_bit(i, configured_nodes);
1102 	}
1103 
1104 	highest_node = o2hb_highest_node(configured_nodes, O2NM_MAX_NODES);
1105 	lowest_node = o2hb_lowest_node(configured_nodes, O2NM_MAX_NODES);
1106 	if (highest_node >= O2NM_MAX_NODES || lowest_node >= O2NM_MAX_NODES) {
1107 		mlog(ML_NOTICE, "o2hb: No configured nodes found!\n");
1108 		ret = -EINVAL;
1109 		goto bail;
1110 	}
1111 
1112 	/* No sense in reading the slots of nodes that don't exist
1113 	 * yet. Of course, if the node definitions have holes in them
1114 	 * then we're reading an empty slot anyway... Consider this
1115 	 * best-effort. */
1116 	ret = o2hb_read_slots(reg, lowest_node, highest_node + 1);
1117 	if (ret < 0) {
1118 		mlog_errno(ret);
1119 		goto bail;
1120 	}
1121 
1122 	/* With an up to date view of the slots, we can check that no
1123 	 * other node has been improperly configured to heartbeat in
1124 	 * our slot. */
1125 	own_slot_ok = o2hb_check_own_slot(reg);
1126 
1127 	/* fill in the proper info for our next heartbeat */
1128 	o2hb_prepare_block(reg, reg->hr_generation);
1129 
1130 	ret = o2hb_issue_node_write(reg, &write_wc);
1131 	if (ret < 0) {
1132 		mlog_errno(ret);
1133 		goto bail;
1134 	}
1135 
1136 	i = -1;
1137 	while((i = find_next_bit(configured_nodes,
1138 				 O2NM_MAX_NODES, i + 1)) < O2NM_MAX_NODES) {
1139 		membership_change |= o2hb_check_slot(reg, &reg->hr_slots[i]);
1140 	}
1141 
1142 	/*
1143 	 * We have to be sure we've advertised ourselves on disk
1144 	 * before we can go to steady state.  This ensures that
1145 	 * people we find in our steady state have seen us.
1146 	 */
1147 	o2hb_wait_on_io(&write_wc);
1148 	if (write_wc.wc_error) {
1149 		/* Do not re-arm the write timeout on I/O error - we
1150 		 * can't be sure that the new block ever made it to
1151 		 * disk */
1152 		mlog(ML_ERROR, "Write error %d on device \"%s\"\n",
1153 		     write_wc.wc_error, reg->hr_dev_name);
1154 		ret = write_wc.wc_error;
1155 		goto bail;
1156 	}
1157 
1158 	/* Skip disarming the timeout if own slot has stale/bad data */
1159 	if (own_slot_ok) {
1160 		o2hb_set_quorum_device(reg);
1161 		o2hb_arm_timeout(reg);
1162 		reg->hr_last_timeout_start = jiffies;
1163 	}
1164 
1165 bail:
1166 	/* let the person who launched us know when things are steady */
1167 	if (atomic_read(&reg->hr_steady_iterations) != 0) {
1168 		if (!ret && own_slot_ok && !membership_change) {
1169 			if (atomic_dec_and_test(&reg->hr_steady_iterations))
1170 				wake_up(&o2hb_steady_queue);
1171 		}
1172 	}
1173 
1174 	if (atomic_read(&reg->hr_steady_iterations) != 0) {
1175 		if (atomic_dec_and_test(&reg->hr_unsteady_iterations)) {
1176 			printk(KERN_NOTICE "o2hb: Unable to stabilize "
1177 			       "heartbeat on region %s (%s)\n",
1178 			       config_item_name(&reg->hr_item),
1179 			       reg->hr_dev_name);
1180 			atomic_set(&reg->hr_steady_iterations, 0);
1181 			reg->hr_aborted_start = 1;
1182 			wake_up(&o2hb_steady_queue);
1183 			ret = -EIO;
1184 		}
1185 	}
1186 
1187 	return ret;
1188 }
1189 
1190 /*
1191  * we ride the region ref that the region dir holds.  before the region
1192  * dir is removed and drops it ref it will wait to tear down this
1193  * thread.
1194  */
1195 static int o2hb_thread(void *data)
1196 {
1197 	int i, ret;
1198 	struct o2hb_region *reg = data;
1199 	struct o2hb_bio_wait_ctxt write_wc;
1200 	ktime_t before_hb, after_hb;
1201 	unsigned int elapsed_msec;
1202 
1203 	mlog(ML_HEARTBEAT|ML_KTHREAD, "hb thread running\n");
1204 
1205 	set_user_nice(current, MIN_NICE);
1206 
1207 	/* Pin node */
1208 	ret = o2nm_depend_this_node();
1209 	if (ret) {
1210 		mlog(ML_ERROR, "Node has been deleted, ret = %d\n", ret);
1211 		reg->hr_node_deleted = 1;
1212 		wake_up(&o2hb_steady_queue);
1213 		return 0;
1214 	}
1215 
1216 	while (!kthread_should_stop() &&
1217 	       !reg->hr_unclean_stop && !reg->hr_aborted_start) {
1218 		/* We track the time spent inside
1219 		 * o2hb_do_disk_heartbeat so that we avoid more than
1220 		 * hr_timeout_ms between disk writes. On busy systems
1221 		 * this should result in a heartbeat which is less
1222 		 * likely to time itself out. */
1223 		before_hb = ktime_get_real();
1224 
1225 		ret = o2hb_do_disk_heartbeat(reg);
1226 		reg->hr_last_hb_status = ret;
1227 
1228 		after_hb = ktime_get_real();
1229 
1230 		elapsed_msec = (unsigned int)
1231 				ktime_ms_delta(after_hb, before_hb);
1232 
1233 		mlog(ML_HEARTBEAT,
1234 		     "start = %lld, end = %lld, msec = %u, ret = %d\n",
1235 		     before_hb, after_hb, elapsed_msec, ret);
1236 
1237 		if (!kthread_should_stop() &&
1238 		    elapsed_msec < reg->hr_timeout_ms) {
1239 			/* the kthread api has blocked signals for us so no
1240 			 * need to record the return value. */
1241 			msleep_interruptible(reg->hr_timeout_ms - elapsed_msec);
1242 		}
1243 	}
1244 
1245 	o2hb_disarm_timeout(reg);
1246 
1247 	/* unclean stop is only used in very bad situation */
1248 	for(i = 0; !reg->hr_unclean_stop && i < reg->hr_blocks; i++)
1249 		o2hb_shutdown_slot(&reg->hr_slots[i]);
1250 
1251 	/* Explicit down notification - avoid forcing the other nodes
1252 	 * to timeout on this region when we could just as easily
1253 	 * write a clear generation - thus indicating to them that
1254 	 * this node has left this region.
1255 	 */
1256 	if (!reg->hr_unclean_stop && !reg->hr_aborted_start) {
1257 		o2hb_prepare_block(reg, 0);
1258 		ret = o2hb_issue_node_write(reg, &write_wc);
1259 		if (ret == 0)
1260 			o2hb_wait_on_io(&write_wc);
1261 		else
1262 			mlog_errno(ret);
1263 	}
1264 
1265 	/* Unpin node */
1266 	o2nm_undepend_this_node();
1267 
1268 	mlog(ML_HEARTBEAT|ML_KTHREAD, "o2hb thread exiting\n");
1269 
1270 	return 0;
1271 }
1272 
1273 #ifdef CONFIG_DEBUG_FS
1274 static int o2hb_debug_open(struct inode *inode, struct file *file)
1275 {
1276 	struct o2hb_debug_buf *db = inode->i_private;
1277 	struct o2hb_region *reg;
1278 	unsigned long map[BITS_TO_LONGS(O2NM_MAX_NODES)];
1279 	unsigned long lts;
1280 	char *buf = NULL;
1281 	int i = -1;
1282 	int out = 0;
1283 
1284 	/* max_nodes should be the largest bitmap we pass here */
1285 	BUG_ON(sizeof(map) < db->db_size);
1286 
1287 	buf = kmalloc(PAGE_SIZE, GFP_KERNEL);
1288 	if (!buf)
1289 		goto bail;
1290 
1291 	switch (db->db_type) {
1292 	case O2HB_DB_TYPE_LIVENODES:
1293 	case O2HB_DB_TYPE_LIVEREGIONS:
1294 	case O2HB_DB_TYPE_QUORUMREGIONS:
1295 	case O2HB_DB_TYPE_FAILEDREGIONS:
1296 		spin_lock(&o2hb_live_lock);
1297 		memcpy(map, db->db_data, db->db_size);
1298 		spin_unlock(&o2hb_live_lock);
1299 		break;
1300 
1301 	case O2HB_DB_TYPE_REGION_LIVENODES:
1302 		spin_lock(&o2hb_live_lock);
1303 		reg = (struct o2hb_region *)db->db_data;
1304 		memcpy(map, reg->hr_live_node_bitmap, db->db_size);
1305 		spin_unlock(&o2hb_live_lock);
1306 		break;
1307 
1308 	case O2HB_DB_TYPE_REGION_NUMBER:
1309 		reg = (struct o2hb_region *)db->db_data;
1310 		out += scnprintf(buf + out, PAGE_SIZE - out, "%d\n",
1311 				reg->hr_region_num);
1312 		goto done;
1313 
1314 	case O2HB_DB_TYPE_REGION_ELAPSED_TIME:
1315 		reg = (struct o2hb_region *)db->db_data;
1316 		lts = reg->hr_last_timeout_start;
1317 		/* If 0, it has never been set before */
1318 		if (lts)
1319 			lts = jiffies_to_msecs(jiffies - lts);
1320 		out += scnprintf(buf + out, PAGE_SIZE - out, "%lu\n", lts);
1321 		goto done;
1322 
1323 	case O2HB_DB_TYPE_REGION_PINNED:
1324 		reg = (struct o2hb_region *)db->db_data;
1325 		out += scnprintf(buf + out, PAGE_SIZE - out, "%u\n",
1326 				!!reg->hr_item_pinned);
1327 		goto done;
1328 
1329 	default:
1330 		goto done;
1331 	}
1332 
1333 	while ((i = find_next_bit(map, db->db_len, i + 1)) < db->db_len)
1334 		out += scnprintf(buf + out, PAGE_SIZE - out, "%d ", i);
1335 	out += scnprintf(buf + out, PAGE_SIZE - out, "\n");
1336 
1337 done:
1338 	i_size_write(inode, out);
1339 
1340 	file->private_data = buf;
1341 
1342 	return 0;
1343 bail:
1344 	return -ENOMEM;
1345 }
1346 
1347 static int o2hb_debug_release(struct inode *inode, struct file *file)
1348 {
1349 	kfree(file->private_data);
1350 	return 0;
1351 }
1352 
1353 static ssize_t o2hb_debug_read(struct file *file, char __user *buf,
1354 				 size_t nbytes, loff_t *ppos)
1355 {
1356 	return simple_read_from_buffer(buf, nbytes, ppos, file->private_data,
1357 				       i_size_read(file->f_mapping->host));
1358 }
1359 #else
1360 static int o2hb_debug_open(struct inode *inode, struct file *file)
1361 {
1362 	return 0;
1363 }
1364 static int o2hb_debug_release(struct inode *inode, struct file *file)
1365 {
1366 	return 0;
1367 }
1368 static ssize_t o2hb_debug_read(struct file *file, char __user *buf,
1369 			       size_t nbytes, loff_t *ppos)
1370 {
1371 	return 0;
1372 }
1373 #endif  /* CONFIG_DEBUG_FS */
1374 
1375 static const struct file_operations o2hb_debug_fops = {
1376 	.open =		o2hb_debug_open,
1377 	.release =	o2hb_debug_release,
1378 	.read =		o2hb_debug_read,
1379 	.llseek =	generic_file_llseek,
1380 };
1381 
1382 void o2hb_exit(void)
1383 {
1384 	debugfs_remove_recursive(o2hb_debug_dir);
1385 	kfree(o2hb_db_livenodes);
1386 	kfree(o2hb_db_liveregions);
1387 	kfree(o2hb_db_quorumregions);
1388 	kfree(o2hb_db_failedregions);
1389 }
1390 
1391 static void o2hb_debug_create(const char *name, struct dentry *dir,
1392 			      struct o2hb_debug_buf **db, int db_len, int type,
1393 			      int size, int len, void *data)
1394 {
1395 	*db = kmalloc(db_len, GFP_KERNEL);
1396 	if (!*db)
1397 		return;
1398 
1399 	(*db)->db_type = type;
1400 	(*db)->db_size = size;
1401 	(*db)->db_len = len;
1402 	(*db)->db_data = data;
1403 
1404 	debugfs_create_file(name, S_IFREG|S_IRUSR, dir, *db, &o2hb_debug_fops);
1405 }
1406 
1407 static void o2hb_debug_init(void)
1408 {
1409 	o2hb_debug_dir = debugfs_create_dir(O2HB_DEBUG_DIR, NULL);
1410 
1411 	o2hb_debug_create(O2HB_DEBUG_LIVENODES, o2hb_debug_dir,
1412 			  &o2hb_db_livenodes, sizeof(*o2hb_db_livenodes),
1413 			  O2HB_DB_TYPE_LIVENODES, sizeof(o2hb_live_node_bitmap),
1414 			  O2NM_MAX_NODES, o2hb_live_node_bitmap);
1415 
1416 	o2hb_debug_create(O2HB_DEBUG_LIVEREGIONS, o2hb_debug_dir,
1417 			  &o2hb_db_liveregions, sizeof(*o2hb_db_liveregions),
1418 			  O2HB_DB_TYPE_LIVEREGIONS,
1419 			  sizeof(o2hb_live_region_bitmap), O2NM_MAX_REGIONS,
1420 			  o2hb_live_region_bitmap);
1421 
1422 	o2hb_debug_create(O2HB_DEBUG_QUORUMREGIONS, o2hb_debug_dir,
1423 			  &o2hb_db_quorumregions,
1424 			  sizeof(*o2hb_db_quorumregions),
1425 			  O2HB_DB_TYPE_QUORUMREGIONS,
1426 			  sizeof(o2hb_quorum_region_bitmap), O2NM_MAX_REGIONS,
1427 			  o2hb_quorum_region_bitmap);
1428 
1429 	o2hb_debug_create(O2HB_DEBUG_FAILEDREGIONS, o2hb_debug_dir,
1430 			  &o2hb_db_failedregions,
1431 			  sizeof(*o2hb_db_failedregions),
1432 			  O2HB_DB_TYPE_FAILEDREGIONS,
1433 			  sizeof(o2hb_failed_region_bitmap), O2NM_MAX_REGIONS,
1434 			  o2hb_failed_region_bitmap);
1435 }
1436 
1437 void o2hb_init(void)
1438 {
1439 	int i;
1440 
1441 	for (i = 0; i < ARRAY_SIZE(o2hb_callbacks); i++)
1442 		INIT_LIST_HEAD(&o2hb_callbacks[i].list);
1443 
1444 	for (i = 0; i < ARRAY_SIZE(o2hb_live_slots); i++)
1445 		INIT_LIST_HEAD(&o2hb_live_slots[i]);
1446 
1447 	INIT_LIST_HEAD(&o2hb_node_events);
1448 
1449 	memset(o2hb_live_node_bitmap, 0, sizeof(o2hb_live_node_bitmap));
1450 	memset(o2hb_region_bitmap, 0, sizeof(o2hb_region_bitmap));
1451 	memset(o2hb_live_region_bitmap, 0, sizeof(o2hb_live_region_bitmap));
1452 	memset(o2hb_quorum_region_bitmap, 0, sizeof(o2hb_quorum_region_bitmap));
1453 	memset(o2hb_failed_region_bitmap, 0, sizeof(o2hb_failed_region_bitmap));
1454 
1455 	o2hb_dependent_users = 0;
1456 
1457 	o2hb_debug_init();
1458 }
1459 
1460 /* if we're already in a callback then we're already serialized by the sem */
1461 static void o2hb_fill_node_map_from_callback(unsigned long *map,
1462 					     unsigned bytes)
1463 {
1464 	BUG_ON(bytes < (BITS_TO_LONGS(O2NM_MAX_NODES) * sizeof(unsigned long)));
1465 
1466 	memcpy(map, &o2hb_live_node_bitmap, bytes);
1467 }
1468 
1469 /*
1470  * get a map of all nodes that are heartbeating in any regions
1471  */
1472 void o2hb_fill_node_map(unsigned long *map, unsigned bytes)
1473 {
1474 	/* callers want to serialize this map and callbacks so that they
1475 	 * can trust that they don't miss nodes coming to the party */
1476 	down_read(&o2hb_callback_sem);
1477 	spin_lock(&o2hb_live_lock);
1478 	o2hb_fill_node_map_from_callback(map, bytes);
1479 	spin_unlock(&o2hb_live_lock);
1480 	up_read(&o2hb_callback_sem);
1481 }
1482 EXPORT_SYMBOL_GPL(o2hb_fill_node_map);
1483 
1484 /*
1485  * heartbeat configfs bits.  The heartbeat set is a default set under
1486  * the cluster set in nodemanager.c.
1487  */
1488 
1489 static struct o2hb_region *to_o2hb_region(struct config_item *item)
1490 {
1491 	return item ? container_of(item, struct o2hb_region, hr_item) : NULL;
1492 }
1493 
1494 /* drop_item only drops its ref after killing the thread, nothing should
1495  * be using the region anymore.  this has to clean up any state that
1496  * attributes might have built up. */
1497 static void o2hb_region_release(struct config_item *item)
1498 {
1499 	int i;
1500 	struct page *page;
1501 	struct o2hb_region *reg = to_o2hb_region(item);
1502 
1503 	mlog(ML_HEARTBEAT, "hb region release (%s)\n", reg->hr_dev_name);
1504 
1505 	kfree(reg->hr_tmp_block);
1506 
1507 	if (reg->hr_slot_data) {
1508 		for (i = 0; i < reg->hr_num_pages; i++) {
1509 			page = reg->hr_slot_data[i];
1510 			if (page)
1511 				__free_page(page);
1512 		}
1513 		kfree(reg->hr_slot_data);
1514 	}
1515 
1516 	if (reg->hr_bdev)
1517 		blkdev_put(reg->hr_bdev, FMODE_READ|FMODE_WRITE);
1518 
1519 	kfree(reg->hr_slots);
1520 
1521 	debugfs_remove_recursive(reg->hr_debug_dir);
1522 	kfree(reg->hr_db_livenodes);
1523 	kfree(reg->hr_db_regnum);
1524 	kfree(reg->hr_db_elapsed_time);
1525 	kfree(reg->hr_db_pinned);
1526 
1527 	spin_lock(&o2hb_live_lock);
1528 	list_del(&reg->hr_all_item);
1529 	spin_unlock(&o2hb_live_lock);
1530 
1531 	o2net_unregister_handler_list(&reg->hr_handler_list);
1532 	kfree(reg);
1533 }
1534 
1535 static int o2hb_read_block_input(struct o2hb_region *reg,
1536 				 const char *page,
1537 				 unsigned long *ret_bytes,
1538 				 unsigned int *ret_bits)
1539 {
1540 	unsigned long bytes;
1541 	char *p = (char *)page;
1542 
1543 	bytes = simple_strtoul(p, &p, 0);
1544 	if (!p || (*p && (*p != '\n')))
1545 		return -EINVAL;
1546 
1547 	/* Heartbeat and fs min / max block sizes are the same. */
1548 	if (bytes > 4096 || bytes < 512)
1549 		return -ERANGE;
1550 	if (hweight16(bytes) != 1)
1551 		return -EINVAL;
1552 
1553 	if (ret_bytes)
1554 		*ret_bytes = bytes;
1555 	if (ret_bits)
1556 		*ret_bits = ffs(bytes) - 1;
1557 
1558 	return 0;
1559 }
1560 
1561 static ssize_t o2hb_region_block_bytes_show(struct config_item *item,
1562 					    char *page)
1563 {
1564 	return sprintf(page, "%u\n", to_o2hb_region(item)->hr_block_bytes);
1565 }
1566 
1567 static ssize_t o2hb_region_block_bytes_store(struct config_item *item,
1568 					     const char *page,
1569 					     size_t count)
1570 {
1571 	struct o2hb_region *reg = to_o2hb_region(item);
1572 	int status;
1573 	unsigned long block_bytes;
1574 	unsigned int block_bits;
1575 
1576 	if (reg->hr_bdev)
1577 		return -EINVAL;
1578 
1579 	status = o2hb_read_block_input(reg, page, &block_bytes,
1580 				       &block_bits);
1581 	if (status)
1582 		return status;
1583 
1584 	reg->hr_block_bytes = (unsigned int)block_bytes;
1585 	reg->hr_block_bits = block_bits;
1586 
1587 	return count;
1588 }
1589 
1590 static ssize_t o2hb_region_start_block_show(struct config_item *item,
1591 					    char *page)
1592 {
1593 	return sprintf(page, "%llu\n", to_o2hb_region(item)->hr_start_block);
1594 }
1595 
1596 static ssize_t o2hb_region_start_block_store(struct config_item *item,
1597 					     const char *page,
1598 					     size_t count)
1599 {
1600 	struct o2hb_region *reg = to_o2hb_region(item);
1601 	unsigned long long tmp;
1602 	char *p = (char *)page;
1603 
1604 	if (reg->hr_bdev)
1605 		return -EINVAL;
1606 
1607 	tmp = simple_strtoull(p, &p, 0);
1608 	if (!p || (*p && (*p != '\n')))
1609 		return -EINVAL;
1610 
1611 	reg->hr_start_block = tmp;
1612 
1613 	return count;
1614 }
1615 
1616 static ssize_t o2hb_region_blocks_show(struct config_item *item, char *page)
1617 {
1618 	return sprintf(page, "%d\n", to_o2hb_region(item)->hr_blocks);
1619 }
1620 
1621 static ssize_t o2hb_region_blocks_store(struct config_item *item,
1622 					const char *page,
1623 					size_t count)
1624 {
1625 	struct o2hb_region *reg = to_o2hb_region(item);
1626 	unsigned long tmp;
1627 	char *p = (char *)page;
1628 
1629 	if (reg->hr_bdev)
1630 		return -EINVAL;
1631 
1632 	tmp = simple_strtoul(p, &p, 0);
1633 	if (!p || (*p && (*p != '\n')))
1634 		return -EINVAL;
1635 
1636 	if (tmp > O2NM_MAX_NODES || tmp == 0)
1637 		return -ERANGE;
1638 
1639 	reg->hr_blocks = (unsigned int)tmp;
1640 
1641 	return count;
1642 }
1643 
1644 static ssize_t o2hb_region_dev_show(struct config_item *item, char *page)
1645 {
1646 	unsigned int ret = 0;
1647 
1648 	if (to_o2hb_region(item)->hr_bdev)
1649 		ret = sprintf(page, "%s\n", to_o2hb_region(item)->hr_dev_name);
1650 
1651 	return ret;
1652 }
1653 
1654 static void o2hb_init_region_params(struct o2hb_region *reg)
1655 {
1656 	reg->hr_slots_per_page = PAGE_SIZE >> reg->hr_block_bits;
1657 	reg->hr_timeout_ms = O2HB_REGION_TIMEOUT_MS;
1658 
1659 	mlog(ML_HEARTBEAT, "hr_start_block = %llu, hr_blocks = %u\n",
1660 	     reg->hr_start_block, reg->hr_blocks);
1661 	mlog(ML_HEARTBEAT, "hr_block_bytes = %u, hr_block_bits = %u\n",
1662 	     reg->hr_block_bytes, reg->hr_block_bits);
1663 	mlog(ML_HEARTBEAT, "hr_timeout_ms = %u\n", reg->hr_timeout_ms);
1664 	mlog(ML_HEARTBEAT, "dead threshold = %u\n", o2hb_dead_threshold);
1665 }
1666 
1667 static int o2hb_map_slot_data(struct o2hb_region *reg)
1668 {
1669 	int i, j;
1670 	unsigned int last_slot;
1671 	unsigned int spp = reg->hr_slots_per_page;
1672 	struct page *page;
1673 	char *raw;
1674 	struct o2hb_disk_slot *slot;
1675 
1676 	reg->hr_tmp_block = kmalloc(reg->hr_block_bytes, GFP_KERNEL);
1677 	if (reg->hr_tmp_block == NULL)
1678 		return -ENOMEM;
1679 
1680 	reg->hr_slots = kcalloc(reg->hr_blocks,
1681 				sizeof(struct o2hb_disk_slot), GFP_KERNEL);
1682 	if (reg->hr_slots == NULL)
1683 		return -ENOMEM;
1684 
1685 	for(i = 0; i < reg->hr_blocks; i++) {
1686 		slot = &reg->hr_slots[i];
1687 		slot->ds_node_num = i;
1688 		INIT_LIST_HEAD(&slot->ds_live_item);
1689 		slot->ds_raw_block = NULL;
1690 	}
1691 
1692 	reg->hr_num_pages = (reg->hr_blocks + spp - 1) / spp;
1693 	mlog(ML_HEARTBEAT, "Going to require %u pages to cover %u blocks "
1694 			   "at %u blocks per page\n",
1695 	     reg->hr_num_pages, reg->hr_blocks, spp);
1696 
1697 	reg->hr_slot_data = kcalloc(reg->hr_num_pages, sizeof(struct page *),
1698 				    GFP_KERNEL);
1699 	if (!reg->hr_slot_data)
1700 		return -ENOMEM;
1701 
1702 	for(i = 0; i < reg->hr_num_pages; i++) {
1703 		page = alloc_page(GFP_KERNEL);
1704 		if (!page)
1705 			return -ENOMEM;
1706 
1707 		reg->hr_slot_data[i] = page;
1708 
1709 		last_slot = i * spp;
1710 		raw = page_address(page);
1711 		for (j = 0;
1712 		     (j < spp) && ((j + last_slot) < reg->hr_blocks);
1713 		     j++) {
1714 			BUG_ON((j + last_slot) >= reg->hr_blocks);
1715 
1716 			slot = &reg->hr_slots[j + last_slot];
1717 			slot->ds_raw_block =
1718 				(struct o2hb_disk_heartbeat_block *) raw;
1719 
1720 			raw += reg->hr_block_bytes;
1721 		}
1722 	}
1723 
1724 	return 0;
1725 }
1726 
1727 /* Read in all the slots available and populate the tracking
1728  * structures so that we can start with a baseline idea of what's
1729  * there. */
1730 static int o2hb_populate_slot_data(struct o2hb_region *reg)
1731 {
1732 	int ret, i;
1733 	struct o2hb_disk_slot *slot;
1734 	struct o2hb_disk_heartbeat_block *hb_block;
1735 
1736 	ret = o2hb_read_slots(reg, 0, reg->hr_blocks);
1737 	if (ret)
1738 		goto out;
1739 
1740 	/* We only want to get an idea of the values initially in each
1741 	 * slot, so we do no verification - o2hb_check_slot will
1742 	 * actually determine if each configured slot is valid and
1743 	 * whether any values have changed. */
1744 	for(i = 0; i < reg->hr_blocks; i++) {
1745 		slot = &reg->hr_slots[i];
1746 		hb_block = (struct o2hb_disk_heartbeat_block *) slot->ds_raw_block;
1747 
1748 		/* Only fill the values that o2hb_check_slot uses to
1749 		 * determine changing slots */
1750 		slot->ds_last_time = le64_to_cpu(hb_block->hb_seq);
1751 		slot->ds_last_generation = le64_to_cpu(hb_block->hb_generation);
1752 	}
1753 
1754 out:
1755 	return ret;
1756 }
1757 
1758 /* this is acting as commit; we set up all of hr_bdev and hr_task or nothing */
1759 static ssize_t o2hb_region_dev_store(struct config_item *item,
1760 				     const char *page,
1761 				     size_t count)
1762 {
1763 	struct o2hb_region *reg = to_o2hb_region(item);
1764 	struct task_struct *hb_task;
1765 	long fd;
1766 	int sectsize;
1767 	char *p = (char *)page;
1768 	struct fd f;
1769 	ssize_t ret = -EINVAL;
1770 	int live_threshold;
1771 
1772 	if (reg->hr_bdev)
1773 		goto out;
1774 
1775 	/* We can't heartbeat without having had our node number
1776 	 * configured yet. */
1777 	if (o2nm_this_node() == O2NM_MAX_NODES)
1778 		goto out;
1779 
1780 	fd = simple_strtol(p, &p, 0);
1781 	if (!p || (*p && (*p != '\n')))
1782 		goto out;
1783 
1784 	if (fd < 0 || fd >= INT_MAX)
1785 		goto out;
1786 
1787 	f = fdget(fd);
1788 	if (f.file == NULL)
1789 		goto out;
1790 
1791 	if (reg->hr_blocks == 0 || reg->hr_start_block == 0 ||
1792 	    reg->hr_block_bytes == 0)
1793 		goto out2;
1794 
1795 	if (!S_ISBLK(f.file->f_mapping->host->i_mode))
1796 		goto out2;
1797 
1798 	reg->hr_bdev = blkdev_get_by_dev(f.file->f_mapping->host->i_rdev,
1799 					 FMODE_WRITE | FMODE_READ, NULL);
1800 	if (IS_ERR(reg->hr_bdev)) {
1801 		ret = PTR_ERR(reg->hr_bdev);
1802 		reg->hr_bdev = NULL;
1803 		goto out2;
1804 	}
1805 
1806 	bdevname(reg->hr_bdev, reg->hr_dev_name);
1807 
1808 	sectsize = bdev_logical_block_size(reg->hr_bdev);
1809 	if (sectsize != reg->hr_block_bytes) {
1810 		mlog(ML_ERROR,
1811 		     "blocksize %u incorrect for device, expected %d",
1812 		     reg->hr_block_bytes, sectsize);
1813 		ret = -EINVAL;
1814 		goto out3;
1815 	}
1816 
1817 	o2hb_init_region_params(reg);
1818 
1819 	/* Generation of zero is invalid */
1820 	do {
1821 		get_random_bytes(&reg->hr_generation,
1822 				 sizeof(reg->hr_generation));
1823 	} while (reg->hr_generation == 0);
1824 
1825 	ret = o2hb_map_slot_data(reg);
1826 	if (ret) {
1827 		mlog_errno(ret);
1828 		goto out3;
1829 	}
1830 
1831 	ret = o2hb_populate_slot_data(reg);
1832 	if (ret) {
1833 		mlog_errno(ret);
1834 		goto out3;
1835 	}
1836 
1837 	INIT_DELAYED_WORK(&reg->hr_write_timeout_work, o2hb_write_timeout);
1838 	INIT_DELAYED_WORK(&reg->hr_nego_timeout_work, o2hb_nego_timeout);
1839 
1840 	/*
1841 	 * A node is considered live after it has beat LIVE_THRESHOLD
1842 	 * times.  We're not steady until we've given them a chance
1843 	 * _after_ our first read.
1844 	 * The default threshold is bare minimum so as to limit the delay
1845 	 * during mounts. For global heartbeat, the threshold doubled for the
1846 	 * first region.
1847 	 */
1848 	live_threshold = O2HB_LIVE_THRESHOLD;
1849 	if (o2hb_global_heartbeat_active()) {
1850 		spin_lock(&o2hb_live_lock);
1851 		if (bitmap_weight(o2hb_region_bitmap, O2NM_MAX_REGIONS) == 1)
1852 			live_threshold <<= 1;
1853 		spin_unlock(&o2hb_live_lock);
1854 	}
1855 	++live_threshold;
1856 	atomic_set(&reg->hr_steady_iterations, live_threshold);
1857 	/* unsteady_iterations is triple the steady_iterations */
1858 	atomic_set(&reg->hr_unsteady_iterations, (live_threshold * 3));
1859 
1860 	hb_task = kthread_run(o2hb_thread, reg, "o2hb-%s",
1861 			      reg->hr_item.ci_name);
1862 	if (IS_ERR(hb_task)) {
1863 		ret = PTR_ERR(hb_task);
1864 		mlog_errno(ret);
1865 		goto out3;
1866 	}
1867 
1868 	spin_lock(&o2hb_live_lock);
1869 	reg->hr_task = hb_task;
1870 	spin_unlock(&o2hb_live_lock);
1871 
1872 	ret = wait_event_interruptible(o2hb_steady_queue,
1873 				atomic_read(&reg->hr_steady_iterations) == 0 ||
1874 				reg->hr_node_deleted);
1875 	if (ret) {
1876 		atomic_set(&reg->hr_steady_iterations, 0);
1877 		reg->hr_aborted_start = 1;
1878 	}
1879 
1880 	if (reg->hr_aborted_start) {
1881 		ret = -EIO;
1882 		goto out3;
1883 	}
1884 
1885 	if (reg->hr_node_deleted) {
1886 		ret = -EINVAL;
1887 		goto out3;
1888 	}
1889 
1890 	/* Ok, we were woken.  Make sure it wasn't by drop_item() */
1891 	spin_lock(&o2hb_live_lock);
1892 	hb_task = reg->hr_task;
1893 	if (o2hb_global_heartbeat_active())
1894 		set_bit(reg->hr_region_num, o2hb_live_region_bitmap);
1895 	spin_unlock(&o2hb_live_lock);
1896 
1897 	if (hb_task)
1898 		ret = count;
1899 	else
1900 		ret = -EIO;
1901 
1902 	if (hb_task && o2hb_global_heartbeat_active())
1903 		printk(KERN_NOTICE "o2hb: Heartbeat started on region %s (%s)\n",
1904 		       config_item_name(&reg->hr_item), reg->hr_dev_name);
1905 
1906 out3:
1907 	if (ret < 0) {
1908 		blkdev_put(reg->hr_bdev, FMODE_READ | FMODE_WRITE);
1909 		reg->hr_bdev = NULL;
1910 	}
1911 out2:
1912 	fdput(f);
1913 out:
1914 	return ret;
1915 }
1916 
1917 static ssize_t o2hb_region_pid_show(struct config_item *item, char *page)
1918 {
1919 	struct o2hb_region *reg = to_o2hb_region(item);
1920 	pid_t pid = 0;
1921 
1922 	spin_lock(&o2hb_live_lock);
1923 	if (reg->hr_task)
1924 		pid = task_pid_nr(reg->hr_task);
1925 	spin_unlock(&o2hb_live_lock);
1926 
1927 	if (!pid)
1928 		return 0;
1929 
1930 	return sprintf(page, "%u\n", pid);
1931 }
1932 
1933 CONFIGFS_ATTR(o2hb_region_, block_bytes);
1934 CONFIGFS_ATTR(o2hb_region_, start_block);
1935 CONFIGFS_ATTR(o2hb_region_, blocks);
1936 CONFIGFS_ATTR(o2hb_region_, dev);
1937 CONFIGFS_ATTR_RO(o2hb_region_, pid);
1938 
1939 static struct configfs_attribute *o2hb_region_attrs[] = {
1940 	&o2hb_region_attr_block_bytes,
1941 	&o2hb_region_attr_start_block,
1942 	&o2hb_region_attr_blocks,
1943 	&o2hb_region_attr_dev,
1944 	&o2hb_region_attr_pid,
1945 	NULL,
1946 };
1947 
1948 static struct configfs_item_operations o2hb_region_item_ops = {
1949 	.release		= o2hb_region_release,
1950 };
1951 
1952 static const struct config_item_type o2hb_region_type = {
1953 	.ct_item_ops	= &o2hb_region_item_ops,
1954 	.ct_attrs	= o2hb_region_attrs,
1955 	.ct_owner	= THIS_MODULE,
1956 };
1957 
1958 /* heartbeat set */
1959 
1960 struct o2hb_heartbeat_group {
1961 	struct config_group hs_group;
1962 	/* some stuff? */
1963 };
1964 
1965 static struct o2hb_heartbeat_group *to_o2hb_heartbeat_group(struct config_group *group)
1966 {
1967 	return group ?
1968 		container_of(group, struct o2hb_heartbeat_group, hs_group)
1969 		: NULL;
1970 }
1971 
1972 static void o2hb_debug_region_init(struct o2hb_region *reg,
1973 				   struct dentry *parent)
1974 {
1975 	struct dentry *dir;
1976 
1977 	dir = debugfs_create_dir(config_item_name(&reg->hr_item), parent);
1978 	reg->hr_debug_dir = dir;
1979 
1980 	o2hb_debug_create(O2HB_DEBUG_LIVENODES, dir, &(reg->hr_db_livenodes),
1981 			  sizeof(*(reg->hr_db_livenodes)),
1982 			  O2HB_DB_TYPE_REGION_LIVENODES,
1983 			  sizeof(reg->hr_live_node_bitmap), O2NM_MAX_NODES,
1984 			  reg);
1985 
1986 	o2hb_debug_create(O2HB_DEBUG_REGION_NUMBER, dir, &(reg->hr_db_regnum),
1987 			  sizeof(*(reg->hr_db_regnum)),
1988 			  O2HB_DB_TYPE_REGION_NUMBER, 0, O2NM_MAX_NODES, reg);
1989 
1990 	o2hb_debug_create(O2HB_DEBUG_REGION_ELAPSED_TIME, dir,
1991 			  &(reg->hr_db_elapsed_time),
1992 			  sizeof(*(reg->hr_db_elapsed_time)),
1993 			  O2HB_DB_TYPE_REGION_ELAPSED_TIME, 0, 0, reg);
1994 
1995 	o2hb_debug_create(O2HB_DEBUG_REGION_PINNED, dir, &(reg->hr_db_pinned),
1996 			  sizeof(*(reg->hr_db_pinned)),
1997 			  O2HB_DB_TYPE_REGION_PINNED, 0, 0, reg);
1998 
1999 }
2000 
2001 static struct config_item *o2hb_heartbeat_group_make_item(struct config_group *group,
2002 							  const char *name)
2003 {
2004 	struct o2hb_region *reg = NULL;
2005 	int ret;
2006 
2007 	reg = kzalloc(sizeof(struct o2hb_region), GFP_KERNEL);
2008 	if (reg == NULL)
2009 		return ERR_PTR(-ENOMEM);
2010 
2011 	if (strlen(name) > O2HB_MAX_REGION_NAME_LEN) {
2012 		ret = -ENAMETOOLONG;
2013 		goto free;
2014 	}
2015 
2016 	spin_lock(&o2hb_live_lock);
2017 	reg->hr_region_num = 0;
2018 	if (o2hb_global_heartbeat_active()) {
2019 		reg->hr_region_num = find_first_zero_bit(o2hb_region_bitmap,
2020 							 O2NM_MAX_REGIONS);
2021 		if (reg->hr_region_num >= O2NM_MAX_REGIONS) {
2022 			spin_unlock(&o2hb_live_lock);
2023 			ret = -EFBIG;
2024 			goto free;
2025 		}
2026 		set_bit(reg->hr_region_num, o2hb_region_bitmap);
2027 	}
2028 	list_add_tail(&reg->hr_all_item, &o2hb_all_regions);
2029 	spin_unlock(&o2hb_live_lock);
2030 
2031 	config_item_init_type_name(&reg->hr_item, name, &o2hb_region_type);
2032 
2033 	/* this is the same way to generate msg key as dlm, for local heartbeat,
2034 	 * name is also the same, so make initial crc value different to avoid
2035 	 * message key conflict.
2036 	 */
2037 	reg->hr_key = crc32_le(reg->hr_region_num + O2NM_MAX_REGIONS,
2038 		name, strlen(name));
2039 	INIT_LIST_HEAD(&reg->hr_handler_list);
2040 	ret = o2net_register_handler(O2HB_NEGO_TIMEOUT_MSG, reg->hr_key,
2041 			sizeof(struct o2hb_nego_msg),
2042 			o2hb_nego_timeout_handler,
2043 			reg, NULL, &reg->hr_handler_list);
2044 	if (ret)
2045 		goto remove_item;
2046 
2047 	ret = o2net_register_handler(O2HB_NEGO_APPROVE_MSG, reg->hr_key,
2048 			sizeof(struct o2hb_nego_msg),
2049 			o2hb_nego_approve_handler,
2050 			reg, NULL, &reg->hr_handler_list);
2051 	if (ret)
2052 		goto unregister_handler;
2053 
2054 	o2hb_debug_region_init(reg, o2hb_debug_dir);
2055 
2056 	return &reg->hr_item;
2057 
2058 unregister_handler:
2059 	o2net_unregister_handler_list(&reg->hr_handler_list);
2060 remove_item:
2061 	spin_lock(&o2hb_live_lock);
2062 	list_del(&reg->hr_all_item);
2063 	if (o2hb_global_heartbeat_active())
2064 		clear_bit(reg->hr_region_num, o2hb_region_bitmap);
2065 	spin_unlock(&o2hb_live_lock);
2066 free:
2067 	kfree(reg);
2068 	return ERR_PTR(ret);
2069 }
2070 
2071 static void o2hb_heartbeat_group_drop_item(struct config_group *group,
2072 					   struct config_item *item)
2073 {
2074 	struct task_struct *hb_task;
2075 	struct o2hb_region *reg = to_o2hb_region(item);
2076 	int quorum_region = 0;
2077 
2078 	/* stop the thread when the user removes the region dir */
2079 	spin_lock(&o2hb_live_lock);
2080 	hb_task = reg->hr_task;
2081 	reg->hr_task = NULL;
2082 	reg->hr_item_dropped = 1;
2083 	spin_unlock(&o2hb_live_lock);
2084 
2085 	if (hb_task)
2086 		kthread_stop(hb_task);
2087 
2088 	if (o2hb_global_heartbeat_active()) {
2089 		spin_lock(&o2hb_live_lock);
2090 		clear_bit(reg->hr_region_num, o2hb_region_bitmap);
2091 		clear_bit(reg->hr_region_num, o2hb_live_region_bitmap);
2092 		if (test_bit(reg->hr_region_num, o2hb_quorum_region_bitmap))
2093 			quorum_region = 1;
2094 		clear_bit(reg->hr_region_num, o2hb_quorum_region_bitmap);
2095 		spin_unlock(&o2hb_live_lock);
2096 		printk(KERN_NOTICE "o2hb: Heartbeat %s on region %s (%s)\n",
2097 		       ((atomic_read(&reg->hr_steady_iterations) == 0) ?
2098 			"stopped" : "start aborted"), config_item_name(item),
2099 		       reg->hr_dev_name);
2100 	}
2101 
2102 	/*
2103 	 * If we're racing a dev_write(), we need to wake them.  They will
2104 	 * check reg->hr_task
2105 	 */
2106 	if (atomic_read(&reg->hr_steady_iterations) != 0) {
2107 		reg->hr_aborted_start = 1;
2108 		atomic_set(&reg->hr_steady_iterations, 0);
2109 		wake_up(&o2hb_steady_queue);
2110 	}
2111 
2112 	config_item_put(item);
2113 
2114 	if (!o2hb_global_heartbeat_active() || !quorum_region)
2115 		return;
2116 
2117 	/*
2118 	 * If global heartbeat active and there are dependent users,
2119 	 * pin all regions if quorum region count <= CUT_OFF
2120 	 */
2121 	spin_lock(&o2hb_live_lock);
2122 
2123 	if (!o2hb_dependent_users)
2124 		goto unlock;
2125 
2126 	if (bitmap_weight(o2hb_quorum_region_bitmap,
2127 			   O2NM_MAX_REGIONS) <= O2HB_PIN_CUT_OFF)
2128 		o2hb_region_pin(NULL);
2129 
2130 unlock:
2131 	spin_unlock(&o2hb_live_lock);
2132 }
2133 
2134 static ssize_t o2hb_heartbeat_group_dead_threshold_show(struct config_item *item,
2135 		char *page)
2136 {
2137 	return sprintf(page, "%u\n", o2hb_dead_threshold);
2138 }
2139 
2140 static ssize_t o2hb_heartbeat_group_dead_threshold_store(struct config_item *item,
2141 		const char *page, size_t count)
2142 {
2143 	unsigned long tmp;
2144 	char *p = (char *)page;
2145 
2146 	tmp = simple_strtoul(p, &p, 10);
2147 	if (!p || (*p && (*p != '\n')))
2148                 return -EINVAL;
2149 
2150 	/* this will validate ranges for us. */
2151 	o2hb_dead_threshold_set((unsigned int) tmp);
2152 
2153 	return count;
2154 }
2155 
2156 static ssize_t o2hb_heartbeat_group_mode_show(struct config_item *item,
2157 		char *page)
2158 {
2159 	return sprintf(page, "%s\n",
2160 		       o2hb_heartbeat_mode_desc[o2hb_heartbeat_mode]);
2161 }
2162 
2163 static ssize_t o2hb_heartbeat_group_mode_store(struct config_item *item,
2164 		const char *page, size_t count)
2165 {
2166 	unsigned int i;
2167 	int ret;
2168 	size_t len;
2169 
2170 	len = (page[count - 1] == '\n') ? count - 1 : count;
2171 	if (!len)
2172 		return -EINVAL;
2173 
2174 	for (i = 0; i < O2HB_HEARTBEAT_NUM_MODES; ++i) {
2175 		if (strncasecmp(page, o2hb_heartbeat_mode_desc[i], len))
2176 			continue;
2177 
2178 		ret = o2hb_global_heartbeat_mode_set(i);
2179 		if (!ret)
2180 			printk(KERN_NOTICE "o2hb: Heartbeat mode set to %s\n",
2181 			       o2hb_heartbeat_mode_desc[i]);
2182 		return count;
2183 	}
2184 
2185 	return -EINVAL;
2186 
2187 }
2188 
2189 CONFIGFS_ATTR(o2hb_heartbeat_group_, dead_threshold);
2190 CONFIGFS_ATTR(o2hb_heartbeat_group_, mode);
2191 
2192 static struct configfs_attribute *o2hb_heartbeat_group_attrs[] = {
2193 	&o2hb_heartbeat_group_attr_dead_threshold,
2194 	&o2hb_heartbeat_group_attr_mode,
2195 	NULL,
2196 };
2197 
2198 static struct configfs_group_operations o2hb_heartbeat_group_group_ops = {
2199 	.make_item	= o2hb_heartbeat_group_make_item,
2200 	.drop_item	= o2hb_heartbeat_group_drop_item,
2201 };
2202 
2203 static const struct config_item_type o2hb_heartbeat_group_type = {
2204 	.ct_group_ops	= &o2hb_heartbeat_group_group_ops,
2205 	.ct_attrs	= o2hb_heartbeat_group_attrs,
2206 	.ct_owner	= THIS_MODULE,
2207 };
2208 
2209 /* this is just here to avoid touching group in heartbeat.h which the
2210  * entire damn world #includes */
2211 struct config_group *o2hb_alloc_hb_set(void)
2212 {
2213 	struct o2hb_heartbeat_group *hs = NULL;
2214 	struct config_group *ret = NULL;
2215 
2216 	hs = kzalloc(sizeof(struct o2hb_heartbeat_group), GFP_KERNEL);
2217 	if (hs == NULL)
2218 		goto out;
2219 
2220 	config_group_init_type_name(&hs->hs_group, "heartbeat",
2221 				    &o2hb_heartbeat_group_type);
2222 
2223 	ret = &hs->hs_group;
2224 out:
2225 	if (ret == NULL)
2226 		kfree(hs);
2227 	return ret;
2228 }
2229 
2230 void o2hb_free_hb_set(struct config_group *group)
2231 {
2232 	struct o2hb_heartbeat_group *hs = to_o2hb_heartbeat_group(group);
2233 	kfree(hs);
2234 }
2235 
2236 /* hb callback registration and issuing */
2237 
2238 static struct o2hb_callback *hbcall_from_type(enum o2hb_callback_type type)
2239 {
2240 	if (type == O2HB_NUM_CB)
2241 		return ERR_PTR(-EINVAL);
2242 
2243 	return &o2hb_callbacks[type];
2244 }
2245 
2246 void o2hb_setup_callback(struct o2hb_callback_func *hc,
2247 			 enum o2hb_callback_type type,
2248 			 o2hb_cb_func *func,
2249 			 void *data,
2250 			 int priority)
2251 {
2252 	INIT_LIST_HEAD(&hc->hc_item);
2253 	hc->hc_func = func;
2254 	hc->hc_data = data;
2255 	hc->hc_priority = priority;
2256 	hc->hc_type = type;
2257 	hc->hc_magic = O2HB_CB_MAGIC;
2258 }
2259 EXPORT_SYMBOL_GPL(o2hb_setup_callback);
2260 
2261 /*
2262  * In local heartbeat mode, region_uuid passed matches the dlm domain name.
2263  * In global heartbeat mode, region_uuid passed is NULL.
2264  *
2265  * In local, we only pin the matching region. In global we pin all the active
2266  * regions.
2267  */
2268 static int o2hb_region_pin(const char *region_uuid)
2269 {
2270 	int ret = 0, found = 0;
2271 	struct o2hb_region *reg;
2272 	char *uuid;
2273 
2274 	assert_spin_locked(&o2hb_live_lock);
2275 
2276 	list_for_each_entry(reg, &o2hb_all_regions, hr_all_item) {
2277 		if (reg->hr_item_dropped)
2278 			continue;
2279 
2280 		uuid = config_item_name(&reg->hr_item);
2281 
2282 		/* local heartbeat */
2283 		if (region_uuid) {
2284 			if (strcmp(region_uuid, uuid))
2285 				continue;
2286 			found = 1;
2287 		}
2288 
2289 		if (reg->hr_item_pinned || reg->hr_item_dropped)
2290 			goto skip_pin;
2291 
2292 		/* Ignore ENOENT only for local hb (userdlm domain) */
2293 		ret = o2nm_depend_item(&reg->hr_item);
2294 		if (!ret) {
2295 			mlog(ML_CLUSTER, "Pin region %s\n", uuid);
2296 			reg->hr_item_pinned = 1;
2297 		} else {
2298 			if (ret == -ENOENT && found)
2299 				ret = 0;
2300 			else {
2301 				mlog(ML_ERROR, "Pin region %s fails with %d\n",
2302 				     uuid, ret);
2303 				break;
2304 			}
2305 		}
2306 skip_pin:
2307 		if (found)
2308 			break;
2309 	}
2310 
2311 	return ret;
2312 }
2313 
2314 /*
2315  * In local heartbeat mode, region_uuid passed matches the dlm domain name.
2316  * In global heartbeat mode, region_uuid passed is NULL.
2317  *
2318  * In local, we only unpin the matching region. In global we unpin all the
2319  * active regions.
2320  */
2321 static void o2hb_region_unpin(const char *region_uuid)
2322 {
2323 	struct o2hb_region *reg;
2324 	char *uuid;
2325 	int found = 0;
2326 
2327 	assert_spin_locked(&o2hb_live_lock);
2328 
2329 	list_for_each_entry(reg, &o2hb_all_regions, hr_all_item) {
2330 		if (reg->hr_item_dropped)
2331 			continue;
2332 
2333 		uuid = config_item_name(&reg->hr_item);
2334 		if (region_uuid) {
2335 			if (strcmp(region_uuid, uuid))
2336 				continue;
2337 			found = 1;
2338 		}
2339 
2340 		if (reg->hr_item_pinned) {
2341 			mlog(ML_CLUSTER, "Unpin region %s\n", uuid);
2342 			o2nm_undepend_item(&reg->hr_item);
2343 			reg->hr_item_pinned = 0;
2344 		}
2345 		if (found)
2346 			break;
2347 	}
2348 }
2349 
2350 static int o2hb_region_inc_user(const char *region_uuid)
2351 {
2352 	int ret = 0;
2353 
2354 	spin_lock(&o2hb_live_lock);
2355 
2356 	/* local heartbeat */
2357 	if (!o2hb_global_heartbeat_active()) {
2358 	    ret = o2hb_region_pin(region_uuid);
2359 	    goto unlock;
2360 	}
2361 
2362 	/*
2363 	 * if global heartbeat active and this is the first dependent user,
2364 	 * pin all regions if quorum region count <= CUT_OFF
2365 	 */
2366 	o2hb_dependent_users++;
2367 	if (o2hb_dependent_users > 1)
2368 		goto unlock;
2369 
2370 	if (bitmap_weight(o2hb_quorum_region_bitmap,
2371 			   O2NM_MAX_REGIONS) <= O2HB_PIN_CUT_OFF)
2372 		ret = o2hb_region_pin(NULL);
2373 
2374 unlock:
2375 	spin_unlock(&o2hb_live_lock);
2376 	return ret;
2377 }
2378 
2379 static void o2hb_region_dec_user(const char *region_uuid)
2380 {
2381 	spin_lock(&o2hb_live_lock);
2382 
2383 	/* local heartbeat */
2384 	if (!o2hb_global_heartbeat_active()) {
2385 	    o2hb_region_unpin(region_uuid);
2386 	    goto unlock;
2387 	}
2388 
2389 	/*
2390 	 * if global heartbeat active and there are no dependent users,
2391 	 * unpin all quorum regions
2392 	 */
2393 	o2hb_dependent_users--;
2394 	if (!o2hb_dependent_users)
2395 		o2hb_region_unpin(NULL);
2396 
2397 unlock:
2398 	spin_unlock(&o2hb_live_lock);
2399 }
2400 
2401 int o2hb_register_callback(const char *region_uuid,
2402 			   struct o2hb_callback_func *hc)
2403 {
2404 	struct o2hb_callback_func *f;
2405 	struct o2hb_callback *hbcall;
2406 	int ret;
2407 
2408 	BUG_ON(hc->hc_magic != O2HB_CB_MAGIC);
2409 	BUG_ON(!list_empty(&hc->hc_item));
2410 
2411 	hbcall = hbcall_from_type(hc->hc_type);
2412 	if (IS_ERR(hbcall)) {
2413 		ret = PTR_ERR(hbcall);
2414 		goto out;
2415 	}
2416 
2417 	if (region_uuid) {
2418 		ret = o2hb_region_inc_user(region_uuid);
2419 		if (ret) {
2420 			mlog_errno(ret);
2421 			goto out;
2422 		}
2423 	}
2424 
2425 	down_write(&o2hb_callback_sem);
2426 
2427 	list_for_each_entry(f, &hbcall->list, hc_item) {
2428 		if (hc->hc_priority < f->hc_priority) {
2429 			list_add_tail(&hc->hc_item, &f->hc_item);
2430 			break;
2431 		}
2432 	}
2433 	if (list_empty(&hc->hc_item))
2434 		list_add_tail(&hc->hc_item, &hbcall->list);
2435 
2436 	up_write(&o2hb_callback_sem);
2437 	ret = 0;
2438 out:
2439 	mlog(ML_CLUSTER, "returning %d on behalf of %p for funcs %p\n",
2440 	     ret, __builtin_return_address(0), hc);
2441 	return ret;
2442 }
2443 EXPORT_SYMBOL_GPL(o2hb_register_callback);
2444 
2445 void o2hb_unregister_callback(const char *region_uuid,
2446 			      struct o2hb_callback_func *hc)
2447 {
2448 	BUG_ON(hc->hc_magic != O2HB_CB_MAGIC);
2449 
2450 	mlog(ML_CLUSTER, "on behalf of %p for funcs %p\n",
2451 	     __builtin_return_address(0), hc);
2452 
2453 	/* XXX Can this happen _with_ a region reference? */
2454 	if (list_empty(&hc->hc_item))
2455 		return;
2456 
2457 	if (region_uuid)
2458 		o2hb_region_dec_user(region_uuid);
2459 
2460 	down_write(&o2hb_callback_sem);
2461 
2462 	list_del_init(&hc->hc_item);
2463 
2464 	up_write(&o2hb_callback_sem);
2465 }
2466 EXPORT_SYMBOL_GPL(o2hb_unregister_callback);
2467 
2468 int o2hb_check_node_heartbeating_no_sem(u8 node_num)
2469 {
2470 	unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
2471 
2472 	spin_lock(&o2hb_live_lock);
2473 	o2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
2474 	spin_unlock(&o2hb_live_lock);
2475 	if (!test_bit(node_num, testing_map)) {
2476 		mlog(ML_HEARTBEAT,
2477 		     "node (%u) does not have heartbeating enabled.\n",
2478 		     node_num);
2479 		return 0;
2480 	}
2481 
2482 	return 1;
2483 }
2484 EXPORT_SYMBOL_GPL(o2hb_check_node_heartbeating_no_sem);
2485 
2486 int o2hb_check_node_heartbeating_from_callback(u8 node_num)
2487 {
2488 	unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
2489 
2490 	o2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
2491 	if (!test_bit(node_num, testing_map)) {
2492 		mlog(ML_HEARTBEAT,
2493 		     "node (%u) does not have heartbeating enabled.\n",
2494 		     node_num);
2495 		return 0;
2496 	}
2497 
2498 	return 1;
2499 }
2500 EXPORT_SYMBOL_GPL(o2hb_check_node_heartbeating_from_callback);
2501 
2502 /*
2503  * this is just a hack until we get the plumbing which flips file systems
2504  * read only and drops the hb ref instead of killing the node dead.
2505  */
2506 void o2hb_stop_all_regions(void)
2507 {
2508 	struct o2hb_region *reg;
2509 
2510 	mlog(ML_ERROR, "stopping heartbeat on all active regions.\n");
2511 
2512 	spin_lock(&o2hb_live_lock);
2513 
2514 	list_for_each_entry(reg, &o2hb_all_regions, hr_all_item)
2515 		reg->hr_unclean_stop = 1;
2516 
2517 	spin_unlock(&o2hb_live_lock);
2518 }
2519 EXPORT_SYMBOL_GPL(o2hb_stop_all_regions);
2520 
2521 int o2hb_get_all_regions(char *region_uuids, u8 max_regions)
2522 {
2523 	struct o2hb_region *reg;
2524 	int numregs = 0;
2525 	char *p;
2526 
2527 	spin_lock(&o2hb_live_lock);
2528 
2529 	p = region_uuids;
2530 	list_for_each_entry(reg, &o2hb_all_regions, hr_all_item) {
2531 		if (reg->hr_item_dropped)
2532 			continue;
2533 
2534 		mlog(0, "Region: %s\n", config_item_name(&reg->hr_item));
2535 		if (numregs < max_regions) {
2536 			memcpy(p, config_item_name(&reg->hr_item),
2537 			       O2HB_MAX_REGION_NAME_LEN);
2538 			p += O2HB_MAX_REGION_NAME_LEN;
2539 		}
2540 		numregs++;
2541 	}
2542 
2543 	spin_unlock(&o2hb_live_lock);
2544 
2545 	return numregs;
2546 }
2547 EXPORT_SYMBOL_GPL(o2hb_get_all_regions);
2548 
2549 int o2hb_global_heartbeat_active(void)
2550 {
2551 	return (o2hb_heartbeat_mode == O2HB_HEARTBEAT_GLOBAL);
2552 }
2553 EXPORT_SYMBOL(o2hb_global_heartbeat_active);
2554