X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_flock.c;h=4090ae09b4351fac783eb7a194970c815f4701b2;hb=e2af7fb3c91dfb13d34d8e1b2f2df8c09621f768;hp=7e8a754947f09a6fa2018438f4b8126803f151be;hpb=8c83e7d75989ef527e43a824a0dbe46bffabd07d;p=fs%2Flustre-release.git diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index 7e8a754..4090ae0 100644 --- a/lustre/ldlm/ldlm_flock.c +++ b/lustre/ldlm/ldlm_flock.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -26,15 +24,14 @@ * GPL HEADER END */ /* - * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. - * Use is subject to license terms. - * * Copyright (c) 2003 Hewlett-Packard Development Company LP. * Developed under the sponsorship of the US Government under * Subcontract No. B514193 * - * Copyright (c) 2011 Whamcloud, Inc. + * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. + * Use is subject to license terms. * + * Copyright (c) 2010, 2011, Whamcloud, Inc. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -58,16 +55,6 @@ #define l_flock_waitq l_lru -/** - * Wait queue for Posix lock deadlock detection, added with - * ldlm_lock::l_flock_waitq. - */ -static CFS_LIST_HEAD(ldlm_flock_waitq); -/** - * Lock protecting access to ldlm_flock_waitq. - */ -cfs_spinlock_t ldlm_flock_waitq_lock = CFS_SPIN_LOCK_UNLOCKED; - int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag); @@ -99,6 +86,43 @@ ldlm_flocks_overlap(struct ldlm_lock *lock, struct ldlm_lock *new) lock->l_policy_data.l_flock.start)); } +static inline void ldlm_flock_blocking_link(struct ldlm_lock *req, + struct ldlm_lock *lock) +{ + /* For server only */ + if (req->l_export == NULL) + return; + + LASSERT(cfs_list_empty(&req->l_flock_waitq)); + cfs_write_lock(&req->l_export->exp_flock_wait_lock); + + req->l_policy_data.l_flock.blocking_owner = + lock->l_policy_data.l_flock.owner; + req->l_policy_data.l_flock.blocking_export = + class_export_get(lock->l_export); + + cfs_list_add_tail(&req->l_flock_waitq, + &req->l_export->exp_flock_wait_list); + cfs_write_unlock(&req->l_export->exp_flock_wait_lock); +} + +static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req) +{ + /* For server only */ + if (req->l_export == NULL) + return; + + cfs_write_lock(&req->l_export->exp_flock_wait_lock); + if (!cfs_list_empty(&req->l_flock_waitq)) { + cfs_list_del_init(&req->l_flock_waitq); + + class_export_put(req->l_policy_data.l_flock.blocking_export); + req->l_policy_data.l_flock.blocking_owner = 0; + req->l_policy_data.l_flock.blocking_export = NULL; + } + cfs_write_unlock(&req->l_export->exp_flock_wait_lock); +} + static inline void ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags) { @@ -126,33 +150,45 @@ ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags) } static int -ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *blocking_lock) +ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock) { - struct obd_export *req_export = req->l_export; - struct obd_export *blocking_export = blocking_lock->l_export; + struct obd_export *req_exp = req->l_export; + struct obd_export *bl_exp = bl_lock->l_export; + struct obd_export *bl_exp_new; __u64 req_owner = req->l_policy_data.l_flock.owner; - __u64 blocking_owner = blocking_lock->l_policy_data.l_flock.owner; + __u64 bl_owner = bl_lock->l_policy_data.l_flock.owner; struct ldlm_lock *lock; - cfs_spin_lock(&ldlm_flock_waitq_lock); + /* For server only */ + if (req_exp == NULL) + return 0; + + class_export_get(bl_exp); restart: - cfs_list_for_each_entry(lock, &ldlm_flock_waitq, l_flock_waitq) { - if ((lock->l_policy_data.l_flock.owner != blocking_owner) || - (lock->l_export != blocking_export)) + cfs_read_lock(&bl_exp->exp_flock_wait_lock); + cfs_list_for_each_entry(lock, &bl_exp->exp_flock_wait_list, + l_flock_waitq) { + struct ldlm_flock *flock = &lock->l_policy_data.l_flock; + + /* want to find something from same client and same process */ + if (flock->owner != bl_owner) continue; - blocking_owner = lock->l_policy_data.l_flock.blocking_owner; - blocking_export = (struct obd_export *) - lock->l_policy_data.l_flock.blocking_export; - if (blocking_owner == req_owner && - blocking_export == req_export) { - cfs_spin_unlock(&ldlm_flock_waitq_lock); + bl_owner = flock->blocking_owner; + bl_exp_new = class_export_get(flock->blocking_export); + cfs_read_unlock(&bl_exp->exp_flock_wait_lock); + class_export_put(bl_exp); + bl_exp = bl_exp_new; + + if (bl_owner == req_owner && bl_exp == req_exp) { + class_export_put(bl_exp); return 1; } goto restart; } - cfs_spin_unlock(&ldlm_flock_waitq_lock); + cfs_read_unlock(&bl_exp->exp_flock_wait_lock); + class_export_put(bl_exp); return 0; } @@ -255,17 +291,8 @@ reprocess: RETURN(LDLM_ITER_STOP); } - req->l_policy_data.l_flock.blocking_owner = - lock->l_policy_data.l_flock.owner; - req->l_policy_data.l_flock.blocking_export = - lock->l_export; - - LASSERT(cfs_list_empty(&req->l_flock_waitq)); - cfs_spin_lock(&ldlm_flock_waitq_lock); - cfs_list_add_tail(&req->l_flock_waitq, - &ldlm_flock_waitq); - cfs_spin_unlock(&ldlm_flock_waitq_lock); + ldlm_flock_blocking_link(req, lock); ldlm_resource_add_lock(res, &res->lr_waiting, req); *flags |= LDLM_FL_BLOCK_GRANTED; RETURN(LDLM_ITER_STOP); @@ -281,9 +308,7 @@ reprocess: /* In case we had slept on this lock request take it off of the * deadlock detection waitq. */ - cfs_spin_lock(&ldlm_flock_waitq_lock); - cfs_list_del_init(&req->l_flock_waitq); - cfs_spin_unlock(&ldlm_flock_waitq_lock); + ldlm_flock_blocking_unlink(req); /* Scan the locks owned by this process that overlap this request. * We may have to merge or split existing locks. */ @@ -442,6 +467,7 @@ reprocess: } if (*flags != LDLM_FL_WAIT_NOREPROC) { +#ifdef HAVE_SERVER_SUPPORT if (first_enq) { /* If this is an unlock, reprocess the waitq and * send completions ASTs for locks that can now be @@ -470,6 +496,13 @@ restart: LASSERT(req->l_completion_ast); ldlm_add_ast_work_item(req, NULL, work_list); } +#else /* !HAVE_SERVER_SUPPORT */ + /* The only one possible case for client-side calls flock + * policy function is ldlm_flock_completion_ast inside which + * carries LDLM_FL_WAIT_NOREPROC flag. */ + CERROR("Illegal parameter for client-side-only module.\n"); + LBUG(); +#endif /* HAVE_SERVER_SUPPORT */ } /* In case we're reprocessing the requested lock we can't destroy @@ -497,12 +530,12 @@ ldlm_flock_interrupted_wait(void *data) lock = ((struct ldlm_flock_wait_data *)data)->fwd_lock; /* take lock off the deadlock detection waitq. */ - cfs_spin_lock(&ldlm_flock_waitq_lock); - cfs_list_del_init(&lock->l_flock_waitq); - cfs_spin_unlock(&ldlm_flock_waitq_lock); + ldlm_flock_blocking_unlink(lock); /* client side - set flag to prevent lock from being put on lru list */ + lock_res_and_lock(lock); lock->l_flags |= LDLM_FL_CBPENDING; + unlock_res_and_lock(lock); EXIT; } @@ -608,9 +641,7 @@ granted: LDLM_DEBUG(lock, "client-side enqueue granted"); /* take lock off the deadlock detection waitq. */ - cfs_spin_lock(&ldlm_flock_waitq_lock); - cfs_list_del_init(&lock->l_flock_waitq); - cfs_spin_unlock(&ldlm_flock_waitq_lock); + ldlm_flock_blocking_unlink(lock); lock_res_and_lock(lock); /* ldlm_lock_enqueue() has already placed lock on the granted list. */ @@ -653,33 +684,38 @@ EXPORT_SYMBOL(ldlm_flock_completion_ast); int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag) { - struct ldlm_namespace *ns; ENTRY; LASSERT(lock); LASSERT(flag == LDLM_CB_CANCELING); - ns = ldlm_lock_to_ns(lock); - /* take lock off the deadlock detection waitq. */ - cfs_spin_lock(&ldlm_flock_waitq_lock); - cfs_list_del_init(&lock->l_flock_waitq); - cfs_spin_unlock(&ldlm_flock_waitq_lock); + ldlm_flock_blocking_unlink(lock); RETURN(0); } -void ldlm_flock_policy_wire_to_local(const ldlm_wire_policy_data_t *wpolicy, - ldlm_policy_data_t *lpolicy) +void ldlm_flock_policy_wire18_to_local(const ldlm_wire_policy_data_t *wpolicy, + ldlm_policy_data_t *lpolicy) { memset(lpolicy, 0, sizeof(*lpolicy)); lpolicy->l_flock.start = wpolicy->l_flock.lfw_start; lpolicy->l_flock.end = wpolicy->l_flock.lfw_end; lpolicy->l_flock.pid = wpolicy->l_flock.lfw_pid; - lpolicy->l_flock.owner = wpolicy->l_flock.lfw_owner; /* Compat code, old clients had no idea about owner field and - * relied solely on pid for ownership. Introduced in 2.1, April 2011 */ - if (!lpolicy->l_flock.owner) - lpolicy->l_flock.owner = wpolicy->l_flock.lfw_pid; + * relied solely on pid for ownership. Introduced in LU-104, 2.1, + * April 2011 */ + lpolicy->l_flock.owner = wpolicy->l_flock.lfw_pid; +} + + +void ldlm_flock_policy_wire21_to_local(const ldlm_wire_policy_data_t *wpolicy, + ldlm_policy_data_t *lpolicy) +{ + memset(lpolicy, 0, sizeof(*lpolicy)); + lpolicy->l_flock.start = wpolicy->l_flock.lfw_start; + lpolicy->l_flock.end = wpolicy->l_flock.lfw_end; + lpolicy->l_flock.pid = wpolicy->l_flock.lfw_pid; + lpolicy->l_flock.owner = wpolicy->l_flock.lfw_owner; } void ldlm_flock_policy_local_to_wire(const ldlm_policy_data_t *lpolicy,