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