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