diff --git a/kmod/src/data.c b/kmod/src/data.c index 661cff5b..3ecb3341 100644 --- a/kmod/src/data.c +++ b/kmod/src/data.c @@ -1238,7 +1238,7 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo, blk_off = ext.blk_off + ext.blocks; } - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR); out: mutex_unlock(&inode->i_mutex); diff --git a/kmod/src/dir.c b/kmod/src/dir.c index 85730f25..8581a8a5 100644 --- a/kmod/src/dir.c +++ b/kmod/src/dir.c @@ -283,7 +283,7 @@ out: else inode = scoutfs_iget(sb, ino); - scoutfs_unlock(sb, dir_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR); scoutfs_key_free(sb, key); @@ -390,7 +390,7 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir) } out: - scoutfs_unlock(sb, dir_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR); kfree(dent); return ret; @@ -545,8 +545,8 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, out: scoutfs_release_trans(sb); out_unlock: - scoutfs_unlock(sb, dir_lock); - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX); /* XXX delete the inode item here */ if (ret && !IS_ERR_OR_NULL(inode)) iput(inode); @@ -614,8 +614,8 @@ static int scoutfs_link(struct dentry *old_dentry, out: scoutfs_release_trans(sb); out_unlock: - scoutfs_unlock(sb, dir_lock); - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX); return ret; } @@ -714,8 +714,8 @@ out_trans: out: scoutfs_key_free(sb, keys[0]); scoutfs_key_free(sb, keys[2]); - scoutfs_unlock(sb, dir_lock); - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX); return ret; } @@ -837,7 +837,7 @@ static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd) nd_set_link(nd, path); } out: - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR); return path; } @@ -934,8 +934,8 @@ out: scoutfs_release_trans(sb); out_unlock: - scoutfs_unlock(sb, dir_lock); - scoutfs_unlock(sb, inode_lock); + scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX); + scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX); return ret; } diff --git a/kmod/src/dlmglue.c b/kmod/src/dlmglue.c index e41d346a..59615a31 100644 --- a/kmod/src/dlmglue.c +++ b/kmod/src/dlmglue.c @@ -97,6 +97,7 @@ static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres); static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres); #endif +#if 0 /* * Return value from ->downconvert_worker functions. * @@ -111,7 +112,7 @@ enum ocfs2_unblock_action { UNBLOCK_STOP_POST = 2, /* Do not downconvert, fire * ->post_unlock() callback. */ }; - +#endif struct ocfs2_unblock_ctl { int requeue; enum ocfs2_unblock_action unblock_action; @@ -169,7 +170,6 @@ static void ocfs2_dump_meta_lvb_info(u64 level, (long long)be64_to_cpu(lvb->lvb_imtime_packed), be32_to_cpu(lvb->lvb_iattr)); } -#endif /* @@ -260,7 +260,6 @@ struct ocfs2_lock_res_ops { */ #define LOCK_TYPE_USES_LVB 0x2 -#if 0 static struct ocfs2_lock_res_ops ocfs2_inode_rw_lops = { .get_osb = ocfs2_get_inode_osb, .flags = 0, @@ -532,10 +531,10 @@ static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw) } #endif -static void ocfs2_lock_res_init_common(struct ocfs2_super *osb, - struct ocfs2_lock_res *res, - struct ocfs2_lock_res_ops *ops, - void *priv) +void ocfs2_lock_res_init_common(struct ocfs2_super *osb, + struct ocfs2_lock_res *res, + struct ocfs2_lock_res_ops *ops, + void *priv) { res->l_ops = ops; res->l_priv = priv; diff --git a/kmod/src/dlmglue.h b/kmod/src/dlmglue.h index 3e25c8c5..7e005da2 100644 --- a/kmod/src/dlmglue.h +++ b/kmod/src/dlmglue.h @@ -172,6 +172,110 @@ struct ocfs2_super /* For s_mount_opt */ #define OCFS2_MOUNT_NOINTR (1 << 2) +/* + * Return value from ->downconvert_worker functions. + * + * These control the precise actions of ocfs2_unblock_lock() + * and ocfs2_process_blocked_lock() + * + */ +enum ocfs2_unblock_action { + UNBLOCK_CONTINUE = 0, /* Continue downconvert */ + UNBLOCK_CONTINUE_POST = 1, /* Continue downconvert, fire + * ->post_unlock callback */ + UNBLOCK_STOP_POST = 2, /* Do not downconvert, fire + * ->post_unlock() callback. */ +}; + +/* + * OCFS2 Lock Resource Operations + * + * These fine tune the behavior of the generic dlmglue locking infrastructure. + * + * The most basic of lock types can point ->l_priv to their respective + * struct ocfs2_super and allow the default actions to manage things. + * + * Right now, each lock type also needs to implement an init function, + * and trivial lock/unlock wrappers. ocfs2_simple_drop_lockres() + * should be called when the lock is no longer needed (i.e., object + * destruction time). + */ +struct ocfs2_lock_res_ops { + /* + * Translate an ocfs2_lock_res * into an ocfs2_super *. Define + * this callback if ->l_priv is not an ocfs2_super pointer + */ + struct ocfs2_super * (*get_osb)(struct ocfs2_lock_res *); + + /* + * Optionally called in the downconvert thread after a + * successful downconvert. The lockres will not be referenced + * after this callback is called, so it is safe to free + * memory, etc. + * + * The exact semantics of when this is called are controlled + * by ->downconvert_worker() + */ + void (*post_unlock)(struct ocfs2_super *, struct ocfs2_lock_res *); + + /* + * Allow a lock type to add checks to determine whether it is + * safe to downconvert a lock. Return 0 to re-queue the + * downconvert at a later time, nonzero to continue. + * + * For most locks, the default checks that there are no + * incompatible holders are sufficient. + * + * Called with the lockres spinlock held. + */ + int (*check_downconvert)(struct ocfs2_lock_res *, int); + + /* + * Allows a lock type to populate the lock value block. This + * is called on downconvert, and when we drop a lock. + * + * Locks that want to use this should set LOCK_TYPE_USES_LVB + * in the flags field. + * + * Called with the lockres spinlock held. + */ + void (*set_lvb)(struct ocfs2_lock_res *); + + /* + * Called from the downconvert thread when it is determined + * that a lock will be downconverted. This is called without + * any locks held so the function can do work that might + * schedule (syncing out data, etc). + * + * This should return any one of the ocfs2_unblock_action + * values, depending on what it wants the thread to do. + */ + int (*downconvert_worker)(struct ocfs2_lock_res *, int); + + /* + * LOCK_TYPE_* flags which describe the specific requirements + * of a lock type. Descriptions of each individual flag follow. + */ + int flags; +}; + +/* + * Some locks want to "refresh" potentially stale data when a + * meaningful (PRMODE or EXMODE) lock level is first obtained. If this + * flag is set, the OCFS2_LOCK_NEEDS_REFRESH flag will be set on the + * individual lockres l_flags member from the ast function. It is + * expected that the locking wrapper will clear the + * OCFS2_LOCK_NEEDS_REFRESH flag when done. + */ +#define LOCK_TYPE_REQUIRES_REFRESH 0x1 + +/* + * Indicate that a lock type makes use of the lock value block. The + * ->set_lvb lock type callback must be defined. + */ +#define LOCK_TYPE_USES_LVB 0x2 + + #if 0 #include "dcache.h" @@ -253,6 +357,10 @@ int ocfs2_dlm_init(struct ocfs2_super *osb, char *cluster_stack, char *cluster_name, char *ls_name, struct dentry *debug_root); void ocfs2_dlm_shutdown(struct ocfs2_super *osb, int hangup_pending); void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res); +void ocfs2_lock_res_init_common(struct ocfs2_super *osb, + struct ocfs2_lock_res *res, + struct ocfs2_lock_res_ops *ops, + void *priv); #if 0 void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res, enum ocfs2_lock_type type, diff --git a/kmod/src/inode.c b/kmod/src/inode.c index 2b7ddb0a..1b9761b6 100644 --- a/kmod/src/inode.c +++ b/kmod/src/inode.c @@ -284,7 +284,7 @@ static int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry, if (ret == 0) generic_fillattr(inode, stat); - scoutfs_unlock(sb, lock); + scoutfs_unlock(sb, lock, DLM_LOCK_PR); return ret; } @@ -416,7 +416,7 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino) } out: - scoutfs_unlock(sb, lock); + scoutfs_unlock(sb, lock, DLM_LOCK_PR); return inode; } diff --git a/kmod/src/ioctl.c b/kmod/src/ioctl.c index 3fe08339..abbf3fa4 100644 --- a/kmod/src/ioctl.c +++ b/kmod/src/ioctl.c @@ -132,7 +132,7 @@ static long scoutfs_ioc_walk_inodes(struct file *file, unsigned long arg) if (ret == -ENOENT) { - scoutfs_unlock(sb, lock); + scoutfs_unlock(sb, lock, DLM_LOCK_PR); /* * XXX This will miss dirty items. We'd need to * force writeouts of dirty items in our @@ -181,7 +181,7 @@ static long scoutfs_ioc_walk_inodes(struct file *file, unsigned long arg) scoutfs_key_inc_cur_len(&key); } - scoutfs_unlock(sb, lock); + scoutfs_unlock(sb, lock, DLM_LOCK_PR); out: scoutfs_key_free(sb, next_key); diff --git a/kmod/src/lock.c b/kmod/src/lock.c index 4ccc4fc3..360fe216 100644 --- a/kmod/src/lock.c +++ b/kmod/src/lock.c @@ -83,7 +83,12 @@ static int invalidate_caches(struct super_block *sb, int mode, static void free_scoutfs_lock(struct scoutfs_lock *lock) { + struct lock_info *linfo; + if (lock) { + linfo = SCOUTFS_SB(lock->sb)->lock_info; + + ocfs2_simple_drop_lockres(&linfo->dlmglue, &lock->lockres); scoutfs_key_free(lock->sb, lock->start); scoutfs_key_free(lock->sb, lock->end); kfree(lock); @@ -113,12 +118,50 @@ static void put_scoutfs_lock(struct super_block *sb, struct scoutfs_lock *lock) } } +static struct ocfs2_super *get_ino_lock_osb(struct ocfs2_lock_res *lockres) +{ + struct scoutfs_lock *lock = lockres->l_priv; + struct super_block *sb = lock->sb; + DECLARE_LOCK_INFO(sb, linfo); + + return &linfo->dlmglue; +} + +static int ino_lock_downconvert(struct ocfs2_lock_res *lockres, int blocking) +{ + struct scoutfs_lock *lock = lockres->l_priv; + struct super_block *sb = lock->sb; + + invalidate_caches(sb, blocking, lock->start, lock->end); + + return UNBLOCK_CONTINUE; +} + +static struct ocfs2_lock_res_ops scoufs_ino_lops = { + .get_osb = get_ino_lock_osb, + .downconvert_worker = ino_lock_downconvert, + /* XXX: .post_unlock for lru */ + /* XXX: .check_downconvert that queries the item cache for dirty items */ + .flags = LOCK_TYPE_REQUIRES_REFRESH, +}; + +static struct ocfs2_lock_res_ops scoufs_ino_index_lops = { + .get_osb = get_ino_lock_osb, + .downconvert_worker = ino_lock_downconvert, + /* XXX: .post_unlock for lru */ + /* XXX: .check_downconvert that queries the item cache for dirty items */ + .flags = 0, +}; + static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb, struct scoutfs_lock_name *lock_name, + struct ocfs2_lock_res_ops *type, struct scoutfs_key_buf *start, struct scoutfs_key_buf *end) { + DECLARE_LOCK_INFO(sb, linfo); +// struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_lock *lock; lock = kzalloc(sizeof(struct scoutfs_lock), GFP_NOFS); @@ -135,6 +178,14 @@ static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb, lock->mode = DLM_LOCK_IV; INIT_WORK(&lock->dc_work, scoutfs_downconvert_func); INIT_LIST_HEAD(&lock->lru_entry); + ocfs2_lock_res_init_once(&lock->lockres); + BUG_ON(sizeof(struct scoutfs_lock_name) >= + OCFS2_LOCK_ID_MAX_LEN); + /* kzalloc above ensures that l_name is NULL terminated */ + memcpy(&lock->lockres.l_name[0], &lock->lock_name, + sizeof(struct scoutfs_lock_name)); + ocfs2_lock_res_init_common(&linfo->dlmglue, + &lock->lockres, type, lock); } } @@ -152,6 +203,7 @@ static int cmp_lock_names(struct scoutfs_lock_name *a, static struct scoutfs_lock *find_alloc_scoutfs_lock(struct super_block *sb, struct scoutfs_lock_name *lock_name, + struct ocfs2_lock_res_ops *type, struct scoutfs_key_buf *start, struct scoutfs_key_buf *end) { @@ -187,7 +239,8 @@ search: if (!found) { if (!new) { spin_unlock(&linfo->lock); - new = alloc_scoutfs_lock(sb, lock_name, start, end); + new = alloc_scoutfs_lock(sb, lock_name, type, start, + end); if (!new) return NULL; @@ -349,6 +402,7 @@ static int lock_blocking(struct lock_info *linfo, struct scoutfs_lock *lock) */ static int lock_name_keys(struct super_block *sb, int mode, struct scoutfs_lock_name *lock_name, + struct ocfs2_lock_res_ops *type, struct scoutfs_key_buf *start, struct scoutfs_key_buf *end, struct scoutfs_lock **ret_lock) @@ -357,12 +411,20 @@ static int lock_name_keys(struct super_block *sb, int mode, struct scoutfs_lock *lock; int ret; - lock = find_alloc_scoutfs_lock(sb, lock_name, start, end); + lock = find_alloc_scoutfs_lock(sb, lock_name, type, start, end); if (!lock) return -ENOMEM; trace_scoutfs_lock_resource(sb, lock); + ret = ocfs2_cluster_lock(&linfo->dlmglue, &lock->lockres, mode, + DLM_LKF_NOORDER, 0); + if (ret) + return ret; + + *ret_lock = lock; + return 0; +#if 0 check_lock_state: spin_lock(&linfo->lock); if (linfo->shutdown) { @@ -413,6 +475,7 @@ check_lock_state: out: *ret_lock = lock; return 0; +#endif } int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino, @@ -441,7 +504,8 @@ int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino, end_ikey.type = ~0; scoutfs_key_init(&end, &end_ikey, sizeof(end_ikey)); - return lock_name_keys(sb, mode, &lock_name, &start, &end, ret_lock); + return lock_name_keys(sb, mode, &lock_name, &scoufs_ino_lops, &start, + &end, ret_lock); } /* @@ -505,10 +569,12 @@ int scoutfs_lock_inode_index(struct super_block *sb, int mode, end_ikey.ino = cpu_to_be64(ino | ino_mask); scoutfs_key_init(&end, &end_ikey, sizeof(end_ikey)); - return lock_name_keys(sb, mode, &lock_name, &start, &end, ret_lock); + return lock_name_keys(sb, mode, &lock_name, &scoufs_ino_index_lops, + &start, &end, ret_lock); } -void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock) +void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, + int level) { DECLARE_LOCK_INFO(sb, linfo); @@ -517,12 +583,15 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock) trace_scoutfs_unlock(sb, lock); + ocfs2_cluster_unlock(&linfo->dlmglue, &lock->lockres, level); + +#if 0 spin_lock(&linfo->lock); lock->holders--; if (lock->holders == 0 && (lock->flags & SCOUTFS_LOCK_BLOCKING)) queue_blocking_work(linfo, lock); spin_unlock(&linfo->lock); - +#endif put_scoutfs_lock(sb, lock); } @@ -666,6 +735,8 @@ void scoutfs_lock_destroy(struct super_block *sb) DECLARE_LOCK_INFO(sb, linfo); if (linfo) { + free_lock_tree(sb); /* Do this before uninitializing the dlm. */ + if (linfo->downconvert_wq) destroy_workqueue(linfo->downconvert_wq); unregister_shrinker(&linfo->shrinker); @@ -674,8 +745,6 @@ void scoutfs_lock_destroy(struct super_block *sb) ocfs2_uninit_super(&linfo->dlmglue); } - free_lock_tree(sb); - sbi->lock_info = NULL; trace_printk("sb %p id %016llx freeing linfo %p linfo %p\n", diff --git a/kmod/src/lock.h b/kmod/src/lock.h index 0b118e04..d47f9a7c 100644 --- a/kmod/src/lock.h +++ b/kmod/src/lock.h @@ -3,6 +3,7 @@ #include #include "key.h" +#include "dlmglue.h" #define SCOUTFS_LOCK_BLOCKING 0x01 /* Blocking another lock request */ #define SCOUTFS_LOCK_QUEUED 0x02 /* Put on drop workqueue */ @@ -22,6 +23,7 @@ struct scoutfs_lock { unsigned int holders; /* Tracks active users of this lock */ unsigned int flags; struct work_struct dc_work; + struct ocfs2_lock_res lockres; }; int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino, @@ -29,7 +31,8 @@ int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino, int scoutfs_lock_inode_index(struct super_block *sb, int mode, u8 type, u64 major, u64 ino, struct scoutfs_lock **ret_lock); -void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock); +void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, + int level); int scoutfs_lock_setup(struct super_block *sb); void scoutfs_lock_shutdown(struct super_block *sb); diff --git a/kmod/src/xattr.c b/kmod/src/xattr.c index cd860bab..ad24cda9 100644 --- a/kmod/src/xattr.c +++ b/kmod/src/xattr.c @@ -229,7 +229,7 @@ ssize_t scoutfs_getxattr(struct dentry *dentry, const char *name, void *buffer, ret = -ERANGE; up_read(&si->xattr_rwsem); - scoutfs_unlock(sb, lck); + scoutfs_unlock(sb, lck, DLM_LOCK_PR); out: scoutfs_key_free(sb, key); @@ -336,7 +336,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name, scoutfs_release_trans(sb); unlock: - scoutfs_unlock(sb, lck); + scoutfs_unlock(sb, lck, DLM_LOCK_EX); out: scoutfs_item_free_batch(sb, &list); @@ -436,7 +436,7 @@ ssize_t scoutfs_listxattr(struct dentry *dentry, char *buffer, size_t size) } up_read(&si->xattr_rwsem); - scoutfs_unlock(sb, lck); + scoutfs_unlock(sb, lck, DLM_LOCK_PR); out: scoutfs_key_free(sb, key); scoutfs_key_free(sb, last); @@ -490,7 +490,7 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino) /* don't need to increment past deleted key */ } - scoutfs_unlock(sb, lck); + scoutfs_unlock(sb, lck, DLM_LOCK_EX); out: scoutfs_key_free(sb, key);