X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_flock.c;h=f59c354bd7031472cfda93c6e098327b63df46bc;hp=c988f28af0e107247b4e9c24534227e3e7d5f301;hb=ea766bd959cc2347b223d597b3f8a31e010bed72;hpb=2179aa60137557a8a289ee3b105a5efa7db90ac8 diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index c988f28..f59c354 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. @@ -33,13 +31,29 @@ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2011, Whamcloud, Inc. + * Copyright (c) 2010, 2012, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. */ +/** + * This file implements POSIX lock type for Lustre. + * Its policy properties are start and end of extent and PID. + * + * These locks are only done through MDS due to POSIX semantics requiring + * e.g. that locks could be only partially released and as such split into + * two parts, and also that two adjacent locks from the same process may be + * merged into a single wider lock. + * + * Lock modes are mapped like this: + * PR and PW for READ and WRITE locks + * NL to request a releasing of a portion of the lock + * + * These flock locks never timeout. + */ + #define DEBUG_SUBSYSTEM S_LDLM #ifdef __KERNEL__ @@ -55,8 +69,6 @@ #include "ldlm_internal.h" -#define l_flock_waitq l_lru - int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag); @@ -88,24 +100,34 @@ 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) +static inline int ldlm_flock_blocking_link(struct ldlm_lock *req, + struct ldlm_lock *lock) { + int rc = 0; + /* For server only */ if (req->l_export == NULL) - return; + return 0; + + if (unlikely(req->l_export->exp_flock_hash == NULL)) { + rc = ldlm_init_flock_export(req->l_export); + if (rc) + goto error; + } - LASSERT(cfs_list_empty(&req->l_flock_waitq)); - cfs_write_lock(&req->l_export->exp_flock_wait_lock); + LASSERT(cfs_hlist_unhashed(&req->l_exp_flock_hash)); 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); + lock->l_export; + req->l_policy_data.l_flock.blocking_refs = 0; + + cfs_hash_add(req->l_export->exp_flock_hash, + &req->l_policy_data.l_flock.owner, + &req->l_exp_flock_hash); +error: + return rc; } static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req) @@ -114,27 +136,24 @@ static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req) 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); + check_res_locked(req->l_resource); + if (req->l_export->exp_flock_hash != NULL && + !cfs_hlist_unhashed(&req->l_exp_flock_hash)) + cfs_hash_del(req->l_export->exp_flock_hash, + &req->l_policy_data.l_flock.owner, + &req->l_exp_flock_hash); } static inline void -ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags) +ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, __u64 flags) { ENTRY; - LDLM_DEBUG(lock, "ldlm_flock_destroy(mode: %d, flags: 0x%x)", - mode, flags); + LDLM_DEBUG(lock, "ldlm_flock_destroy(mode: %d, flags: 0x%llx)", + mode, flags); - /* Safe to not lock here, since it should be empty anyway */ - LASSERT(cfs_list_empty(&lock->l_flock_waitq)); + /* Safe to not lock here, since it should be empty anyway */ + LASSERT(cfs_hlist_unhashed(&lock->l_exp_flock_hash)); cfs_list_del_init(&lock->l_res_link); if (flags == LDLM_FL_WAIT_NOREPROC && @@ -151,53 +170,79 @@ ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags) EXIT; } +/** + * POSIX locks deadlock detection code. + * + * Given a new lock \a req and an existing lock \a bl_lock it conflicts + * with, we need to iterate through all blocked POSIX locks for this + * export and see if there is a deadlock condition arising. (i.e. when + * one client holds a lock on something and want a lock on something + * else and at the same time another client has the opposite situation). + */ static int ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock) { 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; /* For server only */ if (req_exp == NULL) return 0; class_export_get(bl_exp); -restart: - 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; - + while (1) { + struct obd_export *bl_exp_new; + struct ldlm_lock *lock = NULL; + struct ldlm_flock *flock; + + if (bl_exp->exp_flock_hash != NULL) + lock = cfs_hash_lookup(bl_exp->exp_flock_hash, + &bl_owner); + if (lock == NULL) + break; + + flock = &lock->l_policy_data.l_flock; + LASSERT(flock->owner == bl_owner); 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); + + cfs_hash_put(bl_exp->exp_flock_hash, &lock->l_exp_flock_hash); bl_exp = bl_exp_new; if (bl_owner == req_owner && bl_exp == req_exp) { class_export_put(bl_exp); return 1; } - - goto restart; } - cfs_read_unlock(&bl_exp->exp_flock_wait_lock); class_export_put(bl_exp); return 0; } +/** + * Process a granting attempt for flock lock. + * Must be called under ns lock held. + * + * This function looks for any conflicts for \a lock in the granted or + * waiting queues. The lock is granted if no conflicts are found in + * either queue. + * + * It is also responsible for splitting a lock if a portion of the lock + * is released. + * + * If \a first_enq is 0 (ie, called from ldlm_reprocess_queue): + * - blocking ASTs have already been sent + * + * If \a first_enq is 1 (ie, called from ldlm_lock_enqueue): + * - blocking ASTs have not been sent yet, so list of conflicting locks + * would be collected and ASTs sent. + */ int -ldlm_process_flock_lock(struct ldlm_lock *req, int *flags, int first_enq, - ldlm_error_t *err, cfs_list_t *work_list) +ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags, int first_enq, + ldlm_error_t *err, cfs_list_t *work_list) { struct ldlm_resource *res = req->l_resource; struct ldlm_namespace *ns = ldlm_res_to_ns(res); @@ -212,10 +257,12 @@ ldlm_process_flock_lock(struct ldlm_lock *req, int *flags, int first_enq, int overlaps = 0; int splitted = 0; const struct ldlm_callback_suite null_cbs = { NULL }; + int rc; ENTRY; - CDEBUG(D_DLMTRACE, "flags %#x owner "LPU64" pid %u mode %u start "LPU64 - " end "LPU64"\n", *flags, new->l_policy_data.l_flock.owner, + CDEBUG(D_DLMTRACE, "flags %#llx 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); @@ -293,8 +340,12 @@ reprocess: RETURN(LDLM_ITER_STOP); } - - ldlm_flock_blocking_link(req, lock); + rc = ldlm_flock_blocking_link(req, lock); + if (rc) { + ldlm_flock_destroy(req, mode, *flags); + *err = rc; + RETURN(LDLM_ITER_STOP); + } ldlm_resource_add_lock(res, &res->lr_waiting, req); *flags |= LDLM_FL_BLOCK_GRANTED; RETURN(LDLM_ITER_STOP); @@ -309,7 +360,7 @@ reprocess: } /* In case we had slept on this lock request take it off of the - * deadlock detection waitq. */ + * deadlock detection hash list. */ ldlm_flock_blocking_unlink(req); /* Scan the locks owned by this process that overlap this request. @@ -409,9 +460,9 @@ reprocess: * and restart processing this lock. */ if (!new2) { unlock_res_and_lock(req); - new2 = ldlm_lock_create(ns, &res->lr_name, LDLM_FLOCK, - lock->l_granted_mode, &null_cbs, - NULL, 0); + new2 = ldlm_lock_create(ns, &res->lr_name, LDLM_FLOCK, + lock->l_granted_mode, &null_cbs, + NULL, 0, LVB_T_NONE); lock_res_and_lock(req); if (!new2) { ldlm_flock_destroy(req, lock->l_granted_mode, @@ -469,6 +520,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 @@ -497,12 +549,19 @@ 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 - * it until after calling ldlm_ast_work_item() above so that lawi() - * can bump the reference count on req. Otherwise req could be freed - * before the completion AST can be sent. */ + /* In case we're reprocessing the requested lock we can't destroy + * it until after calling ldlm_add_ast_work_item() above so that laawi() + * can bump the reference count on \a req. Otherwise \a req + * could be freed before the completion AST can be sent. */ if (added) ldlm_flock_destroy(req, mode, *flags); @@ -523,17 +582,19 @@ ldlm_flock_interrupted_wait(void *data) lock = ((struct ldlm_flock_wait_data *)data)->fwd_lock; - /* take lock off the deadlock detection waitq. */ + /* take lock off the deadlock detection hash list. */ + lock_res_and_lock(lock); ldlm_flock_blocking_unlink(lock); - /* client side - set flag to prevent lock from being put on lru list */ + /* client side - set flag to prevent lock from being put on LRU list */ lock->l_flags |= LDLM_FL_CBPENDING; + unlock_res_and_lock(lock); EXIT; } /** - * Flock completion calback function. + * Flock completion callback function. * * \param lock [in,out]: A lock to be handled * \param flags [in]: flags @@ -543,9 +604,9 @@ ldlm_flock_interrupted_wait(void *data) * \retval <0 : failure */ int -ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) +ldlm_flock_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data) { - cfs_flock_t *getlk = lock->l_ast_data; + struct file_lock *getlk = lock->l_ast_data; struct obd_device *obd; struct obd_import *imp = NULL; struct ldlm_flock_wait_data fwd; @@ -554,7 +615,7 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) int rc = 0; ENTRY; - CDEBUG(D_DLMTRACE, "flags: 0x%x data: %p getlk: %p\n", + CDEBUG(D_DLMTRACE, "flags: 0x%llx data: %p getlk: %p\n", flags, data, getlk); /* Import invalidation. We need to actually release the lock @@ -595,9 +656,9 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) imp = obd->u.cli.cl_import; if (NULL != imp) { - cfs_spin_lock(&imp->imp_lock); - fwd.fwd_generation = imp->imp_generation; - cfs_spin_unlock(&imp->imp_lock); + spin_lock(&imp->imp_lock); + fwd.fwd_generation = imp->imp_generation; + spin_unlock(&imp->imp_lock); } lwi = LWI_TIMEOUT_INTR(0, NULL, ldlm_flock_interrupted_wait, &fwd); @@ -632,10 +693,11 @@ granted: LDLM_DEBUG(lock, "client-side enqueue granted"); - /* take lock off the deadlock detection waitq. */ + lock_res_and_lock(lock); + + /* take lock off the deadlock detection hash list. */ 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); @@ -643,33 +705,32 @@ granted: /* fcntl(F_GETLK) request */ /* The old mode was saved in getlk->fl_type so that if the mode * in the lock changes we can decref the appropriate refcount.*/ - ldlm_flock_destroy(lock, cfs_flock_type(getlk), - LDLM_FL_WAIT_NOREPROC); - switch (lock->l_granted_mode) { - case LCK_PR: - cfs_flock_set_type(getlk, F_RDLCK); - break; - case LCK_PW: - cfs_flock_set_type(getlk, F_WRLCK); - break; - default: - cfs_flock_set_type(getlk, F_UNLCK); - } - cfs_flock_set_pid(getlk, - (pid_t)lock->l_policy_data.l_flock.pid); - cfs_flock_set_start(getlk, - (loff_t)lock->l_policy_data.l_flock.start); - cfs_flock_set_end(getlk, - (loff_t)lock->l_policy_data.l_flock.end); - } else { - int noreproc = LDLM_FL_WAIT_NOREPROC; - - /* We need to reprocess the lock to do merges or splits - * with existing locks owned by this process. */ - ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL); - } - unlock_res_and_lock(lock); - RETURN(0); + ldlm_flock_destroy(lock, flock_type(getlk), + LDLM_FL_WAIT_NOREPROC); + switch (lock->l_granted_mode) { + case LCK_PR: + flock_set_type(getlk, F_RDLCK); + break; + case LCK_PW: + flock_set_type(getlk, F_WRLCK); + break; + default: + flock_set_type(getlk, F_UNLCK); + } + flock_set_pid(getlk, (pid_t)lock->l_policy_data.l_flock.pid); + flock_set_start(getlk, + (loff_t)lock->l_policy_data.l_flock.start); + flock_set_end(getlk, + (loff_t)lock->l_policy_data.l_flock.end); + } else { + __u64 noreproc = LDLM_FL_WAIT_NOREPROC; + + /* We need to reprocess the lock to do merges or splits + * with existing locks owned by this process. */ + ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL); + } + unlock_res_and_lock(lock); + RETURN(0); } EXPORT_SYMBOL(ldlm_flock_completion_ast); @@ -681,8 +742,10 @@ int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, LASSERT(lock); LASSERT(flag == LDLM_CB_CANCELING); - /* take lock off the deadlock detection waitq. */ + /* take lock off the deadlock detection hash list. */ + lock_res_and_lock(lock); ldlm_flock_blocking_unlink(lock); + unlock_res_and_lock(lock); RETURN(0); } @@ -719,3 +782,104 @@ void ldlm_flock_policy_local_to_wire(const ldlm_policy_data_t *lpolicy, wpolicy->l_flock.lfw_pid = lpolicy->l_flock.pid; wpolicy->l_flock.lfw_owner = lpolicy->l_flock.owner; } + +/* + * Export handle<->flock hash operations. + */ +static unsigned +ldlm_export_flock_hash(cfs_hash_t *hs, const void *key, unsigned mask) +{ + return cfs_hash_u64_hash(*(__u64 *)key, mask); +} + +static void * +ldlm_export_flock_key(cfs_hlist_node_t *hnode) +{ + struct ldlm_lock *lock; + + lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_flock_hash); + return &lock->l_policy_data.l_flock.owner; +} + +static int +ldlm_export_flock_keycmp(const void *key, cfs_hlist_node_t *hnode) +{ + return !memcmp(ldlm_export_flock_key(hnode), key, sizeof(__u64)); +} + +static void * +ldlm_export_flock_object(cfs_hlist_node_t *hnode) +{ + return cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_flock_hash); +} + +static void +ldlm_export_flock_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode) +{ + struct ldlm_lock *lock; + struct ldlm_flock *flock; + + lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_flock_hash); + LDLM_LOCK_GET(lock); + + flock = &lock->l_policy_data.l_flock; + LASSERT(flock->blocking_export != NULL); + class_export_get(flock->blocking_export); + flock->blocking_refs++; +} + +static void +ldlm_export_flock_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode) +{ + struct ldlm_lock *lock; + struct ldlm_flock *flock; + + lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_flock_hash); + LDLM_LOCK_RELEASE(lock); + + flock = &lock->l_policy_data.l_flock; + LASSERT(flock->blocking_export != NULL); + class_export_put(flock->blocking_export); + if (--flock->blocking_refs == 0) { + flock->blocking_owner = 0; + flock->blocking_export = NULL; + } +} + +static cfs_hash_ops_t ldlm_export_flock_ops = { + .hs_hash = ldlm_export_flock_hash, + .hs_key = ldlm_export_flock_key, + .hs_keycmp = ldlm_export_flock_keycmp, + .hs_object = ldlm_export_flock_object, + .hs_get = ldlm_export_flock_get, + .hs_put = ldlm_export_flock_put, + .hs_put_locked = ldlm_export_flock_put, +}; + +int ldlm_init_flock_export(struct obd_export *exp) +{ + exp->exp_flock_hash = + cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid), + HASH_EXP_LOCK_CUR_BITS, + HASH_EXP_LOCK_MAX_BITS, + HASH_EXP_LOCK_BKT_BITS, 0, + CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA, + &ldlm_export_flock_ops, + CFS_HASH_DEFAULT | CFS_HASH_NBLK_CHANGE); + if (!exp->exp_flock_hash) + RETURN(-ENOMEM); + + RETURN(0); +} +EXPORT_SYMBOL(ldlm_init_flock_export); + +void ldlm_destroy_flock_export(struct obd_export *exp) +{ + ENTRY; + if (exp->exp_flock_hash) { + cfs_hash_putref(exp->exp_flock_hash); + exp->exp_flock_hash = NULL; + } + EXIT; +} +EXPORT_SYMBOL(ldlm_destroy_flock_export);