recoverd.c 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352
  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-2011 Red Hat, Inc. All rights reserved.
  7. **
  8. **
  9. *******************************************************************************
  10. ******************************************************************************/
  11. #include "dlm_internal.h"
  12. #include "lockspace.h"
  13. #include "member.h"
  14. #include "dir.h"
  15. #include "ast.h"
  16. #include "recover.h"
  17. #include "lowcomms.h"
  18. #include "lock.h"
  19. #include "requestqueue.h"
  20. #include "recoverd.h"
  21. /* If the start for which we're re-enabling locking (seq) has been superseded
  22. by a newer stop (ls_recover_seq), we need to leave locking disabled.
  23. We suspend dlm_recv threads here to avoid the race where dlm_recv a) sees
  24. locking stopped and b) adds a message to the requestqueue, but dlm_recoverd
  25. enables locking and clears the requestqueue between a and b. */
  26. static int enable_locking(struct dlm_ls *ls, uint64_t seq)
  27. {
  28. int error = -EINTR;
  29. down_write(&ls->ls_recv_active);
  30. spin_lock(&ls->ls_recover_lock);
  31. if (ls->ls_recover_seq == seq) {
  32. set_bit(LSFL_RUNNING, &ls->ls_flags);
  33. /* unblocks processes waiting to enter the dlm */
  34. up_write(&ls->ls_in_recovery);
  35. clear_bit(LSFL_RECOVER_LOCK, &ls->ls_flags);
  36. error = 0;
  37. }
  38. spin_unlock(&ls->ls_recover_lock);
  39. up_write(&ls->ls_recv_active);
  40. return error;
  41. }
  42. static int ls_recover(struct dlm_ls *ls, struct dlm_recover *rv)
  43. {
  44. unsigned long start;
  45. int error, neg = 0;
  46. log_rinfo(ls, "dlm_recover %llu", (unsigned long long)rv->seq);
  47. mutex_lock(&ls->ls_recoverd_active);
  48. dlm_callback_suspend(ls);
  49. dlm_clear_toss(ls);
  50. /*
  51. * This list of root rsb's will be the basis of most of the recovery
  52. * routines.
  53. */
  54. dlm_create_root_list(ls);
  55. /*
  56. * Add or remove nodes from the lockspace's ls_nodes list.
  57. */
  58. error = dlm_recover_members(ls, rv, &neg);
  59. if (error) {
  60. log_rinfo(ls, "dlm_recover_members error %d", error);
  61. goto fail;
  62. }
  63. dlm_recover_dir_nodeid(ls);
  64. ls->ls_recover_dir_sent_res = 0;
  65. ls->ls_recover_dir_sent_msg = 0;
  66. ls->ls_recover_locks_in = 0;
  67. dlm_set_recover_status(ls, DLM_RS_NODES);
  68. error = dlm_recover_members_wait(ls);
  69. if (error) {
  70. log_rinfo(ls, "dlm_recover_members_wait error %d", error);
  71. goto fail;
  72. }
  73. start = jiffies;
  74. /*
  75. * Rebuild our own share of the directory by collecting from all other
  76. * nodes their master rsb names that hash to us.
  77. */
  78. error = dlm_recover_directory(ls);
  79. if (error) {
  80. log_rinfo(ls, "dlm_recover_directory error %d", error);
  81. goto fail;
  82. }
  83. dlm_set_recover_status(ls, DLM_RS_DIR);
  84. error = dlm_recover_directory_wait(ls);
  85. if (error) {
  86. log_rinfo(ls, "dlm_recover_directory_wait error %d", error);
  87. goto fail;
  88. }
  89. log_rinfo(ls, "dlm_recover_directory %u out %u messages",
  90. ls->ls_recover_dir_sent_res, ls->ls_recover_dir_sent_msg);
  91. /*
  92. * We may have outstanding operations that are waiting for a reply from
  93. * a failed node. Mark these to be resent after recovery. Unlock and
  94. * cancel ops can just be completed.
  95. */
  96. dlm_recover_waiters_pre(ls);
  97. error = dlm_recovery_stopped(ls);
  98. if (error)
  99. goto fail;
  100. if (neg || dlm_no_directory(ls)) {
  101. /*
  102. * Clear lkb's for departed nodes.
  103. */
  104. dlm_recover_purge(ls);
  105. /*
  106. * Get new master nodeid's for rsb's that were mastered on
  107. * departed nodes.
  108. */
  109. error = dlm_recover_masters(ls);
  110. if (error) {
  111. log_rinfo(ls, "dlm_recover_masters error %d", error);
  112. goto fail;
  113. }
  114. /*
  115. * Send our locks on remastered rsb's to the new masters.
  116. */
  117. error = dlm_recover_locks(ls);
  118. if (error) {
  119. log_rinfo(ls, "dlm_recover_locks error %d", error);
  120. goto fail;
  121. }
  122. dlm_set_recover_status(ls, DLM_RS_LOCKS);
  123. error = dlm_recover_locks_wait(ls);
  124. if (error) {
  125. log_rinfo(ls, "dlm_recover_locks_wait error %d", error);
  126. goto fail;
  127. }
  128. log_rinfo(ls, "dlm_recover_locks %u in",
  129. ls->ls_recover_locks_in);
  130. /*
  131. * Finalize state in master rsb's now that all locks can be
  132. * checked. This includes conversion resolution and lvb
  133. * settings.
  134. */
  135. dlm_recover_rsbs(ls);
  136. } else {
  137. /*
  138. * Other lockspace members may be going through the "neg" steps
  139. * while also adding us to the lockspace, in which case they'll
  140. * be doing the recover_locks (RS_LOCKS) barrier.
  141. */
  142. dlm_set_recover_status(ls, DLM_RS_LOCKS);
  143. error = dlm_recover_locks_wait(ls);
  144. if (error) {
  145. log_rinfo(ls, "dlm_recover_locks_wait error %d", error);
  146. goto fail;
  147. }
  148. }
  149. dlm_release_root_list(ls);
  150. /*
  151. * Purge directory-related requests that are saved in requestqueue.
  152. * All dir requests from before recovery are invalid now due to the dir
  153. * rebuild and will be resent by the requesting nodes.
  154. */
  155. dlm_purge_requestqueue(ls);
  156. dlm_set_recover_status(ls, DLM_RS_DONE);
  157. error = dlm_recover_done_wait(ls);
  158. if (error) {
  159. log_rinfo(ls, "dlm_recover_done_wait error %d", error);
  160. goto fail;
  161. }
  162. dlm_clear_members_gone(ls);
  163. dlm_adjust_timeouts(ls);
  164. dlm_callback_resume(ls);
  165. error = enable_locking(ls, rv->seq);
  166. if (error) {
  167. log_rinfo(ls, "enable_locking error %d", error);
  168. goto fail;
  169. }
  170. error = dlm_process_requestqueue(ls);
  171. if (error) {
  172. log_rinfo(ls, "dlm_process_requestqueue error %d", error);
  173. goto fail;
  174. }
  175. error = dlm_recover_waiters_post(ls);
  176. if (error) {
  177. log_rinfo(ls, "dlm_recover_waiters_post error %d", error);
  178. goto fail;
  179. }
  180. dlm_recover_grant(ls);
  181. log_rinfo(ls, "dlm_recover %llu generation %u done: %u ms",
  182. (unsigned long long)rv->seq, ls->ls_generation,
  183. jiffies_to_msecs(jiffies - start));
  184. mutex_unlock(&ls->ls_recoverd_active);
  185. dlm_lsop_recover_done(ls);
  186. return 0;
  187. fail:
  188. dlm_release_root_list(ls);
  189. log_rinfo(ls, "dlm_recover %llu error %d",
  190. (unsigned long long)rv->seq, error);
  191. mutex_unlock(&ls->ls_recoverd_active);
  192. return error;
  193. }
  194. /* The dlm_ls_start() that created the rv we take here may already have been
  195. stopped via dlm_ls_stop(); in that case we need to leave the RECOVERY_STOP
  196. flag set. */
  197. static void do_ls_recovery(struct dlm_ls *ls)
  198. {
  199. struct dlm_recover *rv = NULL;
  200. spin_lock(&ls->ls_recover_lock);
  201. rv = ls->ls_recover_args;
  202. ls->ls_recover_args = NULL;
  203. if (rv && ls->ls_recover_seq == rv->seq)
  204. clear_bit(LSFL_RECOVER_STOP, &ls->ls_flags);
  205. spin_unlock(&ls->ls_recover_lock);
  206. if (rv) {
  207. ls_recover(ls, rv);
  208. kfree(rv->nodes);
  209. kfree(rv);
  210. }
  211. }
  212. static int dlm_recoverd(void *arg)
  213. {
  214. struct dlm_ls *ls;
  215. ls = dlm_find_lockspace_local(arg);
  216. if (!ls) {
  217. log_print("dlm_recoverd: no lockspace %p", arg);
  218. return -1;
  219. }
  220. down_write(&ls->ls_in_recovery);
  221. set_bit(LSFL_RECOVER_LOCK, &ls->ls_flags);
  222. wake_up(&ls->ls_recover_lock_wait);
  223. while (1) {
  224. /*
  225. * We call kthread_should_stop() after set_current_state().
  226. * This is because it works correctly if kthread_stop() is
  227. * called just before set_current_state().
  228. */
  229. set_current_state(TASK_INTERRUPTIBLE);
  230. if (kthread_should_stop()) {
  231. set_current_state(TASK_RUNNING);
  232. break;
  233. }
  234. if (!test_bit(LSFL_RECOVER_WORK, &ls->ls_flags) &&
  235. !test_bit(LSFL_RECOVER_DOWN, &ls->ls_flags)) {
  236. if (kthread_should_stop())
  237. break;
  238. schedule();
  239. }
  240. set_current_state(TASK_RUNNING);
  241. if (test_and_clear_bit(LSFL_RECOVER_DOWN, &ls->ls_flags)) {
  242. down_write(&ls->ls_in_recovery);
  243. set_bit(LSFL_RECOVER_LOCK, &ls->ls_flags);
  244. wake_up(&ls->ls_recover_lock_wait);
  245. }
  246. if (test_and_clear_bit(LSFL_RECOVER_WORK, &ls->ls_flags))
  247. do_ls_recovery(ls);
  248. }
  249. if (test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags))
  250. up_write(&ls->ls_in_recovery);
  251. dlm_put_lockspace(ls);
  252. return 0;
  253. }
  254. int dlm_recoverd_start(struct dlm_ls *ls)
  255. {
  256. struct task_struct *p;
  257. int error = 0;
  258. p = kthread_run(dlm_recoverd, ls, "dlm_recoverd");
  259. if (IS_ERR(p))
  260. error = PTR_ERR(p);
  261. else
  262. ls->ls_recoverd_task = p;
  263. return error;
  264. }
  265. void dlm_recoverd_stop(struct dlm_ls *ls)
  266. {
  267. kthread_stop(ls->ls_recoverd_task);
  268. }
  269. void dlm_recoverd_suspend(struct dlm_ls *ls)
  270. {
  271. wake_up(&ls->ls_wait_general);
  272. mutex_lock(&ls->ls_recoverd_active);
  273. }
  274. void dlm_recoverd_resume(struct dlm_ls *ls)
  275. {
  276. mutex_unlock(&ls->ls_recoverd_active);
  277. }