fs/dlm/member.c

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