X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_flock.c;h=4090ae09b4351fac783eb7a194970c815f4701b2;hb=e2af7fb3c91dfb13d34d8e1b2f2df8c09621f768;hp=914ed4eac78a0576aff034efac788fecec4090c5;hpb=158a5a2c08a0b5ff1652331915f1d3cd2f41384a;p=fs%2Flustre-release.git diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index 914ed4e..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,12 +24,14 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. 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) 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/ @@ -55,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); @@ -82,8 +72,8 @@ int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, static inline int ldlm_same_flock_owner(struct ldlm_lock *lock, struct ldlm_lock *new) { - return((new->l_policy_data.l_flock.pid == - lock->l_policy_data.l_flock.pid) && + return((new->l_policy_data.l_flock.owner == + lock->l_policy_data.l_flock.owner) && (new->l_export == lock->l_export)); } @@ -96,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) { @@ -123,32 +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; - pid_t req_pid = req->l_policy_data.l_flock.pid; - pid_t blocking_pid = blocking_lock->l_policy_data.l_flock.pid; + 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 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.pid != blocking_pid) || - (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_pid = lock->l_policy_data.l_flock.blocking_pid; - blocking_export = (struct obd_export *)(long) - lock->l_policy_data.l_flock.blocking_export; - if (blocking_pid == req_pid && 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; } @@ -158,7 +198,7 @@ ldlm_process_flock_lock(struct ldlm_lock *req, int *flags, int first_enq, ldlm_error_t *err, cfs_list_t *work_list) { struct ldlm_resource *res = req->l_resource; - struct ldlm_namespace *ns = res->lr_namespace; + struct ldlm_namespace *ns = ldlm_res_to_ns(res); cfs_list_t *tmp; cfs_list_t *ownlocks = NULL; struct ldlm_lock *lock = NULL; @@ -172,8 +212,9 @@ ldlm_process_flock_lock(struct ldlm_lock *req, int *flags, int first_enq, const struct ldlm_callback_suite null_cbs = { NULL }; ENTRY; - CDEBUG(D_DLMTRACE, "flags %#x pid %u mode %u start "LPU64" end "LPU64 - "\n", *flags, new->l_policy_data.l_flock.pid, mode, + CDEBUG(D_DLMTRACE, "flags %#x owner "LPU64" pid %u mode %u start "LPU64 + " end "LPU64"\n", *flags, new->l_policy_data.l_flock.owner, + new->l_policy_data.l_flock.pid, mode, req->l_policy_data.l_flock.start, req->l_policy_data.l_flock.end); @@ -250,17 +291,8 @@ reprocess: RETURN(LDLM_ITER_STOP); } - req->l_policy_data.l_flock.blocking_pid = - lock->l_policy_data.l_flock.pid; - req->l_policy_data.l_flock.blocking_export = - (long)(void *)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); @@ -276,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. */ @@ -373,7 +403,7 @@ reprocess: * it must see the original lock data in the reply. */ /* XXX - if ldlm_lock_new() can sleep we should - * release the ns_lock, allocate the new lock, + * release the lr_lock, allocate the new lock, * and restart processing this lock. */ if (!new2) { unlock_res_and_lock(req); @@ -395,6 +425,8 @@ reprocess: new2->l_granted_mode = lock->l_granted_mode; new2->l_policy_data.l_flock.pid = new->l_policy_data.l_flock.pid; + new2->l_policy_data.l_flock.owner = + new->l_policy_data.l_flock.owner; new2->l_policy_data.l_flock.start = lock->l_policy_data.l_flock.start; new2->l_policy_data.l_flock.end = @@ -435,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 @@ -453,7 +486,7 @@ restart: &rpc_list); unlock_res_and_lock(req); - rc = ldlm_run_ast_work(&rpc_list, + rc = ldlm_run_ast_work(ns, &rpc_list, LDLM_WORK_CP_AST); lock_res_and_lock(req); if (rc == -ERESTART) @@ -463,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 @@ -490,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; } @@ -601,14 +641,12 @@ 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. */ cfs_list_del_init(&lock->l_res_link); - lock_res_and_lock(lock); if (flags & LDLM_FL_TEST_LOCK) { /* fcntl(F_GETLK) request */ /* The old mode was saved in getlk->fl_type so that if the mode @@ -646,17 +684,46 @@ 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 = lock->l_resource->lr_namespace; - /* 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_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; + /* Compat code, old clients had no idea about owner field and + * 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, + ldlm_wire_policy_data_t *wpolicy) +{ + memset(wpolicy, 0, sizeof(*wpolicy)); + wpolicy->l_flock.lfw_start = lpolicy->l_flock.start; + wpolicy->l_flock.lfw_end = lpolicy->l_flock.end; + wpolicy->l_flock.lfw_pid = lpolicy->l_flock.pid; + wpolicy->l_flock.lfw_owner = lpolicy->l_flock.owner; +}