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
38 #include "heartbeat.h"
39 #include "tcp.h"
40 #include "nodemanager.h"
41 #include "quorum.h"
42
43 #include "masklog.h"
44
45
46 /*
47 * The first heartbeat pass had one global thread that would serialize all hb
48 * callback calls. This global serializing sem should only be removed once
49 * we've made sure that all callees can deal with being called concurrently
50 * from multiple hb region threads.
51 */
52 static DECLARE_RWSEM(o2hb_callback_sem);
53
54 /*
55 * multiple hb threads are watching multiple regions. A node is live
56 * whenever any of the threads sees activity from the node in its region.
57 */
58 static DEFINE_SPINLOCK(o2hb_live_lock);
59 static struct list_head o2hb_live_slots[O2NM_MAX_NODES];
60 static unsigned long o2hb_live_node_bitmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
61 static LIST_HEAD(o2hb_node_events);
62 static DECLARE_WAIT_QUEUE_HEAD(o2hb_steady_queue);
63
64 #define O2HB_DEBUG_DIR "o2hb"
65 #define O2HB_DEBUG_LIVENODES "livenodes"
66 static struct dentry *o2hb_debug_dir;
67 static struct dentry *o2hb_debug_livenodes;
68
69 static LIST_HEAD(o2hb_all_regions);
70
71 static struct o2hb_callback {
72 struct list_head list;
73 } o2hb_callbacks[O2HB_NUM_CB];
74
75 static struct o2hb_callback *hbcall_from_type(enum o2hb_callback_type type);
76
77 #define O2HB_DEFAULT_BLOCK_BITS 9
78
79 unsigned int o2hb_dead_threshold = O2HB_DEFAULT_DEAD_THRESHOLD;
80
81 /* Only sets a new threshold if there are no active regions.
82 *
83 * No locking or otherwise interesting code is required for reading
84 * o2hb_dead_threshold as it can't change once regions are active and
85 * it's not interesting to anyone until then anyway. */
86 static void o2hb_dead_threshold_set(unsigned int threshold)
87 {
88 if (threshold > O2HB_MIN_DEAD_THRESHOLD) {
89 spin_lock(&o2hb_live_lock);
90 if (list_empty(&o2hb_all_regions))
91 o2hb_dead_threshold = threshold;
92 spin_unlock(&o2hb_live_lock);
93 }
94 }
95
96 struct o2hb_node_event {
97 struct list_head hn_item;
98 enum o2hb_callback_type hn_event_type;
99 struct o2nm_node *hn_node;
100 int hn_node_num;
101 };
102
103 struct o2hb_disk_slot {
104 struct o2hb_disk_heartbeat_block *ds_raw_block;
105 u8 ds_node_num;
106 u64 ds_last_time;
107 u64 ds_last_generation;
108 u16 ds_equal_samples;
109 u16 ds_changed_samples;
110 struct list_head ds_live_item;
111 };
112
113 /* each thread owns a region.. when we're asked to tear down the region
114 * we ask the thread to stop, who cleans up the region */
115 struct o2hb_region {
116 struct config_item hr_item;
117
118 struct list_head hr_all_item;
119 unsigned hr_unclean_stop:1;
120
121 /* protected by the hr_callback_sem */
122 struct task_struct *hr_task;
123
124 unsigned int hr_blocks;
125 unsigned long long hr_start_block;
126
127 unsigned int hr_block_bits;
128 unsigned int hr_block_bytes;
129
130 unsigned int hr_slots_per_page;
131 unsigned int hr_num_pages;
132
133 struct page **hr_slot_data;
134 struct block_device *hr_bdev;
135 struct o2hb_disk_slot *hr_slots;
136
137 /* let the person setting up hb wait for it to return until it
138 * has reached a 'steady' state. This will be fixed when we have
139 * a more complete api that doesn't lead to this sort of fragility. */
140 atomic_t hr_steady_iterations;
141
142 char hr_dev_name[BDEVNAME_SIZE];
143
144 unsigned int hr_timeout_ms;
145
146 /* randomized as the region goes up and down so that a node
147 * recognizes a node going up and down in one iteration */
148 u64 hr_generation;
149
150 struct delayed_work hr_write_timeout_work;
151 unsigned long hr_last_timeout_start;
152
153 /* Used during o2hb_check_slot to hold a copy of the block
154 * being checked because we temporarily have to zero out the
155 * crc field. */
156 struct o2hb_disk_heartbeat_block *hr_tmp_block;
157 };
158
159 struct o2hb_bio_wait_ctxt {
160 atomic_t wc_num_reqs;
161 struct completion wc_io_complete;
162 int wc_error;
163 };
164
165 static void o2hb_write_timeout(struct work_struct *work)
166 {
167 struct o2hb_region *reg =
168 container_of(work, struct o2hb_region,
169 hr_write_timeout_work.work);
170
171 mlog(ML_ERROR, "Heartbeat write timeout to device %s after %u "
172 "milliseconds\n", reg->hr_dev_name,
173 jiffies_to_msecs(jiffies - reg->hr_last_timeout_start));
174 o2quo_disk_timeout();
175 }
176
177 static void o2hb_arm_write_timeout(struct o2hb_region *reg)
178 {
179 mlog(0, "Queue write timeout for %u ms\n", O2HB_MAX_WRITE_TIMEOUT_MS);
180
181 cancel_delayed_work(®->hr_write_timeout_work);
182 reg->hr_last_timeout_start = jiffies;
183 schedule_delayed_work(®->hr_write_timeout_work,
184 msecs_to_jiffies(O2HB_MAX_WRITE_TIMEOUT_MS));
185 }
186
187 static void o2hb_disarm_write_timeout(struct o2hb_region *reg)
188 {
189 cancel_delayed_work(®->hr_write_timeout_work);
190 flush_scheduled_work();
191 }
192
193 static inline void o2hb_bio_wait_init(struct o2hb_bio_wait_ctxt *wc)
194 {
195 atomic_set(&wc->wc_num_reqs, 1);
196 init_completion(&wc->wc_io_complete);
197 wc->wc_error = 0;
198 }
199
200 /* Used in error paths too */
201 static inline void o2hb_bio_wait_dec(struct o2hb_bio_wait_ctxt *wc,
202 unsigned int num)
203 {
204 /* sadly atomic_sub_and_test() isn't available on all platforms. The
205 * good news is that the fast path only completes one at a time */
206 while(num--) {
207 if (atomic_dec_and_test(&wc->wc_num_reqs)) {
208 BUG_ON(num > 0);
209 complete(&wc->wc_io_complete);
210 }
211 }
212 }
213
214 static void o2hb_wait_on_io(struct o2hb_region *reg,
215 struct o2hb_bio_wait_ctxt *wc)
216 {
217 struct address_space *mapping = reg->hr_bdev->bd_inode->i_mapping;
218
219 blk_run_address_space(mapping);
220 o2hb_bio_wait_dec(wc, 1);
221
222 wait_for_completion(&wc->wc_io_complete);
223 }
224
225 static void o2hb_bio_end_io(struct bio *bio,
226 int error)
227 {
228 struct o2hb_bio_wait_ctxt *wc = bio->bi_private;
229
230 if (error) {
231 mlog(ML_ERROR, "IO Error %d\n", error);
232 wc->wc_error = error;
233 }
234
235 o2hb_bio_wait_dec(wc, 1);
236 bio_put(bio);
237 }
238
239 /* Setup a Bio to cover I/O against num_slots slots starting at
240 * start_slot. */
241 static struct bio *o2hb_setup_one_bio(struct o2hb_region *reg,
242 struct o2hb_bio_wait_ctxt *wc,
243 unsigned int *current_slot,
244 unsigned int max_slots)
245 {
246 int len, current_page;
247 unsigned int vec_len, vec_start;
248 unsigned int bits = reg->hr_block_bits;
249 unsigned int spp = reg->hr_slots_per_page;
250 unsigned int cs = *current_slot;
251 struct bio *bio;
252 struct page *page;
253
254 /* Testing has shown this allocation to take long enough under
255 * GFP_KERNEL that the local node can get fenced. It would be
256 * nicest if we could pre-allocate these bios and avoid this
257 * all together. */
258 bio = bio_alloc(GFP_ATOMIC, 16);
259 if (!bio) {
260 mlog(ML_ERROR, "Could not alloc slots BIO!\n");
261 bio = ERR_PTR(-ENOMEM);
262 goto bail;
263 }
264
265 /* Must put everything in 512 byte sectors for the bio... */
266 bio->bi_sector = (reg->hr_start_block + cs) << (bits - 9);
267 bio->bi_bdev = reg->hr_bdev;
268 bio->bi_private = wc;
269 bio->bi_end_io = o2hb_bio_end_io;
270
271 vec_start = (cs << bits) % PAGE_CACHE_SIZE;
272 while(cs < max_slots) {
273 current_page = cs / spp;
274 page = reg->hr_slot_data[current_page];
275
276 vec_len = min(PAGE_CACHE_SIZE - vec_start,
277 (max_slots-cs) * (PAGE_CACHE_SIZE/spp) );
278
279 mlog(ML_HB_BIO, "page %d, vec_len = %u, vec_start = %u\n",
280 current_page, vec_len, vec_start);
281
282 len = bio_add_page(bio, page, vec_len, vec_start);
283 if (len != vec_len) break;
284
285 cs += vec_len / (PAGE_CACHE_SIZE/spp);
286 vec_start = 0;
287 }
288
289 bail:
290 *current_slot = cs;
291 return bio;
292 }
293
294 static int o2hb_read_slots(struct o2hb_region *reg,
295 unsigned int max_slots)
296 {
297 unsigned int current_slot=0;
298 int status;
299 struct o2hb_bio_wait_ctxt wc;
300 struct bio *bio;
301
302 o2hb_bio_wait_init(&wc);
303
304 while(current_slot < max_slots) {
305 bio = o2hb_setup_one_bio(reg, &wc, ¤t_slot, max_slots);
306 if (IS_ERR(bio)) {
307 status = PTR_ERR(bio);
308 mlog_errno(status);
309 goto bail_and_wait;
310 }
311
312 atomic_inc(&wc.wc_num_reqs);
313 submit_bio(READ, bio);
314 }
315
316 status = 0;
317
318 bail_and_wait:
319 o2hb_wait_on_io(reg, &wc);
320 if (wc.wc_error && !status)
321 status = wc.wc_error;
322
323 return status;
324 }
325
326 static int o2hb_issue_node_write(struct o2hb_region *reg,
327 struct o2hb_bio_wait_ctxt *write_wc)
328 {
329 int status;
330 unsigned int slot;
331 struct bio *bio;
332
333 o2hb_bio_wait_init(write_wc);
334
335 slot = o2nm_this_node();
336
337 bio = o2hb_setup_one_bio(reg, write_wc, &slot, slot+1);
338 if (IS_ERR(bio)) {
339 status = PTR_ERR(bio);
340 mlog_errno(status);
341 goto bail;
342 }
343
344 atomic_inc(&write_wc->wc_num_reqs);
345 submit_bio(WRITE, bio);
346
347 status = 0;
348 bail:
349 return status;
350 }
351
352 static u32 o2hb_compute_block_crc_le(struct o2hb_region *reg,
353 struct o2hb_disk_heartbeat_block *hb_block)
354 {
355 __le32 old_cksum;
356 u32 ret;
357
358 /* We want to compute the block crc with a 0 value in the
359 * hb_cksum field. Save it off here and replace after the
360 * crc. */
361 old_cksum = hb_block->hb_cksum;
362 hb_block->hb_cksum = 0;
363
364 ret = crc32_le(0, (unsigned char *) hb_block, reg->hr_block_bytes);
365
366 hb_block->hb_cksum = old_cksum;
367
368 return ret;
369 }
370
371 static void o2hb_dump_slot(struct o2hb_disk_heartbeat_block *hb_block)
372 {
373 mlog(ML_ERROR, "Dump slot information: seq = 0x%llx, node = %u, "
374 "cksum = 0x%x, generation 0x%llx\n",
375 (long long)le64_to_cpu(hb_block->hb_seq),
376 hb_block->hb_node, le32_to_cpu(hb_block->hb_cksum),
377 (long long)le64_to_cpu(hb_block->hb_generation));
378 }
379
380 static int o2hb_verify_crc(struct o2hb_region *reg,
381 struct o2hb_disk_heartbeat_block *hb_block)
382 {
383 u32 read, computed;
384
385 read = le32_to_cpu(hb_block->hb_cksum);
386 computed = o2hb_compute_block_crc_le(reg, hb_block);
387
388 return read == computed;
389 }
390
391 /* We want to make sure that nobody is heartbeating on top of us --
392 * this will help detect an invalid configuration. */
393 static int o2hb_check_last_timestamp(struct o2hb_region *reg)
394 {
395 int node_num, ret;
396 struct o2hb_disk_slot *slot;
397 struct o2hb_disk_heartbeat_block *hb_block;
398
399 node_num = o2nm_this_node();
400
401 ret = 1;
402 slot = ®->hr_slots[node_num];
403 /* Don't check on our 1st timestamp */
404 if (slot->ds_last_time) {
405 hb_block = slot->ds_raw_block;
406
407 if (le64_to_cpu(hb_block->hb_seq) != slot->ds_last_time)
408 ret = 0;
409 }
410
411 return ret;
412 }
413
414 static inline void o2hb_prepare_block(struct o2hb_region *reg,
415 u64 generation)
416 {
417 int node_num;
418 u64 cputime;
419 struct o2hb_disk_slot *slot;
420 struct o2hb_disk_heartbeat_block *hb_block;
421
422 node_num = o2nm_this_node();
423 slot = ®->hr_slots[node_num];
424
425 hb_block = (struct o2hb_disk_heartbeat_block *)slot->ds_raw_block;
426 memset(hb_block, 0, reg->hr_block_bytes);
427 /* TODO: time stuff */
428 cputime = CURRENT_TIME.tv_sec;
429 if (!cputime)
430 cputime = 1;
431
432 hb_block->hb_seq = cpu_to_le64(cputime);
433 hb_block->hb_node = node_num;
434 hb_block->hb_generation = cpu_to_le64(generation);
435 hb_block->hb_dead_ms = cpu_to_le32(o2hb_dead_threshold * O2HB_REGION_TIMEOUT_MS);
436
437 /* This step must always happen last! */
438 hb_block->hb_cksum = cpu_to_le32(o2hb_compute_block_crc_le(reg,
439 hb_block));
440
441 mlog(ML_HB_BIO, "our node generation = 0x%llx, cksum = 0x%x\n",
442 (long long)generation,
443 le32_to_cpu(hb_block->hb_cksum));
444 }
445
446 static void o2hb_fire_callbacks(struct o2hb_callback *hbcall,
447 struct o2nm_node *node,
448 int idx)
449 {
450 struct list_head *iter;
451 struct o2hb_callback_func *f;
452
453 list_for_each(iter, &hbcall->list) {
454 f = list_entry(iter, struct o2hb_callback_func, hc_item);
455 mlog(ML_HEARTBEAT, "calling funcs %p\n", f);
456 (f->hc_func)(node, idx, f->hc_data);
457 }
458 }
459
460 /* Will run the list in order until we process the passed event */
461 static void o2hb_run_event_list(struct o2hb_node_event *queued_event)
462 {
463 int empty;
464 struct o2hb_callback *hbcall;
465 struct o2hb_node_event *event;
466
467 spin_lock(&o2hb_live_lock);
468 empty = list_empty(&queued_event->hn_item);
469 spin_unlock(&o2hb_live_lock);
470 if (empty)
471 return;
472
473 /* Holding callback sem assures we don't alter the callback
474 * lists when doing this, and serializes ourselves with other
475 * processes wanting callbacks. */
476 down_write(&o2hb_callback_sem);
477
478 spin_lock(&o2hb_live_lock);
479 while (!list_empty(&o2hb_node_events)
480 && !list_empty(&queued_event->hn_item)) {
481 event = list_entry(o2hb_node_events.next,
482 struct o2hb_node_event,
483 hn_item);
484 list_del_init(&event->hn_item);
485 spin_unlock(&o2hb_live_lock);
486
487 mlog(ML_HEARTBEAT, "Node %s event for %d\n",
488 event->hn_event_type == O2HB_NODE_UP_CB ? "UP" : "DOWN",
489 event->hn_node_num);
490
491 hbcall = hbcall_from_type(event->hn_event_type);
492
493 /* We should *never* have gotten on to the list with a
494 * bad type... This isn't something that we should try
495 * to recover from. */
496 BUG_ON(IS_ERR(hbcall));
497
498 o2hb_fire_callbacks(hbcall, event->hn_node, event->hn_node_num);
499
500 spin_lock(&o2hb_live_lock);
501 }
502 spin_unlock(&o2hb_live_lock);
503
504 up_write(&o2hb_callback_sem);
505 }
506
507 static void o2hb_queue_node_event(struct o2hb_node_event *event,
508 enum o2hb_callback_type type,
509 struct o2nm_node *node,
510 int node_num)
511 {
512 assert_spin_locked(&o2hb_live_lock);
513
514 event->hn_event_type = type;
515 event->hn_node = node;
516 event->hn_node_num = node_num;
517
518 mlog(ML_HEARTBEAT, "Queue node %s event for node %d\n",
519 type == O2HB_NODE_UP_CB ? "UP" : "DOWN", node_num);
520
521 list_add_tail(&event->hn_item, &o2hb_node_events);
522 }
523
524 static void o2hb_shutdown_slot(struct o2hb_disk_slot *slot)
525 {
526 struct o2hb_node_event event =
527 { .hn_item = LIST_HEAD_INIT(event.hn_item), };
528 struct o2nm_node *node;
529
530 node = o2nm_get_node_by_num(slot->ds_node_num);
531 if (!node)
532 return;
533
534 spin_lock(&o2hb_live_lock);
535 if (!list_empty(&slot->ds_live_item)) {
536 mlog(ML_HEARTBEAT, "Shutdown, node %d leaves region\n",
537 slot->ds_node_num);
538
539 list_del_init(&slot->ds_live_item);
540
541 if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
542 clear_bit(slot->ds_node_num, o2hb_live_node_bitmap);
543
544 o2hb_queue_node_event(&event, O2HB_NODE_DOWN_CB, node,
545 slot->ds_node_num);
546 }
547 }
548 spin_unlock(&o2hb_live_lock);
549
550 o2hb_run_event_list(&event);
551
552 o2nm_node_put(node);
553 }
554
555 static int o2hb_check_slot(struct o2hb_region *reg,
556 struct o2hb_disk_slot *slot)
557 {
558 int changed = 0, gen_changed = 0;
559 struct o2hb_node_event event =
560 { .hn_item = LIST_HEAD_INIT(event.hn_item), };
561 struct o2nm_node *node;
562 struct o2hb_disk_heartbeat_block *hb_block = reg->hr_tmp_block;
563 u64 cputime;
564 unsigned int dead_ms = o2hb_dead_threshold * O2HB_REGION_TIMEOUT_MS;
565 unsigned int slot_dead_ms;
566
567 memcpy(hb_block, slot->ds_raw_block, reg->hr_block_bytes);
568
569 /* Is this correct? Do we assume that the node doesn't exist
570 * if we're not configured for him? */
571 node = o2nm_get_node_by_num(slot->ds_node_num);
572 if (!node)
573 return 0;
574
575 if (!o2hb_verify_crc(reg, hb_block)) {
576 /* all paths from here will drop o2hb_live_lock for
577 * us. */
578 spin_lock(&o2hb_live_lock);
579
580 /* Don't print an error on the console in this case -
581 * a freshly formatted heartbeat area will not have a
582 * crc set on it. */
583 if (list_empty(&slot->ds_live_item))
584 goto out;
585
586 /* The node is live but pushed out a bad crc. We
587 * consider it a transient miss but don't populate any
588 * other values as they may be junk. */
589 mlog(ML_ERROR, "Node %d has written a bad crc to %s\n",
590 slot->ds_node_num, reg->hr_dev_name);
591 o2hb_dump_slot(hb_block);
592
593 slot->ds_equal_samples++;
594 goto fire_callbacks;
595 }
596
597 /* we don't care if these wrap.. the state transitions below
598 * clear at the right places */
599 cputime = le64_to_cpu(hb_block->hb_seq);
600 if (slot->ds_last_time != cputime)
601 slot->ds_changed_samples++;
602 else
603 slot->ds_equal_samples++;
604 slot->ds_last_time = cputime;
605
606 /* The node changed heartbeat generations. We assume this to
607 * mean it dropped off but came back before we timed out. We
608 * want to consider it down for the time being but don't want
609 * to lose any changed_samples state we might build up to
610 * considering it live again. */
611 if (slot->ds_last_generation != le64_to_cpu(hb_block->hb_generation)) {
612 gen_changed = 1;
613 slot->ds_equal_samples = 0;
614 mlog(ML_HEARTBEAT, "Node %d changed generation (0x%llx "
615 "to 0x%llx)\n", slot->ds_node_num,
616 (long long)slot->ds_last_generation,
617 (long long)le64_to_cpu(hb_block->hb_generation));
618 }
619
620 slot->ds_last_generation = le64_to_cpu(hb_block->hb_generation);
621
622 mlog(ML_HEARTBEAT, "Slot %d gen 0x%llx cksum 0x%x "
623 "seq %llu last %llu changed %u equal %u\n",
624 slot->ds_node_num, (long long)slot->ds_last_generation,
625 le32_to_cpu(hb_block->hb_cksum),
626 (unsigned long long)le64_to_cpu(hb_block->hb_seq),
627 (unsigned long long)slot->ds_last_time, slot->ds_changed_samples,
628 slot->ds_equal_samples);
629
630 spin_lock(&o2hb_live_lock);
631
632 fire_callbacks:
633 /* dead nodes only come to life after some number of
634 * changes at any time during their dead time */
635 if (list_empty(&slot->ds_live_item) &&
636 slot->ds_changed_samples >= O2HB_LIVE_THRESHOLD) {
637 mlog(ML_HEARTBEAT, "Node %d (id 0x%llx) joined my region\n",
638 slot->ds_node_num, (long long)slot->ds_last_generation);
639
640 /* first on the list generates a callback */
641 if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
642 set_bit(slot->ds_node_num, o2hb_live_node_bitmap);
643
644 o2hb_queue_node_event(&event, O2HB_NODE_UP_CB, node,
645 slot->ds_node_num);
646
647 changed = 1;
648 }
649
650 list_add_tail(&slot->ds_live_item,
651 &o2hb_live_slots[slot->ds_node_num]);
652
653 slot->ds_equal_samples = 0;
654
655 /* We want to be sure that all nodes agree on the
656 * number of milliseconds before a node will be
657 * considered dead. The self-fencing timeout is
658 * computed from this value, and a discrepancy might
659 * result in heartbeat calling a node dead when it
660 * hasn't self-fenced yet. */
661 slot_dead_ms = le32_to_cpu(hb_block->hb_dead_ms);
662 if (slot_dead_ms && slot_dead_ms != dead_ms) {
663 /* TODO: Perhaps we can fail the region here. */
664 mlog(ML_ERROR, "Node %d on device %s has a dead count "
665 "of %u ms, but our count is %u ms.\n"
666 "Please double check your configuration values "
667 "for 'O2CB_HEARTBEAT_THRESHOLD'\n",
668 slot->ds_node_num, reg->hr_dev_name, slot_dead_ms,
669 dead_ms);
670 }
671 goto out;
672 }
673
674 /* if the list is dead, we're done.. */
675 if (list_empty(&slot->ds_live_item))
676 goto out;
677
678 /* live nodes only go dead after enough consequtive missed
679 * samples.. reset the missed counter whenever we see
680 * activity */
681 if (slot->ds_equal_samples >= o2hb_dead_threshold || gen_changed) {
682 mlog(ML_HEARTBEAT, "Node %d left my region\n",
683 slot->ds_node_num);
684
685 /* last off the live_slot generates a callback */
686 list_del_init(&slot->ds_live_item);
687 if (list_empty(&o2hb_live_slots[slot->ds_node_num])) {
688 clear_bit(slot->ds_node_num, o2hb_live_node_bitmap);
689
690 o2hb_queue_node_event(&event, O2HB_NODE_DOWN_CB, node,
691 slot->ds_node_num);
692
693 changed = 1;
694 }
695
696 /* We don't clear this because the node is still
697 * actually writing new blocks. */
698 if (!gen_changed)
699 slot->ds_changed_samples = 0;
700 goto out;
701 }
702 if (slot->ds_changed_samples) {
703 slot->ds_changed_samples = 0;
704 slot->ds_equal_samples = 0;
705 }
706 out:
707 spin_unlock(&o2hb_live_lock);
708
709 o2hb_run_event_list(&event);
710
711 o2nm_node_put(node);
712 return changed;
713 }
714
715 /* This could be faster if we just implmented a find_last_bit, but I
716 * don't think the circumstances warrant it. */
717 static int o2hb_highest_node(unsigned long *nodes,
718 int numbits)
719 {
720 int highest, node;
721
722 highest = numbits;
723 node = -1;
724 while ((node = find_next_bit(nodes, numbits, node + 1)) != -1) {
725 if (node >= numbits)
726 break;
727
728 highest = node;
729 }
730
731 return highest;
732 }
733
734 static int o2hb_do_disk_heartbeat(struct o2hb_region *reg)
735 {
736 int i, ret, highest_node, change = 0;
737 unsigned long configured_nodes[BITS_TO_LONGS(O2NM_MAX_NODES)];
738 struct o2hb_bio_wait_ctxt write_wc;
739
740 ret = o2nm_configured_node_map(configured_nodes,
741 sizeof(configured_nodes));
742 if (ret) {
743 mlog_errno(ret);
744 return ret;
745 }
746
747 highest_node = o2hb_highest_node(configured_nodes, O2NM_MAX_NODES);
748 if (highest_node >= O2NM_MAX_NODES) {
749 mlog(ML_NOTICE, "ocfs2_heartbeat: no configured nodes found!\n");
750 return -EINVAL;
751 }
752
753 /* No sense in reading the slots of nodes that don't exist
754 * yet. Of course, if the node definitions have holes in them
755 * then we're reading an empty slot anyway... Consider this
756 * best-effort. */
757 ret = o2hb_read_slots(reg, highest_node + 1);
758 if (ret < 0) {
759 mlog_errno(ret);
760 return ret;
761 }
762
763 /* With an up to date view of the slots, we can check that no
764 * other node has been improperly configured to heartbeat in
765 * our slot. */
766 if (!o2hb_check_last_timestamp(reg))
767 mlog(ML_ERROR, "Device \"%s\": another node is heartbeating "
768 "in our slot!\n", reg->hr_dev_name);
769
770 /* fill in the proper info for our next heartbeat */
771 o2hb_prepare_block(reg, reg->hr_generation);
772
773 /* And fire off the write. Note that we don't wait on this I/O
774 * until later. */
775 ret = o2hb_issue_node_write(reg, &write_wc);
776 if (ret < 0) {
777 mlog_errno(ret);
778 return ret;
779 }
780
781 i = -1;
782 while((i = find_next_bit(configured_nodes, O2NM_MAX_NODES, i + 1)) < O2NM_MAX_NODES) {
783
784 change |= o2hb_check_slot(reg, ®->hr_slots[i]);
785 }
786
787 /*
788 * We have to be sure we've advertised ourselves on disk
789 * before we can go to steady state. This ensures that
790 * people we find in our steady state have seen us.
791 */
792 o2hb_wait_on_io(reg, &write_wc);
793 if (write_wc.wc_error) {
794 /* Do not re-arm the write timeout on I/O error - we
795 * can't be sure that the new block ever made it to
796 * disk */
797 mlog(ML_ERROR, "Write error %d on device \"%s\"\n",
798 write_wc.wc_error, reg->hr_dev_name);
799 return write_wc.wc_error;
800 }
801
802 o2hb_arm_write_timeout(reg);
803
804 /* let the person who launched us know when things are steady */
805 if (!change && (atomic_read(®->hr_steady_iterations) != 0)) {
806 if (atomic_dec_and_test(®->hr_steady_iterations))
807 wake_up(&o2hb_steady_queue);
808 }
809
810 return 0;
811 }
812
813 /* Subtract b from a, storing the result in a. a *must* have a larger
814 * value than b. */
815 static void o2hb_tv_subtract(struct timeval *a,
816 struct timeval *b)
817 {
818 /* just return 0 when a is after b */
819 if (a->tv_sec < b->tv_sec ||
820 (a->tv_sec == b->tv_sec && a->tv_usec < b->tv_usec)) {
821 a->tv_sec = 0;
822 a->tv_usec = 0;
823 return;
824 }
825
826 a->tv_sec -= b->tv_sec;
827 a->tv_usec -= b->tv_usec;
828 while ( a->tv_usec < 0 ) {
829 a->tv_sec--;
830 a->tv_usec += 1000000;
831 }
832 }
833
834 static unsigned int o2hb_elapsed_msecs(struct timeval *start,
835 struct timeval *end)
836 {
837 struct timeval res = *end;
838
839 o2hb_tv_subtract(&res, start);
840
841 return res.tv_sec * 1000 + res.tv_usec / 1000;
842 }
843
844 /*
845 * we ride the region ref that the region dir holds. before the region
846 * dir is removed and drops it ref it will wait to tear down this
847 * thread.
848 */
849 static int o2hb_thread(void *data)
850 {
851 int i, ret;
852 struct o2hb_region *reg = data;
853 struct o2hb_bio_wait_ctxt write_wc;
854 struct timeval before_hb, after_hb;
855 unsigned int elapsed_msec;
856
857 mlog(ML_HEARTBEAT|ML_KTHREAD, "hb thread running\n");
858
859 set_user_nice(current, -20);
860
861 while (!kthread_should_stop() && !reg->hr_unclean_stop) {
862 /* We track the time spent inside
863 * o2hb_do_disk_heartbeat so that we avoid more than
864 * hr_timeout_ms between disk writes. On busy systems
865 * this should result in a heartbeat which is less
866 * likely to time itself out. */
867 do_gettimeofday(&before_hb);
868
869 i = 0;
870 do {
871 ret = o2hb_do_disk_heartbeat(reg);
872 } while (ret && ++i < 2);
873
874 do_gettimeofday(&after_hb);
875 elapsed_msec = o2hb_elapsed_msecs(&before_hb, &after_hb);
876
877 mlog(0, "start = %lu.%lu, end = %lu.%lu, msec = %u\n",
878 before_hb.tv_sec, (unsigned long) before_hb.tv_usec,
879 after_hb.tv_sec, (unsigned long) after_hb.tv_usec,
880 elapsed_msec);
881
882 if (elapsed_msec < reg->hr_timeout_ms) {
883 /* the kthread api has blocked signals for us so no
884 * need to record the return value. */
885 msleep_interruptible(reg->hr_timeout_ms - elapsed_msec);
886 }
887 }
888
889 o2hb_disarm_write_timeout(reg);
890
891 /* unclean stop is only used in very bad situation */
892 for(i = 0; !reg->hr_unclean_stop && i < reg->hr_blocks; i++)
893 o2hb_shutdown_slot(®->hr_slots[i]);
894
895 /* Explicit down notification - avoid forcing the other nodes
896 * to timeout on this region when we could just as easily
897 * write a clear generation - thus indicating to them that
898 * this node has left this region.
899 *
900 * XXX: Should we skip this on unclean_stop? */
901 o2hb_prepare_block(reg, 0);
902 ret = o2hb_issue_node_write(reg, &write_wc);
903 if (ret == 0) {
904 o2hb_wait_on_io(reg, &write_wc);
905 } else {
906 mlog_errno(ret);
907 }
908
909 mlog(ML_HEARTBEAT|ML_KTHREAD, "hb thread exiting\n");
910
911 return 0;
912 }
913
914 #ifdef CONFIG_DEBUG_FS
915 static int o2hb_debug_open(struct inode *inode, struct file *file)
916 {
917 unsigned long map[BITS_TO_LONGS(O2NM_MAX_NODES)];
918 char *buf = NULL;
919 int i = -1;
920 int out = 0;
921
922 buf = kmalloc(PAGE_SIZE, GFP_KERNEL);
923 if (!buf)
924 goto bail;
925
926 o2hb_fill_node_map(map, sizeof(map));
927
928 while ((i = find_next_bit(map, O2NM_MAX_NODES, i + 1)) < O2NM_MAX_NODES)
929 out += snprintf(buf + out, PAGE_SIZE - out, "%d ", i);
930 out += snprintf(buf + out, PAGE_SIZE - out, "\n");
931
932 i_size_write(inode, out);
933
934 file->private_data = buf;
935
936 return 0;
937 bail:
938 return -ENOMEM;
939 }
940
941 static int o2hb_debug_release(struct inode *inode, struct file *file)
942 {
943 kfree(file->private_data);
944 return 0;
945 }
946
947 static ssize_t o2hb_debug_read(struct file *file, char __user *buf,
948 size_t nbytes, loff_t *ppos)
949 {
950 return simple_read_from_buffer(buf, nbytes, ppos, file->private_data,
951 i_size_read(file->f_mapping->host));
952 }
953 #else
954 static int o2hb_debug_open(struct inode *inode, struct file *file)
955 {
956 return 0;
957 }
958 static int o2hb_debug_release(struct inode *inode, struct file *file)
959 {
960 return 0;
961 }
962 static ssize_t o2hb_debug_read(struct file *file, char __user *buf,
963 size_t nbytes, loff_t *ppos)
964 {
965 return 0;
966 }
967 #endif /* CONFIG_DEBUG_FS */
968
969 static struct file_operations o2hb_debug_fops = {
970 .open = o2hb_debug_open,
971 .release = o2hb_debug_release,
972 .read = o2hb_debug_read,
973 .llseek = generic_file_llseek,
974 };
975
976 void o2hb_exit(void)
977 {
978 if (o2hb_debug_livenodes)
979 debugfs_remove(o2hb_debug_livenodes);
980 if (o2hb_debug_dir)
981 debugfs_remove(o2hb_debug_dir);
982 }
983
984 int o2hb_init(void)
985 {
986 int i;
987
988 for (i = 0; i < ARRAY_SIZE(o2hb_callbacks); i++)
989 INIT_LIST_HEAD(&o2hb_callbacks[i].list);
990
991 for (i = 0; i < ARRAY_SIZE(o2hb_live_slots); i++)
992 INIT_LIST_HEAD(&o2hb_live_slots[i]);
993
994 INIT_LIST_HEAD(&o2hb_node_events);
995
996 memset(o2hb_live_node_bitmap, 0, sizeof(o2hb_live_node_bitmap));
997
998 o2hb_debug_dir = debugfs_create_dir(O2HB_DEBUG_DIR, NULL);
999 if (!o2hb_debug_dir) {
1000 mlog_errno(-ENOMEM);
1001 return -ENOMEM;
1002 }
1003
1004 o2hb_debug_livenodes = debugfs_create_file(O2HB_DEBUG_LIVENODES,
1005 S_IFREG|S_IRUSR,
1006 o2hb_debug_dir, NULL,
1007 &o2hb_debug_fops);
1008 if (!o2hb_debug_livenodes) {
1009 mlog_errno(-ENOMEM);
1010 debugfs_remove(o2hb_debug_dir);
1011 return -ENOMEM;
1012 }
1013
1014 return 0;
1015 }
1016
1017 /* if we're already in a callback then we're already serialized by the sem */
1018 static void o2hb_fill_node_map_from_callback(unsigned long *map,
1019 unsigned bytes)
1020 {
1021 BUG_ON(bytes < (BITS_TO_LONGS(O2NM_MAX_NODES) * sizeof(unsigned long)));
1022
1023 memcpy(map, &o2hb_live_node_bitmap, bytes);
1024 }
1025
1026 /*
1027 * get a map of all nodes that are heartbeating in any regions
1028 */
1029 void o2hb_fill_node_map(unsigned long *map, unsigned bytes)
1030 {
1031 /* callers want to serialize this map and callbacks so that they
1032 * can trust that they don't miss nodes coming to the party */
1033 down_read(&o2hb_callback_sem);
1034 spin_lock(&o2hb_live_lock);
1035 o2hb_fill_node_map_from_callback(map, bytes);
1036 spin_unlock(&o2hb_live_lock);
1037 up_read(&o2hb_callback_sem);
1038 }
1039 EXPORT_SYMBOL_GPL(o2hb_fill_node_map);
1040
1041 /*
1042 * heartbeat configfs bits. The heartbeat set is a default set under
1043 * the cluster set in nodemanager.c.
1044 */
1045
1046 static struct o2hb_region *to_o2hb_region(struct config_item *item)
1047 {
1048 return item ? container_of(item, struct o2hb_region, hr_item) : NULL;
1049 }
1050
1051 /* drop_item only drops its ref after killing the thread, nothing should
1052 * be using the region anymore. this has to clean up any state that
1053 * attributes might have built up. */
1054 static void o2hb_region_release(struct config_item *item)
1055 {
1056 int i;
1057 struct page *page;
1058 struct o2hb_region *reg = to_o2hb_region(item);
1059
1060 if (reg->hr_tmp_block)
1061 kfree(reg->hr_tmp_block);
1062
1063 if (reg->hr_slot_data) {
1064 for (i = 0; i < reg->hr_num_pages; i++) {
1065 page = reg->hr_slot_data[i];
1066 if (page)
1067 __free_page(page);
1068 }
1069 kfree(reg->hr_slot_data);
1070 }
1071
1072 if (reg->hr_bdev)
1073 blkdev_put(reg->hr_bdev, FMODE_READ|FMODE_WRITE);
1074
1075 if (reg->hr_slots)
1076 kfree(reg->hr_slots);
1077
1078 spin_lock(&o2hb_live_lock);
1079 list_del(®->hr_all_item);
1080 spin_unlock(&o2hb_live_lock);
1081
1082 kfree(reg);
1083 }
1084
1085 static int o2hb_read_block_input(struct o2hb_region *reg,
1086 const char *page,
1087 size_t count,
1088 unsigned long *ret_bytes,
1089 unsigned int *ret_bits)
1090 {
1091 unsigned long bytes;
1092 char *p = (char *)page;
1093
1094 bytes = simple_strtoul(p, &p, 0);
1095 if (!p || (*p && (*p != '\n')))
1096 return -EINVAL;
1097
1098 /* Heartbeat and fs min / max block sizes are the same. */
1099 if (bytes > 4096 || bytes < 512)
1100 return -ERANGE;
1101 if (hweight16(bytes) != 1)
1102 return -EINVAL;
1103
1104 if (ret_bytes)
1105 *ret_bytes = bytes;
1106 if (ret_bits)
1107 *ret_bits = ffs(bytes) - 1;
1108
1109 return 0;
1110 }
1111
1112 static ssize_t o2hb_region_block_bytes_read(struct o2hb_region *reg,
1113 char *page)
1114 {
1115 return sprintf(page, "%u\n", reg->hr_block_bytes);
1116 }
1117
1118 static ssize_t o2hb_region_block_bytes_write(struct o2hb_region *reg,
1119 const char *page,
1120 size_t count)
1121 {
1122 int status;
1123 unsigned long block_bytes;
1124 unsigned int block_bits;
1125
1126 if (reg->hr_bdev)
1127 return -EINVAL;
1128
1129 status = o2hb_read_block_input(reg, page, count,
1130 &block_bytes, &block_bits);
1131 if (status)
1132 return status;
1133
1134 reg->hr_block_bytes = (unsigned int)block_bytes;
1135 reg->hr_block_bits = block_bits;
1136
1137 return count;
1138 }
1139
1140 static ssize_t o2hb_region_start_block_read(struct o2hb_region *reg,
1141 char *page)
1142 {
1143 return sprintf(page, "%llu\n", reg->hr_start_block);
1144 }
1145
1146 static ssize_t o2hb_region_start_block_write(struct o2hb_region *reg,
1147 const char *page,
1148 size_t count)
1149 {
1150 unsigned long long tmp;
1151 char *p = (char *)page;
1152
1153 if (reg->hr_bdev)
1154 return -EINVAL;
1155
1156 tmp = simple_strtoull(p, &p, 0);
1157 if (!p || (*p && (*p != '\n')))
1158 return -EINVAL;
1159
1160 reg->hr_start_block = tmp;
1161
1162 return count;
1163 }
1164
1165 static ssize_t o2hb_region_blocks_read(struct o2hb_region *reg,
1166 char *page)
1167 {
1168 return sprintf(page, "%d\n", reg->hr_blocks);
1169 }
1170
1171 static ssize_t o2hb_region_blocks_write(struct o2hb_region *reg,
1172 const char *page,
1173 size_t count)
1174 {
1175 unsigned long tmp;
1176 char *p = (char *)page;
1177
1178 if (reg->hr_bdev)
1179 return -EINVAL;
1180
1181 tmp = simple_strtoul(p, &p, 0);
1182 if (!p || (*p && (*p != '\n')))
1183 return -EINVAL;
1184
1185 if (tmp > O2NM_MAX_NODES || tmp == 0)
1186 return -ERANGE;
1187
1188 reg->hr_blocks = (unsigned int)tmp;
1189
1190 return count;
1191 }
1192
1193 static ssize_t o2hb_region_dev_read(struct o2hb_region *reg,
1194 char *page)
1195 {
1196 unsigned int ret = 0;
1197
1198 if (reg->hr_bdev)
1199 ret = sprintf(page, "%s\n", reg->hr_dev_name);
1200
1201 return ret;
1202 }
1203
1204 static void o2hb_init_region_params(struct o2hb_region *reg)
1205 {
1206 reg->hr_slots_per_page = PAGE_CACHE_SIZE >> reg->hr_block_bits;
1207 reg->hr_timeout_ms = O2HB_REGION_TIMEOUT_MS;
1208
1209 mlog(ML_HEARTBEAT, "hr_start_block = %llu, hr_blocks = %u\n",
1210 reg->hr_start_block, reg->hr_blocks);
1211 mlog(ML_HEARTBEAT, "hr_block_bytes = %u, hr_block_bits = %u\n",
1212 reg->hr_block_bytes, reg->hr_block_bits);
1213 mlog(ML_HEARTBEAT, "hr_timeout_ms = %u\n", reg->hr_timeout_ms);
1214 mlog(ML_HEARTBEAT, "dead threshold = %u\n", o2hb_dead_threshold);
1215 }
1216
1217 static int o2hb_map_slot_data(struct o2hb_region *reg)
1218 {
1219 int i, j;
1220 unsigned int last_slot;
1221 unsigned int spp = reg->hr_slots_per_page;
1222 struct page *page;
1223 char *raw;
1224 struct o2hb_disk_slot *slot;
1225
1226 reg->hr_tmp_block = kmalloc(reg->hr_block_bytes, GFP_KERNEL);
1227 if (reg->hr_tmp_block == NULL) {
1228 mlog_errno(-ENOMEM);
1229 return -ENOMEM;
1230 }
1231
1232 reg->hr_slots = kcalloc(reg->hr_blocks,
1233 sizeof(struct o2hb_disk_slot), GFP_KERNEL);
1234 if (reg->hr_slots == NULL) {
1235 mlog_errno(-ENOMEM);
1236 return -ENOMEM;
1237 }
1238
1239 for(i = 0; i < reg->hr_blocks; i++) {
1240 slot = ®->hr_slots[i];
1241 slot->ds_node_num = i;
1242 INIT_LIST_HEAD(&slot->ds_live_item);
1243 slot->ds_raw_block = NULL;
1244 }
1245
1246 reg->hr_num_pages = (reg->hr_blocks + spp - 1) / spp;
1247 mlog(ML_HEARTBEAT, "Going to require %u pages to cover %u blocks "
1248 "at %u blocks per page\n",
1249 reg->hr_num_pages, reg->hr_blocks, spp);
1250
1251 reg->hr_slot_data = kcalloc(reg->hr_num_pages, sizeof(struct page *),
1252 GFP_KERNEL);
1253 if (!reg->hr_slot_data) {
1254 mlog_errno(-ENOMEM);
1255 return -ENOMEM;
1256 }
1257
1258 for(i = 0; i < reg->hr_num_pages; i++) {
1259 page = alloc_page(GFP_KERNEL);
1260 if (!page) {
1261 mlog_errno(-ENOMEM);
1262 return -ENOMEM;
1263 }
1264
1265 reg->hr_slot_data[i] = page;
1266
1267 last_slot = i * spp;
1268 raw = page_address(page);
1269 for (j = 0;
1270 (j < spp) && ((j + last_slot) < reg->hr_blocks);
1271 j++) {
1272 BUG_ON((j + last_slot) >= reg->hr_blocks);
1273
1274 slot = ®->hr_slots[j + last_slot];
1275 slot->ds_raw_block =
1276 (struct o2hb_disk_heartbeat_block *) raw;
1277
1278 raw += reg->hr_block_bytes;
1279 }
1280 }
1281
1282 return 0;
1283 }
1284
1285 /* Read in all the slots available and populate the tracking
1286 * structures so that we can start with a baseline idea of what's
1287 * there. */
1288 static int o2hb_populate_slot_data(struct o2hb_region *reg)
1289 {
1290 int ret, i;
1291 struct o2hb_disk_slot *slot;
1292 struct o2hb_disk_heartbeat_block *hb_block;
1293
1294 mlog_entry_void();
1295
1296 ret = o2hb_read_slots(reg, reg->hr_blocks);
1297 if (ret) {
1298 mlog_errno(ret);
1299 goto out;
1300 }
1301
1302 /* We only want to get an idea of the values initially in each
1303 * slot, so we do no verification - o2hb_check_slot will
1304 * actually determine if each configured slot is valid and
1305 * whether any values have changed. */
1306 for(i = 0; i < reg->hr_blocks; i++) {
1307 slot = ®->hr_slots[i];
1308 hb_block = (struct o2hb_disk_heartbeat_block *) slot->ds_raw_block;
1309
1310 /* Only fill the values that o2hb_check_slot uses to
1311 * determine changing slots */
1312 slot->ds_last_time = le64_to_cpu(hb_block->hb_seq);
1313 slot->ds_last_generation = le64_to_cpu(hb_block->hb_generation);
1314 }
1315
1316 out:
1317 mlog_exit(ret);
1318 return ret;
1319 }
1320
1321 /* this is acting as commit; we set up all of hr_bdev and hr_task or nothing */
1322 static ssize_t o2hb_region_dev_write(struct o2hb_region *reg,
1323 const char *page,
1324 size_t count)
1325 {
1326 struct task_struct *hb_task;
1327 long fd;
1328 int sectsize;
1329 char *p = (char *)page;
1330 struct file *filp = NULL;
1331 struct inode *inode = NULL;
1332 ssize_t ret = -EINVAL;
1333
1334 if (reg->hr_bdev)
1335 goto out;
1336
1337 /* We can't heartbeat without having had our node number
1338 * configured yet. */
1339 if (o2nm_this_node() == O2NM_MAX_NODES)
1340 goto out;
1341
1342 fd = simple_strtol(p, &p, 0);
1343 if (!p || (*p && (*p != '\n')))
1344 goto out;
1345
1346 if (fd < 0 || fd >= INT_MAX)
1347 goto out;
1348
1349 filp = fget(fd);
1350 if (filp == NULL)
1351 goto out;
1352
1353 if (reg->hr_blocks == 0 || reg->hr_start_block == 0 ||
1354 reg->hr_block_bytes == 0)
1355 goto out;
1356
1357 inode = igrab(filp->f_mapping->host);
1358 if (inode == NULL)
1359 goto out;
1360
1361 if (!S_ISBLK(inode->i_mode))
1362 goto out;
1363
1364 reg->hr_bdev = I_BDEV(filp->f_mapping->host);
1365 ret = blkdev_get(reg->hr_bdev, FMODE_WRITE | FMODE_READ);
1366 if (ret) {
1367 reg->hr_bdev = NULL;
1368 goto out;
1369 }
1370 inode = NULL;
1371
1372 bdevname(reg->hr_bdev, reg->hr_dev_name);
1373
1374 sectsize = bdev_logical_block_size(reg->hr_bdev);
1375 if (sectsize != reg->hr_block_bytes) {
1376 mlog(ML_ERROR,
1377 "blocksize %u incorrect for device, expected %d",
1378 reg->hr_block_bytes, sectsize);
1379 ret = -EINVAL;
1380 goto out;
1381 }
1382
1383 o2hb_init_region_params(reg);
1384
1385 /* Generation of zero is invalid */
1386 do {
1387 get_random_bytes(®->hr_generation,
1388 sizeof(reg->hr_generation));
1389 } while (reg->hr_generation == 0);
1390
1391 ret = o2hb_map_slot_data(reg);
1392 if (ret) {
1393 mlog_errno(ret);
1394 goto out;
1395 }
1396
1397 ret = o2hb_populate_slot_data(reg);
1398 if (ret) {
1399 mlog_errno(ret);
1400 goto out;
1401 }
1402
1403 INIT_DELAYED_WORK(®->hr_write_timeout_work, o2hb_write_timeout);
1404
1405 /*
1406 * A node is considered live after it has beat LIVE_THRESHOLD
1407 * times. We're not steady until we've given them a chance
1408 * _after_ our first read.
1409 */
1410 atomic_set(®->hr_steady_iterations, O2HB_LIVE_THRESHOLD + 1);
1411
1412 hb_task = kthread_run(o2hb_thread, reg, "o2hb-%s",
1413 reg->hr_item.ci_name);
1414 if (IS_ERR(hb_task)) {
1415 ret = PTR_ERR(hb_task);
1416 mlog_errno(ret);
1417 goto out;
1418 }
1419
1420 spin_lock(&o2hb_live_lock);
1421 reg->hr_task = hb_task;
1422 spin_unlock(&o2hb_live_lock);
1423
1424 ret = wait_event_interruptible(o2hb_steady_queue,
1425 atomic_read(®->hr_steady_iterations) == 0);
1426 if (ret) {
1427 /* We got interrupted (hello ptrace!). Clean up */
1428 spin_lock(&o2hb_live_lock);
1429 hb_task = reg->hr_task;
1430 reg->hr_task = NULL;
1431 spin_unlock(&o2hb_live_lock);
1432
1433 if (hb_task)
1434 kthread_stop(hb_task);
1435 goto out;
1436 }
1437
1438 /* Ok, we were woken. Make sure it wasn't by drop_item() */
1439 spin_lock(&o2hb_live_lock);
1440 hb_task = reg->hr_task;
1441 spin_unlock(&o2hb_live_lock);
1442
1443 if (hb_task)
1444 ret = count;
1445 else
1446 ret = -EIO;
1447
1448 out:
1449 if (filp)
1450 fput(filp);
1451 if (inode)
1452 iput(inode);
1453 if (ret < 0) {
1454 if (reg->hr_bdev) {
1455 blkdev_put(reg->hr_bdev, FMODE_READ|FMODE_WRITE);
1456 reg->hr_bdev = NULL;
1457 }
1458 }
1459 return ret;
1460 }
1461
1462 static ssize_t o2hb_region_pid_read(struct o2hb_region *reg,
1463 char *page)
1464 {
1465 pid_t pid = 0;
1466
1467 spin_lock(&o2hb_live_lock);
1468 if (reg->hr_task)
1469 pid = task_pid_nr(reg->hr_task);
1470 spin_unlock(&o2hb_live_lock);
1471
1472 if (!pid)
1473 return 0;
1474
1475 return sprintf(page, "%u\n", pid);
1476 }
1477
1478 struct o2hb_region_attribute {
1479 struct configfs_attribute attr;
1480 ssize_t (*show)(struct o2hb_region *, char *);
1481 ssize_t (*store)(struct o2hb_region *, const char *, size_t);
1482 };
1483
1484 static struct o2hb_region_attribute o2hb_region_attr_block_bytes = {
1485 .attr = { .ca_owner = THIS_MODULE,
1486 .ca_name = "block_bytes",
1487 .ca_mode = S_IRUGO | S_IWUSR },
1488 .show = o2hb_region_block_bytes_read,
1489 .store = o2hb_region_block_bytes_write,
1490 };
1491
1492 static struct o2hb_region_attribute o2hb_region_attr_start_block = {
1493 .attr = { .ca_owner = THIS_MODULE,
1494 .ca_name = "start_block",
1495 .ca_mode = S_IRUGO | S_IWUSR },
1496 .show = o2hb_region_start_block_read,
1497 .store = o2hb_region_start_block_write,
1498 };
1499
1500 static struct o2hb_region_attribute o2hb_region_attr_blocks = {
1501 .attr = { .ca_owner = THIS_MODULE,
1502 .ca_name = "blocks",
1503 .ca_mode = S_IRUGO | S_IWUSR },
1504 .show = o2hb_region_blocks_read,
1505 .store = o2hb_region_blocks_write,
1506 };
1507
1508 static struct o2hb_region_attribute o2hb_region_attr_dev = {
1509 .attr = { .ca_owner = THIS_MODULE,
1510 .ca_name = "dev",
1511 .ca_mode = S_IRUGO | S_IWUSR },
1512 .show = o2hb_region_dev_read,
1513 .store = o2hb_region_dev_write,
1514 };
1515
1516 static struct o2hb_region_attribute o2hb_region_attr_pid = {
1517 .attr = { .ca_owner = THIS_MODULE,
1518 .ca_name = "pid",
1519 .ca_mode = S_IRUGO | S_IRUSR },
1520 .show = o2hb_region_pid_read,
1521 };
1522
1523 static struct configfs_attribute *o2hb_region_attrs[] = {
1524 &o2hb_region_attr_block_bytes.attr,
1525 &o2hb_region_attr_start_block.attr,
1526 &o2hb_region_attr_blocks.attr,
1527 &o2hb_region_attr_dev.attr,
1528 &o2hb_region_attr_pid.attr,
1529 NULL,
1530 };
1531
1532 static ssize_t o2hb_region_show(struct config_item *item,
1533 struct configfs_attribute *attr,
1534 char *page)
1535 {
1536 struct o2hb_region *reg = to_o2hb_region(item);
1537 struct o2hb_region_attribute *o2hb_region_attr =
1538 container_of(attr, struct o2hb_region_attribute, attr);
1539 ssize_t ret = 0;
1540
1541 if (o2hb_region_attr->show)
1542 ret = o2hb_region_attr->show(reg, page);
1543 return ret;
1544 }
1545
1546 static ssize_t o2hb_region_store(struct config_item *item,
1547 struct configfs_attribute *attr,
1548 const char *page, size_t count)
1549 {
1550 struct o2hb_region *reg = to_o2hb_region(item);
1551 struct o2hb_region_attribute *o2hb_region_attr =
1552 container_of(attr, struct o2hb_region_attribute, attr);
1553 ssize_t ret = -EINVAL;
1554
1555 if (o2hb_region_attr->store)
1556 ret = o2hb_region_attr->store(reg, page, count);
1557 return ret;
1558 }
1559
1560 static struct configfs_item_operations o2hb_region_item_ops = {
1561 .release = o2hb_region_release,
1562 .show_attribute = o2hb_region_show,
1563 .store_attribute = o2hb_region_store,
1564 };
1565
1566 static struct config_item_type o2hb_region_type = {
1567 .ct_item_ops = &o2hb_region_item_ops,
1568 .ct_attrs = o2hb_region_attrs,
1569 .ct_owner = THIS_MODULE,
1570 };
1571
1572 /* heartbeat set */
1573
1574 struct o2hb_heartbeat_group {
1575 struct config_group hs_group;
1576 /* some stuff? */
1577 };
1578
1579 static struct o2hb_heartbeat_group *to_o2hb_heartbeat_group(struct config_group *group)
1580 {
1581 return group ?
1582 container_of(group, struct o2hb_heartbeat_group, hs_group)
1583 : NULL;
1584 }
1585
1586 static struct config_item *o2hb_heartbeat_group_make_item(struct config_group *group,
1587 const char *name)
1588 {
1589 struct o2hb_region *reg = NULL;
1590
1591 reg = kzalloc(sizeof(struct o2hb_region), GFP_KERNEL);
1592 if (reg == NULL)
1593 return ERR_PTR(-ENOMEM);
1594
1595 config_item_init_type_name(®->hr_item, name, &o2hb_region_type);
1596
1597 spin_lock(&o2hb_live_lock);
1598 list_add_tail(®->hr_all_item, &o2hb_all_regions);
1599 spin_unlock(&o2hb_live_lock);
1600
1601 return ®->hr_item;
1602 }
1603
1604 static void o2hb_heartbeat_group_drop_item(struct config_group *group,
1605 struct config_item *item)
1606 {
1607 struct task_struct *hb_task;
1608 struct o2hb_region *reg = to_o2hb_region(item);
1609
1610 /* stop the thread when the user removes the region dir */
1611 spin_lock(&o2hb_live_lock);
1612 hb_task = reg->hr_task;
1613 reg->hr_task = NULL;
1614 spin_unlock(&o2hb_live_lock);
1615
1616 if (hb_task)
1617 kthread_stop(hb_task);
1618
1619 /*
1620 * If we're racing a dev_write(), we need to wake them. They will
1621 * check reg->hr_task
1622 */
1623 if (atomic_read(®->hr_steady_iterations) != 0) {
1624 atomic_set(®->hr_steady_iterations, 0);
1625 wake_up(&o2hb_steady_queue);
1626 }
1627
1628 config_item_put(item);
1629 }
1630
1631 struct o2hb_heartbeat_group_attribute {
1632 struct configfs_attribute attr;
1633 ssize_t (*show)(struct o2hb_heartbeat_group *, char *);
1634 ssize_t (*store)(struct o2hb_heartbeat_group *, const char *, size_t);
1635 };
1636
1637 static ssize_t o2hb_heartbeat_group_show(struct config_item *item,
1638 struct configfs_attribute *attr,
1639 char *page)
1640 {
1641 struct o2hb_heartbeat_group *reg = to_o2hb_heartbeat_group(to_config_group(item));
1642 struct o2hb_heartbeat_group_attribute *o2hb_heartbeat_group_attr =
1643 container_of(attr, struct o2hb_heartbeat_group_attribute, attr);
1644 ssize_t ret = 0;
1645
1646 if (o2hb_heartbeat_group_attr->show)
1647 ret = o2hb_heartbeat_group_attr->show(reg, page);
1648 return ret;
1649 }
1650
1651 static ssize_t o2hb_heartbeat_group_store(struct config_item *item,
1652 struct configfs_attribute *attr,
1653 const char *page, size_t count)
1654 {
1655 struct o2hb_heartbeat_group *reg = to_o2hb_heartbeat_group(to_config_group(item));
1656 struct o2hb_heartbeat_group_attribute *o2hb_heartbeat_group_attr =
1657 container_of(attr, struct o2hb_heartbeat_group_attribute, attr);
1658 ssize_t ret = -EINVAL;
1659
1660 if (o2hb_heartbeat_group_attr->store)
1661 ret = o2hb_heartbeat_group_attr->store(reg, page, count);
1662 return ret;
1663 }
1664
1665 static ssize_t o2hb_heartbeat_group_threshold_show(struct o2hb_heartbeat_group *group,
1666 char *page)
1667 {
1668 return sprintf(page, "%u\n", o2hb_dead_threshold);
1669 }
1670
1671 static ssize_t o2hb_heartbeat_group_threshold_store(struct o2hb_heartbeat_group *group,
1672 const char *page,
1673 size_t count)
1674 {
1675 unsigned long tmp;
1676 char *p = (char *)page;
1677
1678 tmp = simple_strtoul(p, &p, 10);
1679 if (!p || (*p && (*p != '\n')))
1680 return -EINVAL;
1681
1682 /* this will validate ranges for us. */
1683 o2hb_dead_threshold_set((unsigned int) tmp);
1684
1685 return count;
1686 }
1687
1688 static struct o2hb_heartbeat_group_attribute o2hb_heartbeat_group_attr_threshold = {
1689 .attr = { .ca_owner = THIS_MODULE,
1690 .ca_name = "dead_threshold",
1691 .ca_mode = S_IRUGO | S_IWUSR },
1692 .show = o2hb_heartbeat_group_threshold_show,
1693 .store = o2hb_heartbeat_group_threshold_store,
1694 };
1695
1696 static struct configfs_attribute *o2hb_heartbeat_group_attrs[] = {
1697 &o2hb_heartbeat_group_attr_threshold.attr,
1698 NULL,
1699 };
1700
1701 static struct configfs_item_operations o2hb_hearbeat_group_item_ops = {
1702 .show_attribute = o2hb_heartbeat_group_show,
1703 .store_attribute = o2hb_heartbeat_group_store,
1704 };
1705
1706 static struct configfs_group_operations o2hb_heartbeat_group_group_ops = {
1707 .make_item = o2hb_heartbeat_group_make_item,
1708 .drop_item = o2hb_heartbeat_group_drop_item,
1709 };
1710
1711 static struct config_item_type o2hb_heartbeat_group_type = {
1712 .ct_group_ops = &o2hb_heartbeat_group_group_ops,
1713 .ct_item_ops = &o2hb_hearbeat_group_item_ops,
1714 .ct_attrs = o2hb_heartbeat_group_attrs,
1715 .ct_owner = THIS_MODULE,
1716 };
1717
1718 /* this is just here to avoid touching group in heartbeat.h which the
1719 * entire damn world #includes */
1720 struct config_group *o2hb_alloc_hb_set(void)
1721 {
1722 struct o2hb_heartbeat_group *hs = NULL;
1723 struct config_group *ret = NULL;
1724
1725 hs = kzalloc(sizeof(struct o2hb_heartbeat_group), GFP_KERNEL);
1726 if (hs == NULL)
1727 goto out;
1728
1729 config_group_init_type_name(&hs->hs_group, "heartbeat",
1730 &o2hb_heartbeat_group_type);
1731
1732 ret = &hs->hs_group;
1733 out:
1734 if (ret == NULL)
1735 kfree(hs);
1736 return ret;
1737 }
1738
1739 void o2hb_free_hb_set(struct config_group *group)
1740 {
1741 struct o2hb_heartbeat_group *hs = to_o2hb_heartbeat_group(group);
1742 kfree(hs);
1743 }
1744
1745 /* hb callback registration and issueing */
1746
1747 static struct o2hb_callback *hbcall_from_type(enum o2hb_callback_type type)
1748 {
1749 if (type == O2HB_NUM_CB)
1750 return ERR_PTR(-EINVAL);
1751
1752 return &o2hb_callbacks[type];
1753 }
1754
1755 void o2hb_setup_callback(struct o2hb_callback_func *hc,
1756 enum o2hb_callback_type type,
1757 o2hb_cb_func *func,
1758 void *data,
1759 int priority)
1760 {
1761 INIT_LIST_HEAD(&hc->hc_item);
1762 hc->hc_func = func;
1763 hc->hc_data = data;
1764 hc->hc_priority = priority;
1765 hc->hc_type = type;
1766 hc->hc_magic = O2HB_CB_MAGIC;
1767 }
1768 EXPORT_SYMBOL_GPL(o2hb_setup_callback);
1769
1770 static struct o2hb_region *o2hb_find_region(const char *region_uuid)
1771 {
1772 struct o2hb_region *p, *reg = NULL;
1773
1774 assert_spin_locked(&o2hb_live_lock);
1775
1776 list_for_each_entry(p, &o2hb_all_regions, hr_all_item) {
1777 if (!strcmp(region_uuid, config_item_name(&p->hr_item))) {
1778 reg = p;
1779 break;
1780 }
1781 }
1782
1783 return reg;
1784 }
1785
1786 static int o2hb_region_get(const char *region_uuid)
1787 {
1788 int ret = 0;
1789 struct o2hb_region *reg;
1790
1791 spin_lock(&o2hb_live_lock);
1792
1793 reg = o2hb_find_region(region_uuid);
1794 if (!reg)
1795 ret = -ENOENT;
1796 spin_unlock(&o2hb_live_lock);
1797
1798 if (ret)
1799 goto out;
1800
1801 ret = o2nm_depend_this_node();
1802 if (ret)
1803 goto out;
1804
1805 ret = o2nm_depend_item(®->hr_item);
1806 if (ret)
1807 o2nm_undepend_this_node();
1808
1809 out:
1810 return ret;
1811 }
1812
1813 static void o2hb_region_put(const char *region_uuid)
1814 {
1815 struct o2hb_region *reg;
1816
1817 spin_lock(&o2hb_live_lock);
1818
1819 reg = o2hb_find_region(region_uuid);
1820
1821 spin_unlock(&o2hb_live_lock);
1822
1823 if (reg) {
1824 o2nm_undepend_item(®->hr_item);
1825 o2nm_undepend_this_node();
1826 }
1827 }
1828
1829 int o2hb_register_callback(const char *region_uuid,
1830 struct o2hb_callback_func *hc)
1831 {
1832 struct o2hb_callback_func *tmp;
1833 struct list_head *iter;
1834 struct o2hb_callback *hbcall;
1835 int ret;
1836
1837 BUG_ON(hc->hc_magic != O2HB_CB_MAGIC);
1838 BUG_ON(!list_empty(&hc->hc_item));
1839
1840 hbcall = hbcall_from_type(hc->hc_type);
1841 if (IS_ERR(hbcall)) {
1842 ret = PTR_ERR(hbcall);
1843 goto out;
1844 }
1845
1846 if (region_uuid) {
1847 ret = o2hb_region_get(region_uuid);
1848 if (ret)
1849 goto out;
1850 }
1851
1852 down_write(&o2hb_callback_sem);
1853
1854 list_for_each(iter, &hbcall->list) {
1855 tmp = list_entry(iter, struct o2hb_callback_func, hc_item);
1856 if (hc->hc_priority < tmp->hc_priority) {
1857 list_add_tail(&hc->hc_item, iter);
1858 break;
1859 }
1860 }
1861 if (list_empty(&hc->hc_item))
1862 list_add_tail(&hc->hc_item, &hbcall->list);
1863
1864 up_write(&o2hb_callback_sem);
1865 ret = 0;
1866 out:
1867 mlog(ML_HEARTBEAT, "returning %d on behalf of %p for funcs %p\n",
1868 ret, __builtin_return_address(0), hc);
1869 return ret;
1870 }
1871 EXPORT_SYMBOL_GPL(o2hb_register_callback);
1872
1873 void o2hb_unregister_callback(const char *region_uuid,
1874 struct o2hb_callback_func *hc)
1875 {
1876 BUG_ON(hc->hc_magic != O2HB_CB_MAGIC);
1877
1878 mlog(ML_HEARTBEAT, "on behalf of %p for funcs %p\n",
1879 __builtin_return_address(0), hc);
1880
1881 /* XXX Can this happen _with_ a region reference? */
1882 if (list_empty(&hc->hc_item))
1883 return;
1884
1885 if (region_uuid)
1886 o2hb_region_put(region_uuid);
1887
1888 down_write(&o2hb_callback_sem);
1889
1890 list_del_init(&hc->hc_item);
1891
1892 up_write(&o2hb_callback_sem);
1893 }
1894 EXPORT_SYMBOL_GPL(o2hb_unregister_callback);
1895
1896 int o2hb_check_node_heartbeating(u8 node_num)
1897 {
1898 unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
1899
1900 o2hb_fill_node_map(testing_map, sizeof(testing_map));
1901 if (!test_bit(node_num, testing_map)) {
1902 mlog(ML_HEARTBEAT,
1903 "node (%u) does not have heartbeating enabled.\n",
1904 node_num);
1905 return 0;
1906 }
1907
1908 return 1;
1909 }
1910 EXPORT_SYMBOL_GPL(o2hb_check_node_heartbeating);
1911
1912 int o2hb_check_node_heartbeating_from_callback(u8 node_num)
1913 {
1914 unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
1915
1916 o2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
1917 if (!test_bit(node_num, testing_map)) {
1918 mlog(ML_HEARTBEAT,
1919 "node (%u) does not have heartbeating enabled.\n",
1920 node_num);
1921 return 0;
1922 }
1923
1924 return 1;
1925 }
1926 EXPORT_SYMBOL_GPL(o2hb_check_node_heartbeating_from_callback);
1927
1928 /* Makes sure our local node is configured with a node number, and is
1929 * heartbeating. */
1930 int o2hb_check_local_node_heartbeating(void)
1931 {
1932 u8 node_num;
1933
1934 /* if this node was set then we have networking */
1935 node_num = o2nm_this_node();
1936 if (node_num == O2NM_MAX_NODES) {
1937 mlog(ML_HEARTBEAT, "this node has not been configured.\n");
1938 return 0;
1939 }
1940
1941 return o2hb_check_node_heartbeating(node_num);
1942 }
1943 EXPORT_SYMBOL_GPL(o2hb_check_local_node_heartbeating);
1944
1945 /*
1946 * this is just a hack until we get the plumbing which flips file systems
1947 * read only and drops the hb ref instead of killing the node dead.
1948 */
1949 void o2hb_stop_all_regions(void)
1950 {
1951 struct o2hb_region *reg;
1952
1953 mlog(ML_ERROR, "stopping heartbeat on all active regions.\n");
1954
1955 spin_lock(&o2hb_live_lock);
1956
1957 list_for_each_entry(reg, &o2hb_all_regions, hr_all_item)
1958 reg->hr_unclean_stop = 1;
1959
1960 spin_unlock(&o2hb_live_lock);
1961 }
1962 EXPORT_SYMBOL_GPL(o2hb_stop_all_regions);
1963
|
This page was automatically generated by the
LXR engine.
|