aboutsummaryrefslogtreecommitdiffstats
path: root/fs
diff options
context:
space:
mode:
authorDavid Teigland <teigland@redhat.com>2010-02-24 12:59:23 -0500
committerDavid Teigland <teigland@redhat.com>2010-02-26 12:57:37 -0500
commitcf6620acc0f6fac57968aafef79ab372bdcf6157 (patch)
treefae381f0ffc0f13124f6138de9843140fc816cb3 /fs
parent7fe2b3190b8b299409f13cf3a6f85c2bd371f8bb (diff)
dlm: send reply before bast
When the lock master processes a successful operation (request, convert, cancel, or unlock), it will process the effects of the change before sending the reply for the operation. The "effects" of the operation are: - blocking callbacks (basts) for any newly granted locks - waiting or converting locks that can now be granted The cast is queued on the local node when the reply from the lock master is received. This means that a lock holder can receive a bast for a lock mode that is doesn't yet know has been granted. Signed-off-by: David Teigland <teigland@redhat.com>
Diffstat (limited to 'fs')
-rw-r--r--fs/dlm/lock.c110
1 files changed, 84 insertions, 26 deletions
diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c
index e08ea93432bc..d0e43a3da887 100644
--- a/fs/dlm/lock.c
+++ b/fs/dlm/lock.c
@@ -2280,20 +2280,30 @@ static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2280 if (can_be_queued(lkb)) { 2280 if (can_be_queued(lkb)) {
2281 error = -EINPROGRESS; 2281 error = -EINPROGRESS;
2282 add_lkb(r, lkb, DLM_LKSTS_WAITING); 2282 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2283 send_blocking_asts(r, lkb);
2284 add_timeout(lkb); 2283 add_timeout(lkb);
2285 goto out; 2284 goto out;
2286 } 2285 }
2287 2286
2288 error = -EAGAIN; 2287 error = -EAGAIN;
2289 if (force_blocking_asts(lkb))
2290 send_blocking_asts_all(r, lkb);
2291 queue_cast(r, lkb, -EAGAIN); 2288 queue_cast(r, lkb, -EAGAIN);
2292
2293 out: 2289 out:
2294 return error; 2290 return error;
2295} 2291}
2296 2292
2293static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2294 int error)
2295{
2296 switch (error) {
2297 case -EAGAIN:
2298 if (force_blocking_asts(lkb))
2299 send_blocking_asts_all(r, lkb);
2300 break;
2301 case -EINPROGRESS:
2302 send_blocking_asts(r, lkb);
2303 break;
2304 }
2305}
2306
2297static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) 2307static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2298{ 2308{
2299 int error = 0; 2309 int error = 0;
@@ -2304,7 +2314,6 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2304 if (can_be_granted(r, lkb, 1, &deadlk)) { 2314 if (can_be_granted(r, lkb, 1, &deadlk)) {
2305 grant_lock(r, lkb); 2315 grant_lock(r, lkb);
2306 queue_cast(r, lkb, 0); 2316 queue_cast(r, lkb, 0);
2307 grant_pending_locks(r);
2308 goto out; 2317 goto out;
2309 } 2318 }
2310 2319
@@ -2334,7 +2343,6 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2334 if (_can_be_granted(r, lkb, 1)) { 2343 if (_can_be_granted(r, lkb, 1)) {
2335 grant_lock(r, lkb); 2344 grant_lock(r, lkb);
2336 queue_cast(r, lkb, 0); 2345 queue_cast(r, lkb, 0);
2337 grant_pending_locks(r);
2338 goto out; 2346 goto out;
2339 } 2347 }
2340 /* else fall through and move to convert queue */ 2348 /* else fall through and move to convert queue */
@@ -2344,28 +2352,47 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2344 error = -EINPROGRESS; 2352 error = -EINPROGRESS;
2345 del_lkb(r, lkb); 2353 del_lkb(r, lkb);
2346 add_lkb(r, lkb, DLM_LKSTS_CONVERT); 2354 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2347 send_blocking_asts(r, lkb);
2348 add_timeout(lkb); 2355 add_timeout(lkb);
2349 goto out; 2356 goto out;
2350 } 2357 }
2351 2358
2352 error = -EAGAIN; 2359 error = -EAGAIN;
2353 if (force_blocking_asts(lkb))
2354 send_blocking_asts_all(r, lkb);
2355 queue_cast(r, lkb, -EAGAIN); 2360 queue_cast(r, lkb, -EAGAIN);
2356
2357 out: 2361 out:
2358 return error; 2362 return error;
2359} 2363}
2360 2364
2365static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2366 int error)
2367{
2368 switch (error) {
2369 case 0:
2370 grant_pending_locks(r);
2371 /* grant_pending_locks also sends basts */
2372 break;
2373 case -EAGAIN:
2374 if (force_blocking_asts(lkb))
2375 send_blocking_asts_all(r, lkb);
2376 break;
2377 case -EINPROGRESS:
2378 send_blocking_asts(r, lkb);
2379 break;
2380 }
2381}
2382
2361static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2383static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2362{ 2384{
2363 remove_lock(r, lkb); 2385 remove_lock(r, lkb);
2364 queue_cast(r, lkb, -DLM_EUNLOCK); 2386 queue_cast(r, lkb, -DLM_EUNLOCK);
2365 grant_pending_locks(r);
2366 return -DLM_EUNLOCK; 2387 return -DLM_EUNLOCK;
2367} 2388}
2368 2389
2390static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2391 int error)
2392{
2393 grant_pending_locks(r);
2394}
2395
2369/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */ 2396/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2370 2397
2371static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb) 2398static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
@@ -2375,12 +2402,18 @@ static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2375 error = revert_lock(r, lkb); 2402 error = revert_lock(r, lkb);
2376 if (error) { 2403 if (error) {
2377 queue_cast(r, lkb, -DLM_ECANCEL); 2404 queue_cast(r, lkb, -DLM_ECANCEL);
2378 grant_pending_locks(r);
2379 return -DLM_ECANCEL; 2405 return -DLM_ECANCEL;
2380 } 2406 }
2381 return 0; 2407 return 0;
2382} 2408}
2383 2409
2410static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2411 int error)
2412{
2413 if (error)
2414 grant_pending_locks(r);
2415}
2416
2384/* 2417/*
2385 * Four stage 3 varieties: 2418 * Four stage 3 varieties:
2386 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock() 2419 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
@@ -2402,11 +2435,15 @@ static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2402 goto out; 2435 goto out;
2403 } 2436 }
2404 2437
2405 if (is_remote(r)) 2438 if (is_remote(r)) {
2406 /* receive_request() calls do_request() on remote node */ 2439 /* receive_request() calls do_request() on remote node */
2407 error = send_request(r, lkb); 2440 error = send_request(r, lkb);
2408 else 2441 } else {
2409 error = do_request(r, lkb); 2442 error = do_request(r, lkb);
2443 /* for remote locks the request_reply is sent
2444 between do_request and do_request_effects */
2445 do_request_effects(r, lkb, error);
2446 }
2410 out: 2447 out:
2411 return error; 2448 return error;
2412} 2449}
@@ -2417,11 +2454,15 @@ static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2417{ 2454{
2418 int error; 2455 int error;
2419 2456
2420 if (is_remote(r)) 2457 if (is_remote(r)) {
2421 /* receive_convert() calls do_convert() on remote node */ 2458 /* receive_convert() calls do_convert() on remote node */
2422 error = send_convert(r, lkb); 2459 error = send_convert(r, lkb);
2423 else 2460 } else {
2424 error = do_convert(r, lkb); 2461 error = do_convert(r, lkb);
2462 /* for remote locks the convert_reply is sent
2463 between do_convert and do_convert_effects */
2464 do_convert_effects(r, lkb, error);
2465 }
2425 2466
2426 return error; 2467 return error;
2427} 2468}
@@ -2432,11 +2473,15 @@ static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2432{ 2473{
2433 int error; 2474 int error;
2434 2475
2435 if (is_remote(r)) 2476 if (is_remote(r)) {
2436 /* receive_unlock() calls do_unlock() on remote node */ 2477 /* receive_unlock() calls do_unlock() on remote node */
2437 error = send_unlock(r, lkb); 2478 error = send_unlock(r, lkb);
2438 else 2479 } else {
2439 error = do_unlock(r, lkb); 2480 error = do_unlock(r, lkb);
2481 /* for remote locks the unlock_reply is sent
2482 between do_unlock and do_unlock_effects */
2483 do_unlock_effects(r, lkb, error);
2484 }
2440 2485
2441 return error; 2486 return error;
2442} 2487}
@@ -2447,11 +2492,15 @@ static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2447{ 2492{
2448 int error; 2493 int error;
2449 2494
2450 if (is_remote(r)) 2495 if (is_remote(r)) {
2451 /* receive_cancel() calls do_cancel() on remote node */ 2496 /* receive_cancel() calls do_cancel() on remote node */
2452 error = send_cancel(r, lkb); 2497 error = send_cancel(r, lkb);
2453 else 2498 } else {
2454 error = do_cancel(r, lkb); 2499 error = do_cancel(r, lkb);
2500 /* for remote locks the cancel_reply is sent
2501 between do_cancel and do_cancel_effects */
2502 do_cancel_effects(r, lkb, error);
2503 }
2455 2504
2456 return error; 2505 return error;
2457} 2506}
@@ -3191,6 +3240,7 @@ static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3191 attach_lkb(r, lkb); 3240 attach_lkb(r, lkb);
3192 error = do_request(r, lkb); 3241 error = do_request(r, lkb);
3193 send_request_reply(r, lkb, error); 3242 send_request_reply(r, lkb, error);
3243 do_request_effects(r, lkb, error);
3194 3244
3195 unlock_rsb(r); 3245 unlock_rsb(r);
3196 put_rsb(r); 3246 put_rsb(r);
@@ -3226,15 +3276,19 @@ static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3226 goto out; 3276 goto out;
3227 3277
3228 receive_flags(lkb, ms); 3278 receive_flags(lkb, ms);
3279
3229 error = receive_convert_args(ls, lkb, ms); 3280 error = receive_convert_args(ls, lkb, ms);
3230 if (error) 3281 if (error) {
3231 goto out_reply; 3282 send_convert_reply(r, lkb, error);
3283 goto out;
3284 }
3285
3232 reply = !down_conversion(lkb); 3286 reply = !down_conversion(lkb);
3233 3287
3234 error = do_convert(r, lkb); 3288 error = do_convert(r, lkb);
3235 out_reply:
3236 if (reply) 3289 if (reply)
3237 send_convert_reply(r, lkb, error); 3290 send_convert_reply(r, lkb, error);
3291 do_convert_effects(r, lkb, error);
3238 out: 3292 out:
3239 unlock_rsb(r); 3293 unlock_rsb(r);
3240 put_rsb(r); 3294 put_rsb(r);
@@ -3266,13 +3320,16 @@ static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3266 goto out; 3320 goto out;
3267 3321
3268 receive_flags(lkb, ms); 3322 receive_flags(lkb, ms);
3323
3269 error = receive_unlock_args(ls, lkb, ms); 3324 error = receive_unlock_args(ls, lkb, ms);
3270 if (error) 3325 if (error) {
3271 goto out_reply; 3326 send_unlock_reply(r, lkb, error);
3327 goto out;
3328 }
3272 3329
3273 error = do_unlock(r, lkb); 3330 error = do_unlock(r, lkb);
3274 out_reply:
3275 send_unlock_reply(r, lkb, error); 3331 send_unlock_reply(r, lkb, error);
3332 do_unlock_effects(r, lkb, error);
3276 out: 3333 out:
3277 unlock_rsb(r); 3334 unlock_rsb(r);
3278 put_rsb(r); 3335 put_rsb(r);
@@ -3307,6 +3364,7 @@ static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3307 3364
3308 error = do_cancel(r, lkb); 3365 error = do_cancel(r, lkb);
3309 send_cancel_reply(r, lkb, error); 3366 send_cancel_reply(r, lkb, error);
3367 do_cancel_effects(r, lkb, error);
3310 out: 3368 out:
3311 unlock_rsb(r); 3369 unlock_rsb(r);
3312 put_rsb(r); 3370 put_rsb(r);