X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_flock.c;h=a765cc7547103d28f2cf5d4374bf2bc509d8b3c9;hb=fef75d4c76745e8cd35df36e4691ed66c9e5902f;hp=7ef134163a4fe753db3f32940c04cd38e287ee4c;hpb=4dad795bcc8f4d99e25563b056cf950b613c2c17;p=fs%2Flustre-release.git diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index 7ef1341..a765cc7 100644 --- a/lustre/ldlm/ldlm_flock.c +++ b/lustre/ldlm/ldlm_flock.c @@ -1,117 +1,293 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: +/* + * GPL HEADER START * - * Copyright (c) 2002, 2003 Cluster File Systems, Inc. - * Author: Peter Braam - * Author: Phil Schwan + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * This file is part of Lustre, http://www.lustre.org. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 only, + * as published by the Free Software Foundation. * - * Lustre is free software; you can redistribute it and/or - * modify it under the terms of version 2 of the GNU General Public - * License as published by the Free Software Foundation. + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License version 2 for more details (a copy is included + * in the LICENSE file that accompanied this code). * - * Lustre is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * You should have received a copy of the GNU General Public License + * version 2 along with this program; If not, see + * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf * - * You should have received a copy of the GNU General Public License - * along with Lustre; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, + * CA 95054 USA or visit www.sun.com if you need additional information or + * have any questions. + * + * GPL HEADER END + */ +/* + * 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, 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__ -#include -#include -#include -#include +#include +#include +#include +#include +#include #else #include +#include #endif +#include "ldlm_internal.h" + +int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, + void *data, int flag); + +/** + * list_for_remaining_safe - iterate over the remaining entries in a list + * and safeguard against removal of a list entry. + * \param pos the &struct list_head to use as a loop counter. pos MUST + * have been initialized prior to using it in this macro. + * \param n another &struct list_head to use as temporary storage + * \param head the head for your list. + */ +#define list_for_remaining_safe(pos, n, head) \ + for (n = pos->next; pos != (head); pos = n, n = pos->next) + static inline int ldlm_same_flock_owner(struct ldlm_lock *lock, struct ldlm_lock *new) { - if ((new->l_data.l_flock.pid == lock->l_data.l_flock.pid) && - (new->l_export == lock->l_export)) - return 1; - else - return 0; + return((new->l_policy_data.l_flock.owner == + lock->l_policy_data.l_flock.owner) && + (new->l_export == lock->l_export)); } static inline int ldlm_flocks_overlap(struct ldlm_lock *lock, struct ldlm_lock *new) { - if ((new->l_data.l_flock.start <= lock->l_data.l_flock.end) && - (new->l_data.l_flock.end >= lock->l_data.l_flock.start)) - return 1; - else - return 0; + return((new->l_policy_data.l_flock.start <= + lock->l_policy_data.l_flock.end) && + (new->l_policy_data.l_flock.end >= + 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_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 = + 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); +} + +static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req) +{ + /* For server only */ + if (req->l_export == NULL) + return; + + 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, int flags) +ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, __u64 flags) { ENTRY; - list_del_init(&lock->l_res_link); - if (flags == LDLM_FL_WAIT_NOREPROC) { - /* client side */ - struct lustre_handle lockh; + 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_hlist_unhashed(&lock->l_exp_flock_hash)); - /* Set a flag to prevent us from sending a CANCEL */ - lock->l_flags |= LDLM_FL_LOCAL_ONLY; + cfs_list_del_init(&lock->l_res_link); + if (flags == LDLM_FL_WAIT_NOREPROC && + !(lock->l_flags & LDLM_FL_FAILED)) { + /* client side - set a flag to prevent sending a CANCEL */ + lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_CBPENDING; - ldlm_lock2handle(lock, &lockh); - ldlm_lock_decref_and_cancel(&lockh, lock->l_granted_mode); + /* when reaching here, it is under lock_res_and_lock(). Thus, + need call the nolock version of ldlm_lock_decref_internal*/ + ldlm_lock_decref_internal_nolock(lock, mode); } - ldlm_lock_destroy(lock); + ldlm_lock_destroy_nolock(lock); 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; + __u64 req_owner = req->l_policy_data.l_flock.owner; + __u64 bl_owner = bl_lock->l_policy_data.l_flock.owner; + + /* For server only */ + if (req_exp == NULL) + return 0; + + class_export_get(bl_exp); + 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; + + LASSERT(req != lock); + 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); + 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; + } + } + 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_flock_enqueue(struct ldlm_lock **reqp, void *req_cookie, int *flags, - int first_enq, ldlm_error_t *err) +ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags, int first_enq, + ldlm_error_t *err, cfs_list_t *work_list) { - struct ldlm_lock *req = *reqp; + struct ldlm_resource *res = req->l_resource; + struct ldlm_namespace *ns = ldlm_res_to_ns(res); + cfs_list_t *tmp; + cfs_list_t *ownlocks = NULL; + struct ldlm_lock *lock = NULL; struct ldlm_lock *new = req; struct ldlm_lock *new2 = NULL; - struct ldlm_lock *lock = NULL; - struct ldlm_resource *res = req->l_resource; - struct ldlm_namespace *ns = res->lr_namespace; - struct list_head *tmp; - struct list_head *ownlocks; ldlm_mode_t mode = req->l_req_mode; - int added = 0; + int local = ns_is_client(ns); + int added = (mode == LCK_NL); int overlaps = 0; + int splitted = 0; + const struct ldlm_callback_suite null_cbs = { NULL }; ENTRY; - CDEBUG(D_FLOCK, "flags: 0x%x pid: %d mode: %d start: %llu end: %llu\n", - *flags, new->l_data.l_flock.pid, mode, - req->l_data.l_flock.start, req->l_data.l_flock.end); + 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); *err = ELDLM_OK; - /* No blocking ASTs are sent for record locks */ - req->l_blocking_ast = NULL; - - ownlocks = NULL; - if ((*flags == LDLM_FL_WAIT_NOREPROC) || (mode == LCK_NL)) { - list_for_each(tmp, &res->lr_granted) { - lock = list_entry(tmp, struct ldlm_lock, l_res_link); + if (local) { + /* No blocking ASTs are sent to the clients for + * Posix file & record locks */ + req->l_blocking_ast = NULL; + } else { + /* Called on the server for lock cancels. */ + req->l_blocking_ast = ldlm_flock_blocking_ast; + } +reprocess: + if ((*flags == LDLM_FL_WAIT_NOREPROC) || (mode == LCK_NL)) { + /* This loop determines where this processes locks start + * in the resource lr_granted list. */ + cfs_list_for_each(tmp, &res->lr_granted) { + lock = cfs_list_entry(tmp, struct ldlm_lock, + l_res_link); if (ldlm_same_flock_owner(lock, req)) { ownlocks = tmp; break; } } } else { - list_for_each(tmp, &res->lr_granted) { - lock = list_entry(tmp, struct ldlm_lock, l_res_link); + lockmode_verify(mode); + + /* This loop determines if there are existing locks + * that conflict with the new lock request. */ + cfs_list_for_each(tmp, &res->lr_granted) { + lock = cfs_list_entry(tmp, struct ldlm_lock, + l_res_link); if (ldlm_same_flock_owner(lock, req)) { if (!ownlocks) @@ -122,120 +298,140 @@ ldlm_flock_enqueue(struct ldlm_lock **reqp, void *req_cookie, int *flags, /* locks are compatible, overlap doesn't matter */ if (lockmode_compat(lock->l_granted_mode, mode)) continue; - + if (!ldlm_flocks_overlap(lock, req)) continue; + if (!first_enq) + RETURN(LDLM_ITER_CONTINUE); + if (*flags & LDLM_FL_BLOCK_NOWAIT) { - ldlm_flock_destroy(req, *flags); - *err = ELDLM_LOCK_ABORTED; + ldlm_flock_destroy(req, mode, *flags); + *err = -EAGAIN; RETURN(LDLM_ITER_STOP); } if (*flags & LDLM_FL_TEST_LOCK) { - req->l_granted_mode = lock->l_granted_mode; - req->l_data.l_flock.pid = - lock->l_data.l_flock.pid; - req->l_data.l_flock.start = - lock->l_data.l_flock.start; - req->l_data.l_flock.end = - lock->l_data.l_flock.end; - ldlm_flock_destroy(req, *flags); + ldlm_flock_destroy(req, mode, *flags); + req->l_req_mode = lock->l_granted_mode; + req->l_policy_data.l_flock.pid = + lock->l_policy_data.l_flock.pid; + req->l_policy_data.l_flock.start = + lock->l_policy_data.l_flock.start; + req->l_policy_data.l_flock.end = + lock->l_policy_data.l_flock.end; + *flags |= LDLM_FL_LOCK_CHANGED; RETURN(LDLM_ITER_STOP); } - if (first_enq) { - /* XXX - add deadlock detection check here */ - } + /* add lock to blocking list before deadlock + * check to prevent race */ + ldlm_flock_blocking_link(req, lock); + + if (ldlm_flock_deadlock(req, lock)) { + ldlm_flock_blocking_unlink(req); + ldlm_flock_destroy(req, mode, *flags); + *err = -EDEADLK; + RETURN(LDLM_ITER_STOP); + } + ldlm_resource_add_lock(res, &res->lr_waiting, req); *flags |= LDLM_FL_BLOCK_GRANTED; - RETURN(LDLM_ITER_CONTINUE); + RETURN(LDLM_ITER_STOP); } } if (*flags & LDLM_FL_TEST_LOCK) { - LASSERT(first_enq); - req->l_granted_mode = req->l_req_mode; + ldlm_flock_destroy(req, mode, *flags); + req->l_req_mode = LCK_NL; + *flags |= LDLM_FL_LOCK_CHANGED; RETURN(LDLM_ITER_STOP); } - added = (mode == LCK_NL); + /* In case we had slept on this lock request take it off of the + * deadlock detection hash list. */ + ldlm_flock_blocking_unlink(req); - /* Insert the new lock into the list */ + /* Scan the locks owned by this process that overlap this request. + * We may have to merge or split existing locks. */ if (!ownlocks) ownlocks = &res->lr_granted; - for (tmp = ownlocks->next; ownlocks != &res->lr_granted; - ownlocks = tmp, tmp = ownlocks->next) { - lock = list_entry(ownlocks, struct ldlm_lock, l_res_link); + list_for_remaining_safe(ownlocks, tmp, &res->lr_granted) { + lock = cfs_list_entry(ownlocks, struct ldlm_lock, l_res_link); if (!ldlm_same_flock_owner(lock, new)) break; - if (lock->l_granted_mode == mode) { - if (lock->l_data.l_flock.end < - (new->l_data.l_flock.start - 1)) - continue; - - if (lock->l_data.l_flock.start > - (new->l_data.l_flock.end + 1)) - break; - - if (lock->l_data.l_flock.start > - new->l_data.l_flock.start) - lock->l_data.l_flock.start = - new->l_data.l_flock.start; - else - new->l_data.l_flock.start = - lock->l_data.l_flock.start; - - if (lock->l_data.l_flock.end < - new->l_data.l_flock.end) - lock->l_data.l_flock.end = - new->l_data.l_flock.end; - else - new->l_data.l_flock.end = - lock->l_data.l_flock.end; - - if (added) { - ldlm_flock_destroy(lock, *flags); - } else { + if (lock->l_granted_mode == mode) { + /* If the modes are the same then we need to process + * locks that overlap OR adjoin the new lock. The extra + * logic condition is necessary to deal with arithmetic + * overflow and underflow. */ + if ((new->l_policy_data.l_flock.start > + (lock->l_policy_data.l_flock.end + 1)) + && (lock->l_policy_data.l_flock.end != + OBD_OBJECT_EOF)) + continue; + + if ((new->l_policy_data.l_flock.end < + (lock->l_policy_data.l_flock.start - 1)) + && (lock->l_policy_data.l_flock.start != 0)) + break; + + if (new->l_policy_data.l_flock.start < + lock->l_policy_data.l_flock.start) { + lock->l_policy_data.l_flock.start = + new->l_policy_data.l_flock.start; + } else { + new->l_policy_data.l_flock.start = + lock->l_policy_data.l_flock.start; + } + + if (new->l_policy_data.l_flock.end > + lock->l_policy_data.l_flock.end) { + lock->l_policy_data.l_flock.end = + new->l_policy_data.l_flock.end; + } else { + new->l_policy_data.l_flock.end = + lock->l_policy_data.l_flock.end; + } + + if (added) { + ldlm_flock_destroy(lock, mode, *flags); + } else { new = lock; added = 1; } continue; - } + } - if (lock->l_data.l_flock.end < new->l_data.l_flock.start) + if (new->l_policy_data.l_flock.start > + lock->l_policy_data.l_flock.end) continue; - if (lock->l_data.l_flock.start > new->l_data.l_flock.end) + + if (new->l_policy_data.l_flock.end < + lock->l_policy_data.l_flock.start) break; ++overlaps; - if (new->l_data.l_flock.start <= - lock->l_data.l_flock.start) { - if (new->l_data.l_flock.end < - lock->l_data.l_flock.end) { - lock->l_data.l_flock.start = - new->l_data.l_flock.end + 1; + if (new->l_policy_data.l_flock.start <= + lock->l_policy_data.l_flock.start) { + if (new->l_policy_data.l_flock.end < + lock->l_policy_data.l_flock.end) { + lock->l_policy_data.l_flock.start = + new->l_policy_data.l_flock.end + 1; break; - } else if (added) { - ldlm_flock_destroy(lock, *flags); - } else { - lock->l_data.l_flock.start = - new->l_data.l_flock.start; - lock->l_data.l_flock.end = - new->l_data.l_flock.end; - new = lock; - added = 1; } + ldlm_flock_destroy(lock, lock->l_req_mode, *flags); continue; } - if (new->l_data.l_flock.end >= lock->l_data.l_flock.end) { - lock->l_data.l_flock.end = - new->l_data.l_flock.start - 1; + if (new->l_policy_data.l_flock.end >= + lock->l_policy_data.l_flock.end) { + lock->l_policy_data.l_flock.end = + new->l_policy_data.l_flock.start - 1; continue; } @@ -245,131 +441,219 @@ ldlm_flock_enqueue(struct ldlm_lock **reqp, void *req_cookie, int *flags, * allocating a new lock and use the req lock passed in * with the request but this would complicate the reply * processing since updates to req get reflected in the - * reply. The client side must see the original lock data - * so that it can process the unlock properly. */ - - /* XXX - if ldlm_lock_new() can sleep we have to - * release the ns_lock, allocate the new lock, and - * restart processing this lock. */ - new2 = ldlm_lock_create(ns, NULL, res->lr_name, LDLM_FLOCK, - lock->l_granted_mode, NULL, NULL); + * reply. The client side replays the lock request so + * it must see the original lock data in the reply. */ + + /* XXX - if ldlm_lock_new() can sleep we should + * release the lr_lock, allocate the new lock, + * and restart processing this lock. */ if (!new2) { - /* LBUG for now */ - LASSERT(0); - RETURN(ENOMEM); + unlock_res_and_lock(req); + 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, + *flags); + *err = -ENOLCK; + RETURN(LDLM_ITER_STOP); + } + goto reprocess; } + splitted = 1; + new2->l_granted_mode = lock->l_granted_mode; - new2->l_data.l_flock.pid = new->l_data.l_flock.pid; - new2->l_data.l_flock.start = lock->l_data.l_flock.start; - new2->l_data.l_flock.end = new->l_data.l_flock.start - 1; - lock->l_data.l_flock.start = new->l_data.l_flock.end + 1; - new2->l_connh = lock->l_connh; - if ((new2->l_export = lock->l_export) != NULL) { - list_add(&new2->l_export_chain, - &new2->l_export-> - exp_ldlm_data.led_held_locks); - } - if (*flags == LDLM_FL_WAIT_NOREPROC) { - /* client side */ - ldlm_lock_addref_internal(new2, 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 = + new->l_policy_data.l_flock.start - 1; + lock->l_policy_data.l_flock.start = + new->l_policy_data.l_flock.end + 1; + new2->l_conn_export = lock->l_conn_export; + if (lock->l_export != NULL) { + new2->l_export = class_export_lock_get(lock->l_export, new2); + if (new2->l_export->exp_lock_hash && + cfs_hlist_unhashed(&new2->l_exp_hash)) + cfs_hash_add(new2->l_export->exp_lock_hash, + &new2->l_remote_handle, + &new2->l_exp_hash); } + if (*flags == LDLM_FL_WAIT_NOREPROC) + ldlm_lock_addref_internal_nolock(new2, + lock->l_granted_mode); /* insert new2 at lock */ - list_add_tail(&new2->l_res_link, ownlocks); - LDLM_LOCK_PUT(new2); + ldlm_resource_add_lock(res, ownlocks, new2); + LDLM_LOCK_RELEASE(new2); break; } - if (added) { - ldlm_flock_destroy(req, *flags); - } else { - /* insert new at ownlocks */ - new->l_granted_mode = new->l_req_mode; - list_del_init(&new->l_res_link); - list_add_tail(&new->l_res_link, ownlocks); - } + /* if new2 is created but never used, destroy it*/ + if (splitted == 0 && new2 != NULL) + ldlm_lock_destroy_nolock(new2); + + /* At this point we're granting the lock request. */ + req->l_granted_mode = req->l_req_mode; - if (*flags != LDLM_FL_WAIT_NOREPROC) { - if (req->l_completion_ast) - ldlm_add_ast_work_item(req, NULL, NULL, 0); - - /* The only problem with doing the reprocessing here is that - * the completion ASTs for newly granted locks will be sent - * before the unlock completion is sent. It shouldn't be an - * issue. Also note that ldlm_flock_enqueue() will recurse, - * but only once because there can't be unlock requests on - * the wait queue. */ - if ((mode == LCK_NL) && overlaps) - ldlm_reprocess_queue(res, &res->lr_waiting); + /* Add req to the granted queue before calling ldlm_reprocess_all(). */ + if (!added) { + cfs_list_del_init(&req->l_res_link); + /* insert new lock before ownlocks in list. */ + ldlm_resource_add_lock(res, ownlocks, req); } - ldlm_resource_dump(res); + 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 + * granted. The only problem with doing this + * reprocessing here is that the completion ASTs for + * newly granted locks will be sent before the unlock + * completion is sent. It shouldn't be an issue. Also + * note that ldlm_process_flock_lock() will recurse, + * but only once because first_enq will be false from + * ldlm_reprocess_queue. */ + if ((mode == LCK_NL) && overlaps) { + CFS_LIST_HEAD(rpc_list); + int rc; +restart: + ldlm_reprocess_queue(res, &res->lr_waiting, + &rpc_list); + + unlock_res_and_lock(req); + rc = ldlm_run_ast_work(ns, &rpc_list, + LDLM_WORK_CP_AST); + lock_res_and_lock(req); + if (rc == -ERESTART) + GOTO(restart, -ERESTART); + } + } else { + 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 */ + } - RETURN(LDLM_ITER_CONTINUE); -} + /* 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); -static void interrupted_flock_completion_wait(void *data) -{ + ldlm_resource_dump(D_INFO, res); + RETURN(LDLM_ITER_CONTINUE); } -struct flock_wait_data { +struct ldlm_flock_wait_data { struct ldlm_lock *fwd_lock; int fwd_generation; }; -int -ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) +static void +ldlm_flock_interrupted_wait(void *data) { - struct ldlm_namespace *ns; - struct file_lock *getlk = data; - struct flock_wait_data fwd; - unsigned long irqflags; - struct obd_device *obd; - struct obd_import *imp = NULL; - ldlm_error_t err; - int rc = 0; - struct l_wait_info lwi; + struct ldlm_lock *lock; ENTRY; - LASSERT(flags != LDLM_FL_WAIT_NOREPROC); + lock = ((struct ldlm_flock_wait_data *)data)->fwd_lock; + + /* 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 */ + lock->l_flags |= LDLM_FL_CBPENDING; + unlock_res_and_lock(lock); - if (flags == 0) { - wake_up(&lock->l_waitq); + EXIT; +} + +/** + * Flock completion callback function. + * + * \param lock [in,out]: A lock to be handled + * \param flags [in]: flags + * \param *data [in]: ldlm_work_cp_ast_lock() will use ldlm_cb_set_arg + * + * \retval 0 : success + * \retval <0 : failure + */ +int +ldlm_flock_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data) +{ + struct file_lock *getlk = lock->l_ast_data; + struct obd_device *obd; + struct obd_import *imp = NULL; + struct ldlm_flock_wait_data fwd; + struct l_wait_info lwi; + ldlm_error_t err; + int rc = 0; + ENTRY; + + CDEBUG(D_DLMTRACE, "flags: 0x%llx data: %p getlk: %p\n", + flags, data, getlk); + + /* Import invalidation. We need to actually release the lock + * references being held, so that it can go away. No point in + * holding the lock even if app still believes it has it, since + * server already dropped it anyway. Only for granted locks too. */ + if ((lock->l_flags & (LDLM_FL_FAILED|LDLM_FL_LOCAL_ONLY)) == + (LDLM_FL_FAILED|LDLM_FL_LOCAL_ONLY)) { + if (lock->l_req_mode == lock->l_granted_mode && + lock->l_granted_mode != LCK_NL && + NULL == data) + ldlm_lock_decref_internal(lock, lock->l_req_mode); + + /* Need to wake up the waiter if we were evicted */ + cfs_waitq_signal(&lock->l_waitq); RETURN(0); } + LASSERT(flags != LDLM_FL_WAIT_NOREPROC); + if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED | - LDLM_FL_BLOCK_CONV))) - goto granted; + LDLM_FL_BLOCK_CONV))) { + if (NULL == data) + /* mds granted the lock in the reply */ + goto granted; + /* CP AST RPC: lock get granted, wake it up */ + cfs_waitq_signal(&lock->l_waitq); + RETURN(0); + } LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, " "sleeping"); - - ldlm_lock_dump(D_OTHER, lock); - fwd.fwd_lock = lock; - obd = class_conn2obd(lock->l_connh); + obd = class_exp2obd(lock->l_conn_export); - /* if this is a local lock, then there is no import */ - if (obd != NULL) + /* if this is a local lock, there is no import */ + if (NULL != obd) imp = obd->u.cli.cl_import; - if (imp != NULL) { - spin_lock_irqsave(&imp->imp_lock, irqflags); - fwd.fwd_generation = imp->imp_generation; - spin_unlock_irqrestore(&imp->imp_lock, irqflags); + if (NULL != imp) { + spin_lock(&imp->imp_lock); + fwd.fwd_generation = imp->imp_generation; + spin_unlock(&imp->imp_lock); } - lwi = LWI_TIMEOUT_INTR(0, NULL, interrupted_flock_completion_wait, - &fwd); + lwi = LWI_TIMEOUT_INTR(0, NULL, ldlm_flock_interrupted_wait, &fwd); /* Go to sleep until the lock is granted. */ - rc = l_wait_event(lock->l_waitq, - ((lock->l_req_mode == lock->l_granted_mode) || - lock->l_destroyed), &lwi); - - LASSERT(!(lock->l_destroyed)); + rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi); if (rc) { LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)", @@ -378,41 +662,216 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) } granted: + OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CP_CB_WAIT, 10); + + if (lock->l_flags & LDLM_FL_DESTROYED) { + LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed"); + RETURN(0); + } + + if (lock->l_flags & LDLM_FL_FAILED) { + LDLM_DEBUG(lock, "client-side enqueue waking up: failed"); + RETURN(-EIO); + } + + if (rc) { + LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)", + rc); + RETURN(rc); + } + + LDLM_DEBUG(lock, "client-side enqueue granted"); + + lock_res_and_lock(lock); - LDLM_DEBUG(lock, "client-side enqueue waking up"); - ns = lock->l_resource->lr_namespace; - l_lock(&ns->ns_lock); + /* take lock off the deadlock detection hash list. */ + ldlm_flock_blocking_unlink(lock); /* ldlm_lock_enqueue() has already placed lock on the granted list. */ - list_del_init(&lock->l_res_link); + cfs_list_del_init(&lock->l_res_link); - if (getlk) { + if (flags & LDLM_FL_TEST_LOCK) { /* fcntl(F_GETLK) request */ - if (lock->l_granted_mode == LCK_PR) - getlk->fl_type = F_RDLCK; - else if (lock->l_granted_mode == LCK_PW) - getlk->fl_type = F_WRLCK; - else - getlk->fl_type = F_UNLCK; - getlk->fl_pid = lock->l_data.l_flock.pid; - getlk->fl_start = lock->l_data.l_flock.start; - getlk->fl_end = lock->l_data.l_flock.end; - /* ldlm_flock_destroy(lock); */ - } else { - flags = LDLM_FL_WAIT_NOREPROC; - /* We need to reprocess the lock to do merges or split */ - ldlm_flock_enqueue(&lock, NULL, &flags, 1, &err); - } - l_unlock(&ns->ns_lock); - RETURN(0); + /* 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, 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); -/* This function is only called on the client when a lock is aborted. */ -int -ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *ld, - void *data, int flag) +int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, + void *data, int flag) { ENTRY; - ldlm_lock_destroy(lock); + + LASSERT(lock); + LASSERT(flag == LDLM_CB_CANCELING); + + /* 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); } + +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; +} + +/* + * 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) +{ + if( strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_MDT_NAME) != 0) + RETURN(0); + + 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);