dlmrecovery.c 87 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936
  1. /* -*- mode: c; c-basic-offset: 8; -*-
  2. * vim: noexpandtab sw=8 ts=8 sts=0:
  3. *
  4. * dlmrecovery.c
  5. *
  6. * recovery stuff
  7. *
  8. * Copyright (C) 2004 Oracle. All rights reserved.
  9. *
  10. * This program is free software; you can redistribute it and/or
  11. * modify it under the terms of the GNU General Public
  12. * License as published by the Free Software Foundation; either
  13. * version 2 of the License, or (at your option) any later version.
  14. *
  15. * This program is distributed in the hope that it will be useful,
  16. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  17. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  18. * General Public License for more details.
  19. *
  20. * You should have received a copy of the GNU General Public
  21. * License along with this program; if not, write to the
  22. * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  23. * Boston, MA 021110-1307, USA.
  24. *
  25. */
  26. #include <linux/module.h>
  27. #include <linux/fs.h>
  28. #include <linux/types.h>
  29. #include <linux/slab.h>
  30. #include <linux/highmem.h>
  31. #include <linux/init.h>
  32. #include <linux/sysctl.h>
  33. #include <linux/random.h>
  34. #include <linux/blkdev.h>
  35. #include <linux/socket.h>
  36. #include <linux/inet.h>
  37. #include <linux/timer.h>
  38. #include <linux/kthread.h>
  39. #include <linux/delay.h>
  40. #include "cluster/heartbeat.h"
  41. #include "cluster/nodemanager.h"
  42. #include "cluster/tcp.h"
  43. #include "dlmapi.h"
  44. #include "dlmcommon.h"
  45. #include "dlmdomain.h"
  46. #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_RECOVERY)
  47. #include "cluster/masklog.h"
  48. static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node);
  49. static int dlm_recovery_thread(void *data);
  50. static int dlm_do_recovery(struct dlm_ctxt *dlm);
  51. static int dlm_pick_recovery_master(struct dlm_ctxt *dlm);
  52. static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node);
  53. static int dlm_init_recovery_area(struct dlm_ctxt *dlm, u8 dead_node);
  54. static int dlm_request_all_locks(struct dlm_ctxt *dlm,
  55. u8 request_from, u8 dead_node);
  56. static void dlm_destroy_recovery_area(struct dlm_ctxt *dlm, u8 dead_node);
  57. static inline int dlm_num_locks_in_lockres(struct dlm_lock_resource *res);
  58. static void dlm_init_migratable_lockres(struct dlm_migratable_lockres *mres,
  59. const char *lockname, int namelen,
  60. int total_locks, u64 cookie,
  61. u8 flags, u8 master);
  62. static int dlm_send_mig_lockres_msg(struct dlm_ctxt *dlm,
  63. struct dlm_migratable_lockres *mres,
  64. u8 send_to,
  65. struct dlm_lock_resource *res,
  66. int total_locks);
  67. static int dlm_process_recovery_data(struct dlm_ctxt *dlm,
  68. struct dlm_lock_resource *res,
  69. struct dlm_migratable_lockres *mres);
  70. static int dlm_send_finalize_reco_message(struct dlm_ctxt *dlm);
  71. static int dlm_send_all_done_msg(struct dlm_ctxt *dlm,
  72. u8 dead_node, u8 send_to);
  73. static int dlm_send_begin_reco_message(struct dlm_ctxt *dlm, u8 dead_node);
  74. static void dlm_move_reco_locks_to_list(struct dlm_ctxt *dlm,
  75. struct list_head *list, u8 dead_node);
  76. static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
  77. u8 dead_node, u8 new_master);
  78. static void dlm_reco_ast(void *astdata);
  79. static void dlm_reco_bast(void *astdata, int blocked_type);
  80. static void dlm_reco_unlock_ast(void *astdata, enum dlm_status st);
  81. static void dlm_request_all_locks_worker(struct dlm_work_item *item,
  82. void *data);
  83. static void dlm_mig_lockres_worker(struct dlm_work_item *item, void *data);
  84. static int dlm_lockres_master_requery(struct dlm_ctxt *dlm,
  85. struct dlm_lock_resource *res,
  86. u8 *real_master);
  87. static u64 dlm_get_next_mig_cookie(void);
  88. static DEFINE_SPINLOCK(dlm_reco_state_lock);
  89. static DEFINE_SPINLOCK(dlm_mig_cookie_lock);
  90. static u64 dlm_mig_cookie = 1;
  91. static u64 dlm_get_next_mig_cookie(void)
  92. {
  93. u64 c;
  94. spin_lock(&dlm_mig_cookie_lock);
  95. c = dlm_mig_cookie;
  96. if (dlm_mig_cookie == (~0ULL))
  97. dlm_mig_cookie = 1;
  98. else
  99. dlm_mig_cookie++;
  100. spin_unlock(&dlm_mig_cookie_lock);
  101. return c;
  102. }
  103. static inline void dlm_set_reco_dead_node(struct dlm_ctxt *dlm,
  104. u8 dead_node)
  105. {
  106. assert_spin_locked(&dlm->spinlock);
  107. if (dlm->reco.dead_node != dead_node)
  108. mlog(0, "%s: changing dead_node from %u to %u\n",
  109. dlm->name, dlm->reco.dead_node, dead_node);
  110. dlm->reco.dead_node = dead_node;
  111. }
  112. static inline void dlm_set_reco_master(struct dlm_ctxt *dlm,
  113. u8 master)
  114. {
  115. assert_spin_locked(&dlm->spinlock);
  116. mlog(0, "%s: changing new_master from %u to %u\n",
  117. dlm->name, dlm->reco.new_master, master);
  118. dlm->reco.new_master = master;
  119. }
  120. static inline void __dlm_reset_recovery(struct dlm_ctxt *dlm)
  121. {
  122. assert_spin_locked(&dlm->spinlock);
  123. clear_bit(dlm->reco.dead_node, dlm->recovery_map);
  124. dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
  125. dlm_set_reco_master(dlm, O2NM_INVALID_NODE_NUM);
  126. }
  127. static inline void dlm_reset_recovery(struct dlm_ctxt *dlm)
  128. {
  129. spin_lock(&dlm->spinlock);
  130. __dlm_reset_recovery(dlm);
  131. spin_unlock(&dlm->spinlock);
  132. }
  133. /* Worker function used during recovery. */
  134. void dlm_dispatch_work(struct work_struct *work)
  135. {
  136. struct dlm_ctxt *dlm =
  137. container_of(work, struct dlm_ctxt, dispatched_work);
  138. LIST_HEAD(tmp_list);
  139. struct dlm_work_item *item, *next;
  140. dlm_workfunc_t *workfunc;
  141. int tot=0;
  142. spin_lock(&dlm->work_lock);
  143. list_splice_init(&dlm->work_list, &tmp_list);
  144. spin_unlock(&dlm->work_lock);
  145. list_for_each_entry(item, &tmp_list, list) {
  146. tot++;
  147. }
  148. mlog(0, "%s: work thread has %d work items\n", dlm->name, tot);
  149. list_for_each_entry_safe(item, next, &tmp_list, list) {
  150. workfunc = item->func;
  151. list_del_init(&item->list);
  152. /* already have ref on dlm to avoid having
  153. * it disappear. just double-check. */
  154. BUG_ON(item->dlm != dlm);
  155. /* this is allowed to sleep and
  156. * call network stuff */
  157. workfunc(item, item->data);
  158. dlm_put(dlm);
  159. kfree(item);
  160. }
  161. }
  162. /*
  163. * RECOVERY THREAD
  164. */
  165. void dlm_kick_recovery_thread(struct dlm_ctxt *dlm)
  166. {
  167. /* wake the recovery thread
  168. * this will wake the reco thread in one of three places
  169. * 1) sleeping with no recovery happening
  170. * 2) sleeping with recovery mastered elsewhere
  171. * 3) recovery mastered here, waiting on reco data */
  172. wake_up(&dlm->dlm_reco_thread_wq);
  173. }
  174. /* Launch the recovery thread */
  175. int dlm_launch_recovery_thread(struct dlm_ctxt *dlm)
  176. {
  177. mlog(0, "starting dlm recovery thread...\n");
  178. dlm->dlm_reco_thread_task = kthread_run(dlm_recovery_thread, dlm,
  179. "dlm_reco_thread");
  180. if (IS_ERR(dlm->dlm_reco_thread_task)) {
  181. mlog_errno(PTR_ERR(dlm->dlm_reco_thread_task));
  182. dlm->dlm_reco_thread_task = NULL;
  183. return -EINVAL;
  184. }
  185. return 0;
  186. }
  187. void dlm_complete_recovery_thread(struct dlm_ctxt *dlm)
  188. {
  189. if (dlm->dlm_reco_thread_task) {
  190. mlog(0, "waiting for dlm recovery thread to exit\n");
  191. kthread_stop(dlm->dlm_reco_thread_task);
  192. dlm->dlm_reco_thread_task = NULL;
  193. }
  194. }
  195. /*
  196. * this is lame, but here's how recovery works...
  197. * 1) all recovery threads cluster wide will work on recovering
  198. * ONE node at a time
  199. * 2) negotiate who will take over all the locks for the dead node.
  200. * thats right... ALL the locks.
  201. * 3) once a new master is chosen, everyone scans all locks
  202. * and moves aside those mastered by the dead guy
  203. * 4) each of these locks should be locked until recovery is done
  204. * 5) the new master collects up all of secondary lock queue info
  205. * one lock at a time, forcing each node to communicate back
  206. * before continuing
  207. * 6) each secondary lock queue responds with the full known lock info
  208. * 7) once the new master has run all its locks, it sends a ALLDONE!
  209. * message to everyone
  210. * 8) upon receiving this message, the secondary queue node unlocks
  211. * and responds to the ALLDONE
  212. * 9) once the new master gets responses from everyone, he unlocks
  213. * everything and recovery for this dead node is done
  214. *10) go back to 2) while there are still dead nodes
  215. *
  216. */
  217. static void dlm_print_reco_node_status(struct dlm_ctxt *dlm)
  218. {
  219. struct dlm_reco_node_data *ndata;
  220. struct dlm_lock_resource *res;
  221. mlog(ML_NOTICE, "%s(%d): recovery info, state=%s, dead=%u, master=%u\n",
  222. dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
  223. dlm->reco.state & DLM_RECO_STATE_ACTIVE ? "ACTIVE" : "inactive",
  224. dlm->reco.dead_node, dlm->reco.new_master);
  225. list_for_each_entry(ndata, &dlm->reco.node_data, list) {
  226. char *st = "unknown";
  227. switch (ndata->state) {
  228. case DLM_RECO_NODE_DATA_INIT:
  229. st = "init";
  230. break;
  231. case DLM_RECO_NODE_DATA_REQUESTING:
  232. st = "requesting";
  233. break;
  234. case DLM_RECO_NODE_DATA_DEAD:
  235. st = "dead";
  236. break;
  237. case DLM_RECO_NODE_DATA_RECEIVING:
  238. st = "receiving";
  239. break;
  240. case DLM_RECO_NODE_DATA_REQUESTED:
  241. st = "requested";
  242. break;
  243. case DLM_RECO_NODE_DATA_DONE:
  244. st = "done";
  245. break;
  246. case DLM_RECO_NODE_DATA_FINALIZE_SENT:
  247. st = "finalize-sent";
  248. break;
  249. default:
  250. st = "bad";
  251. break;
  252. }
  253. mlog(ML_NOTICE, "%s: reco state, node %u, state=%s\n",
  254. dlm->name, ndata->node_num, st);
  255. }
  256. list_for_each_entry(res, &dlm->reco.resources, recovering) {
  257. mlog(ML_NOTICE, "%s: lockres %.*s on recovering list\n",
  258. dlm->name, res->lockname.len, res->lockname.name);
  259. }
  260. }
  261. #define DLM_RECO_THREAD_TIMEOUT_MS (5 * 1000)
  262. static int dlm_recovery_thread(void *data)
  263. {
  264. int status;
  265. struct dlm_ctxt *dlm = data;
  266. unsigned long timeout = msecs_to_jiffies(DLM_RECO_THREAD_TIMEOUT_MS);
  267. mlog(0, "dlm thread running for %s...\n", dlm->name);
  268. while (!kthread_should_stop()) {
  269. if (dlm_domain_fully_joined(dlm)) {
  270. status = dlm_do_recovery(dlm);
  271. if (status == -EAGAIN) {
  272. /* do not sleep, recheck immediately. */
  273. continue;
  274. }
  275. if (status < 0)
  276. mlog_errno(status);
  277. }
  278. wait_event_interruptible_timeout(dlm->dlm_reco_thread_wq,
  279. kthread_should_stop(),
  280. timeout);
  281. }
  282. mlog(0, "quitting DLM recovery thread\n");
  283. return 0;
  284. }
  285. /* returns true when the recovery master has contacted us */
  286. static int dlm_reco_master_ready(struct dlm_ctxt *dlm)
  287. {
  288. int ready;
  289. spin_lock(&dlm->spinlock);
  290. ready = (dlm->reco.new_master != O2NM_INVALID_NODE_NUM);
  291. spin_unlock(&dlm->spinlock);
  292. return ready;
  293. }
  294. /* returns true if node is no longer in the domain
  295. * could be dead or just not joined */
  296. int dlm_is_node_dead(struct dlm_ctxt *dlm, u8 node)
  297. {
  298. int dead;
  299. spin_lock(&dlm->spinlock);
  300. dead = !test_bit(node, dlm->domain_map);
  301. spin_unlock(&dlm->spinlock);
  302. return dead;
  303. }
  304. /* returns true if node is no longer in the domain
  305. * could be dead or just not joined */
  306. static int dlm_is_node_recovered(struct dlm_ctxt *dlm, u8 node)
  307. {
  308. int recovered;
  309. spin_lock(&dlm->spinlock);
  310. recovered = !test_bit(node, dlm->recovery_map);
  311. spin_unlock(&dlm->spinlock);
  312. return recovered;
  313. }
  314. void dlm_wait_for_node_death(struct dlm_ctxt *dlm, u8 node, int timeout)
  315. {
  316. if (dlm_is_node_dead(dlm, node))
  317. return;
  318. printk(KERN_NOTICE "o2dlm: Waiting on the death of node %u in "
  319. "domain %s\n", node, dlm->name);
  320. if (timeout)
  321. wait_event_timeout(dlm->dlm_reco_thread_wq,
  322. dlm_is_node_dead(dlm, node),
  323. msecs_to_jiffies(timeout));
  324. else
  325. wait_event(dlm->dlm_reco_thread_wq,
  326. dlm_is_node_dead(dlm, node));
  327. }
  328. void dlm_wait_for_node_recovery(struct dlm_ctxt *dlm, u8 node, int timeout)
  329. {
  330. if (dlm_is_node_recovered(dlm, node))
  331. return;
  332. printk(KERN_NOTICE "o2dlm: Waiting on the recovery of node %u in "
  333. "domain %s\n", node, dlm->name);
  334. if (timeout)
  335. wait_event_timeout(dlm->dlm_reco_thread_wq,
  336. dlm_is_node_recovered(dlm, node),
  337. msecs_to_jiffies(timeout));
  338. else
  339. wait_event(dlm->dlm_reco_thread_wq,
  340. dlm_is_node_recovered(dlm, node));
  341. }
  342. /* callers of the top-level api calls (dlmlock/dlmunlock) should
  343. * block on the dlm->reco.event when recovery is in progress.
  344. * the dlm recovery thread will set this state when it begins
  345. * recovering a dead node (as the new master or not) and clear
  346. * the state and wake as soon as all affected lock resources have
  347. * been marked with the RECOVERY flag */
  348. static int dlm_in_recovery(struct dlm_ctxt *dlm)
  349. {
  350. int in_recovery;
  351. spin_lock(&dlm->spinlock);
  352. in_recovery = !!(dlm->reco.state & DLM_RECO_STATE_ACTIVE);
  353. spin_unlock(&dlm->spinlock);
  354. return in_recovery;
  355. }
  356. void dlm_wait_for_recovery(struct dlm_ctxt *dlm)
  357. {
  358. if (dlm_in_recovery(dlm)) {
  359. mlog(0, "%s: reco thread %d in recovery: "
  360. "state=%d, master=%u, dead=%u\n",
  361. dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
  362. dlm->reco.state, dlm->reco.new_master,
  363. dlm->reco.dead_node);
  364. }
  365. wait_event(dlm->reco.event, !dlm_in_recovery(dlm));
  366. }
  367. static void dlm_begin_recovery(struct dlm_ctxt *dlm)
  368. {
  369. spin_lock(&dlm->spinlock);
  370. BUG_ON(dlm->reco.state & DLM_RECO_STATE_ACTIVE);
  371. printk(KERN_NOTICE "o2dlm: Begin recovery on domain %s for node %u\n",
  372. dlm->name, dlm->reco.dead_node);
  373. dlm->reco.state |= DLM_RECO_STATE_ACTIVE;
  374. spin_unlock(&dlm->spinlock);
  375. }
  376. static void dlm_end_recovery(struct dlm_ctxt *dlm)
  377. {
  378. spin_lock(&dlm->spinlock);
  379. BUG_ON(!(dlm->reco.state & DLM_RECO_STATE_ACTIVE));
  380. dlm->reco.state &= ~DLM_RECO_STATE_ACTIVE;
  381. spin_unlock(&dlm->spinlock);
  382. printk(KERN_NOTICE "o2dlm: End recovery on domain %s\n", dlm->name);
  383. wake_up(&dlm->reco.event);
  384. }
  385. static void dlm_print_recovery_master(struct dlm_ctxt *dlm)
  386. {
  387. printk(KERN_NOTICE "o2dlm: Node %u (%s) is the Recovery Master for the "
  388. "dead node %u in domain %s\n", dlm->reco.new_master,
  389. (dlm->node_num == dlm->reco.new_master ? "me" : "he"),
  390. dlm->reco.dead_node, dlm->name);
  391. }
  392. static int dlm_do_recovery(struct dlm_ctxt *dlm)
  393. {
  394. int status = 0;
  395. int ret;
  396. spin_lock(&dlm->spinlock);
  397. /* check to see if the new master has died */
  398. if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM &&
  399. test_bit(dlm->reco.new_master, dlm->recovery_map)) {
  400. mlog(0, "new master %u died while recovering %u!\n",
  401. dlm->reco.new_master, dlm->reco.dead_node);
  402. /* unset the new_master, leave dead_node */
  403. dlm_set_reco_master(dlm, O2NM_INVALID_NODE_NUM);
  404. }
  405. /* select a target to recover */
  406. if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
  407. int bit;
  408. bit = find_next_bit (dlm->recovery_map, O2NM_MAX_NODES, 0);
  409. if (bit >= O2NM_MAX_NODES || bit < 0)
  410. dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
  411. else
  412. dlm_set_reco_dead_node(dlm, bit);
  413. } else if (!test_bit(dlm->reco.dead_node, dlm->recovery_map)) {
  414. /* BUG? */
  415. mlog(ML_ERROR, "dead_node %u no longer in recovery map!\n",
  416. dlm->reco.dead_node);
  417. dlm_set_reco_dead_node(dlm, O2NM_INVALID_NODE_NUM);
  418. }
  419. if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
  420. // mlog(0, "nothing to recover! sleeping now!\n");
  421. spin_unlock(&dlm->spinlock);
  422. /* return to main thread loop and sleep. */
  423. return 0;
  424. }
  425. mlog(0, "%s(%d):recovery thread found node %u in the recovery map!\n",
  426. dlm->name, task_pid_nr(dlm->dlm_reco_thread_task),
  427. dlm->reco.dead_node);
  428. spin_unlock(&dlm->spinlock);
  429. /* take write barrier */
  430. /* (stops the list reshuffling thread, proxy ast handling) */
  431. dlm_begin_recovery(dlm);
  432. if (dlm->reco.new_master == dlm->node_num)
  433. goto master_here;
  434. if (dlm->reco.new_master == O2NM_INVALID_NODE_NUM) {
  435. /* choose a new master, returns 0 if this node
  436. * is the master, -EEXIST if it's another node.
  437. * this does not return until a new master is chosen
  438. * or recovery completes entirely. */
  439. ret = dlm_pick_recovery_master(dlm);
  440. if (!ret) {
  441. /* already notified everyone. go. */
  442. goto master_here;
  443. }
  444. mlog(0, "another node will master this recovery session.\n");
  445. }
  446. dlm_print_recovery_master(dlm);
  447. /* it is safe to start everything back up here
  448. * because all of the dead node's lock resources
  449. * have been marked as in-recovery */
  450. dlm_end_recovery(dlm);
  451. /* sleep out in main dlm_recovery_thread loop. */
  452. return 0;
  453. master_here:
  454. dlm_print_recovery_master(dlm);
  455. status = dlm_remaster_locks(dlm, dlm->reco.dead_node);
  456. if (status < 0) {
  457. /* we should never hit this anymore */
  458. mlog(ML_ERROR, "%s: Error %d remastering locks for node %u, "
  459. "retrying.\n", dlm->name, status, dlm->reco.dead_node);
  460. /* yield a bit to allow any final network messages
  461. * to get handled on remaining nodes */
  462. msleep(100);
  463. } else {
  464. /* success! see if any other nodes need recovery */
  465. mlog(0, "DONE mastering recovery of %s:%u here(this=%u)!\n",
  466. dlm->name, dlm->reco.dead_node, dlm->node_num);
  467. spin_lock(&dlm->spinlock);
  468. __dlm_reset_recovery(dlm);
  469. dlm->reco.state &= ~DLM_RECO_STATE_FINALIZE;
  470. spin_unlock(&dlm->spinlock);
  471. }
  472. dlm_end_recovery(dlm);
  473. /* continue and look for another dead node */
  474. return -EAGAIN;
  475. }
  476. static int dlm_remaster_locks(struct dlm_ctxt *dlm, u8 dead_node)
  477. {
  478. int status = 0;
  479. struct dlm_reco_node_data *ndata;
  480. int all_nodes_done;
  481. int destroy = 0;
  482. int pass = 0;
  483. do {
  484. /* we have become recovery master. there is no escaping
  485. * this, so just keep trying until we get it. */
  486. status = dlm_init_recovery_area(dlm, dead_node);
  487. if (status < 0) {
  488. mlog(ML_ERROR, "%s: failed to alloc recovery area, "
  489. "retrying\n", dlm->name);
  490. msleep(1000);
  491. }
  492. } while (status != 0);
  493. /* safe to access the node data list without a lock, since this
  494. * process is the only one to change the list */
  495. list_for_each_entry(ndata, &dlm->reco.node_data, list) {
  496. BUG_ON(ndata->state != DLM_RECO_NODE_DATA_INIT);
  497. ndata->state = DLM_RECO_NODE_DATA_REQUESTING;
  498. mlog(0, "%s: Requesting lock info from node %u\n", dlm->name,
  499. ndata->node_num);
  500. if (ndata->node_num == dlm->node_num) {
  501. ndata->state = DLM_RECO_NODE_DATA_DONE;
  502. continue;
  503. }
  504. do {
  505. status = dlm_request_all_locks(dlm, ndata->node_num,
  506. dead_node);
  507. if (status < 0) {
  508. mlog_errno(status);
  509. if (dlm_is_host_down(status)) {
  510. /* node died, ignore it for recovery */
  511. status = 0;
  512. ndata->state = DLM_RECO_NODE_DATA_DEAD;
  513. /* wait for the domain map to catch up
  514. * with the network state. */
  515. wait_event_timeout(dlm->dlm_reco_thread_wq,
  516. dlm_is_node_dead(dlm,
  517. ndata->node_num),
  518. msecs_to_jiffies(1000));
  519. mlog(0, "waited 1 sec for %u, "
  520. "dead? %s\n", ndata->node_num,
  521. dlm_is_node_dead(dlm, ndata->node_num) ?
  522. "yes" : "no");
  523. } else {
  524. /* -ENOMEM on the other node */
  525. mlog(0, "%s: node %u returned "
  526. "%d during recovery, retrying "
  527. "after a short wait\n",
  528. dlm->name, ndata->node_num,
  529. status);
  530. msleep(100);
  531. }
  532. }
  533. } while (status != 0);
  534. spin_lock(&dlm_reco_state_lock);
  535. switch (ndata->state) {
  536. case DLM_RECO_NODE_DATA_INIT:
  537. case DLM_RECO_NODE_DATA_FINALIZE_SENT:
  538. case DLM_RECO_NODE_DATA_REQUESTED:
  539. BUG();
  540. break;
  541. case DLM_RECO_NODE_DATA_DEAD:
  542. mlog(0, "node %u died after requesting "
  543. "recovery info for node %u\n",
  544. ndata->node_num, dead_node);
  545. /* fine. don't need this node's info.
  546. * continue without it. */
  547. break;
  548. case DLM_RECO_NODE_DATA_REQUESTING:
  549. ndata->state = DLM_RECO_NODE_DATA_REQUESTED;
  550. mlog(0, "now receiving recovery data from "
  551. "node %u for dead node %u\n",
  552. ndata->node_num, dead_node);
  553. break;
  554. case DLM_RECO_NODE_DATA_RECEIVING:
  555. mlog(0, "already receiving recovery data from "
  556. "node %u for dead node %u\n",
  557. ndata->node_num, dead_node);
  558. break;
  559. case DLM_RECO_NODE_DATA_DONE:
  560. mlog(0, "already DONE receiving recovery data "
  561. "from node %u for dead node %u\n",
  562. ndata->node_num, dead_node);
  563. break;
  564. }
  565. spin_unlock(&dlm_reco_state_lock);
  566. }
  567. mlog(0, "%s: Done requesting all lock info\n", dlm->name);
  568. /* nodes should be sending reco data now
  569. * just need to wait */
  570. while (1) {
  571. /* check all the nodes now to see if we are
  572. * done, or if anyone died */
  573. all_nodes_done = 1;
  574. spin_lock(&dlm_reco_state_lock);
  575. list_for_each_entry(ndata, &dlm->reco.node_data, list) {
  576. mlog(0, "checking recovery state of node %u\n",
  577. ndata->node_num);
  578. switch (ndata->state) {
  579. case DLM_RECO_NODE_DATA_INIT:
  580. case DLM_RECO_NODE_DATA_REQUESTING:
  581. mlog(ML_ERROR, "bad ndata state for "
  582. "node %u: state=%d\n",
  583. ndata->node_num, ndata->state);
  584. BUG();
  585. break;
  586. case DLM_RECO_NODE_DATA_DEAD:
  587. mlog(0, "node %u died after "
  588. "requesting recovery info for "
  589. "node %u\n", ndata->node_num,
  590. dead_node);
  591. break;
  592. case DLM_RECO_NODE_DATA_RECEIVING:
  593. case DLM_RECO_NODE_DATA_REQUESTED:
  594. mlog(0, "%s: node %u still in state %s\n",
  595. dlm->name, ndata->node_num,
  596. ndata->state==DLM_RECO_NODE_DATA_RECEIVING ?
  597. "receiving" : "requested");
  598. all_nodes_done = 0;
  599. break;
  600. case DLM_RECO_NODE_DATA_DONE:
  601. mlog(0, "%s: node %u state is done\n",
  602. dlm->name, ndata->node_num);
  603. break;
  604. case DLM_RECO_NODE_DATA_FINALIZE_SENT:
  605. mlog(0, "%s: node %u state is finalize\n",
  606. dlm->name, ndata->node_num);
  607. break;
  608. }
  609. }
  610. spin_unlock(&dlm_reco_state_lock);
  611. mlog(0, "pass #%d, all_nodes_done?: %s\n", ++pass,
  612. all_nodes_done?"yes":"no");
  613. if (all_nodes_done) {
  614. int ret;
  615. /* Set this flag on recovery master to avoid
  616. * a new recovery for another dead node start
  617. * before the recovery is not done. That may
  618. * cause recovery hung.*/
  619. spin_lock(&dlm->spinlock);
  620. dlm->reco.state |= DLM_RECO_STATE_FINALIZE;
  621. spin_unlock(&dlm->spinlock);
  622. /* all nodes are now in DLM_RECO_NODE_DATA_DONE state
  623. * just send a finalize message to everyone and
  624. * clean up */
  625. mlog(0, "all nodes are done! send finalize\n");
  626. ret = dlm_send_finalize_reco_message(dlm);
  627. if (ret < 0)
  628. mlog_errno(ret);
  629. spin_lock(&dlm->spinlock);
  630. dlm_finish_local_lockres_recovery(dlm, dead_node,
  631. dlm->node_num);
  632. spin_unlock(&dlm->spinlock);
  633. mlog(0, "should be done with recovery!\n");
  634. mlog(0, "finishing recovery of %s at %lu, "
  635. "dead=%u, this=%u, new=%u\n", dlm->name,
  636. jiffies, dlm->reco.dead_node,
  637. dlm->node_num, dlm->reco.new_master);
  638. destroy = 1;
  639. status = 0;
  640. /* rescan everything marked dirty along the way */
  641. dlm_kick_thread(dlm, NULL);
  642. break;
  643. }
  644. /* wait to be signalled, with periodic timeout
  645. * to check for node death */
  646. wait_event_interruptible_timeout(dlm->dlm_reco_thread_wq,
  647. kthread_should_stop(),
  648. msecs_to_jiffies(DLM_RECO_THREAD_TIMEOUT_MS));
  649. }
  650. if (destroy)
  651. dlm_destroy_recovery_area(dlm, dead_node);
  652. return status;
  653. }
  654. static int dlm_init_recovery_area(struct dlm_ctxt *dlm, u8 dead_node)
  655. {
  656. int num=0;
  657. struct dlm_reco_node_data *ndata;
  658. spin_lock(&dlm->spinlock);
  659. memcpy(dlm->reco.node_map, dlm->domain_map, sizeof(dlm->domain_map));
  660. /* nodes can only be removed (by dying) after dropping
  661. * this lock, and death will be trapped later, so this should do */
  662. spin_unlock(&dlm->spinlock);
  663. while (1) {
  664. num = find_next_bit (dlm->reco.node_map, O2NM_MAX_NODES, num);
  665. if (num >= O2NM_MAX_NODES) {
  666. break;
  667. }
  668. BUG_ON(num == dead_node);
  669. ndata = kzalloc(sizeof(*ndata), GFP_NOFS);
  670. if (!ndata) {
  671. dlm_destroy_recovery_area(dlm, dead_node);
  672. return -ENOMEM;
  673. }
  674. ndata->node_num = num;
  675. ndata->state = DLM_RECO_NODE_DATA_INIT;
  676. spin_lock(&dlm_reco_state_lock);
  677. list_add_tail(&ndata->list, &dlm->reco.node_data);
  678. spin_unlock(&dlm_reco_state_lock);
  679. num++;
  680. }
  681. return 0;
  682. }
  683. static void dlm_destroy_recovery_area(struct dlm_ctxt *dlm, u8 dead_node)
  684. {
  685. struct dlm_reco_node_data *ndata, *next;
  686. LIST_HEAD(tmplist);
  687. spin_lock(&dlm_reco_state_lock);
  688. list_splice_init(&dlm->reco.node_data, &tmplist);
  689. spin_unlock(&dlm_reco_state_lock);
  690. list_for_each_entry_safe(ndata, next, &tmplist, list) {
  691. list_del_init(&ndata->list);
  692. kfree(ndata);
  693. }
  694. }
  695. static int dlm_request_all_locks(struct dlm_ctxt *dlm, u8 request_from,
  696. u8 dead_node)
  697. {
  698. struct dlm_lock_request lr;
  699. int ret;
  700. int status;
  701. mlog(0, "\n");
  702. mlog(0, "dlm_request_all_locks: dead node is %u, sending request "
  703. "to %u\n", dead_node, request_from);
  704. memset(&lr, 0, sizeof(lr));
  705. lr.node_idx = dlm->node_num;
  706. lr.dead_node = dead_node;
  707. // send message
  708. ret = o2net_send_message(DLM_LOCK_REQUEST_MSG, dlm->key,
  709. &lr, sizeof(lr), request_from, &status);
  710. /* negative status is handled by caller */
  711. if (ret < 0)
  712. mlog(ML_ERROR, "%s: Error %d send LOCK_REQUEST to node %u "
  713. "to recover dead node %u\n", dlm->name, ret,
  714. request_from, dead_node);
  715. else
  716. ret = status;
  717. // return from here, then
  718. // sleep until all received or error
  719. return ret;
  720. }
  721. int dlm_request_all_locks_handler(struct o2net_msg *msg, u32 len, void *data,
  722. void **ret_data)
  723. {
  724. struct dlm_ctxt *dlm = data;
  725. struct dlm_lock_request *lr = (struct dlm_lock_request *)msg->buf;
  726. char *buf = NULL;
  727. struct dlm_work_item *item = NULL;
  728. if (!dlm_grab(dlm))
  729. return -EINVAL;
  730. if (lr->dead_node != dlm->reco.dead_node) {
  731. mlog(ML_ERROR, "%s: node %u sent dead_node=%u, but local "
  732. "dead_node is %u\n", dlm->name, lr->node_idx,
  733. lr->dead_node, dlm->reco.dead_node);
  734. dlm_print_reco_node_status(dlm);
  735. /* this is a hack */
  736. dlm_put(dlm);
  737. return -ENOMEM;
  738. }
  739. BUG_ON(lr->dead_node != dlm->reco.dead_node);
  740. item = kzalloc(sizeof(*item), GFP_NOFS);
  741. if (!item) {
  742. dlm_put(dlm);
  743. return -ENOMEM;
  744. }
  745. /* this will get freed by dlm_request_all_locks_worker */
  746. buf = (char *) __get_free_page(GFP_NOFS);
  747. if (!buf) {
  748. kfree(item);
  749. dlm_put(dlm);
  750. return -ENOMEM;
  751. }
  752. /* queue up work for dlm_request_all_locks_worker */
  753. dlm_grab(dlm); /* get an extra ref for the work item */
  754. dlm_init_work_item(dlm, item, dlm_request_all_locks_worker, buf);
  755. item->u.ral.reco_master = lr->node_idx;
  756. item->u.ral.dead_node = lr->dead_node;
  757. spin_lock(&dlm->work_lock);
  758. list_add_tail(&item->list, &dlm->work_list);
  759. spin_unlock(&dlm->work_lock);
  760. queue_work(dlm->dlm_worker, &dlm->dispatched_work);
  761. dlm_put(dlm);
  762. return 0;
  763. }
  764. static void dlm_request_all_locks_worker(struct dlm_work_item *item, void *data)
  765. {
  766. struct dlm_migratable_lockres *mres;
  767. struct dlm_lock_resource *res;
  768. struct dlm_ctxt *dlm;
  769. LIST_HEAD(resources);
  770. int ret;
  771. u8 dead_node, reco_master;
  772. int skip_all_done = 0;
  773. dlm = item->dlm;
  774. dead_node = item->u.ral.dead_node;
  775. reco_master = item->u.ral.reco_master;
  776. mres = (struct dlm_migratable_lockres *)data;
  777. mlog(0, "%s: recovery worker started, dead=%u, master=%u\n",
  778. dlm->name, dead_node, reco_master);
  779. if (dead_node != dlm->reco.dead_node ||
  780. reco_master != dlm->reco.new_master) {
  781. /* worker could have been created before the recovery master
  782. * died. if so, do not continue, but do not error. */
  783. if (dlm->reco.new_master == O2NM_INVALID_NODE_NUM) {
  784. mlog(ML_NOTICE, "%s: will not send recovery state, "
  785. "recovery master %u died, thread=(dead=%u,mas=%u)"
  786. " current=(dead=%u,mas=%u)\n", dlm->name,
  787. reco_master, dead_node, reco_master,
  788. dlm->reco.dead_node, dlm->reco.new_master);
  789. } else {
  790. mlog(ML_NOTICE, "%s: reco state invalid: reco(dead=%u, "
  791. "master=%u), request(dead=%u, master=%u)\n",
  792. dlm->name, dlm->reco.dead_node,
  793. dlm->reco.new_master, dead_node, reco_master);
  794. }
  795. goto leave;
  796. }
  797. /* lock resources should have already been moved to the
  798. * dlm->reco.resources list. now move items from that list
  799. * to a temp list if the dead owner matches. note that the
  800. * whole cluster recovers only one node at a time, so we
  801. * can safely move UNKNOWN lock resources for each recovery
  802. * session. */
  803. dlm_move_reco_locks_to_list(dlm, &resources, dead_node);
  804. /* now we can begin blasting lockreses without the dlm lock */
  805. /* any errors returned will be due to the new_master dying,
  806. * the dlm_reco_thread should detect this */
  807. list_for_each_entry(res, &resources, recovering) {
  808. ret = dlm_send_one_lockres(dlm, res, mres, reco_master,
  809. DLM_MRES_RECOVERY);
  810. if (ret < 0) {
  811. mlog(ML_ERROR, "%s: node %u went down while sending "
  812. "recovery state for dead node %u, ret=%d\n", dlm->name,
  813. reco_master, dead_node, ret);
  814. skip_all_done = 1;
  815. break;
  816. }
  817. }
  818. /* move the resources back to the list */
  819. spin_lock(&dlm->spinlock);
  820. list_splice_init(&resources, &dlm->reco.resources);
  821. spin_unlock(&dlm->spinlock);
  822. if (!skip_all_done) {
  823. ret = dlm_send_all_done_msg(dlm, dead_node, reco_master);
  824. if (ret < 0) {
  825. mlog(ML_ERROR, "%s: node %u went down while sending "
  826. "recovery all-done for dead node %u, ret=%d\n",
  827. dlm->name, reco_master, dead_node, ret);
  828. }
  829. }
  830. leave:
  831. free_page((unsigned long)data);
  832. }
  833. static int dlm_send_all_done_msg(struct dlm_ctxt *dlm, u8 dead_node, u8 send_to)
  834. {
  835. int ret, tmpret;
  836. struct dlm_reco_data_done done_msg;
  837. memset(&done_msg, 0, sizeof(done_msg));
  838. done_msg.node_idx = dlm->node_num;
  839. done_msg.dead_node = dead_node;
  840. mlog(0, "sending DATA DONE message to %u, "
  841. "my node=%u, dead node=%u\n", send_to, done_msg.node_idx,
  842. done_msg.dead_node);
  843. ret = o2net_send_message(DLM_RECO_DATA_DONE_MSG, dlm->key, &done_msg,
  844. sizeof(done_msg), send_to, &tmpret);
  845. if (ret < 0) {
  846. mlog(ML_ERROR, "%s: Error %d send RECO_DATA_DONE to node %u "
  847. "to recover dead node %u\n", dlm->name, ret, send_to,
  848. dead_node);
  849. if (!dlm_is_host_down(ret)) {
  850. BUG();
  851. }
  852. } else
  853. ret = tmpret;
  854. return ret;
  855. }
  856. int dlm_reco_data_done_handler(struct o2net_msg *msg, u32 len, void *data,
  857. void **ret_data)
  858. {
  859. struct dlm_ctxt *dlm = data;
  860. struct dlm_reco_data_done *done = (struct dlm_reco_data_done *)msg->buf;
  861. struct dlm_reco_node_data *ndata = NULL;
  862. int ret = -EINVAL;
  863. if (!dlm_grab(dlm))
  864. return -EINVAL;
  865. mlog(0, "got DATA DONE: dead_node=%u, reco.dead_node=%u, "
  866. "node_idx=%u, this node=%u\n", done->dead_node,
  867. dlm->reco.dead_node, done->node_idx, dlm->node_num);
  868. mlog_bug_on_msg((done->dead_node != dlm->reco.dead_node),
  869. "Got DATA DONE: dead_node=%u, reco.dead_node=%u, "
  870. "node_idx=%u, this node=%u\n", done->dead_node,
  871. dlm->reco.dead_node, done->node_idx, dlm->node_num);
  872. spin_lock(&dlm_reco_state_lock);
  873. list_for_each_entry(ndata, &dlm->reco.node_data, list) {
  874. if (ndata->node_num != done->node_idx)
  875. continue;
  876. switch (ndata->state) {
  877. /* should have moved beyond INIT but not to FINALIZE yet */
  878. case DLM_RECO_NODE_DATA_INIT:
  879. case DLM_RECO_NODE_DATA_DEAD:
  880. case DLM_RECO_NODE_DATA_FINALIZE_SENT:
  881. mlog(ML_ERROR, "bad ndata state for node %u:"
  882. " state=%d\n", ndata->node_num,
  883. ndata->state);
  884. BUG();
  885. break;
  886. /* these states are possible at this point, anywhere along
  887. * the line of recovery */
  888. case DLM_RECO_NODE_DATA_DONE:
  889. case DLM_RECO_NODE_DATA_RECEIVING:
  890. case DLM_RECO_NODE_DATA_REQUESTED:
  891. case DLM_RECO_NODE_DATA_REQUESTING:
  892. mlog(0, "node %u is DONE sending "
  893. "recovery data!\n",
  894. ndata->node_num);
  895. ndata->state = DLM_RECO_NODE_DATA_DONE;
  896. ret = 0;
  897. break;
  898. }
  899. }
  900. spin_unlock(&dlm_reco_state_lock);
  901. /* wake the recovery thread, some node is done */
  902. if (!ret)
  903. dlm_kick_recovery_thread(dlm);
  904. if (ret < 0)
  905. mlog(ML_ERROR, "failed to find recovery node data for node "
  906. "%u\n", done->node_idx);
  907. dlm_put(dlm);
  908. mlog(0, "leaving reco data done handler, ret=%d\n", ret);
  909. return ret;
  910. }
  911. static void dlm_move_reco_locks_to_list(struct dlm_ctxt *dlm,
  912. struct list_head *list,
  913. u8 dead_node)
  914. {
  915. struct dlm_lock_resource *res, *next;
  916. struct dlm_lock *lock;
  917. spin_lock(&dlm->spinlock);
  918. list_for_each_entry_safe(res, next, &dlm->reco.resources, recovering) {
  919. /* always prune any $RECOVERY entries for dead nodes,
  920. * otherwise hangs can occur during later recovery */
  921. if (dlm_is_recovery_lock(res->lockname.name,
  922. res->lockname.len)) {
  923. spin_lock(&res->spinlock);
  924. list_for_each_entry(lock, &res->granted, list) {
  925. if (lock->ml.node == dead_node) {
  926. mlog(0, "AHA! there was "
  927. "a $RECOVERY lock for dead "
  928. "node %u (%s)!\n",
  929. dead_node, dlm->name);
  930. list_del_init(&lock->list);
  931. dlm_lock_put(lock);
  932. /* Can't schedule DLM_UNLOCK_FREE_LOCK
  933. * - do manually */
  934. dlm_lock_put(lock);
  935. break;
  936. }
  937. }
  938. spin_unlock(&res->spinlock);
  939. continue;
  940. }
  941. if (res->owner == dead_node) {
  942. mlog(0, "found lockres owned by dead node while "
  943. "doing recovery for node %u. sending it.\n",
  944. dead_node);
  945. list_move_tail(&res->recovering, list);
  946. } else if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
  947. mlog(0, "found UNKNOWN owner while doing recovery "
  948. "for node %u. sending it.\n", dead_node);
  949. list_move_tail(&res->recovering, list);
  950. }
  951. }
  952. spin_unlock(&dlm->spinlock);
  953. }
  954. static inline int dlm_num_locks_in_lockres(struct dlm_lock_resource *res)
  955. {
  956. int total_locks = 0;
  957. struct list_head *iter, *queue = &res->granted;
  958. int i;
  959. for (i=0; i<3; i++) {
  960. list_for_each(iter, queue)
  961. total_locks++;
  962. queue++;
  963. }
  964. return total_locks;
  965. }
  966. static int dlm_send_mig_lockres_msg(struct dlm_ctxt *dlm,
  967. struct dlm_migratable_lockres *mres,
  968. u8 send_to,
  969. struct dlm_lock_resource *res,
  970. int total_locks)
  971. {
  972. u64 mig_cookie = be64_to_cpu(mres->mig_cookie);
  973. int mres_total_locks = be32_to_cpu(mres->total_locks);
  974. int sz, ret = 0, status = 0;
  975. u8 orig_flags = mres->flags,
  976. orig_master = mres->master;
  977. BUG_ON(mres->num_locks > DLM_MAX_MIGRATABLE_LOCKS);
  978. if (!mres->num_locks)
  979. return 0;
  980. sz = sizeof(struct dlm_migratable_lockres) +
  981. (mres->num_locks * sizeof(struct dlm_migratable_lock));
  982. /* add an all-done flag if we reached the last lock */
  983. orig_flags = mres->flags;
  984. BUG_ON(total_locks > mres_total_locks);
  985. if (total_locks == mres_total_locks)
  986. mres->flags |= DLM_MRES_ALL_DONE;
  987. mlog(0, "%s:%.*s: sending mig lockres (%s) to %u\n",
  988. dlm->name, res->lockname.len, res->lockname.name,
  989. orig_flags & DLM_MRES_MIGRATION ? "migration" : "recovery",
  990. send_to);
  991. /* send it */
  992. ret = o2net_send_message(DLM_MIG_LOCKRES_MSG, dlm->key, mres,
  993. sz, send_to, &status);
  994. if (ret < 0) {
  995. /* XXX: negative status is not handled.
  996. * this will end up killing this node. */
  997. mlog(ML_ERROR, "%s: res %.*s, Error %d send MIG_LOCKRES to "
  998. "node %u (%s)\n", dlm->name, mres->lockname_len,
  999. mres->lockname, ret, send_to,
  1000. (orig_flags & DLM_MRES_MIGRATION ?
  1001. "migration" : "recovery"));
  1002. } else {
  1003. /* might get an -ENOMEM back here */
  1004. ret = status;
  1005. if (ret < 0) {
  1006. mlog_errno(ret);
  1007. if (ret == -EFAULT) {
  1008. mlog(ML_ERROR, "node %u told me to kill "
  1009. "myself!\n", send_to);
  1010. BUG();
  1011. }
  1012. }
  1013. }
  1014. /* zero and reinit the message buffer */
  1015. dlm_init_migratable_lockres(mres, res->lockname.name,
  1016. res->lockname.len, mres_total_locks,
  1017. mig_cookie, orig_flags, orig_master);
  1018. return ret;
  1019. }
  1020. static void dlm_init_migratable_lockres(struct dlm_migratable_lockres *mres,
  1021. const char *lockname, int namelen,
  1022. int total_locks, u64 cookie,
  1023. u8 flags, u8 master)
  1024. {
  1025. /* mres here is one full page */
  1026. clear_page(mres);
  1027. mres->lockname_len = namelen;
  1028. memcpy(mres->lockname, lockname, namelen);
  1029. mres->num_locks = 0;
  1030. mres->total_locks = cpu_to_be32(total_locks);
  1031. mres->mig_cookie = cpu_to_be64(cookie);
  1032. mres->flags = flags;
  1033. mres->master = master;
  1034. }
  1035. static void dlm_prepare_lvb_for_migration(struct dlm_lock *lock,
  1036. struct dlm_migratable_lockres *mres,
  1037. int queue)
  1038. {
  1039. if (!lock->lksb)
  1040. return;
  1041. /* Ignore lvb in all locks in the blocked list */
  1042. if (queue == DLM_BLOCKED_LIST)
  1043. return;
  1044. /* Only consider lvbs in locks with granted EX or PR lock levels */
  1045. if (lock->ml.type != LKM_EXMODE && lock->ml.type != LKM_PRMODE)
  1046. return;
  1047. if (dlm_lvb_is_empty(mres->lvb)) {
  1048. memcpy(mres->lvb, lock->lksb->lvb, DLM_LVB_LEN);
  1049. return;
  1050. }
  1051. /* Ensure the lvb copied for migration matches in other valid locks */
  1052. if (!memcmp(mres->lvb, lock->lksb->lvb, DLM_LVB_LEN))
  1053. return;
  1054. mlog(ML_ERROR, "Mismatched lvb in lock cookie=%u:%llu, name=%.*s, "
  1055. "node=%u\n",
  1056. dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
  1057. dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
  1058. lock->lockres->lockname.len, lock->lockres->lockname.name,
  1059. lock->ml.node);
  1060. dlm_print_one_lock_resource(lock->lockres);
  1061. BUG();
  1062. }
  1063. /* returns 1 if this lock fills the network structure,
  1064. * 0 otherwise */
  1065. static int dlm_add_lock_to_array(struct dlm_lock *lock,
  1066. struct dlm_migratable_lockres *mres, int queue)
  1067. {
  1068. struct dlm_migratable_lock *ml;
  1069. int lock_num = mres->num_locks;
  1070. ml = &(mres->ml[lock_num]);
  1071. ml->cookie = lock->ml.cookie;
  1072. ml->type = lock->ml.type;
  1073. ml->convert_type = lock->ml.convert_type;
  1074. ml->highest_blocked = lock->ml.highest_blocked;
  1075. ml->list = queue;
  1076. if (lock->lksb) {
  1077. ml->flags = lock->lksb->flags;
  1078. dlm_prepare_lvb_for_migration(lock, mres, queue);
  1079. }
  1080. ml->node = lock->ml.node;
  1081. mres->num_locks++;
  1082. /* we reached the max, send this network message */
  1083. if (mres->num_locks == DLM_MAX_MIGRATABLE_LOCKS)
  1084. return 1;
  1085. return 0;
  1086. }
  1087. static void dlm_add_dummy_lock(struct dlm_ctxt *dlm,
  1088. struct dlm_migratable_lockres *mres)
  1089. {
  1090. struct dlm_lock dummy;
  1091. memset(&dummy, 0, sizeof(dummy));
  1092. dummy.ml.cookie = 0;
  1093. dummy.ml.type = LKM_IVMODE;
  1094. dummy.ml.convert_type = LKM_IVMODE;
  1095. dummy.ml.highest_blocked = LKM_IVMODE;
  1096. dummy.lksb = NULL;
  1097. dummy.ml.node = dlm->node_num;
  1098. dlm_add_lock_to_array(&dummy, mres, DLM_BLOCKED_LIST);
  1099. }
  1100. static inline int dlm_is_dummy_lock(struct dlm_ctxt *dlm,
  1101. struct dlm_migratable_lock *ml,
  1102. u8 *nodenum)
  1103. {
  1104. if (unlikely(ml->cookie == 0 &&
  1105. ml->type == LKM_IVMODE &&
  1106. ml->convert_type == LKM_IVMODE &&
  1107. ml->highest_blocked == LKM_IVMODE &&
  1108. ml->list == DLM_BLOCKED_LIST)) {
  1109. *nodenum = ml->node;
  1110. return 1;
  1111. }
  1112. return 0;
  1113. }
  1114. int dlm_send_one_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
  1115. struct dlm_migratable_lockres *mres,
  1116. u8 send_to, u8 flags)
  1117. {
  1118. struct list_head *queue;
  1119. int total_locks, i;
  1120. u64 mig_cookie = 0;
  1121. struct dlm_lock *lock;
  1122. int ret = 0;
  1123. BUG_ON(!(flags & (DLM_MRES_RECOVERY|DLM_MRES_MIGRATION)));
  1124. mlog(0, "sending to %u\n", send_to);
  1125. total_locks = dlm_num_locks_in_lockres(res);
  1126. if (total_locks > DLM_MAX_MIGRATABLE_LOCKS) {
  1127. /* rare, but possible */
  1128. mlog(0, "argh. lockres has %d locks. this will "
  1129. "require more than one network packet to "
  1130. "migrate\n", total_locks);
  1131. mig_cookie = dlm_get_next_mig_cookie();
  1132. }
  1133. dlm_init_migratable_lockres(mres, res->lockname.name,
  1134. res->lockname.len, total_locks,
  1135. mig_cookie, flags, res->owner);
  1136. total_locks = 0;
  1137. for (i=DLM_GRANTED_LIST; i<=DLM_BLOCKED_LIST; i++) {
  1138. queue = dlm_list_idx_to_ptr(res, i);
  1139. list_for_each_entry(lock, queue, list) {
  1140. /* add another lock. */
  1141. total_locks++;
  1142. if (!dlm_add_lock_to_array(lock, mres, i))
  1143. continue;
  1144. /* this filled the lock message,
  1145. * we must send it immediately. */
  1146. ret = dlm_send_mig_lockres_msg(dlm, mres, send_to,
  1147. res, total_locks);
  1148. if (ret < 0)
  1149. goto error;
  1150. }
  1151. }
  1152. if (total_locks == 0) {
  1153. /* send a dummy lock to indicate a mastery reference only */
  1154. mlog(0, "%s:%.*s: sending dummy lock to %u, %s\n",
  1155. dlm->name, res->lockname.len, res->lockname.name,
  1156. send_to, flags & DLM_MRES_RECOVERY ? "recovery" :
  1157. "migration");
  1158. dlm_add_dummy_lock(dlm, mres);
  1159. }
  1160. /* flush any remaining locks */
  1161. ret = dlm_send_mig_lockres_msg(dlm, mres, send_to, res, total_locks);
  1162. if (ret < 0)
  1163. goto error;
  1164. return ret;
  1165. error:
  1166. mlog(ML_ERROR, "%s: dlm_send_mig_lockres_msg returned %d\n",
  1167. dlm->name, ret);
  1168. if (!dlm_is_host_down(ret))
  1169. BUG();
  1170. mlog(0, "%s: node %u went down while sending %s "
  1171. "lockres %.*s\n", dlm->name, send_to,
  1172. flags & DLM_MRES_RECOVERY ? "recovery" : "migration",
  1173. res->lockname.len, res->lockname.name);
  1174. return ret;
  1175. }
  1176. /*
  1177. * this message will contain no more than one page worth of
  1178. * recovery data, and it will work on only one lockres.
  1179. * there may be many locks in this page, and we may need to wait
  1180. * for additional packets to complete all the locks (rare, but
  1181. * possible).
  1182. */
  1183. /*
  1184. * NOTE: the allocation error cases here are scary
  1185. * we really cannot afford to fail an alloc in recovery
  1186. * do we spin? returning an error only delays the problem really
  1187. */
  1188. int dlm_mig_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
  1189. void **ret_data)
  1190. {
  1191. struct dlm_ctxt *dlm = data;
  1192. struct dlm_migratable_lockres *mres =
  1193. (struct dlm_migratable_lockres *)msg->buf;
  1194. int ret = 0;
  1195. u8 real_master;
  1196. u8 extra_refs = 0;
  1197. char *buf = NULL;
  1198. struct dlm_work_item *item = NULL;
  1199. struct dlm_lock_resource *res = NULL;
  1200. if (!dlm_grab(dlm))
  1201. return -EINVAL;
  1202. BUG_ON(!(mres->flags & (DLM_MRES_RECOVERY|DLM_MRES_MIGRATION)));
  1203. real_master = mres->master;
  1204. if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
  1205. /* cannot migrate a lockres with no master */
  1206. BUG_ON(!(mres->flags & DLM_MRES_RECOVERY));
  1207. }
  1208. mlog(0, "%s message received from node %u\n",
  1209. (mres->flags & DLM_MRES_RECOVERY) ?
  1210. "recovery" : "migration", mres->master);
  1211. if (mres->flags & DLM_MRES_ALL_DONE)
  1212. mlog(0, "all done flag. all lockres data received!\n");
  1213. ret = -ENOMEM;
  1214. buf = kmalloc(be16_to_cpu(msg->data_len), GFP_NOFS);
  1215. item = kzalloc(sizeof(*item), GFP_NOFS);
  1216. if (!buf || !item)
  1217. goto leave;
  1218. /* lookup the lock to see if we have a secondary queue for this
  1219. * already... just add the locks in and this will have its owner
  1220. * and RECOVERY flag changed when it completes. */
  1221. res = dlm_lookup_lockres(dlm, mres->lockname, mres->lockname_len);
  1222. if (res) {
  1223. /* this will get a ref on res */
  1224. /* mark it as recovering/migrating and hash it */
  1225. spin_lock(&res->spinlock);
  1226. if (mres->flags & DLM_MRES_RECOVERY) {
  1227. res->state |= DLM_LOCK_RES_RECOVERING;
  1228. } else {
  1229. if (res->state & DLM_LOCK_RES_MIGRATING) {
  1230. /* this is at least the second
  1231. * lockres message */
  1232. mlog(0, "lock %.*s is already migrating\n",
  1233. mres->lockname_len,
  1234. mres->lockname);
  1235. } else if (res->state & DLM_LOCK_RES_RECOVERING) {
  1236. /* caller should BUG */
  1237. mlog(ML_ERROR, "node is attempting to migrate "
  1238. "lock %.*s, but marked as recovering!\n",
  1239. mres->lockname_len, mres->lockname);
  1240. ret = -EFAULT;
  1241. spin_unlock(&res->spinlock);
  1242. dlm_lockres_put(res);
  1243. goto leave;
  1244. }
  1245. res->state |= DLM_LOCK_RES_MIGRATING;
  1246. }
  1247. spin_unlock(&res->spinlock);
  1248. } else {
  1249. /* need to allocate, just like if it was
  1250. * mastered here normally */
  1251. res = dlm_new_lockres(dlm, mres->lockname, mres->lockname_len);
  1252. if (!res)
  1253. goto leave;
  1254. /* to match the ref that we would have gotten if
  1255. * dlm_lookup_lockres had succeeded */
  1256. dlm_lockres_get(res);
  1257. /* mark it as recovering/migrating and hash it */
  1258. if (mres->flags & DLM_MRES_RECOVERY)
  1259. res->state |= DLM_LOCK_RES_RECOVERING;
  1260. else
  1261. res->state |= DLM_LOCK_RES_MIGRATING;
  1262. spin_lock(&dlm->spinlock);
  1263. __dlm_insert_lockres(dlm, res);
  1264. spin_unlock(&dlm->spinlock);
  1265. /* Add an extra ref for this lock-less lockres lest the
  1266. * dlm_thread purges it before we get the chance to add
  1267. * locks to it */
  1268. dlm_lockres_get(res);
  1269. /* There are three refs that need to be put.
  1270. * 1. Taken above.
  1271. * 2. kref_init in dlm_new_lockres()->dlm_init_lockres().
  1272. * 3. dlm_lookup_lockres()
  1273. * The first one is handled at the end of this function. The
  1274. * other two are handled in the worker thread after locks have
  1275. * been attached. Yes, we don't wait for purge time to match
  1276. * kref_init. The lockres will still have atleast one ref
  1277. * added because it is in the hash __dlm_insert_lockres() */
  1278. extra_refs++;
  1279. /* now that the new lockres is inserted,
  1280. * make it usable by other processes */
  1281. spin_lock(&res->spinlock);
  1282. res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
  1283. spin_unlock(&res->spinlock);
  1284. wake_up(&res->wq);
  1285. }
  1286. /* at this point we have allocated everything we need,
  1287. * and we have a hashed lockres with an extra ref and
  1288. * the proper res->state flags. */
  1289. ret = 0;
  1290. spin_lock(&res->spinlock);
  1291. /* drop this either when master requery finds a different master
  1292. * or when a lock is added by the recovery worker */
  1293. dlm_lockres_grab_inflight_ref(dlm, res);
  1294. if (mres->master == DLM_LOCK_RES_OWNER_UNKNOWN) {
  1295. /* migration cannot have an unknown master */
  1296. BUG_ON(!(mres->flags & DLM_MRES_RECOVERY));
  1297. mlog(0, "recovery has passed me a lockres with an "
  1298. "unknown owner.. will need to requery: "
  1299. "%.*s\n", mres->lockname_len, mres->lockname);
  1300. } else {
  1301. /* take a reference now to pin the lockres, drop it
  1302. * when locks are added in the worker */
  1303. dlm_change_lockres_owner(dlm, res, dlm->node_num);
  1304. }
  1305. spin_unlock(&res->spinlock);
  1306. /* queue up work for dlm_mig_lockres_worker */
  1307. dlm_grab(dlm); /* get an extra ref for the work item */
  1308. memcpy(buf, msg->buf, be16_to_cpu(msg->data_len)); /* copy the whole message */
  1309. dlm_init_work_item(dlm, item, dlm_mig_lockres_worker, buf);
  1310. item->u.ml.lockres = res; /* already have a ref */
  1311. item->u.ml.real_master = real_master;
  1312. item->u.ml.extra_ref = extra_refs;
  1313. spin_lock(&dlm->work_lock);
  1314. list_add_tail(&item->list, &dlm->work_list);
  1315. spin_unlock(&dlm->work_lock);
  1316. queue_work(dlm->dlm_worker, &dlm->dispatched_work);
  1317. leave:
  1318. /* One extra ref taken needs to be put here */
  1319. if (extra_refs)
  1320. dlm_lockres_put(res);
  1321. dlm_put(dlm);
  1322. if (ret < 0) {
  1323. kfree(buf);
  1324. kfree(item);
  1325. mlog_errno(ret);
  1326. }
  1327. return ret;
  1328. }
  1329. static void dlm_mig_lockres_worker(struct dlm_work_item *item, void *data)
  1330. {
  1331. struct dlm_ctxt *dlm;
  1332. struct dlm_migratable_lockres *mres;
  1333. int ret = 0;
  1334. struct dlm_lock_resource *res;
  1335. u8 real_master;
  1336. u8 extra_ref;
  1337. dlm = item->dlm;
  1338. mres = (struct dlm_migratable_lockres *)data;
  1339. res = item->u.ml.lockres;
  1340. real_master = item->u.ml.real_master;
  1341. extra_ref = item->u.ml.extra_ref;
  1342. if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
  1343. /* this case is super-rare. only occurs if
  1344. * node death happens during migration. */
  1345. again:
  1346. ret = dlm_lockres_master_requery(dlm, res, &real_master);
  1347. if (ret < 0) {
  1348. mlog(0, "dlm_lockres_master_requery ret=%d\n",
  1349. ret);
  1350. goto again;
  1351. }
  1352. if (real_master == DLM_LOCK_RES_OWNER_UNKNOWN) {
  1353. mlog(0, "lockres %.*s not claimed. "
  1354. "this node will take it.\n",
  1355. res->lockname.len, res->lockname.name);
  1356. } else {
  1357. spin_lock(&res->spinlock);
  1358. dlm_lockres_drop_inflight_ref(dlm, res);
  1359. spin_unlock(&res->spinlock);
  1360. mlog(0, "master needs to respond to sender "
  1361. "that node %u still owns %.*s\n",
  1362. real_master, res->lockname.len,
  1363. res->lockname.name);
  1364. /* cannot touch this lockres */
  1365. goto leave;
  1366. }
  1367. }
  1368. ret = dlm_process_recovery_data(dlm, res, mres);
  1369. if (ret < 0)
  1370. mlog(0, "dlm_process_recovery_data returned %d\n", ret);
  1371. else
  1372. mlog(0, "dlm_process_recovery_data succeeded\n");
  1373. if ((mres->flags & (DLM_MRES_MIGRATION|DLM_MRES_ALL_DONE)) ==
  1374. (DLM_MRES_MIGRATION|DLM_MRES_ALL_DONE)) {
  1375. ret = dlm_finish_migration(dlm, res, mres->master);
  1376. if (ret < 0)
  1377. mlog_errno(ret);
  1378. }
  1379. leave:
  1380. /* See comment in dlm_mig_lockres_handler() */
  1381. if (res) {
  1382. if (extra_ref)
  1383. dlm_lockres_put(res);
  1384. dlm_lockres_put(res);
  1385. }
  1386. kfree(data);
  1387. }
  1388. static int dlm_lockres_master_requery(struct dlm_ctxt *dlm,
  1389. struct dlm_lock_resource *res,
  1390. u8 *real_master)
  1391. {
  1392. struct dlm_node_iter iter;
  1393. int nodenum;
  1394. int ret = 0;
  1395. *real_master = DLM_LOCK_RES_OWNER_UNKNOWN;
  1396. /* we only reach here if one of the two nodes in a
  1397. * migration died while the migration was in progress.
  1398. * at this point we need to requery the master. we
  1399. * know that the new_master got as far as creating
  1400. * an mle on at least one node, but we do not know
  1401. * if any nodes had actually cleared the mle and set
  1402. * the master to the new_master. the old master
  1403. * is supposed to set the owner to UNKNOWN in the
  1404. * event of a new_master death, so the only possible
  1405. * responses that we can get from nodes here are
  1406. * that the master is new_master, or that the master
  1407. * is UNKNOWN.
  1408. * if all nodes come back with UNKNOWN then we know
  1409. * the lock needs remastering here.
  1410. * if any node comes back with a valid master, check
  1411. * to see if that master is the one that we are
  1412. * recovering. if so, then the new_master died and
  1413. * we need to remaster this lock. if not, then the
  1414. * new_master survived and that node will respond to
  1415. * other nodes about the owner.
  1416. * if there is an owner, this node needs to dump this
  1417. * lockres and alert the sender that this lockres
  1418. * was rejected. */
  1419. spin_lock(&dlm->spinlock);
  1420. dlm_node_iter_init(dlm->domain_map, &iter);
  1421. spin_unlock(&dlm->spinlock);
  1422. while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
  1423. /* do not send to self */
  1424. if (nodenum == dlm->node_num)
  1425. continue;
  1426. ret = dlm_do_master_requery(dlm, res, nodenum, real_master);
  1427. if (ret < 0) {
  1428. mlog_errno(ret);
  1429. if (!dlm_is_host_down(ret))
  1430. BUG();
  1431. /* host is down, so answer for that node would be
  1432. * DLM_LOCK_RES_OWNER_UNKNOWN. continue. */
  1433. }
  1434. if (*real_master != DLM_LOCK_RES_OWNER_UNKNOWN) {
  1435. mlog(0, "lock master is %u\n", *real_master);
  1436. break;
  1437. }
  1438. }
  1439. return ret;
  1440. }
  1441. int dlm_do_master_requery(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
  1442. u8 nodenum, u8 *real_master)
  1443. {
  1444. int ret = -EINVAL;
  1445. struct dlm_master_requery req;
  1446. int status = DLM_LOCK_RES_OWNER_UNKNOWN;
  1447. memset(&req, 0, sizeof(req));
  1448. req.node_idx = dlm->node_num;
  1449. req.namelen = res->lockname.len;
  1450. memcpy(req.name, res->lockname.name, res->lockname.len);
  1451. resend:
  1452. ret = o2net_send_message(DLM_MASTER_REQUERY_MSG, dlm->key,
  1453. &req, sizeof(req), nodenum, &status);
  1454. if (ret < 0)
  1455. mlog(ML_ERROR, "Error %d when sending message %u (key "
  1456. "0x%x) to node %u\n", ret, DLM_MASTER_REQUERY_MSG,
  1457. dlm->key, nodenum);
  1458. else if (status == -ENOMEM) {
  1459. mlog_errno(status);
  1460. msleep(50);
  1461. goto resend;
  1462. } else {
  1463. BUG_ON(status < 0);
  1464. BUG_ON(status > DLM_LOCK_RES_OWNER_UNKNOWN);
  1465. *real_master = (u8) (status & 0xff);
  1466. mlog(0, "node %u responded to master requery with %u\n",
  1467. nodenum, *real_master);
  1468. ret = 0;
  1469. }
  1470. return ret;
  1471. }
  1472. /* this function cannot error, so unless the sending
  1473. * or receiving of the message failed, the owner can
  1474. * be trusted */
  1475. int dlm_master_requery_handler(struct o2net_msg *msg, u32 len, void *data,
  1476. void **ret_data)
  1477. {
  1478. struct dlm_ctxt *dlm = data;
  1479. struct dlm_master_requery *req = (struct dlm_master_requery *)msg->buf;
  1480. struct dlm_lock_resource *res = NULL;
  1481. unsigned int hash;
  1482. int master = DLM_LOCK_RES_OWNER_UNKNOWN;
  1483. u32 flags = DLM_ASSERT_MASTER_REQUERY;
  1484. if (!dlm_grab(dlm)) {
  1485. /* since the domain has gone away on this
  1486. * node, the proper response is UNKNOWN */
  1487. return master;
  1488. }
  1489. hash = dlm_lockid_hash(req->name, req->namelen);
  1490. spin_lock(&dlm->spinlock);
  1491. res = __dlm_lookup_lockres(dlm, req->name, req->namelen, hash);
  1492. if (res) {
  1493. spin_lock(&res->spinlock);
  1494. master = res->owner;
  1495. if (master == dlm->node_num) {
  1496. int ret = dlm_dispatch_assert_master(dlm, res,
  1497. 0, 0, flags);
  1498. if (ret < 0) {
  1499. mlog_errno(ret);
  1500. spin_unlock(&res->spinlock);
  1501. dlm_lockres_put(res);
  1502. spin_unlock(&dlm->spinlock);
  1503. dlm_put(dlm);
  1504. /* sender will take care of this and retry */
  1505. return ret;
  1506. } else
  1507. __dlm_lockres_grab_inflight_worker(dlm, res);
  1508. spin_unlock(&res->spinlock);
  1509. } else {
  1510. /* put.. incase we are not the master */
  1511. spin_unlock(&res->spinlock);
  1512. dlm_lockres_put(res);
  1513. }
  1514. }
  1515. spin_unlock(&dlm->spinlock);
  1516. dlm_put(dlm);
  1517. return master;
  1518. }
  1519. static inline struct list_head *
  1520. dlm_list_num_to_pointer(struct dlm_lock_resource *res, int list_num)
  1521. {
  1522. struct list_head *ret;
  1523. BUG_ON(list_num < 0);
  1524. BUG_ON(list_num > 2);
  1525. ret = &(res->granted);
  1526. ret += list_num;
  1527. return ret;
  1528. }
  1529. /* TODO: do ast flush business
  1530. * TODO: do MIGRATING and RECOVERING spinning
  1531. */
  1532. /*
  1533. * NOTE about in-flight requests during migration:
  1534. *
  1535. * Before attempting the migrate, the master has marked the lockres as
  1536. * MIGRATING and then flushed all of its pending ASTS. So any in-flight
  1537. * requests either got queued before the MIGRATING flag got set, in which
  1538. * case the lock data will reflect the change and a return message is on
  1539. * the way, or the request failed to get in before MIGRATING got set. In
  1540. * this case, the caller will be told to spin and wait for the MIGRATING
  1541. * flag to be dropped, then recheck the master.
  1542. * This holds true for the convert, cancel and unlock cases, and since lvb
  1543. * updates are tied to these same messages, it applies to lvb updates as
  1544. * well. For the lock case, there is no way a lock can be on the master
  1545. * queue and not be on the secondary queue since the lock is always added
  1546. * locally first. This means that the new target node will never be sent
  1547. * a lock that he doesn't already have on the list.
  1548. * In total, this means that the local lock is correct and should not be
  1549. * updated to match the one sent by the master. Any messages sent back
  1550. * from the master before the MIGRATING flag will bring the lock properly
  1551. * up-to-date, and the change will be ordered properly for the waiter.
  1552. * We will *not* attempt to modify the lock underneath the waiter.
  1553. */
  1554. static int dlm_process_recovery_data(struct dlm_ctxt *dlm,
  1555. struct dlm_lock_resource *res,
  1556. struct dlm_migratable_lockres *mres)
  1557. {
  1558. struct dlm_migratable_lock *ml;
  1559. struct list_head *queue, *iter;
  1560. struct list_head *tmpq = NULL;
  1561. struct dlm_lock *newlock = NULL;
  1562. struct dlm_lockstatus *lksb = NULL;
  1563. int ret = 0;
  1564. int i, j, bad;
  1565. struct dlm_lock *lock;
  1566. u8 from = O2NM_MAX_NODES;
  1567. unsigned int added = 0;
  1568. __be64 c;
  1569. mlog(0, "running %d locks for this lockres\n", mres->num_locks);
  1570. for (i=0; i<mres->num_locks; i++) {
  1571. ml = &(mres->ml[i]);
  1572. if (dlm_is_dummy_lock(dlm, ml, &from)) {
  1573. /* placeholder, just need to set the refmap bit */
  1574. BUG_ON(mres->num_locks != 1);
  1575. mlog(0, "%s:%.*s: dummy lock for %u\n",
  1576. dlm->name, mres->lockname_len, mres->lockname,
  1577. from);
  1578. spin_lock(&res->spinlock);
  1579. dlm_lockres_set_refmap_bit(dlm, res, from);
  1580. spin_unlock(&res->spinlock);
  1581. added++;
  1582. break;
  1583. }
  1584. BUG_ON(ml->highest_blocked != LKM_IVMODE);
  1585. newlock = NULL;
  1586. lksb = NULL;
  1587. queue = dlm_list_num_to_pointer(res, ml->list);
  1588. tmpq = NULL;
  1589. /* if the lock is for the local node it needs to
  1590. * be moved to the proper location within the queue.
  1591. * do not allocate a new lock structure. */
  1592. if (ml->node == dlm->node_num) {
  1593. /* MIGRATION ONLY! */
  1594. BUG_ON(!(mres->flags & DLM_MRES_MIGRATION));
  1595. lock = NULL;
  1596. spin_lock(&res->spinlock);
  1597. for (j = DLM_GRANTED_LIST; j <= DLM_BLOCKED_LIST; j++) {
  1598. tmpq = dlm_list_idx_to_ptr(res, j);
  1599. list_for_each(iter, tmpq) {
  1600. lock = list_entry(iter,
  1601. struct dlm_lock, list);
  1602. if (lock->ml.cookie == ml->cookie)
  1603. break;
  1604. lock = NULL;
  1605. }
  1606. if (lock)
  1607. break;
  1608. }
  1609. /* lock is always created locally first, and
  1610. * destroyed locally last. it must be on the list */
  1611. if (!lock) {
  1612. c = ml->cookie;
  1613. mlog(ML_ERROR, "Could not find local lock "
  1614. "with cookie %u:%llu, node %u, "
  1615. "list %u, flags 0x%x, type %d, "
  1616. "conv %d, highest blocked %d\n",
  1617. dlm_get_lock_cookie_node(be64_to_cpu(c)),
  1618. dlm_get_lock_cookie_seq(be64_to_cpu(c)),
  1619. ml->node, ml->list, ml->flags, ml->type,
  1620. ml->convert_type, ml->highest_blocked);
  1621. __dlm_print_one_lock_resource(res);
  1622. BUG();
  1623. }
  1624. if (lock->ml.node != ml->node) {
  1625. c = lock->ml.cookie;
  1626. mlog(ML_ERROR, "Mismatched node# in lock "
  1627. "cookie %u:%llu, name %.*s, node %u\n",
  1628. dlm_get_lock_cookie_node(be64_to_cpu(c)),
  1629. dlm_get_lock_cookie_seq(be64_to_cpu(c)),
  1630. res->lockname.len, res->lockname.name,
  1631. lock->ml.node);
  1632. c = ml->cookie;
  1633. mlog(ML_ERROR, "Migrate lock cookie %u:%llu, "
  1634. "node %u, list %u, flags 0x%x, type %d, "
  1635. "conv %d, highest blocked %d\n",
  1636. dlm_get_lock_cookie_node(be64_to_cpu(c)),
  1637. dlm_get_lock_cookie_seq(be64_to_cpu(c)),
  1638. ml->node, ml->list, ml->flags, ml->type,
  1639. ml->convert_type, ml->highest_blocked);
  1640. __dlm_print_one_lock_resource(res);
  1641. BUG();
  1642. }
  1643. if (tmpq != queue) {
  1644. c = ml->cookie;
  1645. mlog(0, "Lock cookie %u:%llu was on list %u "
  1646. "instead of list %u for %.*s\n",
  1647. dlm_get_lock_cookie_node(be64_to_cpu(c)),
  1648. dlm_get_lock_cookie_seq(be64_to_cpu(c)),
  1649. j, ml->list, res->lockname.len,
  1650. res->lockname.name);
  1651. __dlm_print_one_lock_resource(res);
  1652. spin_unlock(&res->spinlock);
  1653. continue;
  1654. }
  1655. /* see NOTE above about why we do not update
  1656. * to match the master here */
  1657. /* move the lock to its proper place */
  1658. /* do not alter lock refcount. switching lists. */
  1659. list_move_tail(&lock->list, queue);
  1660. spin_unlock(&res->spinlock);
  1661. added++;
  1662. mlog(0, "just reordered a local lock!\n");
  1663. continue;
  1664. }
  1665. /* lock is for another node. */
  1666. newlock = dlm_new_lock(ml->type, ml->node,
  1667. be64_to_cpu(ml->cookie), NULL);
  1668. if (!newlock) {
  1669. ret = -ENOMEM;
  1670. goto leave;
  1671. }
  1672. lksb = newlock->lksb;
  1673. dlm_lock_attach_lockres(newlock, res);
  1674. if (ml->convert_type != LKM_IVMODE) {
  1675. BUG_ON(queue != &res->converting);
  1676. newlock->ml.convert_type = ml->convert_type;
  1677. }
  1678. lksb->flags |= (ml->flags &
  1679. (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
  1680. if (ml->type == LKM_NLMODE)
  1681. goto skip_lvb;
  1682. /*
  1683. * If the lock is in the blocked list it can't have a valid lvb,
  1684. * so skip it
  1685. */
  1686. if (ml->list == DLM_BLOCKED_LIST)
  1687. goto skip_lvb;
  1688. if (!dlm_lvb_is_empty(mres->lvb)) {
  1689. if (lksb->flags & DLM_LKSB_PUT_LVB) {
  1690. /* other node was trying to update
  1691. * lvb when node died. recreate the
  1692. * lksb with the updated lvb. */
  1693. memcpy(lksb->lvb, mres->lvb, DLM_LVB_LEN);
  1694. /* the lock resource lvb update must happen
  1695. * NOW, before the spinlock is dropped.
  1696. * we no longer wait for the AST to update
  1697. * the lvb. */
  1698. memcpy(res->lvb, mres->lvb, DLM_LVB_LEN);
  1699. } else {
  1700. /* otherwise, the node is sending its
  1701. * most recent valid lvb info */
  1702. BUG_ON(ml->type != LKM_EXMODE &&
  1703. ml->type != LKM_PRMODE);
  1704. if (!dlm_lvb_is_empty(res->lvb) &&
  1705. (ml->type == LKM_EXMODE ||
  1706. memcmp(res->lvb, mres->lvb, DLM_LVB_LEN))) {
  1707. int i;
  1708. mlog(ML_ERROR, "%s:%.*s: received bad "
  1709. "lvb! type=%d\n", dlm->name,
  1710. res->lockname.len,
  1711. res->lockname.name, ml->type);
  1712. printk("lockres lvb=[");
  1713. for (i=0; i<DLM_LVB_LEN; i++)
  1714. printk("%02x", res->lvb[i]);
  1715. printk("]\nmigrated lvb=[");
  1716. for (i=0; i<DLM_LVB_LEN; i++)
  1717. printk("%02x", mres->lvb[i]);
  1718. printk("]\n");
  1719. dlm_print_one_lock_resource(res);
  1720. BUG();
  1721. }
  1722. memcpy(res->lvb, mres->lvb, DLM_LVB_LEN);
  1723. }
  1724. }
  1725. skip_lvb:
  1726. /* NOTE:
  1727. * wrt lock queue ordering and recovery:
  1728. * 1. order of locks on granted queue is
  1729. * meaningless.
  1730. * 2. order of locks on converting queue is
  1731. * LOST with the node death. sorry charlie.
  1732. * 3. order of locks on the blocked queue is
  1733. * also LOST.
  1734. * order of locks does not affect integrity, it
  1735. * just means that a lock request may get pushed
  1736. * back in line as a result of the node death.
  1737. * also note that for a given node the lock order
  1738. * for its secondary queue locks is preserved
  1739. * relative to each other, but clearly *not*
  1740. * preserved relative to locks from other nodes.
  1741. */
  1742. bad = 0;
  1743. spin_lock(&res->spinlock);
  1744. list_for_each_entry(lock, queue, list) {
  1745. if (lock->ml.cookie == ml->cookie) {
  1746. c = lock->ml.cookie;
  1747. mlog(ML_ERROR, "%s:%.*s: %u:%llu: lock already "
  1748. "exists on this lockres!\n", dlm->name,
  1749. res->lockname.len, res->lockname.name,
  1750. dlm_get_lock_cookie_node(be64_to_cpu(c)),
  1751. dlm_get_lock_cookie_seq(be64_to_cpu(c)));
  1752. mlog(ML_NOTICE, "sent lock: type=%d, conv=%d, "
  1753. "node=%u, cookie=%u:%llu, queue=%d\n",
  1754. ml->type, ml->convert_type, ml->node,
  1755. dlm_get_lock_cookie_node(be64_to_cpu(ml->cookie)),
  1756. dlm_get_lock_cookie_seq(be64_to_cpu(ml->cookie)),
  1757. ml->list);
  1758. __dlm_print_one_lock_resource(res);
  1759. bad = 1;
  1760. break;
  1761. }
  1762. }
  1763. if (!bad) {
  1764. dlm_lock_get(newlock);
  1765. if (mres->flags & DLM_MRES_RECOVERY &&
  1766. ml->list == DLM_CONVERTING_LIST &&
  1767. newlock->ml.type >
  1768. newlock->ml.convert_type) {
  1769. /* newlock is doing downconvert, add it to the
  1770. * head of converting list */
  1771. list_add(&newlock->list, queue);
  1772. } else
  1773. list_add_tail(&newlock->list, queue);
  1774. mlog(0, "%s:%.*s: added lock for node %u, "
  1775. "setting refmap bit\n", dlm->name,
  1776. res->lockname.len, res->lockname.name, ml->node);
  1777. dlm_lockres_set_refmap_bit(dlm, res, ml->node);
  1778. added++;
  1779. }
  1780. spin_unlock(&res->spinlock);
  1781. }
  1782. mlog(0, "done running all the locks\n");
  1783. leave:
  1784. /* balance the ref taken when the work was queued */
  1785. spin_lock(&res->spinlock);
  1786. dlm_lockres_drop_inflight_ref(dlm, res);
  1787. spin_unlock(&res->spinlock);
  1788. if (ret < 0)
  1789. mlog_errno(ret);
  1790. return ret;
  1791. }
  1792. void dlm_move_lockres_to_recovery_list(struct dlm_ctxt *dlm,
  1793. struct dlm_lock_resource *res)
  1794. {
  1795. int i;
  1796. struct list_head *queue;
  1797. struct dlm_lock *lock, *next;
  1798. assert_spin_locked(&dlm->spinlock);
  1799. assert_spin_locked(&res->spinlock);
  1800. res->state |= DLM_LOCK_RES_RECOVERING;
  1801. if (!list_empty(&res->recovering)) {
  1802. mlog(0,
  1803. "Recovering res %s:%.*s, is already on recovery list!\n",
  1804. dlm->name, res->lockname.len, res->lockname.name);
  1805. list_del_init(&res->recovering);
  1806. dlm_lockres_put(res);
  1807. }
  1808. /* We need to hold a reference while on the recovery list */
  1809. dlm_lockres_get(res);
  1810. list_add_tail(&res->recovering, &dlm->reco.resources);
  1811. /* find any pending locks and put them back on proper list */
  1812. for (i=DLM_BLOCKED_LIST; i>=DLM_GRANTED_LIST; i--) {
  1813. queue = dlm_list_idx_to_ptr(res, i);
  1814. list_for_each_entry_safe(lock, next, queue, list) {
  1815. dlm_lock_get(lock);
  1816. if (lock->convert_pending) {
  1817. /* move converting lock back to granted */
  1818. BUG_ON(i != DLM_CONVERTING_LIST);
  1819. mlog(0, "node died with convert pending "
  1820. "on %.*s. move back to granted list.\n",
  1821. res->lockname.len, res->lockname.name);
  1822. dlm_revert_pending_convert(res, lock);
  1823. lock->convert_pending = 0;
  1824. } else if (lock->lock_pending) {
  1825. /* remove pending lock requests completely */
  1826. BUG_ON(i != DLM_BLOCKED_LIST);
  1827. mlog(0, "node died with lock pending "
  1828. "on %.*s. remove from blocked list and skip.\n",
  1829. res->lockname.len, res->lockname.name);
  1830. /* lock will be floating until ref in
  1831. * dlmlock_remote is freed after the network
  1832. * call returns. ok for it to not be on any
  1833. * list since no ast can be called
  1834. * (the master is dead). */
  1835. dlm_revert_pending_lock(res, lock);
  1836. lock->lock_pending = 0;
  1837. } else if (lock->unlock_pending) {
  1838. /* if an unlock was in progress, treat as
  1839. * if this had completed successfully
  1840. * before sending this lock state to the
  1841. * new master. note that the dlm_unlock
  1842. * call is still responsible for calling
  1843. * the unlockast. that will happen after
  1844. * the network call times out. for now,
  1845. * just move lists to prepare the new
  1846. * recovery master. */
  1847. BUG_ON(i != DLM_GRANTED_LIST);
  1848. mlog(0, "node died with unlock pending "
  1849. "on %.*s. remove from blocked list and skip.\n",
  1850. res->lockname.len, res->lockname.name);
  1851. dlm_commit_pending_unlock(res, lock);
  1852. lock->unlock_pending = 0;
  1853. } else if (lock->cancel_pending) {
  1854. /* if a cancel was in progress, treat as
  1855. * if this had completed successfully
  1856. * before sending this lock state to the
  1857. * new master */
  1858. BUG_ON(i != DLM_CONVERTING_LIST);
  1859. mlog(0, "node died with cancel pending "
  1860. "on %.*s. move back to granted list.\n",
  1861. res->lockname.len, res->lockname.name);
  1862. dlm_commit_pending_cancel(res, lock);
  1863. lock->cancel_pending = 0;
  1864. }
  1865. dlm_lock_put(lock);
  1866. }
  1867. }
  1868. }
  1869. /* removes all recovered locks from the recovery list.
  1870. * sets the res->owner to the new master.
  1871. * unsets the RECOVERY flag and wakes waiters. */
  1872. static void dlm_finish_local_lockres_recovery(struct dlm_ctxt *dlm,
  1873. u8 dead_node, u8 new_master)
  1874. {
  1875. int i;
  1876. struct hlist_head *bucket;
  1877. struct dlm_lock_resource *res, *next;
  1878. assert_spin_locked(&dlm->spinlock);
  1879. list_for_each_entry_safe(res, next, &dlm->reco.resources, recovering) {
  1880. if (res->owner == dead_node) {
  1881. mlog(0, "%s: res %.*s, Changing owner from %u to %u\n",
  1882. dlm->name, res->lockname.len, res->lockname.name,
  1883. res->owner, new_master);
  1884. list_del_init(&res->recovering);
  1885. spin_lock(&res->spinlock);
  1886. /* new_master has our reference from
  1887. * the lock state sent during recovery */
  1888. dlm_change_lockres_owner(dlm, res, new_master);
  1889. res->state &= ~DLM_LOCK_RES_RECOVERING;
  1890. if (__dlm_lockres_has_locks(res))
  1891. __dlm_dirty_lockres(dlm, res);
  1892. spin_unlock(&res->spinlock);
  1893. wake_up(&res->wq);
  1894. dlm_lockres_put(res);
  1895. }
  1896. }
  1897. /* this will become unnecessary eventually, but
  1898. * for now we need to run the whole hash, clear
  1899. * the RECOVERING state and set the owner
  1900. * if necessary */
  1901. for (i = 0; i < DLM_HASH_BUCKETS; i++) {
  1902. bucket = dlm_lockres_hash(dlm, i);
  1903. hlist_for_each_entry(res, bucket, hash_node) {
  1904. if (!(res->state & DLM_LOCK_RES_RECOVERING))
  1905. continue;
  1906. if (res->owner != dead_node &&
  1907. res->owner != dlm->node_num)
  1908. continue;
  1909. if (!list_empty(&res->recovering)) {
  1910. list_del_init(&res->recovering);
  1911. dlm_lockres_put(res);
  1912. }
  1913. /* new_master has our reference from
  1914. * the lock state sent during recovery */
  1915. mlog(0, "%s: res %.*s, Changing owner from %u to %u\n",
  1916. dlm->name, res->lockname.len, res->lockname.name,
  1917. res->owner, new_master);
  1918. spin_lock(&res->spinlock);
  1919. dlm_change_lockres_owner(dlm, res, new_master);
  1920. res->state &= ~DLM_LOCK_RES_RECOVERING;
  1921. if (__dlm_lockres_has_locks(res))
  1922. __dlm_dirty_lockres(dlm, res);
  1923. spin_unlock(&res->spinlock);
  1924. wake_up(&res->wq);
  1925. }
  1926. }
  1927. }
  1928. static inline int dlm_lvb_needs_invalidation(struct dlm_lock *lock, int local)
  1929. {
  1930. if (local) {
  1931. if (lock->ml.type != LKM_EXMODE &&
  1932. lock->ml.type != LKM_PRMODE)
  1933. return 1;
  1934. } else if (lock->ml.type == LKM_EXMODE)
  1935. return 1;
  1936. return 0;
  1937. }
  1938. static void dlm_revalidate_lvb(struct dlm_ctxt *dlm,
  1939. struct dlm_lock_resource *res, u8 dead_node)
  1940. {
  1941. struct list_head *queue;
  1942. struct dlm_lock *lock;
  1943. int blank_lvb = 0, local = 0;
  1944. int i;
  1945. u8 search_node;
  1946. assert_spin_locked(&dlm->spinlock);
  1947. assert_spin_locked(&res->spinlock);
  1948. if (res->owner == dlm->node_num)
  1949. /* if this node owned the lockres, and if the dead node
  1950. * had an EX when he died, blank out the lvb */
  1951. search_node = dead_node;
  1952. else {
  1953. /* if this is a secondary lockres, and we had no EX or PR
  1954. * locks granted, we can no longer trust the lvb */
  1955. search_node = dlm->node_num;
  1956. local = 1; /* check local state for valid lvb */
  1957. }
  1958. for (i=DLM_GRANTED_LIST; i<=DLM_CONVERTING_LIST; i++) {
  1959. queue = dlm_list_idx_to_ptr(res, i);
  1960. list_for_each_entry(lock, queue, list) {
  1961. if (lock->ml.node == search_node) {
  1962. if (dlm_lvb_needs_invalidation(lock, local)) {
  1963. /* zero the lksb lvb and lockres lvb */
  1964. blank_lvb = 1;
  1965. memset(lock->lksb->lvb, 0, DLM_LVB_LEN);
  1966. }
  1967. }
  1968. }
  1969. }
  1970. if (blank_lvb) {
  1971. mlog(0, "clearing %.*s lvb, dead node %u had EX\n",
  1972. res->lockname.len, res->lockname.name, dead_node);
  1973. memset(res->lvb, 0, DLM_LVB_LEN);
  1974. }
  1975. }
  1976. static void dlm_free_dead_locks(struct dlm_ctxt *dlm,
  1977. struct dlm_lock_resource *res, u8 dead_node)
  1978. {
  1979. struct dlm_lock *lock, *next;
  1980. unsigned int freed = 0;
  1981. /* this node is the lockres master:
  1982. * 1) remove any stale locks for the dead node
  1983. * 2) if the dead node had an EX when he died, blank out the lvb
  1984. */
  1985. assert_spin_locked(&dlm->spinlock);
  1986. assert_spin_locked(&res->spinlock);
  1987. /* We do two dlm_lock_put(). One for removing from list and the other is
  1988. * to force the DLM_UNLOCK_FREE_LOCK action so as to free the locks */
  1989. /* TODO: check pending_asts, pending_basts here */
  1990. list_for_each_entry_safe(lock, next, &res->granted, list) {
  1991. if (lock->ml.node == dead_node) {
  1992. list_del_init(&lock->list);
  1993. dlm_lock_put(lock);
  1994. /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
  1995. dlm_lock_put(lock);
  1996. freed++;
  1997. }
  1998. }
  1999. list_for_each_entry_safe(lock, next, &res->converting, list) {
  2000. if (lock->ml.node == dead_node) {
  2001. list_del_init(&lock->list);
  2002. dlm_lock_put(lock);
  2003. /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
  2004. dlm_lock_put(lock);
  2005. freed++;
  2006. }
  2007. }
  2008. list_for_each_entry_safe(lock, next, &res->blocked, list) {
  2009. if (lock->ml.node == dead_node) {
  2010. list_del_init(&lock->list);
  2011. dlm_lock_put(lock);
  2012. /* Can't schedule DLM_UNLOCK_FREE_LOCK - do manually */
  2013. dlm_lock_put(lock);
  2014. freed++;
  2015. }
  2016. }
  2017. if (freed) {
  2018. mlog(0, "%s:%.*s: freed %u locks for dead node %u, "
  2019. "dropping ref from lockres\n", dlm->name,
  2020. res->lockname.len, res->lockname.name, freed, dead_node);
  2021. if(!test_bit(dead_node, res->refmap)) {
  2022. mlog(ML_ERROR, "%s:%.*s: freed %u locks for dead node %u, "
  2023. "but ref was not set\n", dlm->name,
  2024. res->lockname.len, res->lockname.name, freed, dead_node);
  2025. __dlm_print_one_lock_resource(res);
  2026. }
  2027. dlm_lockres_clear_refmap_bit(dlm, res, dead_node);
  2028. } else if (test_bit(dead_node, res->refmap)) {
  2029. mlog(0, "%s:%.*s: dead node %u had a ref, but had "
  2030. "no locks and had not purged before dying\n", dlm->name,
  2031. res->lockname.len, res->lockname.name, dead_node);
  2032. dlm_lockres_clear_refmap_bit(dlm, res, dead_node);
  2033. }
  2034. /* do not kick thread yet */
  2035. __dlm_dirty_lockres(dlm, res);
  2036. }
  2037. /* if this node is the recovery master, and there are no
  2038. * locks for a given lockres owned by this node that are in
  2039. * either PR or EX mode, zero out the lvb before requesting.
  2040. *
  2041. */
  2042. static void dlm_do_local_recovery_cleanup(struct dlm_ctxt *dlm, u8 dead_node)
  2043. {
  2044. struct dlm_lock_resource *res;
  2045. int i;
  2046. struct hlist_head *bucket;
  2047. struct dlm_lock *lock;
  2048. /* purge any stale mles */
  2049. dlm_clean_master_list(dlm, dead_node);
  2050. /*
  2051. * now clean up all lock resources. there are two rules:
  2052. *
  2053. * 1) if the dead node was the master, move the lockres
  2054. * to the recovering list. set the RECOVERING flag.
  2055. * this lockres needs to be cleaned up before it can
  2056. * be used further.
  2057. *
  2058. * 2) if this node was the master, remove all locks from
  2059. * each of the lockres queues that were owned by the
  2060. * dead node. once recovery finishes, the dlm thread
  2061. * can be kicked again to see if any ASTs or BASTs
  2062. * need to be fired as a result.
  2063. */
  2064. for (i = 0; i < DLM_HASH_BUCKETS; i++) {
  2065. bucket = dlm_lockres_hash(dlm, i);
  2066. hlist_for_each_entry(res, bucket, hash_node) {
  2067. /* always prune any $RECOVERY entries for dead nodes,
  2068. * otherwise hangs can occur during later recovery */
  2069. if (dlm_is_recovery_lock(res->lockname.name,
  2070. res->lockname.len)) {
  2071. spin_lock(&res->spinlock);
  2072. list_for_each_entry(lock, &res->granted, list) {
  2073. if (lock->ml.node == dead_node) {
  2074. mlog(0, "AHA! there was "
  2075. "a $RECOVERY lock for dead "
  2076. "node %u (%s)!\n",
  2077. dead_node, dlm->name);
  2078. list_del_init(&lock->list);
  2079. dlm_lock_put(lock);
  2080. /* Can't schedule
  2081. * DLM_UNLOCK_FREE_LOCK
  2082. * - do manually */
  2083. dlm_lock_put(lock);
  2084. break;
  2085. }
  2086. }
  2087. spin_unlock(&res->spinlock);
  2088. continue;
  2089. }
  2090. spin_lock(&res->spinlock);
  2091. /* zero the lvb if necessary */
  2092. dlm_revalidate_lvb(dlm, res, dead_node);
  2093. if (res->owner == dead_node) {
  2094. if (res->state & DLM_LOCK_RES_DROPPING_REF) {
  2095. mlog(ML_NOTICE, "%s: res %.*s, Skip "
  2096. "recovery as it is being freed\n",
  2097. dlm->name, res->lockname.len,
  2098. res->lockname.name);
  2099. } else
  2100. dlm_move_lockres_to_recovery_list(dlm,
  2101. res);
  2102. } else if (res->owner == dlm->node_num) {
  2103. dlm_free_dead_locks(dlm, res, dead_node);
  2104. __dlm_lockres_calc_usage(dlm, res);
  2105. } else if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
  2106. if (test_bit(dead_node, res->refmap)) {
  2107. mlog(0, "%s:%.*s: dead node %u had a ref, but had "
  2108. "no locks and had not purged before dying\n",
  2109. dlm->name, res->lockname.len,
  2110. res->lockname.name, dead_node);
  2111. dlm_lockres_clear_refmap_bit(dlm, res, dead_node);
  2112. }
  2113. }
  2114. spin_unlock(&res->spinlock);
  2115. }
  2116. }
  2117. }
  2118. static void __dlm_hb_node_down(struct dlm_ctxt *dlm, int idx)
  2119. {
  2120. assert_spin_locked(&dlm->spinlock);
  2121. if (dlm->reco.new_master == idx) {
  2122. mlog(0, "%s: recovery master %d just died\n",
  2123. dlm->name, idx);
  2124. if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
  2125. /* finalize1 was reached, so it is safe to clear
  2126. * the new_master and dead_node. that recovery
  2127. * is complete. */
  2128. mlog(0, "%s: dead master %d had reached "
  2129. "finalize1 state, clearing\n", dlm->name, idx);
  2130. dlm->reco.state &= ~DLM_RECO_STATE_FINALIZE;
  2131. __dlm_reset_recovery(dlm);
  2132. }
  2133. }
  2134. /* Clean up join state on node death. */
  2135. if (dlm->joining_node == idx) {
  2136. mlog(0, "Clearing join state for node %u\n", idx);
  2137. __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
  2138. }
  2139. /* check to see if the node is already considered dead */
  2140. if (!test_bit(idx, dlm->live_nodes_map)) {
  2141. mlog(0, "for domain %s, node %d is already dead. "
  2142. "another node likely did recovery already.\n",
  2143. dlm->name, idx);
  2144. return;
  2145. }
  2146. /* check to see if we do not care about this node */
  2147. if (!test_bit(idx, dlm->domain_map)) {
  2148. /* This also catches the case that we get a node down
  2149. * but haven't joined the domain yet. */
  2150. mlog(0, "node %u already removed from domain!\n", idx);
  2151. return;
  2152. }
  2153. clear_bit(idx, dlm->live_nodes_map);
  2154. /* make sure local cleanup occurs before the heartbeat events */
  2155. if (!test_bit(idx, dlm->recovery_map))
  2156. dlm_do_local_recovery_cleanup(dlm, idx);
  2157. /* notify anything attached to the heartbeat events */
  2158. dlm_hb_event_notify_attached(dlm, idx, 0);
  2159. mlog(0, "node %u being removed from domain map!\n", idx);
  2160. clear_bit(idx, dlm->domain_map);
  2161. clear_bit(idx, dlm->exit_domain_map);
  2162. /* wake up migration waiters if a node goes down.
  2163. * perhaps later we can genericize this for other waiters. */
  2164. wake_up(&dlm->migration_wq);
  2165. if (test_bit(idx, dlm->recovery_map))
  2166. mlog(0, "domain %s, node %u already added "
  2167. "to recovery map!\n", dlm->name, idx);
  2168. else
  2169. set_bit(idx, dlm->recovery_map);
  2170. }
  2171. void dlm_hb_node_down_cb(struct o2nm_node *node, int idx, void *data)
  2172. {
  2173. struct dlm_ctxt *dlm = data;
  2174. if (!dlm_grab(dlm))
  2175. return;
  2176. /*
  2177. * This will notify any dlm users that a node in our domain
  2178. * went away without notifying us first.
  2179. */
  2180. if (test_bit(idx, dlm->domain_map))
  2181. dlm_fire_domain_eviction_callbacks(dlm, idx);
  2182. spin_lock(&dlm->spinlock);
  2183. __dlm_hb_node_down(dlm, idx);
  2184. spin_unlock(&dlm->spinlock);
  2185. dlm_put(dlm);
  2186. }
  2187. void dlm_hb_node_up_cb(struct o2nm_node *node, int idx, void *data)
  2188. {
  2189. struct dlm_ctxt *dlm = data;
  2190. if (!dlm_grab(dlm))
  2191. return;
  2192. spin_lock(&dlm->spinlock);
  2193. set_bit(idx, dlm->live_nodes_map);
  2194. /* do NOT notify mle attached to the heartbeat events.
  2195. * new nodes are not interesting in mastery until joined. */
  2196. spin_unlock(&dlm->spinlock);
  2197. dlm_put(dlm);
  2198. }
  2199. static void dlm_reco_ast(void *astdata)
  2200. {
  2201. struct dlm_ctxt *dlm = astdata;
  2202. mlog(0, "ast for recovery lock fired!, this=%u, dlm=%s\n",
  2203. dlm->node_num, dlm->name);
  2204. }
  2205. static void dlm_reco_bast(void *astdata, int blocked_type)
  2206. {
  2207. struct dlm_ctxt *dlm = astdata;
  2208. mlog(0, "bast for recovery lock fired!, this=%u, dlm=%s\n",
  2209. dlm->node_num, dlm->name);
  2210. }
  2211. static void dlm_reco_unlock_ast(void *astdata, enum dlm_status st)
  2212. {
  2213. mlog(0, "unlockast for recovery lock fired!\n");
  2214. }
  2215. /*
  2216. * dlm_pick_recovery_master will continually attempt to use
  2217. * dlmlock() on the special "$RECOVERY" lockres with the
  2218. * LKM_NOQUEUE flag to get an EX. every thread that enters
  2219. * this function on each node racing to become the recovery
  2220. * master will not stop attempting this until either:
  2221. * a) this node gets the EX (and becomes the recovery master),
  2222. * or b) dlm->reco.new_master gets set to some nodenum
  2223. * != O2NM_INVALID_NODE_NUM (another node will do the reco).
  2224. * so each time a recovery master is needed, the entire cluster
  2225. * will sync at this point. if the new master dies, that will
  2226. * be detected in dlm_do_recovery */
  2227. static int dlm_pick_recovery_master(struct dlm_ctxt *dlm)
  2228. {
  2229. enum dlm_status ret;
  2230. struct dlm_lockstatus lksb;
  2231. int status = -EINVAL;
  2232. mlog(0, "starting recovery of %s at %lu, dead=%u, this=%u\n",
  2233. dlm->name, jiffies, dlm->reco.dead_node, dlm->node_num);
  2234. again:
  2235. memset(&lksb, 0, sizeof(lksb));
  2236. ret = dlmlock(dlm, LKM_EXMODE, &lksb, LKM_NOQUEUE|LKM_RECOVERY,
  2237. DLM_RECOVERY_LOCK_NAME, DLM_RECOVERY_LOCK_NAME_LEN,
  2238. dlm_reco_ast, dlm, dlm_reco_bast);
  2239. mlog(0, "%s: dlmlock($RECOVERY) returned %d, lksb=%d\n",
  2240. dlm->name, ret, lksb.status);
  2241. if (ret == DLM_NORMAL) {
  2242. mlog(0, "dlm=%s dlmlock says I got it (this=%u)\n",
  2243. dlm->name, dlm->node_num);
  2244. /* got the EX lock. check to see if another node
  2245. * just became the reco master */
  2246. if (dlm_reco_master_ready(dlm)) {
  2247. mlog(0, "%s: got reco EX lock, but %u will "
  2248. "do the recovery\n", dlm->name,
  2249. dlm->reco.new_master);
  2250. status = -EEXIST;
  2251. } else {
  2252. status = 0;
  2253. /* see if recovery was already finished elsewhere */
  2254. spin_lock(&dlm->spinlock);
  2255. if (dlm->reco.dead_node == O2NM_INVALID_NODE_NUM) {
  2256. status = -EINVAL;
  2257. mlog(0, "%s: got reco EX lock, but "
  2258. "node got recovered already\n", dlm->name);
  2259. if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM) {
  2260. mlog(ML_ERROR, "%s: new master is %u "
  2261. "but no dead node!\n",
  2262. dlm->name, dlm->reco.new_master);
  2263. BUG();
  2264. }
  2265. }
  2266. spin_unlock(&dlm->spinlock);
  2267. }
  2268. /* if this node has actually become the recovery master,
  2269. * set the master and send the messages to begin recovery */
  2270. if (!status) {
  2271. mlog(0, "%s: dead=%u, this=%u, sending "
  2272. "begin_reco now\n", dlm->name,
  2273. dlm->reco.dead_node, dlm->node_num);
  2274. status = dlm_send_begin_reco_message(dlm,
  2275. dlm->reco.dead_node);
  2276. /* this always succeeds */
  2277. BUG_ON(status);
  2278. /* set the new_master to this node */
  2279. spin_lock(&dlm->spinlock);
  2280. dlm_set_reco_master(dlm, dlm->node_num);
  2281. spin_unlock(&dlm->spinlock);
  2282. }
  2283. /* recovery lock is a special case. ast will not get fired,
  2284. * so just go ahead and unlock it. */
  2285. ret = dlmunlock(dlm, &lksb, 0, dlm_reco_unlock_ast, dlm);
  2286. if (ret == DLM_DENIED) {
  2287. mlog(0, "got DLM_DENIED, trying LKM_CANCEL\n");
  2288. ret = dlmunlock(dlm, &lksb, LKM_CANCEL, dlm_reco_unlock_ast, dlm);
  2289. }
  2290. if (ret != DLM_NORMAL) {
  2291. /* this would really suck. this could only happen
  2292. * if there was a network error during the unlock
  2293. * because of node death. this means the unlock
  2294. * is actually "done" and the lock structure is
  2295. * even freed. we can continue, but only
  2296. * because this specific lock name is special. */
  2297. mlog(ML_ERROR, "dlmunlock returned %d\n", ret);
  2298. }
  2299. } else if (ret == DLM_NOTQUEUED) {
  2300. mlog(0, "dlm=%s dlmlock says another node got it (this=%u)\n",
  2301. dlm->name, dlm->node_num);
  2302. /* another node is master. wait on
  2303. * reco.new_master != O2NM_INVALID_NODE_NUM
  2304. * for at most one second */
  2305. wait_event_timeout(dlm->dlm_reco_thread_wq,
  2306. dlm_reco_master_ready(dlm),
  2307. msecs_to_jiffies(1000));
  2308. if (!dlm_reco_master_ready(dlm)) {
  2309. mlog(0, "%s: reco master taking awhile\n",
  2310. dlm->name);
  2311. goto again;
  2312. }
  2313. /* another node has informed this one that it is reco master */
  2314. mlog(0, "%s: reco master %u is ready to recover %u\n",
  2315. dlm->name, dlm->reco.new_master, dlm->reco.dead_node);
  2316. status = -EEXIST;
  2317. } else if (ret == DLM_RECOVERING) {
  2318. mlog(0, "dlm=%s dlmlock says master node died (this=%u)\n",
  2319. dlm->name, dlm->node_num);
  2320. goto again;
  2321. } else {
  2322. struct dlm_lock_resource *res;
  2323. /* dlmlock returned something other than NOTQUEUED or NORMAL */
  2324. mlog(ML_ERROR, "%s: got %s from dlmlock($RECOVERY), "
  2325. "lksb.status=%s\n", dlm->name, dlm_errname(ret),
  2326. dlm_errname(lksb.status));
  2327. res = dlm_lookup_lockres(dlm, DLM_RECOVERY_LOCK_NAME,
  2328. DLM_RECOVERY_LOCK_NAME_LEN);
  2329. if (res) {
  2330. dlm_print_one_lock_resource(res);
  2331. dlm_lockres_put(res);
  2332. } else {
  2333. mlog(ML_ERROR, "recovery lock not found\n");
  2334. }
  2335. BUG();
  2336. }
  2337. return status;
  2338. }
  2339. static int dlm_send_begin_reco_message(struct dlm_ctxt *dlm, u8 dead_node)
  2340. {
  2341. struct dlm_begin_reco br;
  2342. int ret = 0;
  2343. struct dlm_node_iter iter;
  2344. int nodenum;
  2345. int status;
  2346. mlog(0, "%s: dead node is %u\n", dlm->name, dead_node);
  2347. spin_lock(&dlm->spinlock);
  2348. dlm_node_iter_init(dlm->domain_map, &iter);
  2349. spin_unlock(&dlm->spinlock);
  2350. clear_bit(dead_node, iter.node_map);
  2351. memset(&br, 0, sizeof(br));
  2352. br.node_idx = dlm->node_num;
  2353. br.dead_node = dead_node;
  2354. while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
  2355. ret = 0;
  2356. if (nodenum == dead_node) {
  2357. mlog(0, "not sending begin reco to dead node "
  2358. "%u\n", dead_node);
  2359. continue;
  2360. }
  2361. if (nodenum == dlm->node_num) {
  2362. mlog(0, "not sending begin reco to self\n");
  2363. continue;
  2364. }
  2365. retry:
  2366. ret = -EINVAL;
  2367. mlog(0, "attempting to send begin reco msg to %d\n",
  2368. nodenum);
  2369. ret = o2net_send_message(DLM_BEGIN_RECO_MSG, dlm->key,
  2370. &br, sizeof(br), nodenum, &status);
  2371. /* negative status is handled ok by caller here */
  2372. if (ret >= 0)
  2373. ret = status;
  2374. if (dlm_is_host_down(ret)) {
  2375. /* node is down. not involved in recovery
  2376. * so just keep going */
  2377. mlog(ML_NOTICE, "%s: node %u was down when sending "
  2378. "begin reco msg (%d)\n", dlm->name, nodenum, ret);
  2379. ret = 0;
  2380. }
  2381. /*
  2382. * Prior to commit aad1b15310b9bcd59fa81ab8f2b1513b59553ea8,
  2383. * dlm_begin_reco_handler() returned EAGAIN and not -EAGAIN.
  2384. * We are handling both for compatibility reasons.
  2385. */
  2386. if (ret == -EAGAIN || ret == EAGAIN) {
  2387. mlog(0, "%s: trying to start recovery of node "
  2388. "%u, but node %u is waiting for last recovery "
  2389. "to complete, backoff for a bit\n", dlm->name,
  2390. dead_node, nodenum);
  2391. msleep(100);
  2392. goto retry;
  2393. }
  2394. if (ret < 0) {
  2395. struct dlm_lock_resource *res;
  2396. /* this is now a serious problem, possibly ENOMEM
  2397. * in the network stack. must retry */
  2398. mlog_errno(ret);
  2399. mlog(ML_ERROR, "begin reco of dlm %s to node %u "
  2400. "returned %d\n", dlm->name, nodenum, ret);
  2401. res = dlm_lookup_lockres(dlm, DLM_RECOVERY_LOCK_NAME,
  2402. DLM_RECOVERY_LOCK_NAME_LEN);
  2403. if (res) {
  2404. dlm_print_one_lock_resource(res);
  2405. dlm_lockres_put(res);
  2406. } else {
  2407. mlog(ML_ERROR, "recovery lock not found\n");
  2408. }
  2409. /* sleep for a bit in hopes that we can avoid
  2410. * another ENOMEM */
  2411. msleep(100);
  2412. goto retry;
  2413. }
  2414. }
  2415. return ret;
  2416. }
  2417. int dlm_begin_reco_handler(struct o2net_msg *msg, u32 len, void *data,
  2418. void **ret_data)
  2419. {
  2420. struct dlm_ctxt *dlm = data;
  2421. struct dlm_begin_reco *br = (struct dlm_begin_reco *)msg->buf;
  2422. /* ok to return 0, domain has gone away */
  2423. if (!dlm_grab(dlm))
  2424. return 0;
  2425. spin_lock(&dlm->spinlock);
  2426. if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
  2427. mlog(0, "%s: node %u wants to recover node %u (%u:%u) "
  2428. "but this node is in finalize state, waiting on finalize2\n",
  2429. dlm->name, br->node_idx, br->dead_node,
  2430. dlm->reco.dead_node, dlm->reco.new_master);
  2431. spin_unlock(&dlm->spinlock);
  2432. dlm_put(dlm);
  2433. return -EAGAIN;
  2434. }
  2435. spin_unlock(&dlm->spinlock);
  2436. mlog(0, "%s: node %u wants to recover node %u (%u:%u)\n",
  2437. dlm->name, br->node_idx, br->dead_node,
  2438. dlm->reco.dead_node, dlm->reco.new_master);
  2439. dlm_fire_domain_eviction_callbacks(dlm, br->dead_node);
  2440. spin_lock(&dlm->spinlock);
  2441. if (dlm->reco.new_master != O2NM_INVALID_NODE_NUM) {
  2442. if (test_bit(dlm->reco.new_master, dlm->recovery_map)) {
  2443. mlog(0, "%s: new_master %u died, changing "
  2444. "to %u\n", dlm->name, dlm->reco.new_master,
  2445. br->node_idx);
  2446. } else {
  2447. mlog(0, "%s: new_master %u NOT DEAD, changing "
  2448. "to %u\n", dlm->name, dlm->reco.new_master,
  2449. br->node_idx);
  2450. /* may not have seen the new master as dead yet */
  2451. }
  2452. }
  2453. if (dlm->reco.dead_node != O2NM_INVALID_NODE_NUM) {
  2454. mlog(ML_NOTICE, "%s: dead_node previously set to %u, "
  2455. "node %u changing it to %u\n", dlm->name,
  2456. dlm->reco.dead_node, br->node_idx, br->dead_node);
  2457. }
  2458. dlm_set_reco_master(dlm, br->node_idx);
  2459. dlm_set_reco_dead_node(dlm, br->dead_node);
  2460. if (!test_bit(br->dead_node, dlm->recovery_map)) {
  2461. mlog(0, "recovery master %u sees %u as dead, but this "
  2462. "node has not yet. marking %u as dead\n",
  2463. br->node_idx, br->dead_node, br->dead_node);
  2464. if (!test_bit(br->dead_node, dlm->domain_map) ||
  2465. !test_bit(br->dead_node, dlm->live_nodes_map))
  2466. mlog(0, "%u not in domain/live_nodes map "
  2467. "so setting it in reco map manually\n",
  2468. br->dead_node);
  2469. /* force the recovery cleanup in __dlm_hb_node_down
  2470. * both of these will be cleared in a moment */
  2471. set_bit(br->dead_node, dlm->domain_map);
  2472. set_bit(br->dead_node, dlm->live_nodes_map);
  2473. __dlm_hb_node_down(dlm, br->dead_node);
  2474. }
  2475. spin_unlock(&dlm->spinlock);
  2476. dlm_kick_recovery_thread(dlm);
  2477. mlog(0, "%s: recovery started by node %u, for %u (%u:%u)\n",
  2478. dlm->name, br->node_idx, br->dead_node,
  2479. dlm->reco.dead_node, dlm->reco.new_master);
  2480. dlm_put(dlm);
  2481. return 0;
  2482. }
  2483. #define DLM_FINALIZE_STAGE2 0x01
  2484. static int dlm_send_finalize_reco_message(struct dlm_ctxt *dlm)
  2485. {
  2486. int ret = 0;
  2487. struct dlm_finalize_reco fr;
  2488. struct dlm_node_iter iter;
  2489. int nodenum;
  2490. int status;
  2491. int stage = 1;
  2492. mlog(0, "finishing recovery for node %s:%u, "
  2493. "stage %d\n", dlm->name, dlm->reco.dead_node, stage);
  2494. spin_lock(&dlm->spinlock);
  2495. dlm_node_iter_init(dlm->domain_map, &iter);
  2496. spin_unlock(&dlm->spinlock);
  2497. stage2:
  2498. memset(&fr, 0, sizeof(fr));
  2499. fr.node_idx = dlm->node_num;
  2500. fr.dead_node = dlm->reco.dead_node;
  2501. if (stage == 2)
  2502. fr.flags |= DLM_FINALIZE_STAGE2;
  2503. while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
  2504. if (nodenum == dlm->node_num)
  2505. continue;
  2506. ret = o2net_send_message(DLM_FINALIZE_RECO_MSG, dlm->key,
  2507. &fr, sizeof(fr), nodenum, &status);
  2508. if (ret >= 0)
  2509. ret = status;
  2510. if (ret < 0) {
  2511. mlog(ML_ERROR, "Error %d when sending message %u (key "
  2512. "0x%x) to node %u\n", ret, DLM_FINALIZE_RECO_MSG,
  2513. dlm->key, nodenum);
  2514. if (dlm_is_host_down(ret)) {
  2515. /* this has no effect on this recovery
  2516. * session, so set the status to zero to
  2517. * finish out the last recovery */
  2518. mlog(ML_ERROR, "node %u went down after this "
  2519. "node finished recovery.\n", nodenum);
  2520. ret = 0;
  2521. continue;
  2522. }
  2523. break;
  2524. }
  2525. }
  2526. if (stage == 1) {
  2527. /* reset the node_iter back to the top and send finalize2 */
  2528. iter.curnode = -1;
  2529. stage = 2;
  2530. goto stage2;
  2531. }
  2532. return ret;
  2533. }
  2534. int dlm_finalize_reco_handler(struct o2net_msg *msg, u32 len, void *data,
  2535. void **ret_data)
  2536. {
  2537. struct dlm_ctxt *dlm = data;
  2538. struct dlm_finalize_reco *fr = (struct dlm_finalize_reco *)msg->buf;
  2539. int stage = 1;
  2540. /* ok to return 0, domain has gone away */
  2541. if (!dlm_grab(dlm))
  2542. return 0;
  2543. if (fr->flags & DLM_FINALIZE_STAGE2)
  2544. stage = 2;
  2545. mlog(0, "%s: node %u finalizing recovery stage%d of "
  2546. "node %u (%u:%u)\n", dlm->name, fr->node_idx, stage,
  2547. fr->dead_node, dlm->reco.dead_node, dlm->reco.new_master);
  2548. spin_lock(&dlm->spinlock);
  2549. if (dlm->reco.new_master != fr->node_idx) {
  2550. mlog(ML_ERROR, "node %u sent recovery finalize msg, but node "
  2551. "%u is supposed to be the new master, dead=%u\n",
  2552. fr->node_idx, dlm->reco.new_master, fr->dead_node);
  2553. BUG();
  2554. }
  2555. if (dlm->reco.dead_node != fr->dead_node) {
  2556. mlog(ML_ERROR, "node %u sent recovery finalize msg for dead "
  2557. "node %u, but node %u is supposed to be dead\n",
  2558. fr->node_idx, fr->dead_node, dlm->reco.dead_node);
  2559. BUG();
  2560. }
  2561. switch (stage) {
  2562. case 1:
  2563. dlm_finish_local_lockres_recovery(dlm, fr->dead_node, fr->node_idx);
  2564. if (dlm->reco.state & DLM_RECO_STATE_FINALIZE) {
  2565. mlog(ML_ERROR, "%s: received finalize1 from "
  2566. "new master %u for dead node %u, but "
  2567. "this node has already received it!\n",
  2568. dlm->name, fr->node_idx, fr->dead_node);
  2569. dlm_print_reco_node_status(dlm);
  2570. BUG();
  2571. }
  2572. dlm->reco.state |= DLM_RECO_STATE_FINALIZE;
  2573. spin_unlock(&dlm->spinlock);
  2574. break;
  2575. case 2:
  2576. if (!(dlm->reco.state & DLM_RECO_STATE_FINALIZE)) {
  2577. mlog(ML_ERROR, "%s: received finalize2 from "
  2578. "new master %u for dead node %u, but "
  2579. "this node did not have finalize1!\n",
  2580. dlm->name, fr->node_idx, fr->dead_node);
  2581. dlm_print_reco_node_status(dlm);
  2582. BUG();
  2583. }
  2584. dlm->reco.state &= ~DLM_RECO_STATE_FINALIZE;
  2585. __dlm_reset_recovery(dlm);
  2586. spin_unlock(&dlm->spinlock);
  2587. dlm_kick_recovery_thread(dlm);
  2588. break;
  2589. default:
  2590. BUG();
  2591. }
  2592. mlog(0, "%s: recovery done, reco master was %u, dead now %u, master now %u\n",
  2593. dlm->name, fr->node_idx, dlm->reco.dead_node, dlm->reco.new_master);
  2594. dlm_put(dlm);
  2595. return 0;
  2596. }