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