userdlm.c 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682
  1. // SPDX-License-Identifier: GPL-2.0-or-later
  2. /*
  3. * userdlm.c
  4. *
  5. * Code which implements the kernel side of a minimal userspace
  6. * interface to our DLM.
  7. *
  8. * Many of the functions here are pared down versions of dlmglue.c
  9. * functions.
  10. *
  11. * Copyright (C) 2003, 2004 Oracle. All rights reserved.
  12. */
  13. #include <linux/signal.h>
  14. #include <linux/sched/signal.h>
  15. #include <linux/module.h>
  16. #include <linux/fs.h>
  17. #include <linux/types.h>
  18. #include <linux/crc32.h>
  19. #include "../ocfs2_lockingver.h"
  20. #include "../stackglue.h"
  21. #include "userdlm.h"
  22. #define MLOG_MASK_PREFIX ML_DLMFS
  23. #include "../cluster/masklog.h"
  24. static inline struct user_lock_res *user_lksb_to_lock_res(struct ocfs2_dlm_lksb *lksb)
  25. {
  26. return container_of(lksb, struct user_lock_res, l_lksb);
  27. }
  28. static inline int user_check_wait_flag(struct user_lock_res *lockres,
  29. int flag)
  30. {
  31. int ret;
  32. spin_lock(&lockres->l_lock);
  33. ret = lockres->l_flags & flag;
  34. spin_unlock(&lockres->l_lock);
  35. return ret;
  36. }
  37. static inline void user_wait_on_busy_lock(struct user_lock_res *lockres)
  38. {
  39. wait_event(lockres->l_event,
  40. !user_check_wait_flag(lockres, USER_LOCK_BUSY));
  41. }
  42. static inline void user_wait_on_blocked_lock(struct user_lock_res *lockres)
  43. {
  44. wait_event(lockres->l_event,
  45. !user_check_wait_flag(lockres, USER_LOCK_BLOCKED));
  46. }
  47. /* I heart container_of... */
  48. static inline struct ocfs2_cluster_connection *
  49. cluster_connection_from_user_lockres(struct user_lock_res *lockres)
  50. {
  51. struct dlmfs_inode_private *ip;
  52. ip = container_of(lockres,
  53. struct dlmfs_inode_private,
  54. ip_lockres);
  55. return ip->ip_conn;
  56. }
  57. static struct inode *
  58. user_dlm_inode_from_user_lockres(struct user_lock_res *lockres)
  59. {
  60. struct dlmfs_inode_private *ip;
  61. ip = container_of(lockres,
  62. struct dlmfs_inode_private,
  63. ip_lockres);
  64. return &ip->ip_vfs_inode;
  65. }
  66. static inline void user_recover_from_dlm_error(struct user_lock_res *lockres)
  67. {
  68. spin_lock(&lockres->l_lock);
  69. lockres->l_flags &= ~USER_LOCK_BUSY;
  70. spin_unlock(&lockres->l_lock);
  71. }
  72. #define user_log_dlm_error(_func, _stat, _lockres) do { \
  73. mlog(ML_ERROR, "Dlm error %d while calling %s on " \
  74. "resource %.*s\n", _stat, _func, \
  75. _lockres->l_namelen, _lockres->l_name); \
  76. } while (0)
  77. /* WARNING: This function lives in a world where the only three lock
  78. * levels are EX, PR, and NL. It *will* have to be adjusted when more
  79. * lock types are added. */
  80. static inline int user_highest_compat_lock_level(int level)
  81. {
  82. int new_level = DLM_LOCK_EX;
  83. if (level == DLM_LOCK_EX)
  84. new_level = DLM_LOCK_NL;
  85. else if (level == DLM_LOCK_PR)
  86. new_level = DLM_LOCK_PR;
  87. return new_level;
  88. }
  89. static void user_ast(struct ocfs2_dlm_lksb *lksb)
  90. {
  91. struct user_lock_res *lockres = user_lksb_to_lock_res(lksb);
  92. int status;
  93. mlog(ML_BASTS, "AST fired for lockres %.*s, level %d => %d\n",
  94. lockres->l_namelen, lockres->l_name, lockres->l_level,
  95. lockres->l_requested);
  96. spin_lock(&lockres->l_lock);
  97. status = ocfs2_dlm_lock_status(&lockres->l_lksb);
  98. if (status) {
  99. mlog(ML_ERROR, "lksb status value of %u on lockres %.*s\n",
  100. status, lockres->l_namelen, lockres->l_name);
  101. spin_unlock(&lockres->l_lock);
  102. return;
  103. }
  104. mlog_bug_on_msg(lockres->l_requested == DLM_LOCK_IV,
  105. "Lockres %.*s, requested ivmode. flags 0x%x\n",
  106. lockres->l_namelen, lockres->l_name, lockres->l_flags);
  107. /* we're downconverting. */
  108. if (lockres->l_requested < lockres->l_level) {
  109. if (lockres->l_requested <=
  110. user_highest_compat_lock_level(lockres->l_blocking)) {
  111. lockres->l_blocking = DLM_LOCK_NL;
  112. lockres->l_flags &= ~USER_LOCK_BLOCKED;
  113. }
  114. }
  115. lockres->l_level = lockres->l_requested;
  116. lockres->l_requested = DLM_LOCK_IV;
  117. lockres->l_flags |= USER_LOCK_ATTACHED;
  118. lockres->l_flags &= ~USER_LOCK_BUSY;
  119. spin_unlock(&lockres->l_lock);
  120. wake_up(&lockres->l_event);
  121. }
  122. static inline void user_dlm_grab_inode_ref(struct user_lock_res *lockres)
  123. {
  124. struct inode *inode;
  125. inode = user_dlm_inode_from_user_lockres(lockres);
  126. if (!igrab(inode))
  127. BUG();
  128. }
  129. static void user_dlm_unblock_lock(struct work_struct *work);
  130. static void __user_dlm_queue_lockres(struct user_lock_res *lockres)
  131. {
  132. if (!(lockres->l_flags & USER_LOCK_QUEUED)) {
  133. user_dlm_grab_inode_ref(lockres);
  134. INIT_WORK(&lockres->l_work, user_dlm_unblock_lock);
  135. queue_work(user_dlm_worker, &lockres->l_work);
  136. lockres->l_flags |= USER_LOCK_QUEUED;
  137. }
  138. }
  139. static void __user_dlm_cond_queue_lockres(struct user_lock_res *lockres)
  140. {
  141. int queue = 0;
  142. if (!(lockres->l_flags & USER_LOCK_BLOCKED))
  143. return;
  144. switch (lockres->l_blocking) {
  145. case DLM_LOCK_EX:
  146. if (!lockres->l_ex_holders && !lockres->l_ro_holders)
  147. queue = 1;
  148. break;
  149. case DLM_LOCK_PR:
  150. if (!lockres->l_ex_holders)
  151. queue = 1;
  152. break;
  153. default:
  154. BUG();
  155. }
  156. if (queue)
  157. __user_dlm_queue_lockres(lockres);
  158. }
  159. static void user_bast(struct ocfs2_dlm_lksb *lksb, int level)
  160. {
  161. struct user_lock_res *lockres = user_lksb_to_lock_res(lksb);
  162. mlog(ML_BASTS, "BAST fired for lockres %.*s, blocking %d, level %d\n",
  163. lockres->l_namelen, lockres->l_name, level, lockres->l_level);
  164. spin_lock(&lockres->l_lock);
  165. lockres->l_flags |= USER_LOCK_BLOCKED;
  166. if (level > lockres->l_blocking)
  167. lockres->l_blocking = level;
  168. __user_dlm_queue_lockres(lockres);
  169. spin_unlock(&lockres->l_lock);
  170. wake_up(&lockres->l_event);
  171. }
  172. static void user_unlock_ast(struct ocfs2_dlm_lksb *lksb, int status)
  173. {
  174. struct user_lock_res *lockres = user_lksb_to_lock_res(lksb);
  175. mlog(ML_BASTS, "UNLOCK AST fired for lockres %.*s, flags 0x%x\n",
  176. lockres->l_namelen, lockres->l_name, lockres->l_flags);
  177. if (status)
  178. mlog(ML_ERROR, "dlm returns status %d\n", status);
  179. spin_lock(&lockres->l_lock);
  180. /* The teardown flag gets set early during the unlock process,
  181. * so test the cancel flag to make sure that this ast isn't
  182. * for a concurrent cancel. */
  183. if (lockres->l_flags & USER_LOCK_IN_TEARDOWN
  184. && !(lockres->l_flags & USER_LOCK_IN_CANCEL)) {
  185. lockres->l_level = DLM_LOCK_IV;
  186. } else if (status == DLM_CANCELGRANT) {
  187. /* We tried to cancel a convert request, but it was
  188. * already granted. Don't clear the busy flag - the
  189. * ast should've done this already. */
  190. BUG_ON(!(lockres->l_flags & USER_LOCK_IN_CANCEL));
  191. lockres->l_flags &= ~USER_LOCK_IN_CANCEL;
  192. goto out_noclear;
  193. } else {
  194. BUG_ON(!(lockres->l_flags & USER_LOCK_IN_CANCEL));
  195. /* Cancel succeeded, we want to re-queue */
  196. lockres->l_requested = DLM_LOCK_IV; /* cancel an
  197. * upconvert
  198. * request. */
  199. lockres->l_flags &= ~USER_LOCK_IN_CANCEL;
  200. /* we want the unblock thread to look at it again
  201. * now. */
  202. if (lockres->l_flags & USER_LOCK_BLOCKED)
  203. __user_dlm_queue_lockres(lockres);
  204. }
  205. lockres->l_flags &= ~USER_LOCK_BUSY;
  206. out_noclear:
  207. spin_unlock(&lockres->l_lock);
  208. wake_up(&lockres->l_event);
  209. }
  210. /*
  211. * This is the userdlmfs locking protocol version.
  212. *
  213. * See fs/ocfs2/dlmglue.c for more details on locking versions.
  214. */
  215. static struct ocfs2_locking_protocol user_dlm_lproto = {
  216. .lp_max_version = {
  217. .pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
  218. .pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
  219. },
  220. .lp_lock_ast = user_ast,
  221. .lp_blocking_ast = user_bast,
  222. .lp_unlock_ast = user_unlock_ast,
  223. };
  224. static inline void user_dlm_drop_inode_ref(struct user_lock_res *lockres)
  225. {
  226. struct inode *inode;
  227. inode = user_dlm_inode_from_user_lockres(lockres);
  228. iput(inode);
  229. }
  230. static void user_dlm_unblock_lock(struct work_struct *work)
  231. {
  232. int new_level, status;
  233. struct user_lock_res *lockres =
  234. container_of(work, struct user_lock_res, l_work);
  235. struct ocfs2_cluster_connection *conn =
  236. cluster_connection_from_user_lockres(lockres);
  237. mlog(0, "lockres %.*s\n", lockres->l_namelen, lockres->l_name);
  238. spin_lock(&lockres->l_lock);
  239. mlog_bug_on_msg(!(lockres->l_flags & USER_LOCK_QUEUED),
  240. "Lockres %.*s, flags 0x%x\n",
  241. lockres->l_namelen, lockres->l_name, lockres->l_flags);
  242. /* notice that we don't clear USER_LOCK_BLOCKED here. If it's
  243. * set, we want user_ast clear it. */
  244. lockres->l_flags &= ~USER_LOCK_QUEUED;
  245. /* It's valid to get here and no longer be blocked - if we get
  246. * several basts in a row, we might be queued by the first
  247. * one, the unblock thread might run and clear the queued
  248. * flag, and finally we might get another bast which re-queues
  249. * us before our ast for the downconvert is called. */
  250. if (!(lockres->l_flags & USER_LOCK_BLOCKED)) {
  251. mlog(ML_BASTS, "lockres %.*s USER_LOCK_BLOCKED\n",
  252. lockres->l_namelen, lockres->l_name);
  253. spin_unlock(&lockres->l_lock);
  254. goto drop_ref;
  255. }
  256. if (lockres->l_flags & USER_LOCK_IN_TEARDOWN) {
  257. mlog(ML_BASTS, "lockres %.*s USER_LOCK_IN_TEARDOWN\n",
  258. lockres->l_namelen, lockres->l_name);
  259. spin_unlock(&lockres->l_lock);
  260. goto drop_ref;
  261. }
  262. if (lockres->l_flags & USER_LOCK_BUSY) {
  263. if (lockres->l_flags & USER_LOCK_IN_CANCEL) {
  264. mlog(ML_BASTS, "lockres %.*s USER_LOCK_IN_CANCEL\n",
  265. lockres->l_namelen, lockres->l_name);
  266. spin_unlock(&lockres->l_lock);
  267. goto drop_ref;
  268. }
  269. lockres->l_flags |= USER_LOCK_IN_CANCEL;
  270. spin_unlock(&lockres->l_lock);
  271. status = ocfs2_dlm_unlock(conn, &lockres->l_lksb,
  272. DLM_LKF_CANCEL);
  273. if (status)
  274. user_log_dlm_error("ocfs2_dlm_unlock", status, lockres);
  275. goto drop_ref;
  276. }
  277. /* If there are still incompat holders, we can exit safely
  278. * without worrying about re-queueing this lock as that will
  279. * happen on the last call to user_cluster_unlock. */
  280. if ((lockres->l_blocking == DLM_LOCK_EX)
  281. && (lockres->l_ex_holders || lockres->l_ro_holders)) {
  282. spin_unlock(&lockres->l_lock);
  283. mlog(ML_BASTS, "lockres %.*s, EX/PR Holders %u,%u\n",
  284. lockres->l_namelen, lockres->l_name,
  285. lockres->l_ex_holders, lockres->l_ro_holders);
  286. goto drop_ref;
  287. }
  288. if ((lockres->l_blocking == DLM_LOCK_PR)
  289. && lockres->l_ex_holders) {
  290. spin_unlock(&lockres->l_lock);
  291. mlog(ML_BASTS, "lockres %.*s, EX Holders %u\n",
  292. lockres->l_namelen, lockres->l_name,
  293. lockres->l_ex_holders);
  294. goto drop_ref;
  295. }
  296. /* yay, we can downconvert now. */
  297. new_level = user_highest_compat_lock_level(lockres->l_blocking);
  298. lockres->l_requested = new_level;
  299. lockres->l_flags |= USER_LOCK_BUSY;
  300. mlog(ML_BASTS, "lockres %.*s, downconvert %d => %d\n",
  301. lockres->l_namelen, lockres->l_name, lockres->l_level, new_level);
  302. spin_unlock(&lockres->l_lock);
  303. /* need lock downconvert request now... */
  304. status = ocfs2_dlm_lock(conn, new_level, &lockres->l_lksb,
  305. DLM_LKF_CONVERT|DLM_LKF_VALBLK,
  306. lockres->l_name,
  307. lockres->l_namelen);
  308. if (status) {
  309. user_log_dlm_error("ocfs2_dlm_lock", status, lockres);
  310. user_recover_from_dlm_error(lockres);
  311. }
  312. drop_ref:
  313. user_dlm_drop_inode_ref(lockres);
  314. }
  315. static inline void user_dlm_inc_holders(struct user_lock_res *lockres,
  316. int level)
  317. {
  318. switch(level) {
  319. case DLM_LOCK_EX:
  320. lockres->l_ex_holders++;
  321. break;
  322. case DLM_LOCK_PR:
  323. lockres->l_ro_holders++;
  324. break;
  325. default:
  326. BUG();
  327. }
  328. }
  329. /* predict what lock level we'll be dropping down to on behalf
  330. * of another node, and return true if the currently wanted
  331. * level will be compatible with it. */
  332. static inline int
  333. user_may_continue_on_blocked_lock(struct user_lock_res *lockres,
  334. int wanted)
  335. {
  336. BUG_ON(!(lockres->l_flags & USER_LOCK_BLOCKED));
  337. return wanted <= user_highest_compat_lock_level(lockres->l_blocking);
  338. }
  339. int user_dlm_cluster_lock(struct user_lock_res *lockres,
  340. int level,
  341. int lkm_flags)
  342. {
  343. int status, local_flags;
  344. struct ocfs2_cluster_connection *conn =
  345. cluster_connection_from_user_lockres(lockres);
  346. if (level != DLM_LOCK_EX &&
  347. level != DLM_LOCK_PR) {
  348. mlog(ML_ERROR, "lockres %.*s: invalid request!\n",
  349. lockres->l_namelen, lockres->l_name);
  350. status = -EINVAL;
  351. goto bail;
  352. }
  353. mlog(ML_BASTS, "lockres %.*s, level %d, flags = 0x%x\n",
  354. lockres->l_namelen, lockres->l_name, level, lkm_flags);
  355. again:
  356. if (signal_pending(current)) {
  357. status = -ERESTARTSYS;
  358. goto bail;
  359. }
  360. spin_lock(&lockres->l_lock);
  361. if (lockres->l_flags & USER_LOCK_IN_TEARDOWN) {
  362. spin_unlock(&lockres->l_lock);
  363. status = -EAGAIN;
  364. goto bail;
  365. }
  366. /* We only compare against the currently granted level
  367. * here. If the lock is blocked waiting on a downconvert,
  368. * we'll get caught below. */
  369. if ((lockres->l_flags & USER_LOCK_BUSY) &&
  370. (level > lockres->l_level)) {
  371. /* is someone sitting in dlm_lock? If so, wait on
  372. * them. */
  373. spin_unlock(&lockres->l_lock);
  374. user_wait_on_busy_lock(lockres);
  375. goto again;
  376. }
  377. if ((lockres->l_flags & USER_LOCK_BLOCKED) &&
  378. (!user_may_continue_on_blocked_lock(lockres, level))) {
  379. /* is the lock is currently blocked on behalf of
  380. * another node */
  381. spin_unlock(&lockres->l_lock);
  382. user_wait_on_blocked_lock(lockres);
  383. goto again;
  384. }
  385. if (level > lockres->l_level) {
  386. local_flags = lkm_flags | DLM_LKF_VALBLK;
  387. if (lockres->l_level != DLM_LOCK_IV)
  388. local_flags |= DLM_LKF_CONVERT;
  389. lockres->l_requested = level;
  390. lockres->l_flags |= USER_LOCK_BUSY;
  391. spin_unlock(&lockres->l_lock);
  392. BUG_ON(level == DLM_LOCK_IV);
  393. BUG_ON(level == DLM_LOCK_NL);
  394. /* call dlm_lock to upgrade lock now */
  395. status = ocfs2_dlm_lock(conn, level, &lockres->l_lksb,
  396. local_flags, lockres->l_name,
  397. lockres->l_namelen);
  398. if (status) {
  399. if ((lkm_flags & DLM_LKF_NOQUEUE) &&
  400. (status != -EAGAIN))
  401. user_log_dlm_error("ocfs2_dlm_lock",
  402. status, lockres);
  403. user_recover_from_dlm_error(lockres);
  404. goto bail;
  405. }
  406. user_wait_on_busy_lock(lockres);
  407. goto again;
  408. }
  409. user_dlm_inc_holders(lockres, level);
  410. spin_unlock(&lockres->l_lock);
  411. status = 0;
  412. bail:
  413. return status;
  414. }
  415. static inline void user_dlm_dec_holders(struct user_lock_res *lockres,
  416. int level)
  417. {
  418. switch(level) {
  419. case DLM_LOCK_EX:
  420. BUG_ON(!lockres->l_ex_holders);
  421. lockres->l_ex_holders--;
  422. break;
  423. case DLM_LOCK_PR:
  424. BUG_ON(!lockres->l_ro_holders);
  425. lockres->l_ro_holders--;
  426. break;
  427. default:
  428. BUG();
  429. }
  430. }
  431. void user_dlm_cluster_unlock(struct user_lock_res *lockres,
  432. int level)
  433. {
  434. if (level != DLM_LOCK_EX &&
  435. level != DLM_LOCK_PR) {
  436. mlog(ML_ERROR, "lockres %.*s: invalid request!\n",
  437. lockres->l_namelen, lockres->l_name);
  438. return;
  439. }
  440. spin_lock(&lockres->l_lock);
  441. user_dlm_dec_holders(lockres, level);
  442. __user_dlm_cond_queue_lockres(lockres);
  443. spin_unlock(&lockres->l_lock);
  444. }
  445. void user_dlm_write_lvb(struct inode *inode,
  446. const char *val,
  447. unsigned int len)
  448. {
  449. struct user_lock_res *lockres = &DLMFS_I(inode)->ip_lockres;
  450. char *lvb;
  451. BUG_ON(len > DLM_LVB_LEN);
  452. spin_lock(&lockres->l_lock);
  453. BUG_ON(lockres->l_level < DLM_LOCK_EX);
  454. lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
  455. memcpy(lvb, val, len);
  456. spin_unlock(&lockres->l_lock);
  457. }
  458. bool user_dlm_read_lvb(struct inode *inode, char *val)
  459. {
  460. struct user_lock_res *lockres = &DLMFS_I(inode)->ip_lockres;
  461. char *lvb;
  462. bool ret = true;
  463. spin_lock(&lockres->l_lock);
  464. BUG_ON(lockres->l_level < DLM_LOCK_PR);
  465. if (ocfs2_dlm_lvb_valid(&lockres->l_lksb)) {
  466. lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
  467. memcpy(val, lvb, DLM_LVB_LEN);
  468. } else
  469. ret = false;
  470. spin_unlock(&lockres->l_lock);
  471. return ret;
  472. }
  473. void user_dlm_lock_res_init(struct user_lock_res *lockres,
  474. struct dentry *dentry)
  475. {
  476. memset(lockres, 0, sizeof(*lockres));
  477. spin_lock_init(&lockres->l_lock);
  478. init_waitqueue_head(&lockres->l_event);
  479. lockres->l_level = DLM_LOCK_IV;
  480. lockres->l_requested = DLM_LOCK_IV;
  481. lockres->l_blocking = DLM_LOCK_IV;
  482. /* should have been checked before getting here. */
  483. BUG_ON(dentry->d_name.len >= USER_DLM_LOCK_ID_MAX_LEN);
  484. memcpy(lockres->l_name,
  485. dentry->d_name.name,
  486. dentry->d_name.len);
  487. lockres->l_namelen = dentry->d_name.len;
  488. }
  489. int user_dlm_destroy_lock(struct user_lock_res *lockres)
  490. {
  491. int status = -EBUSY;
  492. struct ocfs2_cluster_connection *conn =
  493. cluster_connection_from_user_lockres(lockres);
  494. mlog(ML_BASTS, "lockres %.*s\n", lockres->l_namelen, lockres->l_name);
  495. spin_lock(&lockres->l_lock);
  496. if (lockres->l_flags & USER_LOCK_IN_TEARDOWN) {
  497. spin_unlock(&lockres->l_lock);
  498. goto bail;
  499. }
  500. lockres->l_flags |= USER_LOCK_IN_TEARDOWN;
  501. while (lockres->l_flags & USER_LOCK_BUSY) {
  502. spin_unlock(&lockres->l_lock);
  503. user_wait_on_busy_lock(lockres);
  504. spin_lock(&lockres->l_lock);
  505. }
  506. if (lockres->l_ro_holders || lockres->l_ex_holders) {
  507. lockres->l_flags &= ~USER_LOCK_IN_TEARDOWN;
  508. spin_unlock(&lockres->l_lock);
  509. goto bail;
  510. }
  511. status = 0;
  512. if (!(lockres->l_flags & USER_LOCK_ATTACHED)) {
  513. /*
  514. * lock is never requested, leave USER_LOCK_IN_TEARDOWN set
  515. * to avoid new lock request coming in.
  516. */
  517. spin_unlock(&lockres->l_lock);
  518. goto bail;
  519. }
  520. lockres->l_flags |= USER_LOCK_BUSY;
  521. spin_unlock(&lockres->l_lock);
  522. status = ocfs2_dlm_unlock(conn, &lockres->l_lksb, DLM_LKF_VALBLK);
  523. if (status) {
  524. spin_lock(&lockres->l_lock);
  525. lockres->l_flags &= ~USER_LOCK_IN_TEARDOWN;
  526. lockres->l_flags &= ~USER_LOCK_BUSY;
  527. spin_unlock(&lockres->l_lock);
  528. user_log_dlm_error("ocfs2_dlm_unlock", status, lockres);
  529. goto bail;
  530. }
  531. user_wait_on_busy_lock(lockres);
  532. status = 0;
  533. bail:
  534. return status;
  535. }
  536. static void user_dlm_recovery_handler_noop(int node_num,
  537. void *recovery_data)
  538. {
  539. /* We ignore recovery events */
  540. return;
  541. }
  542. void user_dlm_set_locking_protocol(void)
  543. {
  544. ocfs2_stack_glue_set_max_proto_version(&user_dlm_lproto.lp_max_version);
  545. }
  546. struct ocfs2_cluster_connection *user_dlm_register(const struct qstr *name)
  547. {
  548. int rc;
  549. struct ocfs2_cluster_connection *conn;
  550. rc = ocfs2_cluster_connect_agnostic(name->name, name->len,
  551. &user_dlm_lproto,
  552. user_dlm_recovery_handler_noop,
  553. NULL, &conn);
  554. if (rc)
  555. mlog_errno(rc);
  556. return rc ? ERR_PTR(rc) : conn;
  557. }
  558. void user_dlm_unregister(struct ocfs2_cluster_connection *conn)
  559. {
  560. ocfs2_cluster_disconnect(conn, 0);
  561. }