#define LDLM_FL_CANCEL (1 << 8)
#define LDLM_FL_REPLAY (1 << 9)
#define LDLM_FL_INTENT_ONLY (1 << 10) /* don't grant lock, just do intent */
+#define LDLM_FL_LOCAL_ONLY (1 << 11) /* see ldlm_cli_cancel_unused */
+#define LDLM_FL_NO_CALLBACK (1 << 12) /* see ldlm_cli_cancel_unused */
#define LDLM_CB_BLOCKING 1
#define LDLM_CB_CANCELING 2
} \
if (lock->l_resource->lr_type == LDLM_EXTENT) { \
CDEBUG(D_DLMTRACE, "### " format \
- " (%s: lock %p(rc=%d/%d,%d) mode %s/%s on res " \
- LPU64" (rc=%d) type %s ["LPU64"->"LPU64"] remote " \
+ " (%s: lock %p(rc=%d/%d,%d) mode %s/%s on res "LPU64 \
+ "/"LPU64" (rc=%d) type %s ["LPU64"->"LPU64"] remote " \
LPX64")\n" , ## a, \
lock->l_resource->lr_namespace->ns_name, lock, \
lock->l_refc, lock->l_readers, lock->l_writers, \
ldlm_lockname[lock->l_granted_mode], \
ldlm_lockname[lock->l_req_mode], \
lock->l_resource->lr_name[0], \
+ lock->l_resource->lr_name[1], \
atomic_read(&lock->l_resource->lr_refcount), \
ldlm_typename[lock->l_resource->lr_type], \
lock->l_extent.start, lock->l_extent.end, \
} \
{ \
CDEBUG(D_DLMTRACE, "### " format \
- " (%s: lock %p(rc=%d/%d,%d) mode %s/%s on res " \
- LPU64" (rc=%d) type %s remote "LPX64")\n" , ## a, \
+ " (%s: lock %p(rc=%d/%d,%d) mode %s/%s on res "LPU64 \
+ "/"LPU64" (rc=%d) type %s remote "LPX64")\n" , ## a, \
lock->l_resource->lr_namespace->ns_name, lock, \
lock->l_refc, lock->l_readers, lock->l_writers, \
ldlm_lockname[lock->l_granted_mode], \
ldlm_lockname[lock->l_req_mode], \
lock->l_resource->lr_name[0], \
+ lock->l_resource->lr_name[1], \
atomic_read(&lock->l_resource->lr_refcount), \
ldlm_typename[lock->l_resource->lr_type], \
lock->l_remote_handle.addr); \
void ldlm_lock2handle(struct ldlm_lock *lock, struct lustre_handle *lockh);
struct ldlm_lock *__ldlm_handle2lock(struct lustre_handle *, int strict);
void ldlm_lock2handle(struct ldlm_lock *lock, struct lustre_handle *lockh);
-void ldlm_cancel_callback(struct ldlm_lock *lock);
+void ldlm_cancel_callback(struct ldlm_lock *);
+int ldlm_lock_set_data(struct lustre_handle *, void *data, int datalen);
static inline struct ldlm_lock *ldlm_handle2lock(struct lustre_handle *h)
{
void *data, __u32 data_len);
int ldlm_cli_convert(struct lustre_handle *, int new_mode, int *flags);
int ldlm_cli_cancel(struct lustre_handle *lockh);
-int ldlm_cli_cancel_unused(struct ldlm_namespace *, __u64 *, int local_only);
+int ldlm_cli_cancel_unused(struct ldlm_namespace *, __u64 *, int flags);
/* mds/handler.c */
/* This has to be here because recurisve inclusion sucks. */
struct lookup_intent *it, struct lustre_handle *lockh);
int ll_unlock(__u32 mode, struct lustre_handle *lockh);
+typedef int (*intent_finish_cb)(int flag, struct ptlrpc_request *,
+ struct dentry **, struct lookup_intent *,
+ int offset, obd_id ino);
+int ll_intent_lock(struct inode *parent, struct dentry **,
+ struct lookup_intent *, intent_finish_cb);
+
/* dcache.c */
void ll_intent_release(struct dentry *, struct lookup_intent *);
int ll_set_dd(struct dentry *de);
struct lookup_intent *it, int lock_mode, struct inode *dir,
struct dentry *de, struct lustre_handle *lockh, char *tgt,
int tgtlen, void *data, int datalen);
+int mdc_cancel_unused(struct lustre_handle *conn, struct inode *, int flags);
int mdc_getlovinfo(struct obd_device *obd, struct lustre_handle *mdc_connh,
struct ptlrpc_request **request);
int mdc_getstatus(struct lustre_handle *conn,
EXIT;
}
+int ldlm_lock_set_data(struct lustre_handle *lockh, void *data, int datalen)
+{
+ struct ldlm_lock *lock = ldlm_handle2lock(lockh);
+ ENTRY;
+
+ if (lock == NULL)
+ RETURN(-EINVAL);
+
+ lock->l_data = data;
+ lock->l_data_len = datalen;
+
+ LDLM_LOCK_PUT(lock);
+
+ RETURN(0);
+}
+
void ldlm_cancel_locks_for_export(struct obd_export *exp)
{
struct list_head *iter, *n; /* MUST BE CALLED "n"! */
EXPORT_SYMBOL(ldlm_lock_addref);
EXPORT_SYMBOL(ldlm_lock_decref);
EXPORT_SYMBOL(ldlm_lock_change_resource);
+EXPORT_SYMBOL(ldlm_lock_set_data);
EXPORT_SYMBOL(ldlm_cli_convert);
EXPORT_SYMBOL(ldlm_cli_enqueue);
EXPORT_SYMBOL(ldlm_cli_cancel);
body->lock_desc.l_extent.start,
body->lock_desc.l_extent.end,
reply->lock_extent.start, reply->lock_extent.end);
- cookie = &reply->lock_extent; /* FIXME bug 629281 */
+ cookie = &reply->lock_extent; /* FIXME bug 267 */
cookielen = sizeof(reply->lock_extent);
}
if (!req)
GOTO(out, rc = -ENOMEM);
- /* XXX FIXME bug 625069 */
+ /* XXX FIXME bug 249 */
req->rq_request_portal = LDLM_REQUEST_PORTAL;
req->rq_reply_portal = LDLM_REPLY_PORTAL;
}
static int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
- __u64 *res_id, int local_only)
+ __u64 *res_id, int flags)
{
struct ldlm_resource *res;
struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
int rc;
w = list_entry(tmp, struct ldlm_ast_work, w_list);
- if (local_only)
+ /* Prevent the cancel callback from being called by setting
+ * LDLM_FL_CANCEL in the lock. Very sneaky. -p */
+ if (flags & LDLM_FL_NO_CALLBACK)
+ w->w_lock->l_flags |= LDLM_FL_CANCEL;
+
+ if (flags & LDLM_FL_LOCAL_ONLY) {
ldlm_lock_cancel(w->w_lock);
- else {
+ } else {
ldlm_lock2handle(w->w_lock, &lockh);
rc = ldlm_cli_cancel(&lockh);
if (rc != ELDLM_OK)
* If 'local_only' is true, throw the locks away without trying to notify the
* server. */
int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id,
- int local_only)
+ int flags)
{
int i;
if (res_id)
- RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, local_only));
+ RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, flags));
l_lock(&ns->ns_lock);
for (i = 0; i < RES_HASH_SIZE; i++) {
ldlm_resource_getref(res);
rc = ldlm_cli_cancel_unused_resource(ns, res->lr_name,
- local_only);
+ flags);
if (rc)
CERROR("cancel_unused_res ("LPU64"): %d\n",
lock = list_entry(tmp, struct ldlm_lock, l_res_link);
LDLM_LOCK_GET(lock);
+ /* At shutdown time, don't call the cancellation callback */
+ lock->l_flags |= LDLM_FL_CANCEL;
+
if (client) {
struct lustre_handle lockh;
ldlm_lock2handle(lock, &lockh);
int ll_revalidate2(struct dentry *de, int flags, struct lookup_intent *it)
{
- int ll_intent_lock(struct inode *parent, struct dentry *dentry,
- struct lookup_intent *it, void *);
int rc;
ENTRY;
- rc = ll_intent_lock(de->d_parent->d_inode, de, it, NULL);
+ /* We don't want to cache negative dentries, so return 0 immediately.
+ * We believe that this is safe, that negative dentries cannot be
+ * pinned by someone else */
+ if (de->d_inode == NULL) {
+ CDEBUG(D_INODE, "negative dentry: ret 0 to force lookup2\n");
+ RETURN(0);
+ }
+
+ rc = ll_intent_lock(de->d_parent->d_inode, &de, it, NULL);
if (rc < 0) {
/* Something bad happened; overwrite it_status? */
+ CERROR("ll_intent_lock: %d\n", rc);
}
if (it != NULL && it->it_status == 0) {
/* If the dentry is busy, we won't get called in lookup2 if we
* return 0, so return 1.
*
- * This is a temporary fix for bug 618962, but is one of the causes of
- * 619078. */
+ * This is a temporary fix for bug 224, but is one of the causes of
+ * 225. */
CDEBUG(D_INFO, "d_count: %d\n", atomic_read(&de->d_count));
if (it && atomic_read(&de->d_count) > 1) {
CDEBUG(D_INFO, "returning 1 for %*s during %s because d_count "
#define LL_LOOKUP_POSITIVE 1
#define LL_LOOKUP_NEGATIVE 2
-typedef int (*intent_finish_cb)(int flag, struct ptlrpc_request *,
- struct dentry *, struct lookup_intent *,
- int offset, obd_id ino);
-
-int ll_intent_lock(struct inode *parent, struct dentry *dentry,
+int ll_intent_lock(struct inode *parent, struct dentry **de,
struct lookup_intent *it,
intent_finish_cb intent_finish)
{
+ struct dentry *dentry = *de;
struct ll_sb_info *sbi = ll_i2sbi(parent);
struct lustre_handle lockh;
struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
atomic_inc(&request->rq_refcount);
GOTO(out, flag = LL_LOOKUP_NEGATIVE);
}
+ /* Fall through to update attibutes. */
} else if (it->it_op & (IT_GETATTR | IT_SETATTR | IT_LOOKUP |
IT_READLINK)) {
/* For check ops, we want the lookup to succeed */
it->it_data = NULL;
if (it->it_status)
GOTO(out, flag = LL_LOOKUP_NEGATIVE);
+ /* Fall through to update attibutes. */
} else if (it->it_op & (IT_RENAME | IT_LINK)) {
/* For rename, we want the source lookup to succeed */
if (it->it_status) {
GOTO(drop_req, rc = it->it_status);
}
it->it_data = dentry;
+ /* Fall through to update attibutes. */
} else if (it->it_op & (IT_UNLINK | IT_RMDIR)) {
/* For remove ops, we want the lookup to succeed unless
* the file truly doesn't exist */
it->it_data = NULL;
if (it->it_status == -ENOENT)
GOTO(out, flag = LL_LOOKUP_NEGATIVE);
+ /* No point in updating attributes that we're about to
+ * unlink. -phil */
GOTO(out, flag = LL_LOOKUP_POSITIVE);
} else if (it->it_op == IT_OPEN) {
it->it_data = NULL;
if (it->it_status && it->it_status != -EEXIST)
GOTO(out, flag = LL_LOOKUP_NEGATIVE);
+ /* Fall through to update attibutes. */
} else if (it->it_op & (IT_RENAME2 | IT_LINK2)) {
it->it_data = NULL;
/* This means the target lookup is negative */
if (mds_body->valid == 0)
GOTO(out, flag = LL_LOOKUP_NEGATIVE);
+ /* XXX bug 289: should we maybe fall through here? -p */
GOTO(out, flag = LL_LOOKUP_POSITIVE);
}
EXIT;
out:
- if (intent_finish != NULL)
- rc = intent_finish(flag, request, dentry, it, offset, ino);
- else
+ if (intent_finish != NULL) {
+ rc = intent_finish(flag, request, de, it, offset, ino);
+ dentry = *de; /* intent_finish may change *de */
+ } else {
ptlrpc_req_finished(request);
+ }
if (it->it_op == IT_LOOKUP || rc < 0)
ll_intent_release(dentry, it);
return rc;
}
-static int lookup2_finish(int flag, struct ptlrpc_request *request,
- struct dentry *dentry, struct lookup_intent *it,
- int offset, obd_id ino)
+/* Search "inode"'s alias list for a dentry that has the same name and parent as
+ * de. If found, return it. If not found, return de. */
+static struct dentry *ll_find_alias(struct inode *inode, struct dentry *de)
+{
+ struct list_head *tmp;
+
+ spin_lock(&dcache_lock);
+ list_for_each(tmp, &inode->i_dentry) {
+ struct dentry *dentry = list_entry(tmp, struct dentry, d_alias);
+
+ /* We are called here with 'de' already on the aliases list. */
+ if (dentry == de)
+ continue;
+
+ if (dentry->d_parent != de->d_parent)
+ continue;
+
+ if (dentry->d_name.len != de->d_name.len)
+ continue;
+
+ if (memcmp(dentry->d_name.name, de->d_name.name,
+ de->d_name.len) != 0)
+ continue;
+
+ spin_unlock(&dcache_lock);
+ d_rehash(dentry);
+ return dget(dentry);
+ }
+ spin_unlock(&dcache_lock);
+ return de;
+}
+
+static int
+lookup2_finish(int flag, struct ptlrpc_request *request, struct dentry **de,
+ struct lookup_intent *it, int offset, obd_id ino)
{
+ struct dentry *dentry = *de, *saved = *de;
struct inode *inode = NULL;
struct ll_read_inode2_cookie lic;
* I think it is, but double-check refcounts. -phil */
RETURN(-ENOMEM);
}
+
+ dentry = *de = ll_find_alias(inode, dentry);
+
+ /* We asked for a lock on the directory, and may have been
+ * granted a lock on the inode. Just in case, fixup the data
+ * pointer. */
+ ldlm_lock_set_data((struct lustre_handle *)it->it_lock_handle,
+ inode, sizeof(*inode));
+
EXIT;
} else {
ENTRY;
} else
CERROR("NOT allocating fsdata - already set\n");
- d_add(dentry, inode);
+ if (dentry == saved)
+ d_add(dentry, inode);
if (it->it_status == 0) {
LL_SAVE_INTENT(dentry, it);
static struct dentry *ll_lookup2(struct inode *parent, struct dentry *dentry,
struct lookup_intent *it)
{
+ struct dentry *save = dentry;
int rc;
- rc = ll_intent_lock(parent, dentry, it, lookup2_finish);
+ rc = ll_intent_lock(parent, &dentry, it, lookup2_finish);
if (rc < 0) {
CERROR("ll_intent_lock: %d\n", rc);
return ERR_PTR(rc);
}
- return 0;
+ if (dentry == save)
+ return NULL;
+ else
+ return dentry;
}
static struct inode *ll_create_node(struct inode *dir, const char *name,
GOTO(out, -EIO);
}
+ if (it && it->it_disposition) {
+ /* We asked for a lock on the directory, but were
+ * granted a lock on the inode. Since we finally have
+ * an inode pointer, stuff it in the lock. */
+ ldlm_lock_set_data((struct lustre_handle *)it->it_lock_handle,
+ inode, sizeof(*inode));
+ }
+
EXIT;
out:
if (lsm && lmm)
static void prepare_mdc(struct obd_import *imp)
{
struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
- ldlm_cli_cancel_unused(ns, NULL, 1 /* local only */);
+ ldlm_cli_cancel_unused(ns, NULL, LDLM_FL_LOCAL_ONLY);
}
static int ll_prepare_recovery(struct ptlrpc_connection *conn)
static void ll_clear_inode(struct inode *inode)
{
+ struct ll_sb_info *sbi = ll_i2sbi(inode);
+ int rc;
ENTRY;
+ rc = mdc_cancel_unused(&sbi->ll_mdc_conn, inode, LDLM_FL_NO_CALLBACK);
+ if (rc < 0) {
+ CERROR("obd_cancel_unused: %d\n", rc);
+ /* XXX FIXME do something dramatic */
+ }
+
if (atomic_read(&inode->i_count) == 0) {
struct ll_inode_info *lli = ll_i2info(inode);
struct lov_stripe_md *lsm = lli->lli_smd;
lli->lli_symlink_name = NULL;
}
}
+
EXIT;
}
}
static int lov_cancel_unused(struct lustre_handle *conn,
- struct lov_stripe_md *lsm, int local_only)
+ struct lov_stripe_md *lsm, int flags)
{
struct obd_export *export = class_conn2export(conn);
struct lov_obd *lov;
submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
submd.lsm_stripe_count = 0;
rc = obd_cancel_unused(&lov->tgts[loi->loi_ost_idx].conn,
- &submd, local_only);
+ &submd, flags);
if (rc)
CERROR("Error cancel unused objid "LPX64" subobj "LPX64
" on OST idx %d: rc = %d\n", lsm->lsm_object_id,
return rc;
}
+static void d_delete_aliases(struct inode *inode)
+{
+ struct dentry *dentry;
+ struct list_head *tmp;
+ int dentry_count = 0;
+ ENTRY;
+
+ spin_lock(&dcache_lock);
+ list_for_each(tmp, &inode->i_dentry) {
+ dentry = list_entry(tmp, struct dentry, d_alias);
+ dentry_count++;
+ }
+
+ /* XXX FIXME tell phil/peter that you see this -- unless you're playing
+ * with hard links, in which case, stop. */
+ LASSERT(dentry_count <= 1);
+
+ if (dentry_count == 0) {
+ spin_unlock(&dcache_lock);
+ EXIT;
+ return;
+ }
+
+ CDEBUG(D_INODE, "d_deleting dentry %p\n", dentry);
+ dget_locked(dentry);
+ spin_unlock(&dcache_lock);
+ d_delete(dentry);
+ dput(dentry);
+ EXIT;
+}
+
static int mdc_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
void *data, __u32 data_len, int flag)
{
int rc;
- struct inode *inode = data;
struct lustre_handle lockh;
ENTRY;
- if (data_len != sizeof(*inode)) {
- CERROR("data_len should be %d, but is %d\n", sizeof(*inode),
- data_len);
- LBUG();
- RETURN(-EINVAL);
- }
-
switch (flag) {
case LDLM_CB_BLOCKING:
ldlm_lock2handle(lock, &lockh);
RETURN(rc);
}
break;
- case LDLM_CB_CANCELING:
- /* FIXME: do something better than throwing away everything */
- if (inode == NULL)
- LBUG();
+ case LDLM_CB_CANCELING: {
+ /* Invalidate all dentries associated with this inode */
+ struct inode *inode = data;
+
+ LASSERT(inode != NULL);
+ LASSERT(data_len == sizeof(*inode));
+
if (S_ISDIR(inode->i_mode)) {
CDEBUG(D_INODE, "invalidating inode %ld\n",
inode->i_ino);
ll_invalidate_inode_pages(inode);
}
+
+ LASSERT(igrab(inode) == inode);
+ d_delete_aliases(inode);
+ iput(inode);
break;
+ }
default:
LBUG();
}
{
struct ptlrpc_request *req;
struct obd_device *obddev = class_conn2obd(conn);
- __u64 res_id[RES_NAME_SIZE] = {dir->i_ino};
+ __u64 res_id[RES_NAME_SIZE] = {dir->i_ino, (__u64)dir->i_generation};
int size[6] = {sizeof(struct ldlm_request), sizeof(struct ldlm_intent)};
int rc, flags = 0;
int repsize[3] = {sizeof(struct ldlm_reply),
RETURN(0);
}
+int mdc_cancel_unused(struct lustre_handle *conn, struct inode *inode,
+ int flags)
+{
+ __u64 res_id[RES_NAME_SIZE] = {inode->i_ino, inode->i_generation};
+ struct obd_device *obddev = class_conn2obd(conn);
+ ENTRY;
+ RETURN(ldlm_cli_cancel_unused(obddev->obd_namespace, res_id, flags));
+}
+
struct replay_open_data {
struct lustre_handle *fh;
};
EXPORT_SYMBOL(mdc_getstatus);
EXPORT_SYMBOL(mdc_getlovinfo);
EXPORT_SYMBOL(mdc_enqueue);
+EXPORT_SYMBOL(mdc_cancel_unused);
EXPORT_SYMBOL(mdc_getattr);
EXPORT_SYMBOL(mdc_statfs);
EXPORT_SYMBOL(mdc_create);
}
}
-
if (flags & LDLM_FL_INTENT_ONLY) {
LDLM_DEBUG(lock, "INTENT_ONLY, aborting lock");
RETURN(ELDLM_LOCK_ABORTED);
/* Give the client a lock on the child object, instead of the
* parent that it requested. */
new_resid[0] = NTOH__u32(mds_rep->ino);
+ new_resid[1] = NTOH__u32(mds_rep->generation);
if (new_resid[0] == 0)
LBUG();
old_res = lock->l_resource->lr_name[0];
}
static int osc_cancel_unused(struct lustre_handle *connh,
- struct lov_stripe_md *lsm, int local)
+ struct lov_stripe_md *lsm, int flags)
{
struct obd_device *obddev = class_conn2obd(connh);
__u64 res_id[RES_NAME_SIZE] = { lsm->lsm_object_id };
- return ldlm_cli_cancel_unused(obddev->obd_namespace, res_id, local);
+ return ldlm_cli_cancel_unused(obddev->obd_namespace, res_id, flags);
}
static int osc_statfs(struct lustre_handle *conn, struct obd_statfs *osfs)
git://git.whamcloud.com / fs / lustre-release.git / commitdiff