Merge branch 'master'
[linux-2.6] / fs / gfs2 / locking / dlm / thread.c
1 /*
2  * Copyright (C) Sistina Software, Inc.  1997-2003 All rights reserved.
3  * Copyright (C) 2004-2005 Red Hat, Inc.  All rights reserved.
4  *
5  * This copyrighted material is made available to anyone wishing to use,
6  * modify, copy, or redistribute it subject to the terms and conditions
7  * of the GNU General Public License v.2.
8  */
9
10 #include "lock_dlm.h"
11
12 /* A lock placed on this queue is re-submitted to DLM as soon as the lock_dlm
13    thread gets to it. */
14
15 static void queue_submit(struct gdlm_lock *lp)
16 {
17         struct gdlm_ls *ls = lp->ls;
18
19         spin_lock(&ls->async_lock);
20         list_add_tail(&lp->delay_list, &ls->submit);
21         spin_unlock(&ls->async_lock);
22         wake_up(&ls->thread_wait);
23 }
24
25 static void process_blocking(struct gdlm_lock *lp, int bast_mode)
26 {
27         struct gdlm_ls *ls = lp->ls;
28         unsigned int cb = 0;
29
30         switch (gdlm_make_lmstate(bast_mode)) {
31         case LM_ST_EXCLUSIVE:
32                 cb = LM_CB_NEED_E;
33                 break;
34         case LM_ST_DEFERRED:
35                 cb = LM_CB_NEED_D;
36                 break;
37         case LM_ST_SHARED:
38                 cb = LM_CB_NEED_S;
39                 break;
40         default:
41                 gdlm_assert(0, "unknown bast mode %u", lp->bast_mode);
42         }
43
44         ls->fscb(ls->fsdata, cb, &lp->lockname);
45 }
46
47 static void process_complete(struct gdlm_lock *lp)
48 {
49         struct gdlm_ls *ls = lp->ls;
50         struct lm_async_cb acb;
51         int16_t prev_mode = lp->cur;
52
53         memset(&acb, 0, sizeof(acb));
54
55         if (lp->lksb.sb_status == -DLM_ECANCEL) {
56                 log_info("complete dlm cancel %x,%llx flags %lx",
57                          lp->lockname.ln_type, lp->lockname.ln_number,
58                          lp->flags);
59
60                 lp->req = lp->cur;
61                 acb.lc_ret |= LM_OUT_CANCELED;
62                 if (lp->cur == DLM_LOCK_IV)
63                         lp->lksb.sb_lkid = 0;
64                 goto out;
65         }
66
67         if (test_and_clear_bit(LFL_DLM_UNLOCK, &lp->flags)) {
68                 if (lp->lksb.sb_status != -DLM_EUNLOCK) {
69                         log_info("unlock sb_status %d %x,%llx flags %lx",
70                                  lp->lksb.sb_status, lp->lockname.ln_type,
71                                  lp->lockname.ln_number, lp->flags);
72                         return;
73                 }
74
75                 lp->cur = DLM_LOCK_IV;
76                 lp->req = DLM_LOCK_IV;
77                 lp->lksb.sb_lkid = 0;
78
79                 if (test_and_clear_bit(LFL_UNLOCK_DELETE, &lp->flags)) {
80                         gdlm_delete_lp(lp);
81                         return;
82                 }
83                 goto out;
84         }
85
86         if (lp->lksb.sb_flags & DLM_SBF_VALNOTVALID)
87                 memset(lp->lksb.sb_lvbptr, 0, GDLM_LVB_SIZE);
88
89         if (lp->lksb.sb_flags & DLM_SBF_ALTMODE) {
90                 if (lp->req == DLM_LOCK_PR)
91                         lp->req = DLM_LOCK_CW;
92                 else if (lp->req == DLM_LOCK_CW)
93                         lp->req = DLM_LOCK_PR;
94         }
95
96         /*
97          * A canceled lock request.  The lock was just taken off the delayed
98          * list and was never even submitted to dlm.
99          */
100
101         if (test_and_clear_bit(LFL_CANCEL, &lp->flags)) {
102                 log_info("complete internal cancel %x,%llx",
103                          lp->lockname.ln_type, lp->lockname.ln_number);
104                 lp->req = lp->cur;
105                 acb.lc_ret |= LM_OUT_CANCELED;
106                 goto out;
107         }
108
109         /*
110          * An error occured.
111          */
112
113         if (lp->lksb.sb_status) {
114                 /* a "normal" error */
115                 if ((lp->lksb.sb_status == -EAGAIN) &&
116                     (lp->lkf & DLM_LKF_NOQUEUE)) {
117                         lp->req = lp->cur;
118                         if (lp->cur == DLM_LOCK_IV)
119                                 lp->lksb.sb_lkid = 0;
120                         goto out;
121                 }
122
123                 /* this could only happen with cancels I think */
124                 log_info("ast sb_status %d %x,%llx flags %lx",
125                          lp->lksb.sb_status, lp->lockname.ln_type,
126                          lp->lockname.ln_number, lp->flags);
127                 return;
128         }
129
130         /*
131          * This is an AST for an EX->EX conversion for sync_lvb from GFS.
132          */
133
134         if (test_and_clear_bit(LFL_SYNC_LVB, &lp->flags)) {
135                 complete(&lp->ast_wait);
136                 return;
137         }
138
139         /*
140          * A lock has been demoted to NL because it initially completed during
141          * BLOCK_LOCKS.  Now it must be requested in the originally requested
142          * mode.
143          */
144
145         if (test_and_clear_bit(LFL_REREQUEST, &lp->flags)) {
146                 gdlm_assert(lp->req == DLM_LOCK_NL, "%x,%llx",
147                             lp->lockname.ln_type, lp->lockname.ln_number);
148                 gdlm_assert(lp->prev_req > DLM_LOCK_NL, "%x,%llx",
149                             lp->lockname.ln_type, lp->lockname.ln_number);
150
151                 lp->cur = DLM_LOCK_NL;
152                 lp->req = lp->prev_req;
153                 lp->prev_req = DLM_LOCK_IV;
154                 lp->lkf &= ~DLM_LKF_CONVDEADLK;
155
156                 set_bit(LFL_NOCACHE, &lp->flags);
157
158                 if (test_bit(DFL_BLOCK_LOCKS, &ls->flags) &&
159                     !test_bit(LFL_NOBLOCK, &lp->flags))
160                         gdlm_queue_delayed(lp);
161                 else
162                         queue_submit(lp);
163                 return;
164         }
165
166         /*
167          * A request is granted during dlm recovery.  It may be granted
168          * because the locks of a failed node were cleared.  In that case,
169          * there may be inconsistent data beneath this lock and we must wait
170          * for recovery to complete to use it.  When gfs recovery is done this
171          * granted lock will be converted to NL and then reacquired in this
172          * granted state.
173          */
174
175         if (test_bit(DFL_BLOCK_LOCKS, &ls->flags) &&
176             !test_bit(LFL_NOBLOCK, &lp->flags) &&
177             lp->req != DLM_LOCK_NL) {
178
179                 lp->cur = lp->req;
180                 lp->prev_req = lp->req;
181                 lp->req = DLM_LOCK_NL;
182                 lp->lkf |= DLM_LKF_CONVERT;
183                 lp->lkf &= ~DLM_LKF_CONVDEADLK;
184
185                 log_debug("rereq %x,%llx id %x %d,%d",
186                           lp->lockname.ln_type, lp->lockname.ln_number,
187                           lp->lksb.sb_lkid, lp->cur, lp->req);
188
189                 set_bit(LFL_REREQUEST, &lp->flags);
190                 queue_submit(lp);
191                 return;
192         }
193
194         /*
195          * DLM demoted the lock to NL before it was granted so GFS must be
196          * told it cannot cache data for this lock.
197          */
198
199         if (lp->lksb.sb_flags & DLM_SBF_DEMOTED)
200                 set_bit(LFL_NOCACHE, &lp->flags);
201
202  out:
203         /*
204          * This is an internal lock_dlm lock
205          */
206
207         if (test_bit(LFL_INLOCK, &lp->flags)) {
208                 clear_bit(LFL_NOBLOCK, &lp->flags);
209                 lp->cur = lp->req;
210                 complete(&lp->ast_wait);
211                 return;
212         }
213
214         /*
215          * Normal completion of a lock request.  Tell GFS it now has the lock.
216          */
217
218         clear_bit(LFL_NOBLOCK, &lp->flags);
219         lp->cur = lp->req;
220
221         acb.lc_name = lp->lockname;
222         acb.lc_ret |= gdlm_make_lmstate(lp->cur);
223
224         if (!test_and_clear_bit(LFL_NOCACHE, &lp->flags) &&
225             (lp->cur > DLM_LOCK_NL) && (prev_mode > DLM_LOCK_NL))
226                 acb.lc_ret |= LM_OUT_CACHEABLE;
227
228         ls->fscb(ls->fsdata, LM_CB_ASYNC, &acb);
229 }
230
231 static inline int no_work(struct gdlm_ls *ls, int blocking)
232 {
233         int ret;
234
235         spin_lock(&ls->async_lock);
236         ret = list_empty(&ls->complete) && list_empty(&ls->submit);
237         if (ret && blocking)
238                 ret = list_empty(&ls->blocking);
239         spin_unlock(&ls->async_lock);
240
241         return ret;
242 }
243
244 static inline int check_drop(struct gdlm_ls *ls)
245 {
246         if (!ls->drop_locks_count)
247                 return 0;
248
249         if (time_after(jiffies, ls->drop_time + ls->drop_locks_period * HZ)) {
250                 ls->drop_time = jiffies;
251                 if (ls->all_locks_count >= ls->drop_locks_count)
252                         return 1;
253         }
254         return 0;
255 }
256
257 static int gdlm_thread(void *data)
258 {
259         struct gdlm_ls *ls = (struct gdlm_ls *) data;
260         struct gdlm_lock *lp = NULL;
261         int blist = 0;
262         uint8_t complete, blocking, submit, drop;
263         DECLARE_WAITQUEUE(wait, current);
264
265         /* Only thread1 is allowed to do blocking callbacks since gfs
266            may wait for a completion callback within a blocking cb. */
267
268         if (current == ls->thread1)
269                 blist = 1;
270
271         while (!kthread_should_stop()) {
272                 set_current_state(TASK_INTERRUPTIBLE);
273                 add_wait_queue(&ls->thread_wait, &wait);
274                 if (no_work(ls, blist))
275                         schedule();
276                 remove_wait_queue(&ls->thread_wait, &wait);
277                 set_current_state(TASK_RUNNING);
278
279                 complete = blocking = submit = drop = 0;
280
281                 spin_lock(&ls->async_lock);
282
283                 if (blist && !list_empty(&ls->blocking)) {
284                         lp = list_entry(ls->blocking.next, struct gdlm_lock,
285                                         blist);
286                         list_del_init(&lp->blist);
287                         blocking = lp->bast_mode;
288                         lp->bast_mode = 0;
289                 } else if (!list_empty(&ls->complete)) {
290                         lp = list_entry(ls->complete.next, struct gdlm_lock,
291                                         clist);
292                         list_del_init(&lp->clist);
293                         complete = 1;
294                 } else if (!list_empty(&ls->submit)) {
295                         lp = list_entry(ls->submit.next, struct gdlm_lock,
296                                         delay_list);
297                         list_del_init(&lp->delay_list);
298                         submit = 1;
299                 }
300
301                 drop = check_drop(ls);
302                 spin_unlock(&ls->async_lock);
303
304                 if (complete)
305                         process_complete(lp);
306
307                 else if (blocking)
308                         process_blocking(lp, blocking);
309
310                 else if (submit)
311                         gdlm_do_lock(lp);
312
313                 if (drop)
314                         ls->fscb(ls->fsdata, LM_CB_DROPLOCKS, NULL);
315
316                 schedule();
317         }
318
319         return 0;
320 }
321
322 int gdlm_init_threads(struct gdlm_ls *ls)
323 {
324         struct task_struct *p;
325         int error;
326
327         p = kthread_run(gdlm_thread, ls, "lock_dlm1");
328         error = IS_ERR(p);
329         if (error) {
330                 log_error("can't start lock_dlm1 thread %d", error);
331                 return error;
332         }
333         ls->thread1 = p;
334
335         p = kthread_run(gdlm_thread, ls, "lock_dlm2");
336         error = IS_ERR(p);
337         if (error) {
338                 log_error("can't start lock_dlm2 thread %d", error);
339                 kthread_stop(ls->thread1);
340                 return error;
341         }
342         ls->thread2 = p;
343
344         return 0;
345 }
346
347 void gdlm_release_threads(struct gdlm_ls *ls)
348 {
349         kthread_stop(ls->thread1);
350         kthread_stop(ls->thread2);
351 }
352