[PATCH] OCFS2: The Second Oracle Cluster Filesystem
[sfrench/cifs-2.6.git] / fs / ocfs2 / dlm / userdlm.c
1 /* -*- mode: c; c-basic-offset: 8; -*-
2  * vim: noexpandtab sw=8 ts=8 sts=0:
3  *
4  * userdlm.c
5  *
6  * Code which implements the kernel side of a minimal userspace
7  * interface to our DLM.
8  *
9  * Many of the functions here are pared down versions of dlmglue.c
10  * functions.
11  *
12  * Copyright (C) 2003, 2004 Oracle.  All rights reserved.
13  *
14  * This program is free software; you can redistribute it and/or
15  * modify it under the terms of the GNU General Public
16  * License as published by the Free Software Foundation; either
17  * version 2 of the License, or (at your option) any later version.
18  *
19  * This program is distributed in the hope that it will be useful,
20  * but WITHOUT ANY WARRANTY; without even the implied warranty of
21  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
22  * General Public License for more details.
23  *
24  * You should have received a copy of the GNU General Public
25  * License along with this program; if not, write to the
26  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
27  * Boston, MA 021110-1307, USA.
28  */
29
30 #include <asm/signal.h>
31
32 #include <linux/module.h>
33 #include <linux/fs.h>
34 #include <linux/types.h>
35 #include <linux/crc32.h>
36
37
38 #include "cluster/nodemanager.h"
39 #include "cluster/heartbeat.h"
40 #include "cluster/tcp.h"
41
42 #include "dlmapi.h"
43
44 #include "userdlm.h"
45
46 #define MLOG_MASK_PREFIX ML_DLMFS
47 #include "cluster/masklog.h"
48
49 static inline int user_check_wait_flag(struct user_lock_res *lockres,
50                                        int flag)
51 {
52         int ret;
53
54         spin_lock(&lockres->l_lock);
55         ret = lockres->l_flags & flag;
56         spin_unlock(&lockres->l_lock);
57
58         return ret;
59 }
60
61 static inline void user_wait_on_busy_lock(struct user_lock_res *lockres)
62
63 {
64         wait_event(lockres->l_event,
65                    !user_check_wait_flag(lockres, USER_LOCK_BUSY));
66 }
67
68 static inline void user_wait_on_blocked_lock(struct user_lock_res *lockres)
69
70 {
71         wait_event(lockres->l_event,
72                    !user_check_wait_flag(lockres, USER_LOCK_BLOCKED));
73 }
74
75 /* I heart container_of... */
76 static inline struct dlm_ctxt *
77 dlm_ctxt_from_user_lockres(struct user_lock_res *lockres)
78 {
79         struct dlmfs_inode_private *ip;
80
81         ip = container_of(lockres,
82                           struct dlmfs_inode_private,
83                           ip_lockres);
84         return ip->ip_dlm;
85 }
86
87 static struct inode *
88 user_dlm_inode_from_user_lockres(struct user_lock_res *lockres)
89 {
90         struct dlmfs_inode_private *ip;
91
92         ip = container_of(lockres,
93                           struct dlmfs_inode_private,
94                           ip_lockres);
95         return &ip->ip_vfs_inode;
96 }
97
98 static inline void user_recover_from_dlm_error(struct user_lock_res *lockres)
99 {
100         spin_lock(&lockres->l_lock);
101         lockres->l_flags &= ~USER_LOCK_BUSY;
102         spin_unlock(&lockres->l_lock);
103 }
104
105 #define user_log_dlm_error(_func, _stat, _lockres) do {         \
106         mlog(ML_ERROR, "Dlm error \"%s\" while calling %s on "  \
107                 "resource %s: %s\n", dlm_errname(_stat), _func, \
108                 _lockres->l_name, dlm_errmsg(_stat));           \
109 } while (0)
110
111 /* WARNING: This function lives in a world where the only three lock
112  * levels are EX, PR, and NL. It *will* have to be adjusted when more
113  * lock types are added. */
114 static inline int user_highest_compat_lock_level(int level)
115 {
116         int new_level = LKM_EXMODE;
117
118         if (level == LKM_EXMODE)
119                 new_level = LKM_NLMODE;
120         else if (level == LKM_PRMODE)
121                 new_level = LKM_PRMODE;
122         return new_level;
123 }
124
125 static void user_ast(void *opaque)
126 {
127         struct user_lock_res *lockres = opaque;
128         struct dlm_lockstatus *lksb;
129
130         mlog(0, "AST fired for lockres %s\n", lockres->l_name);
131
132         spin_lock(&lockres->l_lock);
133
134         lksb = &(lockres->l_lksb);
135         if (lksb->status != DLM_NORMAL) {
136                 mlog(ML_ERROR, "lksb status value of %u on lockres %s\n",
137                      lksb->status, lockres->l_name);
138                 spin_unlock(&lockres->l_lock);
139                 return;
140         }
141
142         /* we're downconverting. */
143         if (lockres->l_requested < lockres->l_level) {
144                 if (lockres->l_requested <=
145                     user_highest_compat_lock_level(lockres->l_blocking)) {
146                         lockres->l_blocking = LKM_NLMODE;
147                         lockres->l_flags &= ~USER_LOCK_BLOCKED;
148                 }
149         }
150
151         lockres->l_level = lockres->l_requested;
152         lockres->l_requested = LKM_IVMODE;
153         lockres->l_flags |= USER_LOCK_ATTACHED;
154         lockres->l_flags &= ~USER_LOCK_BUSY;
155
156         spin_unlock(&lockres->l_lock);
157
158         wake_up(&lockres->l_event);
159 }
160
161 static inline void user_dlm_grab_inode_ref(struct user_lock_res *lockres)
162 {
163         struct inode *inode;
164         inode = user_dlm_inode_from_user_lockres(lockres);
165         if (!igrab(inode))
166                 BUG();
167 }
168
169 static void user_dlm_unblock_lock(void *opaque);
170
171 static void __user_dlm_queue_lockres(struct user_lock_res *lockres)
172 {
173         if (!(lockres->l_flags & USER_LOCK_QUEUED)) {
174                 user_dlm_grab_inode_ref(lockres);
175
176                 INIT_WORK(&lockres->l_work, user_dlm_unblock_lock,
177                           lockres);
178
179                 queue_work(user_dlm_worker, &lockres->l_work);
180                 lockres->l_flags |= USER_LOCK_QUEUED;
181         }
182 }
183
184 static void __user_dlm_cond_queue_lockres(struct user_lock_res *lockres)
185 {
186         int queue = 0;
187
188         if (!(lockres->l_flags & USER_LOCK_BLOCKED))
189                 return;
190
191         switch (lockres->l_blocking) {
192         case LKM_EXMODE:
193                 if (!lockres->l_ex_holders && !lockres->l_ro_holders)
194                         queue = 1;
195                 break;
196         case LKM_PRMODE:
197                 if (!lockres->l_ex_holders)
198                         queue = 1;
199                 break;
200         default:
201                 BUG();
202         }
203
204         if (queue)
205                 __user_dlm_queue_lockres(lockres);
206 }
207
208 static void user_bast(void *opaque, int level)
209 {
210         struct user_lock_res *lockres = opaque;
211
212         mlog(0, "Blocking AST fired for lockres %s. Blocking level %d\n",
213                 lockres->l_name, level);
214
215         spin_lock(&lockres->l_lock);
216         lockres->l_flags |= USER_LOCK_BLOCKED;
217         if (level > lockres->l_blocking)
218                 lockres->l_blocking = level;
219
220         __user_dlm_queue_lockres(lockres);
221         spin_unlock(&lockres->l_lock);
222
223         wake_up(&lockres->l_event);
224 }
225
226 static void user_unlock_ast(void *opaque, enum dlm_status status)
227 {
228         struct user_lock_res *lockres = opaque;
229
230         mlog(0, "UNLOCK AST called on lock %s\n", lockres->l_name);
231
232         if (status != DLM_NORMAL)
233                 mlog(ML_ERROR, "Dlm returns status %d\n", status);
234
235         spin_lock(&lockres->l_lock);
236         if (lockres->l_flags & USER_LOCK_IN_TEARDOWN)
237                 lockres->l_level = LKM_IVMODE;
238         else {
239                 lockres->l_requested = LKM_IVMODE; /* cancel an
240                                                     * upconvert
241                                                     * request. */
242                 lockres->l_flags &= ~USER_LOCK_IN_CANCEL;
243                 /* we want the unblock thread to look at it again
244                  * now. */
245                 __user_dlm_queue_lockres(lockres);
246         }
247
248         lockres->l_flags &= ~USER_LOCK_BUSY;
249         spin_unlock(&lockres->l_lock);
250
251         wake_up(&lockres->l_event);
252 }
253
254 static inline void user_dlm_drop_inode_ref(struct user_lock_res *lockres)
255 {
256         struct inode *inode;
257         inode = user_dlm_inode_from_user_lockres(lockres);
258         iput(inode);
259 }
260
261 static void user_dlm_unblock_lock(void *opaque)
262 {
263         int new_level, status;
264         struct user_lock_res *lockres = (struct user_lock_res *) opaque;
265         struct dlm_ctxt *dlm = dlm_ctxt_from_user_lockres(lockres);
266
267         mlog(0, "processing lockres %s\n", lockres->l_name);
268
269         spin_lock(&lockres->l_lock);
270
271         BUG_ON(!(lockres->l_flags & USER_LOCK_BLOCKED));
272         BUG_ON(!(lockres->l_flags & USER_LOCK_QUEUED));
273
274         /* notice that we don't clear USER_LOCK_BLOCKED here. That's
275          * for user_ast to do. */
276         lockres->l_flags &= ~USER_LOCK_QUEUED;
277
278         if (lockres->l_flags & USER_LOCK_IN_TEARDOWN) {
279                 mlog(0, "lock is in teardown so we do nothing\n");
280                 spin_unlock(&lockres->l_lock);
281                 goto drop_ref;
282         }
283
284         if (lockres->l_flags & USER_LOCK_BUSY) {
285                 mlog(0, "BUSY flag detected...\n");
286                 if (lockres->l_flags & USER_LOCK_IN_CANCEL) {
287                         spin_unlock(&lockres->l_lock);
288                         goto drop_ref;
289                 }
290
291                 lockres->l_flags |= USER_LOCK_IN_CANCEL;
292                 spin_unlock(&lockres->l_lock);
293
294                 status = dlmunlock(dlm,
295                                    &lockres->l_lksb,
296                                    LKM_CANCEL,
297                                    user_unlock_ast,
298                                    lockres);
299                 if (status == DLM_CANCELGRANT) {
300                         /* If we got this, then the ast was fired
301                          * before we could cancel. We cleanup our
302                          * state, and restart the function. */
303                         spin_lock(&lockres->l_lock);
304                         lockres->l_flags &= ~USER_LOCK_IN_CANCEL;
305                         spin_unlock(&lockres->l_lock);
306                 } else if (status != DLM_NORMAL)
307                         user_log_dlm_error("dlmunlock", status, lockres);
308                 goto drop_ref;
309         }
310
311         /* If there are still incompat holders, we can exit safely
312          * without worrying about re-queueing this lock as that will
313          * happen on the last call to user_cluster_unlock. */
314         if ((lockres->l_blocking == LKM_EXMODE)
315             && (lockres->l_ex_holders || lockres->l_ro_holders)) {
316                 spin_unlock(&lockres->l_lock);
317                 mlog(0, "can't downconvert for ex: ro = %u, ex = %u\n",
318                         lockres->l_ro_holders, lockres->l_ex_holders);
319                 goto drop_ref;
320         }
321
322         if ((lockres->l_blocking == LKM_PRMODE)
323             && lockres->l_ex_holders) {
324                 spin_unlock(&lockres->l_lock);
325                 mlog(0, "can't downconvert for pr: ex = %u\n",
326                         lockres->l_ex_holders);
327                 goto drop_ref;
328         }
329
330         /* yay, we can downconvert now. */
331         new_level = user_highest_compat_lock_level(lockres->l_blocking);
332         lockres->l_requested = new_level;
333         lockres->l_flags |= USER_LOCK_BUSY;
334         mlog(0, "Downconvert lock from %d to %d\n",
335                 lockres->l_level, new_level);
336         spin_unlock(&lockres->l_lock);
337
338         /* need lock downconvert request now... */
339         status = dlmlock(dlm,
340                          new_level,
341                          &lockres->l_lksb,
342                          LKM_CONVERT|LKM_VALBLK,
343                          lockres->l_name,
344                          user_ast,
345                          lockres,
346                          user_bast);
347         if (status != DLM_NORMAL) {
348                 user_log_dlm_error("dlmlock", status, lockres);
349                 user_recover_from_dlm_error(lockres);
350         }
351
352 drop_ref:
353         user_dlm_drop_inode_ref(lockres);
354 }
355
356 static inline void user_dlm_inc_holders(struct user_lock_res *lockres,
357                                         int level)
358 {
359         switch(level) {
360         case LKM_EXMODE:
361                 lockres->l_ex_holders++;
362                 break;
363         case LKM_PRMODE:
364                 lockres->l_ro_holders++;
365                 break;
366         default:
367                 BUG();
368         }
369 }
370
371 /* predict what lock level we'll be dropping down to on behalf
372  * of another node, and return true if the currently wanted
373  * level will be compatible with it. */
374 static inline int
375 user_may_continue_on_blocked_lock(struct user_lock_res *lockres,
376                                   int wanted)
377 {
378         BUG_ON(!(lockres->l_flags & USER_LOCK_BLOCKED));
379
380         return wanted <= user_highest_compat_lock_level(lockres->l_blocking);
381 }
382
383 int user_dlm_cluster_lock(struct user_lock_res *lockres,
384                           int level,
385                           int lkm_flags)
386 {
387         int status, local_flags;
388         struct dlm_ctxt *dlm = dlm_ctxt_from_user_lockres(lockres);
389
390         if (level != LKM_EXMODE &&
391             level != LKM_PRMODE) {
392                 mlog(ML_ERROR, "lockres %s: invalid request!\n",
393                      lockres->l_name);
394                 status = -EINVAL;
395                 goto bail;
396         }
397
398         mlog(0, "lockres %s: asking for %s lock, passed flags = 0x%x\n",
399                 lockres->l_name,
400                 (level == LKM_EXMODE) ? "LKM_EXMODE" : "LKM_PRMODE",
401                 lkm_flags);
402
403 again:
404         if (signal_pending(current)) {
405                 status = -ERESTARTSYS;
406                 goto bail;
407         }
408
409         spin_lock(&lockres->l_lock);
410
411         /* We only compare against the currently granted level
412          * here. If the lock is blocked waiting on a downconvert,
413          * we'll get caught below. */
414         if ((lockres->l_flags & USER_LOCK_BUSY) &&
415             (level > lockres->l_level)) {
416                 /* is someone sitting in dlm_lock? If so, wait on
417                  * them. */
418                 spin_unlock(&lockres->l_lock);
419
420                 user_wait_on_busy_lock(lockres);
421                 goto again;
422         }
423
424         if ((lockres->l_flags & USER_LOCK_BLOCKED) &&
425             (!user_may_continue_on_blocked_lock(lockres, level))) {
426                 /* is the lock is currently blocked on behalf of
427                  * another node */
428                 spin_unlock(&lockres->l_lock);
429
430                 user_wait_on_blocked_lock(lockres);
431                 goto again;
432         }
433
434         if (level > lockres->l_level) {
435                 local_flags = lkm_flags | LKM_VALBLK;
436                 if (lockres->l_level != LKM_IVMODE)
437                         local_flags |= LKM_CONVERT;
438
439                 lockres->l_requested = level;
440                 lockres->l_flags |= USER_LOCK_BUSY;
441                 spin_unlock(&lockres->l_lock);
442
443                 BUG_ON(level == LKM_IVMODE);
444                 BUG_ON(level == LKM_NLMODE);
445
446                 mlog(0, "lock %s, get lock from %d to level = %d\n",
447                         lockres->l_name, lockres->l_level, level);
448
449                 /* call dlm_lock to upgrade lock now */
450                 status = dlmlock(dlm,
451                                  level,
452                                  &lockres->l_lksb,
453                                  local_flags,
454                                  lockres->l_name,
455                                  user_ast,
456                                  lockres,
457                                  user_bast);
458                 if (status != DLM_NORMAL) {
459                         if ((lkm_flags & LKM_NOQUEUE) &&
460                             (status == DLM_NOTQUEUED))
461                                 status = -EAGAIN;
462                         else {
463                                 user_log_dlm_error("dlmlock", status, lockres);
464                                 status = -EINVAL;
465                         }
466                         user_recover_from_dlm_error(lockres);
467                         goto bail;
468                 }
469
470                 mlog(0, "lock %s, successfull return from dlmlock\n",
471                         lockres->l_name);
472
473                 user_wait_on_busy_lock(lockres);
474                 goto again;
475         }
476
477         user_dlm_inc_holders(lockres, level);
478         spin_unlock(&lockres->l_lock);
479
480         mlog(0, "lockres %s: Got %s lock!\n", lockres->l_name,
481                 (level == LKM_EXMODE) ? "LKM_EXMODE" : "LKM_PRMODE");
482
483         status = 0;
484 bail:
485         return status;
486 }
487
488 static inline void user_dlm_dec_holders(struct user_lock_res *lockres,
489                                         int level)
490 {
491         switch(level) {
492         case LKM_EXMODE:
493                 BUG_ON(!lockres->l_ex_holders);
494                 lockres->l_ex_holders--;
495                 break;
496         case LKM_PRMODE:
497                 BUG_ON(!lockres->l_ro_holders);
498                 lockres->l_ro_holders--;
499                 break;
500         default:
501                 BUG();
502         }
503 }
504
505 void user_dlm_cluster_unlock(struct user_lock_res *lockres,
506                              int level)
507 {
508         if (level != LKM_EXMODE &&
509             level != LKM_PRMODE) {
510                 mlog(ML_ERROR, "lockres %s: invalid request!\n", lockres->l_name);
511                 return;
512         }
513
514         mlog(0, "lockres %s: dropping %s lock\n", lockres->l_name,
515                 (level == LKM_EXMODE) ? "LKM_EXMODE" : "LKM_PRMODE");
516
517         spin_lock(&lockres->l_lock);
518         user_dlm_dec_holders(lockres, level);
519         __user_dlm_cond_queue_lockres(lockres);
520         spin_unlock(&lockres->l_lock);
521 }
522
523 void user_dlm_write_lvb(struct inode *inode,
524                         const char *val,
525                         unsigned int len)
526 {
527         struct user_lock_res *lockres = &DLMFS_I(inode)->ip_lockres;
528         char *lvb = lockres->l_lksb.lvb;
529
530         BUG_ON(len > DLM_LVB_LEN);
531
532         spin_lock(&lockres->l_lock);
533
534         BUG_ON(lockres->l_level < LKM_EXMODE);
535         memcpy(lvb, val, len);
536
537         spin_unlock(&lockres->l_lock);
538 }
539
540 void user_dlm_read_lvb(struct inode *inode,
541                        char *val,
542                        unsigned int len)
543 {
544         struct user_lock_res *lockres = &DLMFS_I(inode)->ip_lockres;
545         char *lvb = lockres->l_lksb.lvb;
546
547         BUG_ON(len > DLM_LVB_LEN);
548
549         spin_lock(&lockres->l_lock);
550
551         BUG_ON(lockres->l_level < LKM_PRMODE);
552         memcpy(val, lvb, len);
553
554         spin_unlock(&lockres->l_lock);
555 }
556
557 void user_dlm_lock_res_init(struct user_lock_res *lockres,
558                             struct dentry *dentry)
559 {
560         memset(lockres, 0, sizeof(*lockres));
561
562         spin_lock_init(&lockres->l_lock);
563         init_waitqueue_head(&lockres->l_event);
564         lockres->l_level = LKM_IVMODE;
565         lockres->l_requested = LKM_IVMODE;
566         lockres->l_blocking = LKM_IVMODE;
567
568         /* should have been checked before getting here. */
569         BUG_ON(dentry->d_name.len >= USER_DLM_LOCK_ID_MAX_LEN);
570
571         memcpy(lockres->l_name,
572                dentry->d_name.name,
573                dentry->d_name.len);
574 }
575
576 int user_dlm_destroy_lock(struct user_lock_res *lockres)
577 {
578         int status = -EBUSY;
579         struct dlm_ctxt *dlm = dlm_ctxt_from_user_lockres(lockres);
580
581         mlog(0, "asked to destroy %s\n", lockres->l_name);
582
583         spin_lock(&lockres->l_lock);
584         while (lockres->l_flags & USER_LOCK_BUSY) {
585                 spin_unlock(&lockres->l_lock);
586
587                 mlog(0, "lock %s is busy\n", lockres->l_name);
588
589                 user_wait_on_busy_lock(lockres);
590
591                 spin_lock(&lockres->l_lock);
592         }
593
594         if (lockres->l_ro_holders || lockres->l_ex_holders) {
595                 spin_unlock(&lockres->l_lock);
596                 mlog(0, "lock %s has holders\n", lockres->l_name);
597                 goto bail;
598         }
599
600         status = 0;
601         if (!(lockres->l_flags & USER_LOCK_ATTACHED)) {
602                 spin_unlock(&lockres->l_lock);
603                 mlog(0, "lock %s is not attached\n", lockres->l_name);
604                 goto bail;
605         }
606
607         lockres->l_flags &= ~USER_LOCK_ATTACHED;
608         lockres->l_flags |= USER_LOCK_BUSY;
609         lockres->l_flags |= USER_LOCK_IN_TEARDOWN;
610         spin_unlock(&lockres->l_lock);
611
612         mlog(0, "unlocking lockres %s\n", lockres->l_name);
613         status = dlmunlock(dlm,
614                            &lockres->l_lksb,
615                            LKM_VALBLK,
616                            user_unlock_ast,
617                            lockres);
618         if (status != DLM_NORMAL) {
619                 user_log_dlm_error("dlmunlock", status, lockres);
620                 status = -EINVAL;
621                 goto bail;
622         }
623
624         user_wait_on_busy_lock(lockres);
625
626         status = 0;
627 bail:
628         return status;
629 }
630
631 struct dlm_ctxt *user_dlm_register_context(struct qstr *name)
632 {
633         struct dlm_ctxt *dlm;
634         u32 dlm_key;
635         char *domain;
636
637         domain = kmalloc(name->len + 1, GFP_KERNEL);
638         if (!domain) {
639                 mlog_errno(-ENOMEM);
640                 return ERR_PTR(-ENOMEM);
641         }
642
643         dlm_key = crc32_le(0, name->name, name->len);
644
645         snprintf(domain, name->len + 1, "%.*s", name->len, name->name);
646
647         dlm = dlm_register_domain(domain, dlm_key);
648         if (IS_ERR(dlm))
649                 mlog_errno(PTR_ERR(dlm));
650
651         kfree(domain);
652         return dlm;
653 }
654
655 void user_dlm_unregister_context(struct dlm_ctxt *dlm)
656 {
657         dlm_unregister_domain(dlm);
658 }