recoverd.c 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310
  1. /******************************************************************************
  2. *******************************************************************************
  3. **
  4. ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
  5. ** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
  6. **
  7. ** This copyrighted material is made available to anyone wishing to use,
  8. ** modify, copy, or redistribute it subject to the terms and conditions
  9. ** of the GNU General Public License v.2.
  10. **
  11. *******************************************************************************
  12. ******************************************************************************/
  13. #include "dlm_internal.h"
  14. #include "lockspace.h"
  15. #include "member.h"
  16. #include "dir.h"
  17. #include "ast.h"
  18. #include "recover.h"
  19. #include "lowcomms.h"
  20. #include "lock.h"
  21. #include "requestqueue.h"
  22. #include "recoverd.h"
  23. /* If the start for which we're re-enabling locking (seq) has been superseded
  24. by a newer stop (ls_recover_seq), we need to leave locking disabled. */
  25. static int enable_locking(struct dlm_ls *ls, uint64_t seq)
  26. {
  27. int error = -EINTR;
  28. spin_lock(&ls->ls_recover_lock);
  29. if (ls->ls_recover_seq == seq) {
  30. set_bit(LSFL_RUNNING, &ls->ls_flags);
  31. up_write(&ls->ls_in_recovery);
  32. error = 0;
  33. }
  34. spin_unlock(&ls->ls_recover_lock);
  35. return error;
  36. }
  37. static int ls_recover(struct dlm_ls *ls, struct dlm_recover *rv)
  38. {
  39. unsigned long start;
  40. int error, neg = 0;
  41. log_debug(ls, "recover %llx", (unsigned long long)rv->seq);
  42. mutex_lock(&ls->ls_recoverd_active);
  43. /*
  44. * Suspending and resuming dlm_astd ensures that no lkb's from this ls
  45. * will be processed by dlm_astd during recovery.
  46. */
  47. dlm_astd_suspend();
  48. dlm_astd_resume();
  49. /*
  50. * This list of root rsb's will be the basis of most of the recovery
  51. * routines.
  52. */
  53. dlm_create_root_list(ls);
  54. /*
  55. * Free all the tossed rsb's so we don't have to recover them.
  56. */
  57. dlm_clear_toss_list(ls);
  58. /*
  59. * Add or remove nodes from the lockspace's ls_nodes list.
  60. * Also waits for all nodes to complete dlm_recover_members.
  61. */
  62. error = dlm_recover_members(ls, rv, &neg);
  63. if (error) {
  64. log_debug(ls, "recover_members failed %d", error);
  65. goto fail;
  66. }
  67. start = jiffies;
  68. /*
  69. * Rebuild our own share of the directory by collecting from all other
  70. * nodes their master rsb names that hash to us.
  71. */
  72. error = dlm_recover_directory(ls);
  73. if (error) {
  74. log_debug(ls, "recover_directory failed %d", error);
  75. goto fail;
  76. }
  77. /*
  78. * Wait for all nodes to complete directory rebuild.
  79. */
  80. error = dlm_recover_directory_wait(ls);
  81. if (error) {
  82. log_debug(ls, "recover_directory_wait failed %d", error);
  83. goto fail;
  84. }
  85. /*
  86. * We may have outstanding operations that are waiting for a reply from
  87. * a failed node. Mark these to be resent after recovery. Unlock and
  88. * cancel ops can just be completed.
  89. */
  90. dlm_recover_waiters_pre(ls);
  91. error = dlm_recovery_stopped(ls);
  92. if (error)
  93. goto fail;
  94. if (neg || dlm_no_directory(ls)) {
  95. /*
  96. * Clear lkb's for departed nodes.
  97. */
  98. dlm_purge_locks(ls);
  99. /*
  100. * Get new master nodeid's for rsb's that were mastered on
  101. * departed nodes.
  102. */
  103. error = dlm_recover_masters(ls);
  104. if (error) {
  105. log_debug(ls, "recover_masters failed %d", error);
  106. goto fail;
  107. }
  108. /*
  109. * Send our locks on remastered rsb's to the new masters.
  110. */
  111. error = dlm_recover_locks(ls);
  112. if (error) {
  113. log_debug(ls, "recover_locks failed %d", error);
  114. goto fail;
  115. }
  116. error = dlm_recover_locks_wait(ls);
  117. if (error) {
  118. log_debug(ls, "recover_locks_wait failed %d", error);
  119. goto fail;
  120. }
  121. /*
  122. * Finalize state in master rsb's now that all locks can be
  123. * checked. This includes conversion resolution and lvb
  124. * settings.
  125. */
  126. dlm_recover_rsbs(ls);
  127. } else {
  128. /*
  129. * Other lockspace members may be going through the "neg" steps
  130. * while also adding us to the lockspace, in which case they'll
  131. * be doing the recover_locks (RS_LOCKS) barrier.
  132. */
  133. dlm_set_recover_status(ls, DLM_RS_LOCKS);
  134. error = dlm_recover_locks_wait(ls);
  135. if (error) {
  136. log_debug(ls, "recover_locks_wait failed %d", error);
  137. goto fail;
  138. }
  139. }
  140. dlm_release_root_list(ls);
  141. /*
  142. * Purge directory-related requests that are saved in requestqueue.
  143. * All dir requests from before recovery are invalid now due to the dir
  144. * rebuild and will be resent by the requesting nodes.
  145. */
  146. dlm_purge_requestqueue(ls);
  147. dlm_set_recover_status(ls, DLM_RS_DONE);
  148. error = dlm_recover_done_wait(ls);
  149. if (error) {
  150. log_debug(ls, "recover_done_wait failed %d", error);
  151. goto fail;
  152. }
  153. dlm_clear_members_gone(ls);
  154. error = enable_locking(ls, rv->seq);
  155. if (error) {
  156. log_debug(ls, "enable_locking failed %d", error);
  157. goto fail;
  158. }
  159. error = dlm_process_requestqueue(ls);
  160. if (error) {
  161. log_debug(ls, "process_requestqueue failed %d", error);
  162. goto fail;
  163. }
  164. error = dlm_recover_waiters_post(ls);
  165. if (error) {
  166. log_debug(ls, "recover_waiters_post failed %d", error);
  167. goto fail;
  168. }
  169. dlm_grant_after_purge(ls);
  170. dlm_astd_wake();
  171. log_debug(ls, "recover %llx done: %u ms",
  172. (unsigned long long)rv->seq,
  173. jiffies_to_msecs(jiffies - start));
  174. mutex_unlock(&ls->ls_recoverd_active);
  175. return 0;
  176. fail:
  177. dlm_release_root_list(ls);
  178. log_debug(ls, "recover %llx error %d",
  179. (unsigned long long)rv->seq, error);
  180. mutex_unlock(&ls->ls_recoverd_active);
  181. return error;
  182. }
  183. /* The dlm_ls_start() that created the rv we take here may already have been
  184. stopped via dlm_ls_stop(); in that case we need to leave the RECOVERY_STOP
  185. flag set. */
  186. static void do_ls_recovery(struct dlm_ls *ls)
  187. {
  188. struct dlm_recover *rv = NULL;
  189. spin_lock(&ls->ls_recover_lock);
  190. rv = ls->ls_recover_args;
  191. ls->ls_recover_args = NULL;
  192. if (rv && ls->ls_recover_seq == rv->seq)
  193. clear_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
  194. spin_unlock(&ls->ls_recover_lock);
  195. if (rv) {
  196. ls_recover(ls, rv);
  197. kfree(rv->nodeids);
  198. kfree(rv);
  199. }
  200. }
  201. static int dlm_recoverd(void *arg)
  202. {
  203. struct dlm_ls *ls;
  204. ls = dlm_find_lockspace_local(arg);
  205. if (!ls) {
  206. log_print("dlm_recoverd: no lockspace %p", arg);
  207. return -1;
  208. }
  209. while (!kthread_should_stop()) {
  210. set_current_state(TASK_INTERRUPTIBLE);
  211. if (!test_bit(LSFL_WORK, &ls->ls_flags))
  212. schedule();
  213. set_current_state(TASK_RUNNING);
  214. if (test_and_clear_bit(LSFL_WORK, &ls->ls_flags))
  215. do_ls_recovery(ls);
  216. }
  217. dlm_put_lockspace(ls);
  218. return 0;
  219. }
  220. void dlm_recoverd_kick(struct dlm_ls *ls)
  221. {
  222. set_bit(LSFL_WORK, &ls->ls_flags);
  223. wake_up_process(ls->ls_recoverd_task);
  224. }
  225. int dlm_recoverd_start(struct dlm_ls *ls)
  226. {
  227. struct task_struct *p;
  228. int error = 0;
  229. p = kthread_run(dlm_recoverd, ls, "dlm_recoverd");
  230. if (IS_ERR(p))
  231. error = PTR_ERR(p);
  232. else
  233. ls->ls_recoverd_task = p;
  234. return error;
  235. }
  236. void dlm_recoverd_stop(struct dlm_ls *ls)
  237. {
  238. kthread_stop(ls->ls_recoverd_task);
  239. }
  240. void dlm_recoverd_suspend(struct dlm_ls *ls)
  241. {
  242. wake_up(&ls->ls_wait_general);
  243. mutex_lock(&ls->ls_recoverd_active);
  244. }
  245. void dlm_recoverd_resume(struct dlm_ls *ls)
  246. {
  247. mutex_unlock(&ls->ls_recoverd_active);
  248. }