recover.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957
  1. // SPDX-License-Identifier: GPL-2.0-only
  2. /******************************************************************************
  3. *******************************************************************************
  4. **
  5. ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
  6. ** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
  7. **
  8. **
  9. *******************************************************************************
  10. ******************************************************************************/
  11. #include "dlm_internal.h"
  12. #include "lockspace.h"
  13. #include "dir.h"
  14. #include "config.h"
  15. #include "ast.h"
  16. #include "memory.h"
  17. #include "rcom.h"
  18. #include "lock.h"
  19. #include "lowcomms.h"
  20. #include "member.h"
  21. #include "recover.h"
  22. /*
  23. * Recovery waiting routines: these functions wait for a particular reply from
  24. * a remote node, or for the remote node to report a certain status. They need
  25. * to abort if the lockspace is stopped indicating a node has failed (perhaps
  26. * the one being waited for).
  27. */
  28. /*
  29. * Wait until given function returns non-zero or lockspace is stopped
  30. * (LS_RECOVERY_STOP set due to failure of a node in ls_nodes). When another
  31. * function thinks it could have completed the waited-on task, they should wake
  32. * up ls_wait_general to get an immediate response rather than waiting for the
  33. * timeout. This uses a timeout so it can check periodically if the wait
  34. * should abort due to node failure (which doesn't cause a wake_up).
  35. * This should only be called by the dlm_recoverd thread.
  36. */
  37. int dlm_wait_function(struct dlm_ls *ls, int (*testfn) (struct dlm_ls *ls))
  38. {
  39. int error = 0;
  40. int rv;
  41. while (1) {
  42. rv = wait_event_timeout(ls->ls_wait_general,
  43. testfn(ls) || dlm_recovery_stopped(ls),
  44. dlm_config.ci_recover_timer * HZ);
  45. if (rv)
  46. break;
  47. if (test_bit(LSFL_RCOM_WAIT, &ls->ls_flags)) {
  48. log_debug(ls, "dlm_wait_function timed out");
  49. return -ETIMEDOUT;
  50. }
  51. }
  52. if (dlm_recovery_stopped(ls)) {
  53. log_debug(ls, "dlm_wait_function aborted");
  54. error = -EINTR;
  55. }
  56. return error;
  57. }
  58. /*
  59. * An efficient way for all nodes to wait for all others to have a certain
  60. * status. The node with the lowest nodeid polls all the others for their
  61. * status (wait_status_all) and all the others poll the node with the low id
  62. * for its accumulated result (wait_status_low). When all nodes have set
  63. * status flag X, then status flag X_ALL will be set on the low nodeid.
  64. */
  65. uint32_t dlm_recover_status(struct dlm_ls *ls)
  66. {
  67. uint32_t status;
  68. spin_lock(&ls->ls_recover_lock);
  69. status = ls->ls_recover_status;
  70. spin_unlock(&ls->ls_recover_lock);
  71. return status;
  72. }
  73. static void _set_recover_status(struct dlm_ls *ls, uint32_t status)
  74. {
  75. ls->ls_recover_status |= status;
  76. }
  77. void dlm_set_recover_status(struct dlm_ls *ls, uint32_t status)
  78. {
  79. spin_lock(&ls->ls_recover_lock);
  80. _set_recover_status(ls, status);
  81. spin_unlock(&ls->ls_recover_lock);
  82. }
  83. static int wait_status_all(struct dlm_ls *ls, uint32_t wait_status,
  84. int save_slots)
  85. {
  86. struct dlm_rcom *rc = ls->ls_recover_buf;
  87. struct dlm_member *memb;
  88. int error = 0, delay;
  89. list_for_each_entry(memb, &ls->ls_nodes, list) {
  90. delay = 0;
  91. for (;;) {
  92. if (dlm_recovery_stopped(ls)) {
  93. error = -EINTR;
  94. goto out;
  95. }
  96. error = dlm_rcom_status(ls, memb->nodeid, 0);
  97. if (error)
  98. goto out;
  99. if (save_slots)
  100. dlm_slot_save(ls, rc, memb);
  101. if (rc->rc_result & wait_status)
  102. break;
  103. if (delay < 1000)
  104. delay += 20;
  105. msleep(delay);
  106. }
  107. }
  108. out:
  109. return error;
  110. }
  111. static int wait_status_low(struct dlm_ls *ls, uint32_t wait_status,
  112. uint32_t status_flags)
  113. {
  114. struct dlm_rcom *rc = ls->ls_recover_buf;
  115. int error = 0, delay = 0, nodeid = ls->ls_low_nodeid;
  116. for (;;) {
  117. if (dlm_recovery_stopped(ls)) {
  118. error = -EINTR;
  119. goto out;
  120. }
  121. error = dlm_rcom_status(ls, nodeid, status_flags);
  122. if (error)
  123. break;
  124. if (rc->rc_result & wait_status)
  125. break;
  126. if (delay < 1000)
  127. delay += 20;
  128. msleep(delay);
  129. }
  130. out:
  131. return error;
  132. }
  133. static int wait_status(struct dlm_ls *ls, uint32_t status)
  134. {
  135. uint32_t status_all = status << 1;
  136. int error;
  137. if (ls->ls_low_nodeid == dlm_our_nodeid()) {
  138. error = wait_status_all(ls, status, 0);
  139. if (!error)
  140. dlm_set_recover_status(ls, status_all);
  141. } else
  142. error = wait_status_low(ls, status_all, 0);
  143. return error;
  144. }
  145. int dlm_recover_members_wait(struct dlm_ls *ls)
  146. {
  147. struct dlm_member *memb;
  148. struct dlm_slot *slots;
  149. int num_slots, slots_size;
  150. int error, rv;
  151. uint32_t gen;
  152. list_for_each_entry(memb, &ls->ls_nodes, list) {
  153. memb->slot = -1;
  154. memb->generation = 0;
  155. }
  156. if (ls->ls_low_nodeid == dlm_our_nodeid()) {
  157. error = wait_status_all(ls, DLM_RS_NODES, 1);
  158. if (error)
  159. goto out;
  160. /* slots array is sparse, slots_size may be > num_slots */
  161. rv = dlm_slots_assign(ls, &num_slots, &slots_size, &slots, &gen);
  162. if (!rv) {
  163. spin_lock(&ls->ls_recover_lock);
  164. _set_recover_status(ls, DLM_RS_NODES_ALL);
  165. ls->ls_num_slots = num_slots;
  166. ls->ls_slots_size = slots_size;
  167. ls->ls_slots = slots;
  168. ls->ls_generation = gen;
  169. spin_unlock(&ls->ls_recover_lock);
  170. } else {
  171. dlm_set_recover_status(ls, DLM_RS_NODES_ALL);
  172. }
  173. } else {
  174. error = wait_status_low(ls, DLM_RS_NODES_ALL, DLM_RSF_NEED_SLOTS);
  175. if (error)
  176. goto out;
  177. dlm_slots_copy_in(ls);
  178. }
  179. out:
  180. return error;
  181. }
  182. int dlm_recover_directory_wait(struct dlm_ls *ls)
  183. {
  184. return wait_status(ls, DLM_RS_DIR);
  185. }
  186. int dlm_recover_locks_wait(struct dlm_ls *ls)
  187. {
  188. return wait_status(ls, DLM_RS_LOCKS);
  189. }
  190. int dlm_recover_done_wait(struct dlm_ls *ls)
  191. {
  192. return wait_status(ls, DLM_RS_DONE);
  193. }
  194. /*
  195. * The recover_list contains all the rsb's for which we've requested the new
  196. * master nodeid. As replies are returned from the resource directories the
  197. * rsb's are removed from the list. When the list is empty we're done.
  198. *
  199. * The recover_list is later similarly used for all rsb's for which we've sent
  200. * new lkb's and need to receive new corresponding lkid's.
  201. *
  202. * We use the address of the rsb struct as a simple local identifier for the
  203. * rsb so we can match an rcom reply with the rsb it was sent for.
  204. */
  205. static int recover_list_empty(struct dlm_ls *ls)
  206. {
  207. int empty;
  208. spin_lock(&ls->ls_recover_list_lock);
  209. empty = list_empty(&ls->ls_recover_list);
  210. spin_unlock(&ls->ls_recover_list_lock);
  211. return empty;
  212. }
  213. static void recover_list_add(struct dlm_rsb *r)
  214. {
  215. struct dlm_ls *ls = r->res_ls;
  216. spin_lock(&ls->ls_recover_list_lock);
  217. if (list_empty(&r->res_recover_list)) {
  218. list_add_tail(&r->res_recover_list, &ls->ls_recover_list);
  219. ls->ls_recover_list_count++;
  220. dlm_hold_rsb(r);
  221. }
  222. spin_unlock(&ls->ls_recover_list_lock);
  223. }
  224. static void recover_list_del(struct dlm_rsb *r)
  225. {
  226. struct dlm_ls *ls = r->res_ls;
  227. spin_lock(&ls->ls_recover_list_lock);
  228. list_del_init(&r->res_recover_list);
  229. ls->ls_recover_list_count--;
  230. spin_unlock(&ls->ls_recover_list_lock);
  231. dlm_put_rsb(r);
  232. }
  233. static void recover_list_clear(struct dlm_ls *ls)
  234. {
  235. struct dlm_rsb *r, *s;
  236. spin_lock(&ls->ls_recover_list_lock);
  237. list_for_each_entry_safe(r, s, &ls->ls_recover_list, res_recover_list) {
  238. list_del_init(&r->res_recover_list);
  239. r->res_recover_locks_count = 0;
  240. dlm_put_rsb(r);
  241. ls->ls_recover_list_count--;
  242. }
  243. if (ls->ls_recover_list_count != 0) {
  244. log_error(ls, "warning: recover_list_count %d",
  245. ls->ls_recover_list_count);
  246. ls->ls_recover_list_count = 0;
  247. }
  248. spin_unlock(&ls->ls_recover_list_lock);
  249. }
  250. static int recover_idr_empty(struct dlm_ls *ls)
  251. {
  252. int empty = 1;
  253. spin_lock(&ls->ls_recover_idr_lock);
  254. if (ls->ls_recover_list_count)
  255. empty = 0;
  256. spin_unlock(&ls->ls_recover_idr_lock);
  257. return empty;
  258. }
  259. static int recover_idr_add(struct dlm_rsb *r)
  260. {
  261. struct dlm_ls *ls = r->res_ls;
  262. int rv;
  263. idr_preload(GFP_NOFS);
  264. spin_lock(&ls->ls_recover_idr_lock);
  265. if (r->res_id) {
  266. rv = -1;
  267. goto out_unlock;
  268. }
  269. rv = idr_alloc(&ls->ls_recover_idr, r, 1, 0, GFP_NOWAIT);
  270. if (rv < 0)
  271. goto out_unlock;
  272. r->res_id = rv;
  273. ls->ls_recover_list_count++;
  274. dlm_hold_rsb(r);
  275. rv = 0;
  276. out_unlock:
  277. spin_unlock(&ls->ls_recover_idr_lock);
  278. idr_preload_end();
  279. return rv;
  280. }
  281. static void recover_idr_del(struct dlm_rsb *r)
  282. {
  283. struct dlm_ls *ls = r->res_ls;
  284. spin_lock(&ls->ls_recover_idr_lock);
  285. idr_remove(&ls->ls_recover_idr, r->res_id);
  286. r->res_id = 0;
  287. ls->ls_recover_list_count--;
  288. spin_unlock(&ls->ls_recover_idr_lock);
  289. dlm_put_rsb(r);
  290. }
  291. static struct dlm_rsb *recover_idr_find(struct dlm_ls *ls, uint64_t id)
  292. {
  293. struct dlm_rsb *r;
  294. spin_lock(&ls->ls_recover_idr_lock);
  295. r = idr_find(&ls->ls_recover_idr, (int)id);
  296. spin_unlock(&ls->ls_recover_idr_lock);
  297. return r;
  298. }
  299. static void recover_idr_clear(struct dlm_ls *ls)
  300. {
  301. struct dlm_rsb *r;
  302. int id;
  303. spin_lock(&ls->ls_recover_idr_lock);
  304. idr_for_each_entry(&ls->ls_recover_idr, r, id) {
  305. idr_remove(&ls->ls_recover_idr, id);
  306. r->res_id = 0;
  307. r->res_recover_locks_count = 0;
  308. ls->ls_recover_list_count--;
  309. dlm_put_rsb(r);
  310. }
  311. if (ls->ls_recover_list_count != 0) {
  312. log_error(ls, "warning: recover_list_count %d",
  313. ls->ls_recover_list_count);
  314. ls->ls_recover_list_count = 0;
  315. }
  316. spin_unlock(&ls->ls_recover_idr_lock);
  317. }
  318. /* Master recovery: find new master node for rsb's that were
  319. mastered on nodes that have been removed.
  320. dlm_recover_masters
  321. recover_master
  322. dlm_send_rcom_lookup -> receive_rcom_lookup
  323. dlm_dir_lookup
  324. receive_rcom_lookup_reply <-
  325. dlm_recover_master_reply
  326. set_new_master
  327. set_master_lkbs
  328. set_lock_master
  329. */
  330. /*
  331. * Set the lock master for all LKBs in a lock queue
  332. * If we are the new master of the rsb, we may have received new
  333. * MSTCPY locks from other nodes already which we need to ignore
  334. * when setting the new nodeid.
  335. */
  336. static void set_lock_master(struct list_head *queue, int nodeid)
  337. {
  338. struct dlm_lkb *lkb;
  339. list_for_each_entry(lkb, queue, lkb_statequeue) {
  340. if (!(lkb->lkb_flags & DLM_IFL_MSTCPY)) {
  341. lkb->lkb_nodeid = nodeid;
  342. lkb->lkb_remid = 0;
  343. }
  344. }
  345. }
  346. static void set_master_lkbs(struct dlm_rsb *r)
  347. {
  348. set_lock_master(&r->res_grantqueue, r->res_nodeid);
  349. set_lock_master(&r->res_convertqueue, r->res_nodeid);
  350. set_lock_master(&r->res_waitqueue, r->res_nodeid);
  351. }
  352. /*
  353. * Propagate the new master nodeid to locks
  354. * The NEW_MASTER flag tells dlm_recover_locks() which rsb's to consider.
  355. * The NEW_MASTER2 flag tells recover_lvb() and recover_grant() which
  356. * rsb's to consider.
  357. */
  358. static void set_new_master(struct dlm_rsb *r)
  359. {
  360. set_master_lkbs(r);
  361. rsb_set_flag(r, RSB_NEW_MASTER);
  362. rsb_set_flag(r, RSB_NEW_MASTER2);
  363. }
  364. /*
  365. * We do async lookups on rsb's that need new masters. The rsb's
  366. * waiting for a lookup reply are kept on the recover_list.
  367. *
  368. * Another node recovering the master may have sent us a rcom lookup,
  369. * and our dlm_master_lookup() set it as the new master, along with
  370. * NEW_MASTER so that we'll recover it here (this implies dir_nodeid
  371. * equals our_nodeid below).
  372. */
  373. static int recover_master(struct dlm_rsb *r, unsigned int *count)
  374. {
  375. struct dlm_ls *ls = r->res_ls;
  376. int our_nodeid, dir_nodeid;
  377. int is_removed = 0;
  378. int error;
  379. if (is_master(r))
  380. return 0;
  381. is_removed = dlm_is_removed(ls, r->res_nodeid);
  382. if (!is_removed && !rsb_flag(r, RSB_NEW_MASTER))
  383. return 0;
  384. our_nodeid = dlm_our_nodeid();
  385. dir_nodeid = dlm_dir_nodeid(r);
  386. if (dir_nodeid == our_nodeid) {
  387. if (is_removed) {
  388. r->res_master_nodeid = our_nodeid;
  389. r->res_nodeid = 0;
  390. }
  391. /* set master of lkbs to ourself when is_removed, or to
  392. another new master which we set along with NEW_MASTER
  393. in dlm_master_lookup */
  394. set_new_master(r);
  395. error = 0;
  396. } else {
  397. recover_idr_add(r);
  398. error = dlm_send_rcom_lookup(r, dir_nodeid);
  399. }
  400. (*count)++;
  401. return error;
  402. }
  403. /*
  404. * All MSTCPY locks are purged and rebuilt, even if the master stayed the same.
  405. * This is necessary because recovery can be started, aborted and restarted,
  406. * causing the master nodeid to briefly change during the aborted recovery, and
  407. * change back to the original value in the second recovery. The MSTCPY locks
  408. * may or may not have been purged during the aborted recovery. Another node
  409. * with an outstanding request in waiters list and a request reply saved in the
  410. * requestqueue, cannot know whether it should ignore the reply and resend the
  411. * request, or accept the reply and complete the request. It must do the
  412. * former if the remote node purged MSTCPY locks, and it must do the later if
  413. * the remote node did not. This is solved by always purging MSTCPY locks, in
  414. * which case, the request reply would always be ignored and the request
  415. * resent.
  416. */
  417. static int recover_master_static(struct dlm_rsb *r, unsigned int *count)
  418. {
  419. int dir_nodeid = dlm_dir_nodeid(r);
  420. int new_master = dir_nodeid;
  421. if (dir_nodeid == dlm_our_nodeid())
  422. new_master = 0;
  423. dlm_purge_mstcpy_locks(r);
  424. r->res_master_nodeid = dir_nodeid;
  425. r->res_nodeid = new_master;
  426. set_new_master(r);
  427. (*count)++;
  428. return 0;
  429. }
  430. /*
  431. * Go through local root resources and for each rsb which has a master which
  432. * has departed, get the new master nodeid from the directory. The dir will
  433. * assign mastery to the first node to look up the new master. That means
  434. * we'll discover in this lookup if we're the new master of any rsb's.
  435. *
  436. * We fire off all the dir lookup requests individually and asynchronously to
  437. * the correct dir node.
  438. */
  439. int dlm_recover_masters(struct dlm_ls *ls)
  440. {
  441. struct dlm_rsb *r;
  442. unsigned int total = 0;
  443. unsigned int count = 0;
  444. int nodir = dlm_no_directory(ls);
  445. int error;
  446. log_rinfo(ls, "dlm_recover_masters");
  447. down_read(&ls->ls_root_sem);
  448. list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
  449. if (dlm_recovery_stopped(ls)) {
  450. up_read(&ls->ls_root_sem);
  451. error = -EINTR;
  452. goto out;
  453. }
  454. lock_rsb(r);
  455. if (nodir)
  456. error = recover_master_static(r, &count);
  457. else
  458. error = recover_master(r, &count);
  459. unlock_rsb(r);
  460. cond_resched();
  461. total++;
  462. if (error) {
  463. up_read(&ls->ls_root_sem);
  464. goto out;
  465. }
  466. }
  467. up_read(&ls->ls_root_sem);
  468. log_rinfo(ls, "dlm_recover_masters %u of %u", count, total);
  469. error = dlm_wait_function(ls, &recover_idr_empty);
  470. out:
  471. if (error)
  472. recover_idr_clear(ls);
  473. return error;
  474. }
  475. int dlm_recover_master_reply(struct dlm_ls *ls, struct dlm_rcom *rc)
  476. {
  477. struct dlm_rsb *r;
  478. int ret_nodeid, new_master;
  479. r = recover_idr_find(ls, rc->rc_id);
  480. if (!r) {
  481. log_error(ls, "dlm_recover_master_reply no id %llx",
  482. (unsigned long long)rc->rc_id);
  483. goto out;
  484. }
  485. ret_nodeid = rc->rc_result;
  486. if (ret_nodeid == dlm_our_nodeid())
  487. new_master = 0;
  488. else
  489. new_master = ret_nodeid;
  490. lock_rsb(r);
  491. r->res_master_nodeid = ret_nodeid;
  492. r->res_nodeid = new_master;
  493. set_new_master(r);
  494. unlock_rsb(r);
  495. recover_idr_del(r);
  496. if (recover_idr_empty(ls))
  497. wake_up(&ls->ls_wait_general);
  498. out:
  499. return 0;
  500. }
  501. /* Lock recovery: rebuild the process-copy locks we hold on a
  502. remastered rsb on the new rsb master.
  503. dlm_recover_locks
  504. recover_locks
  505. recover_locks_queue
  506. dlm_send_rcom_lock -> receive_rcom_lock
  507. dlm_recover_master_copy
  508. receive_rcom_lock_reply <-
  509. dlm_recover_process_copy
  510. */
  511. /*
  512. * keep a count of the number of lkb's we send to the new master; when we get
  513. * an equal number of replies then recovery for the rsb is done
  514. */
  515. static int recover_locks_queue(struct dlm_rsb *r, struct list_head *head)
  516. {
  517. struct dlm_lkb *lkb;
  518. int error = 0;
  519. list_for_each_entry(lkb, head, lkb_statequeue) {
  520. error = dlm_send_rcom_lock(r, lkb);
  521. if (error)
  522. break;
  523. r->res_recover_locks_count++;
  524. }
  525. return error;
  526. }
  527. static int recover_locks(struct dlm_rsb *r)
  528. {
  529. int error = 0;
  530. lock_rsb(r);
  531. DLM_ASSERT(!r->res_recover_locks_count, dlm_dump_rsb(r););
  532. error = recover_locks_queue(r, &r->res_grantqueue);
  533. if (error)
  534. goto out;
  535. error = recover_locks_queue(r, &r->res_convertqueue);
  536. if (error)
  537. goto out;
  538. error = recover_locks_queue(r, &r->res_waitqueue);
  539. if (error)
  540. goto out;
  541. if (r->res_recover_locks_count)
  542. recover_list_add(r);
  543. else
  544. rsb_clear_flag(r, RSB_NEW_MASTER);
  545. out:
  546. unlock_rsb(r);
  547. return error;
  548. }
  549. int dlm_recover_locks(struct dlm_ls *ls)
  550. {
  551. struct dlm_rsb *r;
  552. int error, count = 0;
  553. down_read(&ls->ls_root_sem);
  554. list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
  555. if (is_master(r)) {
  556. rsb_clear_flag(r, RSB_NEW_MASTER);
  557. continue;
  558. }
  559. if (!rsb_flag(r, RSB_NEW_MASTER))
  560. continue;
  561. if (dlm_recovery_stopped(ls)) {
  562. error = -EINTR;
  563. up_read(&ls->ls_root_sem);
  564. goto out;
  565. }
  566. error = recover_locks(r);
  567. if (error) {
  568. up_read(&ls->ls_root_sem);
  569. goto out;
  570. }
  571. count += r->res_recover_locks_count;
  572. }
  573. up_read(&ls->ls_root_sem);
  574. log_rinfo(ls, "dlm_recover_locks %d out", count);
  575. error = dlm_wait_function(ls, &recover_list_empty);
  576. out:
  577. if (error)
  578. recover_list_clear(ls);
  579. return error;
  580. }
  581. void dlm_recovered_lock(struct dlm_rsb *r)
  582. {
  583. DLM_ASSERT(rsb_flag(r, RSB_NEW_MASTER), dlm_dump_rsb(r););
  584. r->res_recover_locks_count--;
  585. if (!r->res_recover_locks_count) {
  586. rsb_clear_flag(r, RSB_NEW_MASTER);
  587. recover_list_del(r);
  588. }
  589. if (recover_list_empty(r->res_ls))
  590. wake_up(&r->res_ls->ls_wait_general);
  591. }
  592. /*
  593. * The lvb needs to be recovered on all master rsb's. This includes setting
  594. * the VALNOTVALID flag if necessary, and determining the correct lvb contents
  595. * based on the lvb's of the locks held on the rsb.
  596. *
  597. * RSB_VALNOTVALID is set in two cases:
  598. *
  599. * 1. we are master, but not new, and we purged an EX/PW lock held by a
  600. * failed node (in dlm_recover_purge which set RSB_RECOVER_LVB_INVAL)
  601. *
  602. * 2. we are a new master, and there are only NL/CR locks left.
  603. * (We could probably improve this by only invaliding in this way when
  604. * the previous master left uncleanly. VMS docs mention that.)
  605. *
  606. * The LVB contents are only considered for changing when this is a new master
  607. * of the rsb (NEW_MASTER2). Then, the rsb's lvb is taken from any lkb with
  608. * mode > CR. If no lkb's exist with mode above CR, the lvb contents are taken
  609. * from the lkb with the largest lvb sequence number.
  610. */
  611. static void recover_lvb(struct dlm_rsb *r)
  612. {
  613. struct dlm_lkb *lkb, *high_lkb = NULL;
  614. uint32_t high_seq = 0;
  615. int lock_lvb_exists = 0;
  616. int big_lock_exists = 0;
  617. int lvblen = r->res_ls->ls_lvblen;
  618. if (!rsb_flag(r, RSB_NEW_MASTER2) &&
  619. rsb_flag(r, RSB_RECOVER_LVB_INVAL)) {
  620. /* case 1 above */
  621. rsb_set_flag(r, RSB_VALNOTVALID);
  622. return;
  623. }
  624. if (!rsb_flag(r, RSB_NEW_MASTER2))
  625. return;
  626. /* we are the new master, so figure out if VALNOTVALID should
  627. be set, and set the rsb lvb from the best lkb available. */
  628. list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) {
  629. if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
  630. continue;
  631. lock_lvb_exists = 1;
  632. if (lkb->lkb_grmode > DLM_LOCK_CR) {
  633. big_lock_exists = 1;
  634. goto setflag;
  635. }
  636. if (((int)lkb->lkb_lvbseq - (int)high_seq) >= 0) {
  637. high_lkb = lkb;
  638. high_seq = lkb->lkb_lvbseq;
  639. }
  640. }
  641. list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) {
  642. if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
  643. continue;
  644. lock_lvb_exists = 1;
  645. if (lkb->lkb_grmode > DLM_LOCK_CR) {
  646. big_lock_exists = 1;
  647. goto setflag;
  648. }
  649. if (((int)lkb->lkb_lvbseq - (int)high_seq) >= 0) {
  650. high_lkb = lkb;
  651. high_seq = lkb->lkb_lvbseq;
  652. }
  653. }
  654. setflag:
  655. if (!lock_lvb_exists)
  656. goto out;
  657. /* lvb is invalidated if only NL/CR locks remain */
  658. if (!big_lock_exists)
  659. rsb_set_flag(r, RSB_VALNOTVALID);
  660. if (!r->res_lvbptr) {
  661. r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
  662. if (!r->res_lvbptr)
  663. goto out;
  664. }
  665. if (big_lock_exists) {
  666. r->res_lvbseq = lkb->lkb_lvbseq;
  667. memcpy(r->res_lvbptr, lkb->lkb_lvbptr, lvblen);
  668. } else if (high_lkb) {
  669. r->res_lvbseq = high_lkb->lkb_lvbseq;
  670. memcpy(r->res_lvbptr, high_lkb->lkb_lvbptr, lvblen);
  671. } else {
  672. r->res_lvbseq = 0;
  673. memset(r->res_lvbptr, 0, lvblen);
  674. }
  675. out:
  676. return;
  677. }
  678. /* All master rsb's flagged RECOVER_CONVERT need to be looked at. The locks
  679. converting PR->CW or CW->PR need to have their lkb_grmode set. */
  680. static void recover_conversion(struct dlm_rsb *r)
  681. {
  682. struct dlm_ls *ls = r->res_ls;
  683. struct dlm_lkb *lkb;
  684. int grmode = -1;
  685. list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) {
  686. if (lkb->lkb_grmode == DLM_LOCK_PR ||
  687. lkb->lkb_grmode == DLM_LOCK_CW) {
  688. grmode = lkb->lkb_grmode;
  689. break;
  690. }
  691. }
  692. list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) {
  693. if (lkb->lkb_grmode != DLM_LOCK_IV)
  694. continue;
  695. if (grmode == -1) {
  696. log_debug(ls, "recover_conversion %x set gr to rq %d",
  697. lkb->lkb_id, lkb->lkb_rqmode);
  698. lkb->lkb_grmode = lkb->lkb_rqmode;
  699. } else {
  700. log_debug(ls, "recover_conversion %x set gr %d",
  701. lkb->lkb_id, grmode);
  702. lkb->lkb_grmode = grmode;
  703. }
  704. }
  705. }
  706. /* We've become the new master for this rsb and waiting/converting locks may
  707. need to be granted in dlm_recover_grant() due to locks that may have
  708. existed from a removed node. */
  709. static void recover_grant(struct dlm_rsb *r)
  710. {
  711. if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
  712. rsb_set_flag(r, RSB_RECOVER_GRANT);
  713. }
  714. void dlm_recover_rsbs(struct dlm_ls *ls)
  715. {
  716. struct dlm_rsb *r;
  717. unsigned int count = 0;
  718. down_read(&ls->ls_root_sem);
  719. list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
  720. lock_rsb(r);
  721. if (is_master(r)) {
  722. if (rsb_flag(r, RSB_RECOVER_CONVERT))
  723. recover_conversion(r);
  724. /* recover lvb before granting locks so the updated
  725. lvb/VALNOTVALID is presented in the completion */
  726. recover_lvb(r);
  727. if (rsb_flag(r, RSB_NEW_MASTER2))
  728. recover_grant(r);
  729. count++;
  730. } else {
  731. rsb_clear_flag(r, RSB_VALNOTVALID);
  732. }
  733. rsb_clear_flag(r, RSB_RECOVER_CONVERT);
  734. rsb_clear_flag(r, RSB_RECOVER_LVB_INVAL);
  735. rsb_clear_flag(r, RSB_NEW_MASTER2);
  736. unlock_rsb(r);
  737. }
  738. up_read(&ls->ls_root_sem);
  739. if (count)
  740. log_rinfo(ls, "dlm_recover_rsbs %d done", count);
  741. }
  742. /* Create a single list of all root rsb's to be used during recovery */
  743. int dlm_create_root_list(struct dlm_ls *ls)
  744. {
  745. struct rb_node *n;
  746. struct dlm_rsb *r;
  747. int i, error = 0;
  748. down_write(&ls->ls_root_sem);
  749. if (!list_empty(&ls->ls_root_list)) {
  750. log_error(ls, "root list not empty");
  751. error = -EINVAL;
  752. goto out;
  753. }
  754. for (i = 0; i < ls->ls_rsbtbl_size; i++) {
  755. spin_lock(&ls->ls_rsbtbl[i].lock);
  756. for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
  757. r = rb_entry(n, struct dlm_rsb, res_hashnode);
  758. list_add(&r->res_root_list, &ls->ls_root_list);
  759. dlm_hold_rsb(r);
  760. }
  761. if (!RB_EMPTY_ROOT(&ls->ls_rsbtbl[i].toss))
  762. log_error(ls, "dlm_create_root_list toss not empty");
  763. spin_unlock(&ls->ls_rsbtbl[i].lock);
  764. }
  765. out:
  766. up_write(&ls->ls_root_sem);
  767. return error;
  768. }
  769. void dlm_release_root_list(struct dlm_ls *ls)
  770. {
  771. struct dlm_rsb *r, *safe;
  772. down_write(&ls->ls_root_sem);
  773. list_for_each_entry_safe(r, safe, &ls->ls_root_list, res_root_list) {
  774. list_del_init(&r->res_root_list);
  775. dlm_put_rsb(r);
  776. }
  777. up_write(&ls->ls_root_sem);
  778. }
  779. void dlm_clear_toss(struct dlm_ls *ls)
  780. {
  781. struct rb_node *n, *next;
  782. struct dlm_rsb *r;
  783. unsigned int count = 0;
  784. int i;
  785. for (i = 0; i < ls->ls_rsbtbl_size; i++) {
  786. spin_lock(&ls->ls_rsbtbl[i].lock);
  787. for (n = rb_first(&ls->ls_rsbtbl[i].toss); n; n = next) {
  788. next = rb_next(n);
  789. r = rb_entry(n, struct dlm_rsb, res_hashnode);
  790. rb_erase(n, &ls->ls_rsbtbl[i].toss);
  791. dlm_free_rsb(r);
  792. count++;
  793. }
  794. spin_unlock(&ls->ls_rsbtbl[i].lock);
  795. }
  796. if (count)
  797. log_rinfo(ls, "dlm_clear_toss %u done", count);
  798. }