dlmunlock.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713
  1. /* -*- mode: c; c-basic-offset: 8; -*-
  2. * vim: noexpandtab sw=8 ts=8 sts=0:
  3. *
  4. * dlmunlock.c
  5. *
  6. * underlying calls for unlocking locks
  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/highmem.h>
  30. #include <linux/init.h>
  31. #include <linux/sysctl.h>
  32. #include <linux/random.h>
  33. #include <linux/blkdev.h>
  34. #include <linux/socket.h>
  35. #include <linux/inet.h>
  36. #include <linux/spinlock.h>
  37. #include <linux/delay.h>
  38. #include "cluster/heartbeat.h"
  39. #include "cluster/nodemanager.h"
  40. #include "cluster/tcp.h"
  41. #include "dlmapi.h"
  42. #include "dlmcommon.h"
  43. #define MLOG_MASK_PREFIX ML_DLM
  44. #include "cluster/masklog.h"
  45. #define DLM_UNLOCK_FREE_LOCK 0x00000001
  46. #define DLM_UNLOCK_CALL_AST 0x00000002
  47. #define DLM_UNLOCK_REMOVE_LOCK 0x00000004
  48. #define DLM_UNLOCK_REGRANT_LOCK 0x00000008
  49. #define DLM_UNLOCK_CLEAR_CONVERT_TYPE 0x00000010
  50. static enum dlm_status dlm_get_cancel_actions(struct dlm_ctxt *dlm,
  51. struct dlm_lock_resource *res,
  52. struct dlm_lock *lock,
  53. struct dlm_lockstatus *lksb,
  54. int *actions);
  55. static enum dlm_status dlm_get_unlock_actions(struct dlm_ctxt *dlm,
  56. struct dlm_lock_resource *res,
  57. struct dlm_lock *lock,
  58. struct dlm_lockstatus *lksb,
  59. int *actions);
  60. static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
  61. struct dlm_lock_resource *res,
  62. struct dlm_lock *lock,
  63. struct dlm_lockstatus *lksb,
  64. int flags,
  65. u8 owner);
  66. /*
  67. * according to the spec:
  68. * http://opendlm.sourceforge.net/cvsmirror/opendlm/docs/dlmbook_final.pdf
  69. *
  70. * flags & LKM_CANCEL != 0: must be converting or blocked
  71. * flags & LKM_CANCEL == 0: must be granted
  72. *
  73. * So to unlock a converting lock, you must first cancel the
  74. * convert (passing LKM_CANCEL in flags), then call the unlock
  75. * again (with no LKM_CANCEL in flags).
  76. */
  77. /*
  78. * locking:
  79. * caller needs: none
  80. * taken: res->spinlock and lock->spinlock taken and dropped
  81. * held on exit: none
  82. * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
  83. * all callers should have taken an extra ref on lock coming in
  84. */
  85. static enum dlm_status dlmunlock_common(struct dlm_ctxt *dlm,
  86. struct dlm_lock_resource *res,
  87. struct dlm_lock *lock,
  88. struct dlm_lockstatus *lksb,
  89. int flags, int *call_ast,
  90. int master_node)
  91. {
  92. enum dlm_status status;
  93. int actions = 0;
  94. int in_use;
  95. u8 owner;
  96. int recovery_wait = 0;
  97. mlog(0, "master_node = %d, valblk = %d\n", master_node,
  98. flags & LKM_VALBLK);
  99. if (master_node)
  100. BUG_ON(res->owner != dlm->node_num);
  101. else
  102. BUG_ON(res->owner == dlm->node_num);
  103. spin_lock(&dlm->ast_lock);
  104. /* We want to be sure that we're not freeing a lock
  105. * that still has AST's pending... */
  106. in_use = !list_empty(&lock->ast_list);
  107. spin_unlock(&dlm->ast_lock);
  108. if (in_use && !(flags & LKM_CANCEL)) {
  109. mlog(ML_ERROR, "lockres %.*s: Someone is calling dlmunlock "
  110. "while waiting for an ast!", res->lockname.len,
  111. res->lockname.name);
  112. return DLM_BADPARAM;
  113. }
  114. spin_lock(&res->spinlock);
  115. if (res->state & DLM_LOCK_RES_IN_PROGRESS) {
  116. if (master_node && !(flags & LKM_CANCEL)) {
  117. mlog(ML_ERROR, "lockres in progress!\n");
  118. spin_unlock(&res->spinlock);
  119. return DLM_FORWARD;
  120. }
  121. /* ok for this to sleep if not in a network handler */
  122. __dlm_wait_on_lockres(res);
  123. res->state |= DLM_LOCK_RES_IN_PROGRESS;
  124. }
  125. spin_lock(&lock->spinlock);
  126. if (res->state & DLM_LOCK_RES_RECOVERING) {
  127. status = DLM_RECOVERING;
  128. goto leave;
  129. }
  130. if (res->state & DLM_LOCK_RES_MIGRATING) {
  131. status = DLM_MIGRATING;
  132. goto leave;
  133. }
  134. /* see above for what the spec says about
  135. * LKM_CANCEL and the lock queue state */
  136. if (flags & LKM_CANCEL)
  137. status = dlm_get_cancel_actions(dlm, res, lock, lksb, &actions);
  138. else
  139. status = dlm_get_unlock_actions(dlm, res, lock, lksb, &actions);
  140. if (status != DLM_NORMAL && (status != DLM_CANCELGRANT || !master_node))
  141. goto leave;
  142. /* By now this has been masked out of cancel requests. */
  143. if (flags & LKM_VALBLK) {
  144. /* make the final update to the lvb */
  145. if (master_node)
  146. memcpy(res->lvb, lksb->lvb, DLM_LVB_LEN);
  147. else
  148. flags |= LKM_PUT_LVB; /* let the send function
  149. * handle it. */
  150. }
  151. if (!master_node) {
  152. owner = res->owner;
  153. /* drop locks and send message */
  154. if (flags & LKM_CANCEL)
  155. lock->cancel_pending = 1;
  156. else
  157. lock->unlock_pending = 1;
  158. spin_unlock(&lock->spinlock);
  159. spin_unlock(&res->spinlock);
  160. status = dlm_send_remote_unlock_request(dlm, res, lock, lksb,
  161. flags, owner);
  162. spin_lock(&res->spinlock);
  163. spin_lock(&lock->spinlock);
  164. /* if the master told us the lock was already granted,
  165. * let the ast handle all of these actions */
  166. if (status == DLM_CANCELGRANT) {
  167. actions &= ~(DLM_UNLOCK_REMOVE_LOCK|
  168. DLM_UNLOCK_REGRANT_LOCK|
  169. DLM_UNLOCK_CLEAR_CONVERT_TYPE);
  170. } else if (status == DLM_RECOVERING ||
  171. status == DLM_MIGRATING ||
  172. status == DLM_FORWARD ||
  173. status == DLM_NOLOCKMGR
  174. ) {
  175. /* must clear the actions because this unlock
  176. * is about to be retried. cannot free or do
  177. * any list manipulation. */
  178. mlog(0, "%s:%.*s: clearing actions, %s\n",
  179. dlm->name, res->lockname.len,
  180. res->lockname.name,
  181. status==DLM_RECOVERING?"recovering":
  182. (status==DLM_MIGRATING?"migrating":
  183. (status == DLM_FORWARD ? "forward" :
  184. "nolockmanager")));
  185. actions = 0;
  186. }
  187. if (flags & LKM_CANCEL)
  188. lock->cancel_pending = 0;
  189. else {
  190. if (!lock->unlock_pending)
  191. recovery_wait = 1;
  192. else
  193. lock->unlock_pending = 0;
  194. }
  195. }
  196. /* get an extra ref on lock. if we are just switching
  197. * lists here, we dont want the lock to go away. */
  198. dlm_lock_get(lock);
  199. if (actions & DLM_UNLOCK_REMOVE_LOCK) {
  200. list_del_init(&lock->list);
  201. dlm_lock_put(lock);
  202. }
  203. if (actions & DLM_UNLOCK_REGRANT_LOCK) {
  204. dlm_lock_get(lock);
  205. list_add_tail(&lock->list, &res->granted);
  206. }
  207. if (actions & DLM_UNLOCK_CLEAR_CONVERT_TYPE) {
  208. mlog(0, "clearing convert_type at %smaster node\n",
  209. master_node ? "" : "non-");
  210. lock->ml.convert_type = LKM_IVMODE;
  211. }
  212. /* remove the extra ref on lock */
  213. dlm_lock_put(lock);
  214. leave:
  215. res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
  216. if (!dlm_lock_on_list(&res->converting, lock))
  217. BUG_ON(lock->ml.convert_type != LKM_IVMODE);
  218. else
  219. BUG_ON(lock->ml.convert_type == LKM_IVMODE);
  220. spin_unlock(&lock->spinlock);
  221. spin_unlock(&res->spinlock);
  222. wake_up(&res->wq);
  223. if (recovery_wait) {
  224. spin_lock(&res->spinlock);
  225. /* Unlock request will directly succeed after owner dies,
  226. * and the lock is already removed from grant list. We have to
  227. * wait for RECOVERING done or we miss the chance to purge it
  228. * since the removement is much faster than RECOVERING proc.
  229. */
  230. __dlm_wait_on_lockres_flags(res, DLM_LOCK_RES_RECOVERING);
  231. spin_unlock(&res->spinlock);
  232. }
  233. /* let the caller's final dlm_lock_put handle the actual kfree */
  234. if (actions & DLM_UNLOCK_FREE_LOCK) {
  235. /* this should always be coupled with list removal */
  236. BUG_ON(!(actions & DLM_UNLOCK_REMOVE_LOCK));
  237. mlog(0, "lock %u:%llu should be gone now! refs=%d\n",
  238. dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
  239. dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
  240. kref_read(&lock->lock_refs)-1);
  241. dlm_lock_put(lock);
  242. }
  243. if (actions & DLM_UNLOCK_CALL_AST)
  244. *call_ast = 1;
  245. /* if cancel or unlock succeeded, lvb work is done */
  246. if (status == DLM_NORMAL)
  247. lksb->flags &= ~(DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB);
  248. return status;
  249. }
  250. void dlm_commit_pending_unlock(struct dlm_lock_resource *res,
  251. struct dlm_lock *lock)
  252. {
  253. /* leave DLM_LKSB_PUT_LVB on the lksb so any final
  254. * update of the lvb will be sent to the new master */
  255. list_del_init(&lock->list);
  256. }
  257. void dlm_commit_pending_cancel(struct dlm_lock_resource *res,
  258. struct dlm_lock *lock)
  259. {
  260. list_move_tail(&lock->list, &res->granted);
  261. lock->ml.convert_type = LKM_IVMODE;
  262. }
  263. static inline enum dlm_status dlmunlock_master(struct dlm_ctxt *dlm,
  264. struct dlm_lock_resource *res,
  265. struct dlm_lock *lock,
  266. struct dlm_lockstatus *lksb,
  267. int flags,
  268. int *call_ast)
  269. {
  270. return dlmunlock_common(dlm, res, lock, lksb, flags, call_ast, 1);
  271. }
  272. static inline enum dlm_status dlmunlock_remote(struct dlm_ctxt *dlm,
  273. struct dlm_lock_resource *res,
  274. struct dlm_lock *lock,
  275. struct dlm_lockstatus *lksb,
  276. int flags, int *call_ast)
  277. {
  278. return dlmunlock_common(dlm, res, lock, lksb, flags, call_ast, 0);
  279. }
  280. /*
  281. * locking:
  282. * caller needs: none
  283. * taken: none
  284. * held on exit: none
  285. * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
  286. */
  287. static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
  288. struct dlm_lock_resource *res,
  289. struct dlm_lock *lock,
  290. struct dlm_lockstatus *lksb,
  291. int flags,
  292. u8 owner)
  293. {
  294. struct dlm_unlock_lock unlock;
  295. int tmpret;
  296. enum dlm_status ret;
  297. int status = 0;
  298. struct kvec vec[2];
  299. size_t veclen = 1;
  300. mlog(0, "%.*s\n", res->lockname.len, res->lockname.name);
  301. if (owner == dlm->node_num) {
  302. /* ended up trying to contact ourself. this means
  303. * that the lockres had been remote but became local
  304. * via a migration. just retry it, now as local */
  305. mlog(0, "%s:%.*s: this node became the master due to a "
  306. "migration, re-evaluate now\n", dlm->name,
  307. res->lockname.len, res->lockname.name);
  308. return DLM_FORWARD;
  309. }
  310. memset(&unlock, 0, sizeof(unlock));
  311. unlock.node_idx = dlm->node_num;
  312. unlock.flags = cpu_to_be32(flags);
  313. unlock.cookie = lock->ml.cookie;
  314. unlock.namelen = res->lockname.len;
  315. memcpy(unlock.name, res->lockname.name, unlock.namelen);
  316. vec[0].iov_len = sizeof(struct dlm_unlock_lock);
  317. vec[0].iov_base = &unlock;
  318. if (flags & LKM_PUT_LVB) {
  319. /* extra data to send if we are updating lvb */
  320. vec[1].iov_len = DLM_LVB_LEN;
  321. vec[1].iov_base = lock->lksb->lvb;
  322. veclen++;
  323. }
  324. tmpret = o2net_send_message_vec(DLM_UNLOCK_LOCK_MSG, dlm->key,
  325. vec, veclen, owner, &status);
  326. if (tmpret >= 0) {
  327. // successfully sent and received
  328. if (status == DLM_FORWARD)
  329. mlog(0, "master was in-progress. retry\n");
  330. ret = status;
  331. } else {
  332. mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
  333. "node %u\n", tmpret, DLM_UNLOCK_LOCK_MSG, dlm->key, owner);
  334. if (dlm_is_host_down(tmpret)) {
  335. /* NOTE: this seems strange, but it is what we want.
  336. * when the master goes down during a cancel or
  337. * unlock, the recovery code completes the operation
  338. * as if the master had not died, then passes the
  339. * updated state to the recovery master. this thread
  340. * just needs to finish out the operation and call
  341. * the unlockast. */
  342. if (dlm_is_node_dead(dlm, owner))
  343. ret = DLM_NORMAL;
  344. else
  345. ret = DLM_NOLOCKMGR;
  346. } else {
  347. /* something bad. this will BUG in ocfs2 */
  348. ret = dlm_err_to_dlm_status(tmpret);
  349. }
  350. }
  351. return ret;
  352. }
  353. /*
  354. * locking:
  355. * caller needs: none
  356. * taken: takes and drops res->spinlock
  357. * held on exit: none
  358. * returns: DLM_NORMAL, DLM_BADARGS, DLM_IVLOCKID,
  359. * return value from dlmunlock_master
  360. */
  361. int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data,
  362. void **ret_data)
  363. {
  364. struct dlm_ctxt *dlm = data;
  365. struct dlm_unlock_lock *unlock = (struct dlm_unlock_lock *)msg->buf;
  366. struct dlm_lock_resource *res = NULL;
  367. struct dlm_lock *lock = NULL;
  368. enum dlm_status status = DLM_NORMAL;
  369. int found = 0, i;
  370. struct dlm_lockstatus *lksb = NULL;
  371. int ignore;
  372. u32 flags;
  373. struct list_head *queue;
  374. flags = be32_to_cpu(unlock->flags);
  375. if (flags & LKM_GET_LVB) {
  376. mlog(ML_ERROR, "bad args! GET_LVB specified on unlock!\n");
  377. return DLM_BADARGS;
  378. }
  379. if ((flags & (LKM_PUT_LVB|LKM_CANCEL)) == (LKM_PUT_LVB|LKM_CANCEL)) {
  380. mlog(ML_ERROR, "bad args! cannot modify lvb on a CANCEL "
  381. "request!\n");
  382. return DLM_BADARGS;
  383. }
  384. if (unlock->namelen > DLM_LOCKID_NAME_MAX) {
  385. mlog(ML_ERROR, "Invalid name length in unlock handler!\n");
  386. return DLM_IVBUFLEN;
  387. }
  388. if (!dlm_grab(dlm))
  389. return DLM_FORWARD;
  390. mlog_bug_on_msg(!dlm_domain_fully_joined(dlm),
  391. "Domain %s not fully joined!\n", dlm->name);
  392. mlog(0, "lvb: %s\n", flags & LKM_PUT_LVB ? "put lvb" : "none");
  393. res = dlm_lookup_lockres(dlm, unlock->name, unlock->namelen);
  394. if (!res) {
  395. /* We assume here that a no lock resource simply means
  396. * it was migrated away and destroyed before the other
  397. * node could detect it. */
  398. mlog(0, "returning DLM_FORWARD -- res no longer exists\n");
  399. status = DLM_FORWARD;
  400. goto not_found;
  401. }
  402. queue=&res->granted;
  403. found = 0;
  404. spin_lock(&res->spinlock);
  405. if (res->state & DLM_LOCK_RES_RECOVERING) {
  406. spin_unlock(&res->spinlock);
  407. mlog(0, "returning DLM_RECOVERING\n");
  408. status = DLM_RECOVERING;
  409. goto leave;
  410. }
  411. if (res->state & DLM_LOCK_RES_MIGRATING) {
  412. spin_unlock(&res->spinlock);
  413. mlog(0, "returning DLM_MIGRATING\n");
  414. status = DLM_MIGRATING;
  415. goto leave;
  416. }
  417. if (res->owner != dlm->node_num) {
  418. spin_unlock(&res->spinlock);
  419. mlog(0, "returning DLM_FORWARD -- not master\n");
  420. status = DLM_FORWARD;
  421. goto leave;
  422. }
  423. for (i=0; i<3; i++) {
  424. list_for_each_entry(lock, queue, list) {
  425. if (lock->ml.cookie == unlock->cookie &&
  426. lock->ml.node == unlock->node_idx) {
  427. dlm_lock_get(lock);
  428. found = 1;
  429. break;
  430. }
  431. }
  432. if (found)
  433. break;
  434. /* scan granted -> converting -> blocked queues */
  435. queue++;
  436. }
  437. spin_unlock(&res->spinlock);
  438. if (!found) {
  439. status = DLM_IVLOCKID;
  440. goto not_found;
  441. }
  442. /* lock was found on queue */
  443. lksb = lock->lksb;
  444. if (flags & (LKM_VALBLK|LKM_PUT_LVB) &&
  445. lock->ml.type != LKM_EXMODE)
  446. flags &= ~(LKM_VALBLK|LKM_PUT_LVB);
  447. /* unlockast only called on originating node */
  448. if (flags & LKM_PUT_LVB) {
  449. lksb->flags |= DLM_LKSB_PUT_LVB;
  450. memcpy(&lksb->lvb[0], &unlock->lvb[0], DLM_LVB_LEN);
  451. }
  452. /* if this is in-progress, propagate the DLM_FORWARD
  453. * all the way back out */
  454. status = dlmunlock_master(dlm, res, lock, lksb, flags, &ignore);
  455. if (status == DLM_FORWARD)
  456. mlog(0, "lockres is in progress\n");
  457. if (flags & LKM_PUT_LVB)
  458. lksb->flags &= ~DLM_LKSB_PUT_LVB;
  459. dlm_lockres_calc_usage(dlm, res);
  460. dlm_kick_thread(dlm, res);
  461. not_found:
  462. if (!found)
  463. mlog(ML_ERROR, "failed to find lock to unlock! "
  464. "cookie=%u:%llu\n",
  465. dlm_get_lock_cookie_node(be64_to_cpu(unlock->cookie)),
  466. dlm_get_lock_cookie_seq(be64_to_cpu(unlock->cookie)));
  467. else
  468. dlm_lock_put(lock);
  469. leave:
  470. if (res)
  471. dlm_lockres_put(res);
  472. dlm_put(dlm);
  473. return status;
  474. }
  475. static enum dlm_status dlm_get_cancel_actions(struct dlm_ctxt *dlm,
  476. struct dlm_lock_resource *res,
  477. struct dlm_lock *lock,
  478. struct dlm_lockstatus *lksb,
  479. int *actions)
  480. {
  481. enum dlm_status status;
  482. if (dlm_lock_on_list(&res->blocked, lock)) {
  483. /* cancel this outright */
  484. status = DLM_NORMAL;
  485. *actions = (DLM_UNLOCK_CALL_AST |
  486. DLM_UNLOCK_REMOVE_LOCK);
  487. } else if (dlm_lock_on_list(&res->converting, lock)) {
  488. /* cancel the request, put back on granted */
  489. status = DLM_NORMAL;
  490. *actions = (DLM_UNLOCK_CALL_AST |
  491. DLM_UNLOCK_REMOVE_LOCK |
  492. DLM_UNLOCK_REGRANT_LOCK |
  493. DLM_UNLOCK_CLEAR_CONVERT_TYPE);
  494. } else if (dlm_lock_on_list(&res->granted, lock)) {
  495. /* too late, already granted. */
  496. status = DLM_CANCELGRANT;
  497. *actions = DLM_UNLOCK_CALL_AST;
  498. } else {
  499. mlog(ML_ERROR, "lock to cancel is not on any list!\n");
  500. status = DLM_IVLOCKID;
  501. *actions = 0;
  502. }
  503. return status;
  504. }
  505. static enum dlm_status dlm_get_unlock_actions(struct dlm_ctxt *dlm,
  506. struct dlm_lock_resource *res,
  507. struct dlm_lock *lock,
  508. struct dlm_lockstatus *lksb,
  509. int *actions)
  510. {
  511. enum dlm_status status;
  512. /* unlock request */
  513. if (!dlm_lock_on_list(&res->granted, lock)) {
  514. status = DLM_DENIED;
  515. dlm_error(status);
  516. *actions = 0;
  517. } else {
  518. /* unlock granted lock */
  519. status = DLM_NORMAL;
  520. *actions = (DLM_UNLOCK_FREE_LOCK |
  521. DLM_UNLOCK_CALL_AST |
  522. DLM_UNLOCK_REMOVE_LOCK);
  523. }
  524. return status;
  525. }
  526. /* there seems to be no point in doing this async
  527. * since (even for the remote case) there is really
  528. * no work to queue up... so just do it and fire the
  529. * unlockast by hand when done... */
  530. enum dlm_status dlmunlock(struct dlm_ctxt *dlm, struct dlm_lockstatus *lksb,
  531. int flags, dlm_astunlockfunc_t *unlockast, void *data)
  532. {
  533. enum dlm_status status;
  534. struct dlm_lock_resource *res;
  535. struct dlm_lock *lock = NULL;
  536. int call_ast, is_master;
  537. if (!lksb) {
  538. dlm_error(DLM_BADARGS);
  539. return DLM_BADARGS;
  540. }
  541. if (flags & ~(LKM_CANCEL | LKM_VALBLK | LKM_INVVALBLK)) {
  542. dlm_error(DLM_BADPARAM);
  543. return DLM_BADPARAM;
  544. }
  545. if ((flags & (LKM_VALBLK | LKM_CANCEL)) == (LKM_VALBLK | LKM_CANCEL)) {
  546. mlog(0, "VALBLK given with CANCEL: ignoring VALBLK\n");
  547. flags &= ~LKM_VALBLK;
  548. }
  549. if (!lksb->lockid || !lksb->lockid->lockres) {
  550. dlm_error(DLM_BADPARAM);
  551. return DLM_BADPARAM;
  552. }
  553. lock = lksb->lockid;
  554. BUG_ON(!lock);
  555. dlm_lock_get(lock);
  556. res = lock->lockres;
  557. BUG_ON(!res);
  558. dlm_lockres_get(res);
  559. retry:
  560. call_ast = 0;
  561. /* need to retry up here because owner may have changed */
  562. mlog(0, "lock=%p res=%p\n", lock, res);
  563. spin_lock(&res->spinlock);
  564. is_master = (res->owner == dlm->node_num);
  565. if (flags & LKM_VALBLK && lock->ml.type != LKM_EXMODE)
  566. flags &= ~LKM_VALBLK;
  567. spin_unlock(&res->spinlock);
  568. if (is_master) {
  569. status = dlmunlock_master(dlm, res, lock, lksb, flags,
  570. &call_ast);
  571. mlog(0, "done calling dlmunlock_master: returned %d, "
  572. "call_ast is %d\n", status, call_ast);
  573. } else {
  574. status = dlmunlock_remote(dlm, res, lock, lksb, flags,
  575. &call_ast);
  576. mlog(0, "done calling dlmunlock_remote: returned %d, "
  577. "call_ast is %d\n", status, call_ast);
  578. }
  579. if (status == DLM_RECOVERING ||
  580. status == DLM_MIGRATING ||
  581. status == DLM_FORWARD ||
  582. status == DLM_NOLOCKMGR) {
  583. /* We want to go away for a tiny bit to allow recovery
  584. * / migration to complete on this resource. I don't
  585. * know of any wait queue we could sleep on as this
  586. * may be happening on another node. Perhaps the
  587. * proper solution is to queue up requests on the
  588. * other end? */
  589. /* do we want to yield(); ?? */
  590. msleep(50);
  591. mlog(0, "retrying unlock due to pending recovery/"
  592. "migration/in-progress/reconnect\n");
  593. goto retry;
  594. }
  595. if (call_ast) {
  596. mlog(0, "calling unlockast(%p, %d)\n", data, status);
  597. if (is_master) {
  598. /* it is possible that there is one last bast
  599. * pending. make sure it is flushed, then
  600. * call the unlockast.
  601. * not an issue if this is a mastered remotely,
  602. * since this lock has been removed from the
  603. * lockres queues and cannot be found. */
  604. dlm_kick_thread(dlm, NULL);
  605. wait_event(dlm->ast_wq,
  606. dlm_lock_basts_flushed(dlm, lock));
  607. }
  608. (*unlockast)(data, status);
  609. }
  610. if (status == DLM_CANCELGRANT)
  611. status = DLM_NORMAL;
  612. if (status == DLM_NORMAL) {
  613. mlog(0, "kicking the thread\n");
  614. dlm_kick_thread(dlm, res);
  615. } else
  616. dlm_error(status);
  617. dlm_lockres_calc_usage(dlm, res);
  618. dlm_lockres_put(res);
  619. dlm_lock_put(lock);
  620. mlog(0, "returning status=%d!\n", status);
  621. return status;
  622. }
  623. EXPORT_SYMBOL_GPL(dlmunlock);