aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorSunil Mushran <sunil.mushran@oracle.com>2007-01-29 18:19:16 -0500
committerMark Fasheh <mark.fasheh@oracle.com>2007-02-07 15:08:13 -0500
commitf3f854648de64c4b6f13f6f13113bc9525c621e5 (patch)
treeaeacd37bd30e2f416cb5eef88d2e2bc00b457f88
parentab81afd30bc154bb1e8749e5aeeffe9b93c90834 (diff)
ocfs2_dlm: Ensure correct ordering of set/clear refmap bit on lockres
Eventhough the set refmap bit message is sent before the clear refmap message, currently there is no guarentee that the set message will be handled before the clear. This patch prevents the clear refmap to be processed while the node is sending assert master messages to other nodes. (The set refmap message is sent as a response to the assert master request). Signed-off-by: Sunil Mushran <sunil.mushran@oracle.com> Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>
-rw-r--r--fs/ocfs2/dlm/dlmcommon.h6
-rw-r--r--fs/ocfs2/dlm/dlmmaster.c99
2 files changed, 94 insertions, 11 deletions
diff --git a/fs/ocfs2/dlm/dlmcommon.h b/fs/ocfs2/dlm/dlmcommon.h
index 3f554711efe5..2f4f5d4edb07 100644
--- a/fs/ocfs2/dlm/dlmcommon.h
+++ b/fs/ocfs2/dlm/dlmcommon.h
@@ -180,6 +180,11 @@ struct dlm_assert_master_priv
180 unsigned ignore_higher:1; 180 unsigned ignore_higher:1;
181}; 181};
182 182
183struct dlm_deref_lockres_priv
184{
185 struct dlm_lock_resource *deref_res;
186 u8 deref_node;
187};
183 188
184struct dlm_work_item 189struct dlm_work_item
185{ 190{
@@ -191,6 +196,7 @@ struct dlm_work_item
191 struct dlm_request_all_locks_priv ral; 196 struct dlm_request_all_locks_priv ral;
192 struct dlm_mig_lockres_priv ml; 197 struct dlm_mig_lockres_priv ml;
193 struct dlm_assert_master_priv am; 198 struct dlm_assert_master_priv am;
199 struct dlm_deref_lockres_priv dl;
194 } u; 200 } u;
195}; 201};
196 202
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 84f36db8ada3..77e4e6169a0d 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -102,6 +102,7 @@ static void dlm_assert_master_worker(struct dlm_work_item *item, void *data);
102static int dlm_do_assert_master(struct dlm_ctxt *dlm, 102static int dlm_do_assert_master(struct dlm_ctxt *dlm,
103 struct dlm_lock_resource *res, 103 struct dlm_lock_resource *res,
104 void *nodemap, u32 flags); 104 void *nodemap, u32 flags);
105static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data);
105 106
106static inline int dlm_mle_equal(struct dlm_ctxt *dlm, 107static inline int dlm_mle_equal(struct dlm_ctxt *dlm,
107 struct dlm_master_list_entry *mle, 108 struct dlm_master_list_entry *mle,
@@ -1717,6 +1718,11 @@ int dlm_do_assert_master(struct dlm_ctxt *dlm,
1717 unsigned int namelen = res->lockname.len; 1718 unsigned int namelen = res->lockname.len;
1718 1719
1719 BUG_ON(namelen > O2NM_MAX_NAME_LEN); 1720 BUG_ON(namelen > O2NM_MAX_NAME_LEN);
1721
1722 spin_lock(&res->spinlock);
1723 res->state |= DLM_LOCK_RES_SETREF_INPROG;
1724 spin_unlock(&res->spinlock);
1725
1720again: 1726again:
1721 reassert = 0; 1727 reassert = 0;
1722 1728
@@ -1789,6 +1795,11 @@ again:
1789 if (reassert) 1795 if (reassert)
1790 goto again; 1796 goto again;
1791 1797
1798 spin_lock(&res->spinlock);
1799 res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
1800 spin_unlock(&res->spinlock);
1801 wake_up(&res->wq);
1802
1792 return ret; 1803 return ret;
1793} 1804}
1794 1805
@@ -2296,6 +2307,9 @@ int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
2296 int ret = -EINVAL; 2307 int ret = -EINVAL;
2297 u8 node; 2308 u8 node;
2298 unsigned int hash; 2309 unsigned int hash;
2310 struct dlm_work_item *item;
2311 int cleared = 0;
2312 int dispatch = 0;
2299 2313
2300 if (!dlm_grab(dlm)) 2314 if (!dlm_grab(dlm))
2301 return 0; 2315 return 0;
@@ -2326,27 +2340,90 @@ int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
2326 spin_unlock(&dlm->spinlock); 2340 spin_unlock(&dlm->spinlock);
2327 2341
2328 spin_lock(&res->spinlock); 2342 spin_lock(&res->spinlock);
2329 BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF); 2343 if (res->state & DLM_LOCK_RES_SETREF_INPROG)
2330 if (test_bit(node, res->refmap)) { 2344 dispatch = 1;
2331 ret = 0; 2345 else {
2332 dlm_lockres_clear_refmap_bit(node, res); 2346 BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2333 } else { 2347 if (test_bit(node, res->refmap)) {
2334 mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref " 2348 dlm_lockres_clear_refmap_bit(node, res);
2335 "but it is already dropped!\n", dlm->name, namelen, 2349 cleared = 1;
2336 name, node); 2350 }
2337 __dlm_print_one_lock_resource(res);
2338 } 2351 }
2339 spin_unlock(&res->spinlock); 2352 spin_unlock(&res->spinlock);
2340 2353
2341 if (!ret) 2354 if (!dispatch) {
2342 dlm_lockres_calc_usage(dlm, res); 2355 if (cleared)
2356 dlm_lockres_calc_usage(dlm, res);
2357 else {
2358 mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2359 "but it is already dropped!\n", dlm->name,
2360 res->lockname.len, res->lockname.name, node);
2361 __dlm_print_one_lock_resource(res);
2362 }
2363 ret = 0;
2364 goto done;
2365 }
2366
2367 item = kzalloc(sizeof(*item), GFP_NOFS);
2368 if (!item) {
2369 ret = -ENOMEM;
2370 mlog_errno(ret);
2371 goto done;
2372 }
2373
2374 dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);
2375 item->u.dl.deref_res = res;
2376 item->u.dl.deref_node = node;
2377
2378 spin_lock(&dlm->work_lock);
2379 list_add_tail(&item->list, &dlm->work_list);
2380 spin_unlock(&dlm->work_lock);
2381
2382 queue_work(dlm->dlm_worker, &dlm->dispatched_work);
2383 return 0;
2384
2343done: 2385done:
2344 if (res) 2386 if (res)
2345 dlm_lockres_put(res); 2387 dlm_lockres_put(res);
2346 dlm_put(dlm); 2388 dlm_put(dlm);
2389
2347 return ret; 2390 return ret;
2348} 2391}
2349 2392
2393static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
2394{
2395 struct dlm_ctxt *dlm;
2396 struct dlm_lock_resource *res;
2397 u8 node;
2398 u8 cleared = 0;
2399
2400 dlm = item->dlm;
2401 res = item->u.dl.deref_res;
2402 node = item->u.dl.deref_node;
2403
2404 spin_lock(&res->spinlock);
2405 BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2406 if (test_bit(node, res->refmap)) {
2407 __dlm_wait_on_lockres_flags(res, DLM_LOCK_RES_SETREF_INPROG);
2408 dlm_lockres_clear_refmap_bit(node, res);
2409 cleared = 1;
2410 }
2411 spin_unlock(&res->spinlock);
2412
2413 if (cleared) {
2414 mlog(0, "%s:%.*s node %u ref dropped in dispatch\n",
2415 dlm->name, res->lockname.len, res->lockname.name, node);
2416 dlm_lockres_calc_usage(dlm, res);
2417 } else {
2418 mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2419 "but it is already dropped!\n", dlm->name,
2420 res->lockname.len, res->lockname.name, node);
2421 __dlm_print_one_lock_resource(res);
2422 }
2423
2424 dlm_lockres_put(res);
2425}
2426
2350 2427
2351/* 2428/*
2352 * DLM_MIGRATE_LOCKRES 2429 * DLM_MIGRATE_LOCKRES