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