member.c 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. /******************************************************************************
  2. *******************************************************************************
  3. **
  4. ** Copyright (C) 2005 Red Hat, Inc. All rights reserved.
  5. **
  6. ** This copyrighted material is made available to anyone wishing to use,
  7. ** modify, copy, or redistribute it subject to the terms and conditions
  8. ** of the GNU General Public License v.2.
  9. **
  10. *******************************************************************************
  11. ******************************************************************************/
  12. #include "dlm_internal.h"
  13. #include "lockspace.h"
  14. #include "member.h"
  15. #include "recoverd.h"
  16. #include "recover.h"
  17. #include "rcom.h"
  18. #include "config.h"
  19. /*
  20. * Following called by dlm_recoverd thread
  21. */
  22. static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
  23. {
  24. struct dlm_member *memb = NULL;
  25. struct list_head *tmp;
  26. struct list_head *newlist = &new->list;
  27. struct list_head *head = &ls->ls_nodes;
  28. list_for_each(tmp, head) {
  29. memb = list_entry(tmp, struct dlm_member, list);
  30. if (new->nodeid < memb->nodeid)
  31. break;
  32. }
  33. if (!memb)
  34. list_add_tail(newlist, head);
  35. else {
  36. /* FIXME: can use list macro here */
  37. newlist->prev = tmp->prev;
  38. newlist->next = tmp;
  39. tmp->prev->next = newlist;
  40. tmp->prev = newlist;
  41. }
  42. }
  43. static int dlm_add_member(struct dlm_ls *ls, int nodeid)
  44. {
  45. struct dlm_member *memb;
  46. int w;
  47. memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
  48. if (!memb)
  49. return -ENOMEM;
  50. w = dlm_node_weight(ls->ls_name, nodeid);
  51. if (w < 0)
  52. return w;
  53. memb->nodeid = nodeid;
  54. memb->weight = w;
  55. add_ordered_member(ls, memb);
  56. ls->ls_num_nodes++;
  57. return 0;
  58. }
  59. static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
  60. {
  61. list_move(&memb->list, &ls->ls_nodes_gone);
  62. ls->ls_num_nodes--;
  63. }
  64. static int dlm_is_member(struct dlm_ls *ls, int nodeid)
  65. {
  66. struct dlm_member *memb;
  67. list_for_each_entry(memb, &ls->ls_nodes, list) {
  68. if (memb->nodeid == nodeid)
  69. return 1;
  70. }
  71. return 0;
  72. }
  73. int dlm_is_removed(struct dlm_ls *ls, int nodeid)
  74. {
  75. struct dlm_member *memb;
  76. list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
  77. if (memb->nodeid == nodeid)
  78. return 1;
  79. }
  80. return 0;
  81. }
  82. static void clear_memb_list(struct list_head *head)
  83. {
  84. struct dlm_member *memb;
  85. while (!list_empty(head)) {
  86. memb = list_entry(head->next, struct dlm_member, list);
  87. list_del(&memb->list);
  88. kfree(memb);
  89. }
  90. }
  91. void dlm_clear_members(struct dlm_ls *ls)
  92. {
  93. clear_memb_list(&ls->ls_nodes);
  94. ls->ls_num_nodes = 0;
  95. }
  96. void dlm_clear_members_gone(struct dlm_ls *ls)
  97. {
  98. clear_memb_list(&ls->ls_nodes_gone);
  99. }
  100. static void make_member_array(struct dlm_ls *ls)
  101. {
  102. struct dlm_member *memb;
  103. int i, w, x = 0, total = 0, all_zero = 0, *array;
  104. kfree(ls->ls_node_array);
  105. ls->ls_node_array = NULL;
  106. list_for_each_entry(memb, &ls->ls_nodes, list) {
  107. if (memb->weight)
  108. total += memb->weight;
  109. }
  110. /* all nodes revert to weight of 1 if all have weight 0 */
  111. if (!total) {
  112. total = ls->ls_num_nodes;
  113. all_zero = 1;
  114. }
  115. ls->ls_total_weight = total;
  116. array = kmalloc(sizeof(int) * total, GFP_KERNEL);
  117. if (!array)
  118. return;
  119. list_for_each_entry(memb, &ls->ls_nodes, list) {
  120. if (!all_zero && !memb->weight)
  121. continue;
  122. if (all_zero)
  123. w = 1;
  124. else
  125. w = memb->weight;
  126. DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
  127. for (i = 0; i < w; i++)
  128. array[x++] = memb->nodeid;
  129. }
  130. ls->ls_node_array = array;
  131. }
  132. /* send a status request to all members just to establish comms connections */
  133. static int ping_members(struct dlm_ls *ls)
  134. {
  135. struct dlm_member *memb;
  136. int error = 0;
  137. list_for_each_entry(memb, &ls->ls_nodes, list) {
  138. error = dlm_recovery_stopped(ls);
  139. if (error)
  140. break;
  141. error = dlm_rcom_status(ls, memb->nodeid);
  142. if (error)
  143. break;
  144. }
  145. if (error)
  146. log_debug(ls, "ping_members aborted %d last nodeid %d",
  147. error, ls->ls_recover_nodeid);
  148. return error;
  149. }
  150. int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
  151. {
  152. struct dlm_member *memb, *safe;
  153. int i, error, found, pos = 0, neg = 0, low = -1;
  154. /* previously removed members that we've not finished removing need to
  155. count as a negative change so the "neg" recovery steps will happen */
  156. list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
  157. log_debug(ls, "prev removed member %d", memb->nodeid);
  158. neg++;
  159. }
  160. /* move departed members from ls_nodes to ls_nodes_gone */
  161. list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
  162. found = 0;
  163. for (i = 0; i < rv->node_count; i++) {
  164. if (memb->nodeid == rv->nodeids[i]) {
  165. found = 1;
  166. break;
  167. }
  168. }
  169. if (!found) {
  170. neg++;
  171. dlm_remove_member(ls, memb);
  172. log_debug(ls, "remove member %d", memb->nodeid);
  173. }
  174. }
  175. /* add new members to ls_nodes */
  176. for (i = 0; i < rv->node_count; i++) {
  177. if (dlm_is_member(ls, rv->nodeids[i]))
  178. continue;
  179. dlm_add_member(ls, rv->nodeids[i]);
  180. pos++;
  181. log_debug(ls, "add member %d", rv->nodeids[i]);
  182. }
  183. list_for_each_entry(memb, &ls->ls_nodes, list) {
  184. if (low == -1 || memb->nodeid < low)
  185. low = memb->nodeid;
  186. }
  187. ls->ls_low_nodeid = low;
  188. make_member_array(ls);
  189. dlm_set_recover_status(ls, DLM_RS_NODES);
  190. *neg_out = neg;
  191. error = ping_members(ls);
  192. if (error)
  193. goto out;
  194. error = dlm_recover_members_wait(ls);
  195. out:
  196. log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
  197. return error;
  198. }
  199. /*
  200. * Following called from lockspace.c
  201. */
  202. int dlm_ls_stop(struct dlm_ls *ls)
  203. {
  204. int new;
  205. /*
  206. * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
  207. * dlm_recovery_stopped()) and prevents any new locks from being
  208. * processed (see RUNNING, dlm_locking_stopped()).
  209. */
  210. spin_lock(&ls->ls_recover_lock);
  211. set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
  212. new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
  213. ls->ls_recover_seq++;
  214. spin_unlock(&ls->ls_recover_lock);
  215. /*
  216. * This in_recovery lock does two things:
  217. *
  218. * 1) Keeps this function from returning until all threads are out
  219. * of locking routines and locking is truely stopped.
  220. * 2) Keeps any new requests from being processed until it's unlocked
  221. * when recovery is complete.
  222. */
  223. if (new)
  224. down_write(&ls->ls_in_recovery);
  225. /*
  226. * The recoverd suspend/resume makes sure that dlm_recoverd (if
  227. * running) has noticed the clearing of RUNNING above and quit
  228. * processing the previous recovery. This will be true for all nodes
  229. * before any nodes start the new recovery.
  230. */
  231. dlm_recoverd_suspend(ls);
  232. ls->ls_recover_status = 0;
  233. dlm_recoverd_resume(ls);
  234. return 0;
  235. }
  236. int dlm_ls_start(struct dlm_ls *ls)
  237. {
  238. struct dlm_recover *rv = NULL, *rv_old;
  239. int *ids = NULL;
  240. int error, count;
  241. rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
  242. if (!rv)
  243. return -ENOMEM;
  244. error = count = dlm_nodeid_list(ls->ls_name, &ids);
  245. if (error <= 0)
  246. goto fail;
  247. spin_lock(&ls->ls_recover_lock);
  248. /* the lockspace needs to be stopped before it can be started */
  249. if (!dlm_locking_stopped(ls)) {
  250. spin_unlock(&ls->ls_recover_lock);
  251. log_error(ls, "start ignored: lockspace running");
  252. error = -EINVAL;
  253. goto fail;
  254. }
  255. rv->nodeids = ids;
  256. rv->node_count = count;
  257. rv->seq = ++ls->ls_recover_seq;
  258. rv_old = ls->ls_recover_args;
  259. ls->ls_recover_args = rv;
  260. spin_unlock(&ls->ls_recover_lock);
  261. if (rv_old) {
  262. kfree(rv_old->nodeids);
  263. kfree(rv_old);
  264. }
  265. dlm_recoverd_kick(ls);
  266. return 0;
  267. fail:
  268. kfree(rv);
  269. kfree(ids);
  270. return error;
  271. }