Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/tiwai/sound
[platform/adaptation/renesas_rcar/renesas_kernel.git] / fs / dlm / member.c
1 /******************************************************************************
2 *******************************************************************************
3 **
4 **  Copyright (C) 2005-2011 Red Hat, Inc.  All rights reserved.
5 **
6 **  This copyrighted material is made available to anyone wishing to use,
7 **  modify, copy, or redistribute it subject to the terms and conditions
8 **  of the GNU General Public License v.2.
9 **
10 *******************************************************************************
11 ******************************************************************************/
12
13 #include "dlm_internal.h"
14 #include "lockspace.h"
15 #include "member.h"
16 #include "recoverd.h"
17 #include "recover.h"
18 #include "rcom.h"
19 #include "config.h"
20 #include "lowcomms.h"
21
22 int dlm_slots_version(struct dlm_header *h)
23 {
24         if ((h->h_version & 0x0000FFFF) < DLM_HEADER_SLOTS)
25                 return 0;
26         return 1;
27 }
28
29 void dlm_slot_save(struct dlm_ls *ls, struct dlm_rcom *rc,
30                    struct dlm_member *memb)
31 {
32         struct rcom_config *rf = (struct rcom_config *)rc->rc_buf;
33
34         if (!dlm_slots_version(&rc->rc_header))
35                 return;
36
37         memb->slot = le16_to_cpu(rf->rf_our_slot);
38         memb->generation = le32_to_cpu(rf->rf_generation);
39 }
40
41 void dlm_slots_copy_out(struct dlm_ls *ls, struct dlm_rcom *rc)
42 {
43         struct dlm_slot *slot;
44         struct rcom_slot *ro;
45         int i;
46
47         ro = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));
48
49         /* ls_slots array is sparse, but not rcom_slots */
50
51         for (i = 0; i < ls->ls_slots_size; i++) {
52                 slot = &ls->ls_slots[i];
53                 if (!slot->nodeid)
54                         continue;
55                 ro->ro_nodeid = cpu_to_le32(slot->nodeid);
56                 ro->ro_slot = cpu_to_le16(slot->slot);
57                 ro++;
58         }
59 }
60
61 #define SLOT_DEBUG_LINE 128
62
63 static void log_debug_slots(struct dlm_ls *ls, uint32_t gen, int num_slots,
64                             struct rcom_slot *ro0, struct dlm_slot *array,
65                             int array_size)
66 {
67         char line[SLOT_DEBUG_LINE];
68         int len = SLOT_DEBUG_LINE - 1;
69         int pos = 0;
70         int ret, i;
71
72         if (!dlm_config.ci_log_debug)
73                 return;
74
75         memset(line, 0, sizeof(line));
76
77         if (array) {
78                 for (i = 0; i < array_size; i++) {
79                         if (!array[i].nodeid)
80                                 continue;
81
82                         ret = snprintf(line + pos, len - pos, " %d:%d",
83                                        array[i].slot, array[i].nodeid);
84                         if (ret >= len - pos)
85                                 break;
86                         pos += ret;
87                 }
88         } else if (ro0) {
89                 for (i = 0; i < num_slots; i++) {
90                         ret = snprintf(line + pos, len - pos, " %d:%d",
91                                        ro0[i].ro_slot, ro0[i].ro_nodeid);
92                         if (ret >= len - pos)
93                                 break;
94                         pos += ret;
95                 }
96         }
97
98         log_debug(ls, "generation %u slots %d%s", gen, num_slots, line);
99 }
100
101 int dlm_slots_copy_in(struct dlm_ls *ls)
102 {
103         struct dlm_member *memb;
104         struct dlm_rcom *rc = ls->ls_recover_buf;
105         struct rcom_config *rf = (struct rcom_config *)rc->rc_buf;
106         struct rcom_slot *ro0, *ro;
107         int our_nodeid = dlm_our_nodeid();
108         int i, num_slots;
109         uint32_t gen;
110
111         if (!dlm_slots_version(&rc->rc_header))
112                 return -1;
113
114         gen = le32_to_cpu(rf->rf_generation);
115         if (gen <= ls->ls_generation) {
116                 log_error(ls, "dlm_slots_copy_in gen %u old %u",
117                           gen, ls->ls_generation);
118         }
119         ls->ls_generation = gen;
120
121         num_slots = le16_to_cpu(rf->rf_num_slots);
122         if (!num_slots)
123                 return -1;
124
125         ro0 = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config));
126
127         for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
128                 ro->ro_nodeid = le32_to_cpu(ro->ro_nodeid);
129                 ro->ro_slot = le16_to_cpu(ro->ro_slot);
130         }
131
132         log_debug_slots(ls, gen, num_slots, ro0, NULL, 0);
133
134         list_for_each_entry(memb, &ls->ls_nodes, list) {
135                 for (i = 0, ro = ro0; i < num_slots; i++, ro++) {
136                         if (ro->ro_nodeid != memb->nodeid)
137                                 continue;
138                         memb->slot = ro->ro_slot;
139                         memb->slot_prev = memb->slot;
140                         break;
141                 }
142
143                 if (memb->nodeid == our_nodeid) {
144                         if (ls->ls_slot && ls->ls_slot != memb->slot) {
145                                 log_error(ls, "dlm_slots_copy_in our slot "
146                                           "changed %d %d", ls->ls_slot,
147                                           memb->slot);
148                                 return -1;
149                         }
150
151                         if (!ls->ls_slot)
152                                 ls->ls_slot = memb->slot;
153                 }
154
155                 if (!memb->slot) {
156                         log_error(ls, "dlm_slots_copy_in nodeid %d no slot",
157                                    memb->nodeid);
158                         return -1;
159                 }
160         }
161
162         return 0;
163 }
164
165 /* for any nodes that do not support slots, we will not have set memb->slot
166    in wait_status_all(), so memb->slot will remain -1, and we will not
167    assign slots or set ls_num_slots here */
168
169 int dlm_slots_assign(struct dlm_ls *ls, int *num_slots, int *slots_size,
170                      struct dlm_slot **slots_out, uint32_t *gen_out)
171 {
172         struct dlm_member *memb;
173         struct dlm_slot *array;
174         int our_nodeid = dlm_our_nodeid();
175         int array_size, max_slots, i;
176         int need = 0;
177         int max = 0;
178         int num = 0;
179         uint32_t gen = 0;
180
181         /* our own memb struct will have slot -1 gen 0 */
182
183         list_for_each_entry(memb, &ls->ls_nodes, list) {
184                 if (memb->nodeid == our_nodeid) {
185                         memb->slot = ls->ls_slot;
186                         memb->generation = ls->ls_generation;
187                         break;
188                 }
189         }
190
191         list_for_each_entry(memb, &ls->ls_nodes, list) {
192                 if (memb->generation > gen)
193                         gen = memb->generation;
194
195                 /* node doesn't support slots */
196
197                 if (memb->slot == -1)
198                         return -1;
199
200                 /* node needs a slot assigned */
201
202                 if (!memb->slot)
203                         need++;
204
205                 /* node has a slot assigned */
206
207                 num++;
208
209                 if (!max || max < memb->slot)
210                         max = memb->slot;
211
212                 /* sanity check, once slot is assigned it shouldn't change */
213
214                 if (memb->slot_prev && memb->slot && memb->slot_prev != memb->slot) {
215                         log_error(ls, "nodeid %d slot changed %d %d",
216                                   memb->nodeid, memb->slot_prev, memb->slot);
217                         return -1;
218                 }
219                 memb->slot_prev = memb->slot;
220         }
221
222         array_size = max + need;
223
224         array = kzalloc(array_size * sizeof(struct dlm_slot), GFP_NOFS);
225         if (!array)
226                 return -ENOMEM;
227
228         num = 0;
229
230         /* fill in slots (offsets) that are used */
231
232         list_for_each_entry(memb, &ls->ls_nodes, list) {
233                 if (!memb->slot)
234                         continue;
235
236                 if (memb->slot > array_size) {
237                         log_error(ls, "invalid slot number %d", memb->slot);
238                         kfree(array);
239                         return -1;
240                 }
241
242                 array[memb->slot - 1].nodeid = memb->nodeid;
243                 array[memb->slot - 1].slot = memb->slot;
244                 num++;
245         }
246
247         /* assign new slots from unused offsets */
248
249         list_for_each_entry(memb, &ls->ls_nodes, list) {
250                 if (memb->slot)
251                         continue;
252
253                 for (i = 0; i < array_size; i++) {
254                         if (array[i].nodeid)
255                                 continue;
256
257                         memb->slot = i + 1;
258                         memb->slot_prev = memb->slot;
259                         array[i].nodeid = memb->nodeid;
260                         array[i].slot = memb->slot;
261                         num++;
262
263                         if (!ls->ls_slot && memb->nodeid == our_nodeid)
264                                 ls->ls_slot = memb->slot;
265                         break;
266                 }
267
268                 if (!memb->slot) {
269                         log_error(ls, "no free slot found");
270                         kfree(array);
271                         return -1;
272                 }
273         }
274
275         gen++;
276
277         log_debug_slots(ls, gen, num, NULL, array, array_size);
278
279         max_slots = (dlm_config.ci_buffer_size - sizeof(struct dlm_rcom) -
280                      sizeof(struct rcom_config)) / sizeof(struct rcom_slot);
281
282         if (num > max_slots) {
283                 log_error(ls, "num_slots %d exceeds max_slots %d",
284                           num, max_slots);
285                 kfree(array);
286                 return -1;
287         }
288
289         *gen_out = gen;
290         *slots_out = array;
291         *slots_size = array_size;
292         *num_slots = num;
293         return 0;
294 }
295
296 static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
297 {
298         struct dlm_member *memb = NULL;
299         struct list_head *tmp;
300         struct list_head *newlist = &new->list;
301         struct list_head *head = &ls->ls_nodes;
302
303         list_for_each(tmp, head) {
304                 memb = list_entry(tmp, struct dlm_member, list);
305                 if (new->nodeid < memb->nodeid)
306                         break;
307         }
308
309         if (!memb)
310                 list_add_tail(newlist, head);
311         else {
312                 /* FIXME: can use list macro here */
313                 newlist->prev = tmp->prev;
314                 newlist->next = tmp;
315                 tmp->prev->next = newlist;
316                 tmp->prev = newlist;
317         }
318 }
319
320 static int dlm_add_member(struct dlm_ls *ls, struct dlm_config_node *node)
321 {
322         struct dlm_member *memb;
323         int error;
324
325         memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS);
326         if (!memb)
327                 return -ENOMEM;
328
329         error = dlm_lowcomms_connect_node(node->nodeid);
330         if (error < 0) {
331                 kfree(memb);
332                 return error;
333         }
334
335         memb->nodeid = node->nodeid;
336         memb->weight = node->weight;
337         memb->comm_seq = node->comm_seq;
338         add_ordered_member(ls, memb);
339         ls->ls_num_nodes++;
340         return 0;
341 }
342
343 static struct dlm_member *find_memb(struct list_head *head, int nodeid)
344 {
345         struct dlm_member *memb;
346
347         list_for_each_entry(memb, head, list) {
348                 if (memb->nodeid == nodeid)
349                         return memb;
350         }
351         return NULL;
352 }
353
354 int dlm_is_member(struct dlm_ls *ls, int nodeid)
355 {
356         if (find_memb(&ls->ls_nodes, nodeid))
357                 return 1;
358         return 0;
359 }
360
361 int dlm_is_removed(struct dlm_ls *ls, int nodeid)
362 {
363         if (find_memb(&ls->ls_nodes_gone, nodeid))
364                 return 1;
365         return 0;
366 }
367
368 static void clear_memb_list(struct list_head *head)
369 {
370         struct dlm_member *memb;
371
372         while (!list_empty(head)) {
373                 memb = list_entry(head->next, struct dlm_member, list);
374                 list_del(&memb->list);
375                 kfree(memb);
376         }
377 }
378
379 void dlm_clear_members(struct dlm_ls *ls)
380 {
381         clear_memb_list(&ls->ls_nodes);
382         ls->ls_num_nodes = 0;
383 }
384
385 void dlm_clear_members_gone(struct dlm_ls *ls)
386 {
387         clear_memb_list(&ls->ls_nodes_gone);
388 }
389
390 static void make_member_array(struct dlm_ls *ls)
391 {
392         struct dlm_member *memb;
393         int i, w, x = 0, total = 0, all_zero = 0, *array;
394
395         kfree(ls->ls_node_array);
396         ls->ls_node_array = NULL;
397
398         list_for_each_entry(memb, &ls->ls_nodes, list) {
399                 if (memb->weight)
400                         total += memb->weight;
401         }
402
403         /* all nodes revert to weight of 1 if all have weight 0 */
404
405         if (!total) {
406                 total = ls->ls_num_nodes;
407                 all_zero = 1;
408         }
409
410         ls->ls_total_weight = total;
411
412         array = kmalloc(sizeof(int) * total, GFP_NOFS);
413         if (!array)
414                 return;
415
416         list_for_each_entry(memb, &ls->ls_nodes, list) {
417                 if (!all_zero && !memb->weight)
418                         continue;
419
420                 if (all_zero)
421                         w = 1;
422                 else
423                         w = memb->weight;
424
425                 DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
426
427                 for (i = 0; i < w; i++)
428                         array[x++] = memb->nodeid;
429         }
430
431         ls->ls_node_array = array;
432 }
433
434 /* send a status request to all members just to establish comms connections */
435
436 static int ping_members(struct dlm_ls *ls)
437 {
438         struct dlm_member *memb;
439         int error = 0;
440
441         list_for_each_entry(memb, &ls->ls_nodes, list) {
442                 error = dlm_recovery_stopped(ls);
443                 if (error)
444                         break;
445                 error = dlm_rcom_status(ls, memb->nodeid, 0);
446                 if (error)
447                         break;
448         }
449         if (error)
450                 log_debug(ls, "ping_members aborted %d last nodeid %d",
451                           error, ls->ls_recover_nodeid);
452         return error;
453 }
454
455 static void dlm_lsop_recover_prep(struct dlm_ls *ls)
456 {
457         if (!ls->ls_ops || !ls->ls_ops->recover_prep)
458                 return;
459         ls->ls_ops->recover_prep(ls->ls_ops_arg);
460 }
461
462 static void dlm_lsop_recover_slot(struct dlm_ls *ls, struct dlm_member *memb)
463 {
464         struct dlm_slot slot;
465         uint32_t seq;
466         int error;
467
468         if (!ls->ls_ops || !ls->ls_ops->recover_slot)
469                 return;
470
471         /* if there is no comms connection with this node
472            or the present comms connection is newer
473            than the one when this member was added, then
474            we consider the node to have failed (versus
475            being removed due to dlm_release_lockspace) */
476
477         error = dlm_comm_seq(memb->nodeid, &seq);
478
479         if (!error && seq == memb->comm_seq)
480                 return;
481
482         slot.nodeid = memb->nodeid;
483         slot.slot = memb->slot;
484
485         ls->ls_ops->recover_slot(ls->ls_ops_arg, &slot);
486 }
487
488 void dlm_lsop_recover_done(struct dlm_ls *ls)
489 {
490         struct dlm_member *memb;
491         struct dlm_slot *slots;
492         int i, num;
493
494         if (!ls->ls_ops || !ls->ls_ops->recover_done)
495                 return;
496
497         num = ls->ls_num_nodes;
498
499         slots = kzalloc(num * sizeof(struct dlm_slot), GFP_KERNEL);
500         if (!slots)
501                 return;
502
503         i = 0;
504         list_for_each_entry(memb, &ls->ls_nodes, list) {
505                 if (i == num) {
506                         log_error(ls, "dlm_lsop_recover_done bad num %d", num);
507                         goto out;
508                 }
509                 slots[i].nodeid = memb->nodeid;
510                 slots[i].slot = memb->slot;
511                 i++;
512         }
513
514         ls->ls_ops->recover_done(ls->ls_ops_arg, slots, num,
515                                  ls->ls_slot, ls->ls_generation);
516  out:
517         kfree(slots);
518 }
519
520 static struct dlm_config_node *find_config_node(struct dlm_recover *rv,
521                                                 int nodeid)
522 {
523         int i;
524
525         for (i = 0; i < rv->nodes_count; i++) {
526                 if (rv->nodes[i].nodeid == nodeid)
527                         return &rv->nodes[i];
528         }
529         return NULL;
530 }
531
532 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
533 {
534         struct dlm_member *memb, *safe;
535         struct dlm_config_node *node;
536         int i, error, neg = 0, low = -1;
537
538         /* previously removed members that we've not finished removing need to
539            count as a negative change so the "neg" recovery steps will happen */
540
541         list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
542                 log_debug(ls, "prev removed member %d", memb->nodeid);
543                 neg++;
544         }
545
546         /* move departed members from ls_nodes to ls_nodes_gone */
547
548         list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
549                 node = find_config_node(rv, memb->nodeid);
550                 if (node && !node->new)
551                         continue;
552
553                 if (!node) {
554                         log_debug(ls, "remove member %d", memb->nodeid);
555                 } else {
556                         /* removed and re-added */
557                         log_debug(ls, "remove member %d comm_seq %u %u",
558                                   memb->nodeid, memb->comm_seq, node->comm_seq);
559                 }
560
561                 neg++;
562                 list_move(&memb->list, &ls->ls_nodes_gone);
563                 ls->ls_num_nodes--;
564                 dlm_lsop_recover_slot(ls, memb);
565         }
566
567         /* add new members to ls_nodes */
568
569         for (i = 0; i < rv->nodes_count; i++) {
570                 node = &rv->nodes[i];
571                 if (dlm_is_member(ls, node->nodeid))
572                         continue;
573                 dlm_add_member(ls, node);
574                 log_debug(ls, "add member %d", node->nodeid);
575         }
576
577         list_for_each_entry(memb, &ls->ls_nodes, list) {
578                 if (low == -1 || memb->nodeid < low)
579                         low = memb->nodeid;
580         }
581         ls->ls_low_nodeid = low;
582
583         make_member_array(ls);
584         *neg_out = neg;
585
586         error = ping_members(ls);
587         if (!error || error == -EPROTO) {
588                 /* new_lockspace() may be waiting to know if the config
589                    is good or bad */
590                 ls->ls_members_result = error;
591                 complete(&ls->ls_members_done);
592         }
593
594         log_debug(ls, "dlm_recover_members %d nodes", ls->ls_num_nodes);
595         return error;
596 }
597
598 /* Userspace guarantees that dlm_ls_stop() has completed on all nodes before
599    dlm_ls_start() is called on any of them to start the new recovery. */
600
601 int dlm_ls_stop(struct dlm_ls *ls)
602 {
603         int new;
604
605         /*
606          * Prevent dlm_recv from being in the middle of something when we do
607          * the stop.  This includes ensuring dlm_recv isn't processing a
608          * recovery message (rcom), while dlm_recoverd is aborting and
609          * resetting things from an in-progress recovery.  i.e. we want
610          * dlm_recoverd to abort its recovery without worrying about dlm_recv
611          * processing an rcom at the same time.  Stopping dlm_recv also makes
612          * it easy for dlm_receive_message() to check locking stopped and add a
613          * message to the requestqueue without races.
614          */
615
616         down_write(&ls->ls_recv_active);
617
618         /*
619          * Abort any recovery that's in progress (see RECOVERY_STOP,
620          * dlm_recovery_stopped()) and tell any other threads running in the
621          * dlm to quit any processing (see RUNNING, dlm_locking_stopped()).
622          */
623
624         spin_lock(&ls->ls_recover_lock);
625         set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
626         new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
627         ls->ls_recover_seq++;
628         spin_unlock(&ls->ls_recover_lock);
629
630         /*
631          * Let dlm_recv run again, now any normal messages will be saved on the
632          * requestqueue for later.
633          */
634
635         up_write(&ls->ls_recv_active);
636
637         /*
638          * This in_recovery lock does two things:
639          * 1) Keeps this function from returning until all threads are out
640          *    of locking routines and locking is truly stopped.
641          * 2) Keeps any new requests from being processed until it's unlocked
642          *    when recovery is complete.
643          */
644
645         if (new)
646                 down_write(&ls->ls_in_recovery);
647
648         /*
649          * The recoverd suspend/resume makes sure that dlm_recoverd (if
650          * running) has noticed RECOVERY_STOP above and quit processing the
651          * previous recovery.
652          */
653
654         dlm_recoverd_suspend(ls);
655
656         spin_lock(&ls->ls_recover_lock);
657         kfree(ls->ls_slots);
658         ls->ls_slots = NULL;
659         ls->ls_num_slots = 0;
660         ls->ls_slots_size = 0;
661         ls->ls_recover_status = 0;
662         spin_unlock(&ls->ls_recover_lock);
663
664         dlm_recoverd_resume(ls);
665
666         if (!ls->ls_recover_begin)
667                 ls->ls_recover_begin = jiffies;
668
669         dlm_lsop_recover_prep(ls);
670         return 0;
671 }
672
673 int dlm_ls_start(struct dlm_ls *ls)
674 {
675         struct dlm_recover *rv = NULL, *rv_old;
676         struct dlm_config_node *nodes;
677         int error, count;
678
679         rv = kzalloc(sizeof(struct dlm_recover), GFP_NOFS);
680         if (!rv)
681                 return -ENOMEM;
682
683         error = dlm_config_nodes(ls->ls_name, &nodes, &count);
684         if (error < 0)
685                 goto fail;
686
687         spin_lock(&ls->ls_recover_lock);
688
689         /* the lockspace needs to be stopped before it can be started */
690
691         if (!dlm_locking_stopped(ls)) {
692                 spin_unlock(&ls->ls_recover_lock);
693                 log_error(ls, "start ignored: lockspace running");
694                 error = -EINVAL;
695                 goto fail;
696         }
697
698         rv->nodes = nodes;
699         rv->nodes_count = count;
700         rv->seq = ++ls->ls_recover_seq;
701         rv_old = ls->ls_recover_args;
702         ls->ls_recover_args = rv;
703         spin_unlock(&ls->ls_recover_lock);
704
705         if (rv_old) {
706                 log_error(ls, "unused recovery %llx %d",
707                           (unsigned long long)rv_old->seq, rv_old->nodes_count);
708                 kfree(rv_old->nodes);
709                 kfree(rv_old);
710         }
711
712         dlm_recoverd_kick(ls);
713         return 0;
714
715  fail:
716         kfree(rv);
717         kfree(nodes);
718         return error;
719 }
720