/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * 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. * * 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). * * 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 * * 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 #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) { 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) { 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, ldlm_mode_t mode, __u64 flags) { ENTRY; 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)); 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; /* 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_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; } static void ldlm_flock_cancel_on_deadlock(struct ldlm_lock *lock, cfs_list_t *work_list) { CDEBUG(D_INFO, "reprocess deadlock req=%p\n", lock); if ((exp_connect_flags(lock->l_export) & OBD_CONNECT_FLOCK_DEAD) == 0) { CERROR("deadlock found, but client doesn't " "support flock canceliation\n"); } else { LASSERT(lock->l_completion_ast); LASSERT((lock->l_flags & LDLM_FL_AST_SENT) == 0); lock->l_flags |= LDLM_FL_AST_SENT | LDLM_FL_CANCEL_ON_BLOCK | LDLM_FL_FLOCK_DEADLOCK; ldlm_flock_blocking_unlink(lock); ldlm_resource_unlink_lock(lock); ldlm_add_ast_work_item(lock, NULL, work_list); } } /** * 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, __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); cfs_list_t *tmp; cfs_list_t *ownlocks = NULL; struct ldlm_lock *lock = NULL; struct ldlm_lock *new = req; struct ldlm_lock *new2 = NULL; ldlm_mode_t mode = req->l_req_mode; 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_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; 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 { int reprocess_failed = 0; 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) ownlocks = tmp; continue; } /* 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) { reprocess_failed = 1; if (ldlm_flock_deadlock(req, lock)) { ldlm_flock_cancel_on_deadlock(req, work_list); RETURN(LDLM_ITER_CONTINUE); } continue; } if (*flags & LDLM_FL_BLOCK_NOWAIT) { ldlm_flock_destroy(req, mode, *flags); *err = -EAGAIN; RETURN(LDLM_ITER_STOP); } if (*flags & LDLM_FL_TEST_LOCK) { 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); } /* 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_STOP); } if (reprocess_failed) RETURN(LDLM_ITER_CONTINUE); } if (*flags & LDLM_FL_TEST_LOCK) { ldlm_flock_destroy(req, mode, *flags); req->l_req_mode = LCK_NL; *flags |= LDLM_FL_LOCK_CHANGED; RETURN(LDLM_ITER_STOP); } /* In case we had slept on this lock request take it off of the * deadlock detection hash list. */ 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. */ if (!ownlocks) ownlocks = &res->lr_granted; 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 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 (new->l_policy_data.l_flock.start > lock->l_policy_data.l_flock.end) continue; if (new->l_policy_data.l_flock.end < lock->l_policy_data.l_flock.start) break; ++overlaps; 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; } ldlm_flock_destroy(lock, lock->l_req_mode, *flags); continue; } 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; } /* split the existing lock into two locks */ /* if this is an F_UNLCK operation then we could avoid * 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 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) { 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_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 */ ldlm_resource_add_lock(res, ownlocks, new2); LDLM_LOCK_RELEASE(new2); break; } /* 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; /* 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); } 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 */ } /* 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); ldlm_resource_dump(D_INFO, res); RETURN(LDLM_ITER_CONTINUE); } struct ldlm_flock_wait_data { struct ldlm_lock *fwd_lock; int fwd_generation; }; static void ldlm_flock_interrupted_wait(void *data) { struct ldlm_lock *lock; ENTRY; 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); 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))) { 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"); fwd.fwd_lock = lock; obd = class_exp2obd(lock->l_conn_export); /* if this is a local lock, there is no import */ if (NULL != obd) imp = obd->u.cli.cl_import; 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, ldlm_flock_interrupted_wait, &fwd); /* Go to sleep until the lock is granted. */ 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)", rc); RETURN(rc); } 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); /* take lock off the deadlock detection hash list. */ ldlm_flock_blocking_unlink(lock); /* ldlm_lock_enqueue() has already placed lock on the granted list. */ cfs_list_del_init(&lock->l_res_link); if (lock->l_flags & LDLM_FL_FLOCK_DEADLOCK) { LDLM_DEBUG(lock, "client-side enqueue deadlock received"); rc = -EDEADLK; } else if (flags & LDLM_FL_TEST_LOCK) { /* 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, 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(rc); } EXPORT_SYMBOL(ldlm_flock_completion_ast); int ldlm_flock_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag) { ENTRY; 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);