mirrors/scoutfs
1
0
Fork 0
mirror of https://github.com/versity/scoutfs.git synced 2026-07-19 22:42:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

scoutfs: simplify lock use of kernel dlm

Browse Source 
We had an excessive number of layers between scoutfs and the dlm code in
the kernel.  We had dlmglue, the scoutfs locks, and task refs.  Each
layer had structs that track the lifetime of the layer below it.  We
were about to add another layer to hold on to locks just a bit longer so
that we can avoid down conversion and transaction commit storms under
contention.

This collapses all those layers into simple state machine in lock.c that
manages the mode of dlm locks on behalf of the file system.

The users of the lock interface are mainly unchanged.  We did change
from a heavier trylock to a lighter nonblock lock attempt and have to
change the single rare readpage use.  Lock fields change so a few
external users of those fields change.

This not only removes a lot of code it also contains functional
improvements.  For example, it can now convert directly to CW locks with
a single lock request instead of having to use two by first converting
to NL.

It introduces the concept of an unlock grace period.  Locks won't be
dropped on behalf of other nodes soon after being unlocked so that tasks
have a chance to batch up work before the other node gets a chance.
This can result in two orders of magnitude improvements in the time it
takes to, say, change a set of xattrs on the same file population from
two nodes concurrently.

There are significant changes to trace points, counters, and debug files
that follow the implementation changes.

Signed-off-by: Zach Brown <zab@versity.com>
This commit is contained in:
Zach Brown
2018-02-02 09:18:08 -08:00
committed by Zach Brown
parent d42a3115c9
commit f52dc28322
15 changed files with 989 additions and 4697 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kmod/Makefile
+1 -1
View File
@@ -22,7 +22,7 @@ SCOUTFS_FORMAT_HASH := \
SCOUTFS_ARGS := SCOUTFS_GIT_DESCRIBE=$(SCOUTFS_GIT_DESCRIBE) \
SCOUTFS_FORMAT_HASH=$(SCOUTFS_FORMAT_HASH) \
CONFIG_SCOUTFS_FS=m -C $(SK_KSRC) M=$(CURDIR)/src \
EXTRA_CFLAGS="-Werror -DCONFIG_OCFS2_FS_STATS"
EXTRA_CFLAGS="-Werror"
all: module
kmod/src/Makefile
+4 -4
View File
@@ -6,10 +6,10 @@ CFLAGS_super.o = -DSCOUTFS_GIT_DESCRIBE=\"$(SCOUTFS_GIT_DESCRIBE)\" \
CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
scoutfs-y += alloc.o bio.o btree.o client.o compact.o counters.o data.o dir.o \
dlmglue.o export.o file.o kvec.o inode.o ioctl.o item.o key.o \
lock.o manifest.o msg.o options.o per_task.o seg.o server.o \
scoutfs_trace.o sock.o sort_priv.o stackglue.o super.o sysfs.o \
trans.o triggers.o xattr.o
export.o file.o kvec.o inode.o ioctl.o item.o key.o lock.o \
manifest.o msg.o options.o per_task.o seg.o server.o \
scoutfs_trace.o sock.o sort_priv.o super.o sysfs.o trans.o \
triggers.o xattr.o
#
# The raw types aren't available in userspace headers. Make sure all
kmod/src/counters.h
+16 -12
View File
@@ -29,13 +29,6 @@
EXPAND_COUNTER(data_write_begin) \
EXPAND_COUNTER(data_write_end) \
EXPAND_COUNTER(data_writepage) \
EXPAND_COUNTER(dlm_cancel_convert) \
EXPAND_COUNTER(dlm_convert_request) \
EXPAND_COUNTER(dlm_cw_downconvert) \
EXPAND_COUNTER(dlm_ex_downconvert) \
EXPAND_COUNTER(dlm_lock_request) \
EXPAND_COUNTER(dlm_pr_downconvert) \
EXPAND_COUNTER(dlm_unlock_request) \
EXPAND_COUNTER(item_alloc) \
EXPAND_COUNTER(item_create) \
EXPAND_COUNTER(item_delete) \
@@ -56,12 +49,23 @@
EXPAND_COUNTER(item_shrink_small_split) \
EXPAND_COUNTER(item_shrink_split_range) \
EXPAND_COUNTER(lock_alloc) \
EXPAND_COUNTER(lock_blocked_wait) \
EXPAND_COUNTER(lock_busy_wait) \
EXPAND_COUNTER(lock_ast) \
EXPAND_COUNTER(lock_ast_edeadlk) \
EXPAND_COUNTER(lock_ast_error) \
EXPAND_COUNTER(lock_bast) \
EXPAND_COUNTER(lock_dlm_call) \
EXPAND_COUNTER(lock_dlm_call_error) \
EXPAND_COUNTER(lock_free) \
EXPAND_COUNTER(lock_incompat_wait) \
EXPAND_COUNTER(lock_type_ino_downconvert) \
EXPAND_COUNTER(lock_type_idx_downconvert) \
EXPAND_COUNTER(lock_grace_enforced) \
EXPAND_COUNTER(lock_grace_expired) \
EXPAND_COUNTER(lock_grace_extended) \
EXPAND_COUNTER(lock_invalidate_clean_item) \
EXPAND_COUNTER(lock_lock) \
EXPAND_COUNTER(lock_lock_error) \
EXPAND_COUNTER(lock_nonblock_eagain) \
EXPAND_COUNTER(lock_shrink) \
EXPAND_COUNTER(lock_write_dirty_item) \
EXPAND_COUNTER(lock_unlock) \
EXPAND_COUNTER(manifest_compact_migrate) \
EXPAND_COUNTER(manifest_hard_stale_error) \
EXPAND_COUNTER(seg_alloc) \
kmod/src/data.c
+21 -12
View File
@@ -1076,29 +1076,38 @@ out:
return ret;
}
/*
* This is almost never used. We can't block on a cluster lock while
* holding the page lock because lock invalidation gets the page lock
* while blocking locks. If we can't use an existing lock then we drop
* the page lock and try again.
*/
static int scoutfs_readpage(struct file *file, struct page *page)
{
struct inode *inode = file->f_inode;
struct super_block *sb = inode->i_sb;
struct scoutfs_lock *inode_lock = NULL;
int unlock = 1;
int flags;
int ret;
ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, SCOUTFS_LKF_REFRESH_INODE |
SCOUTFS_LKF_TRYLOCK, inode, &inode_lock);
if (ret) {
if (ret == -EAGAIN)
ret = AOP_TRUNCATED_PAGE;
goto out;
flags = SCOUTFS_LKF_REFRESH_INODE | SCOUTFS_LKF_NONBLOCK;
ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, flags, inode, &inode_lock);
if (ret < 0) {
unlock_page(page);
if (ret == -EAGAIN) {
flags &= ~SCOUTFS_LKF_NONBLOCK;
ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, flags, inode,
&inode_lock);
if (ret == 0) {
scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
ret = AOP_TRUNCATED_PAGE;
}
}
return ret;
}
ret = mpage_readpage(page, scoutfs_get_block);
unlock = 0;
scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
out:
if (unlock)
unlock_page(page);
return ret;
}
kmod/src/dlmglue.c
-2803
View File
File diff suppressed because it is too large Load Diff
kmod/src/dlmglue.h
-390
View File
@@ -1,390 +0,0 @@
/* -*- mode: c; c-basic-offset: 8; -*-
* vim: noexpandtab sw=8 ts=8 sts=0:
*
* dlmglue.h
*
* description here
*
* Copyright (C) 2002, 2004 Oracle. All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* 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 for more details.
*
* You should have received a copy of the GNU General Public
* License along with this program; if not, write to the
* Free Software Foundation, Inc., 59 Temple Place - Suite 330,
* Boston, MA 021110-1307, USA.
*/
#ifndef DLMGLUE_H
#define DLMGLUE_H
#include "stackglue.h"
/* Max length of lockid name */
#define OCFS2_LOCK_ID_MAX_LEN 32
#define OCFS2_LOCK_ID_PRETTY_LEN 64
enum ocfs2_ast_action {
OCFS2_AST_INVALID = 0,
OCFS2_AST_ATTACH,
OCFS2_AST_CONVERT,
OCFS2_AST_DOWNCONVERT,
};
/* actions for an unlockast function to take. */
enum ocfs2_unlock_action {
OCFS2_UNLOCK_INVALID = 0,
OCFS2_UNLOCK_CANCEL_CONVERT,
OCFS2_UNLOCK_DROP_LOCK,
};
/* ocfs2_lock_res->l_flags flags. */
#define OCFS2_LOCK_ATTACHED (0x00000001) /* we have initialized
* the lvb */
#define OCFS2_LOCK_BUSY (0x00000002) /* we are currently in
* dlm_lock */
#define OCFS2_LOCK_BLOCKED (0x00000004) /* blocked waiting to
* downconvert*/
#define OCFS2_LOCK_LOCAL (0x00000008) /* newly created inode */
#define OCFS2_LOCK_NEEDS_REFRESH (0x00000010)
#define OCFS2_LOCK_REFRESHING (0x00000020)
#define OCFS2_LOCK_INITIALIZED (0x00000040) /* track initialization
* for shutdown paths */
#define OCFS2_LOCK_FREEING (0x00000080) /* help dlmglue track
* when to skip queueing
* a lock because it's
* about to be
* dropped. */
#define OCFS2_LOCK_QUEUED (0x00000100) /* queued for downconvert */
#define OCFS2_LOCK_NOCACHE (0x00000200) /* don't use a holder count */
#define OCFS2_LOCK_PENDING (0x00000400) /* This lockres is pending a
call to dlm_lock. Only
exists with BUSY set. */
#define OCFS2_LOCK_UPCONVERT_FINISHING (0x00000800) /* blocks the dc thread
* from downconverting
* before the upconvert
* has completed */
#define OCFS2_LOCK_NONBLOCK_FINISHED (0x00001000) /* NONBLOCK cluster
* lock has already
* returned, do not block
* dc thread from
* downconverting */
struct ocfs2_lock_res_ops;
typedef void (*ocfs2_lock_callback)(int status, unsigned long data);
#ifdef CONFIG_OCFS2_FS_STATS
struct ocfs2_lock_stats {
u64 ls_total; /* Total wait in NSEC */
u32 ls_gets; /* Num acquires */
u32 ls_fail; /* Num failed acquires */
/* Storing max wait in usecs saves 24 bytes per inode */
u32 ls_max; /* Max wait in USEC */
};
#endif
struct ocfs2_lock_res {
void *l_priv;
struct ocfs2_lock_res_ops *l_ops;
struct list_head l_blocked_list;
struct list_head l_mask_waiters;
struct list_head l_holders;
u64 l_refresh_gen;
unsigned long l_flags;
char l_name[OCFS2_LOCK_ID_MAX_LEN];
char l_pretty_name[OCFS2_LOCK_ID_PRETTY_LEN];
unsigned int l_ro_holders;
unsigned int l_cw_holders;
unsigned int l_ex_holders;
signed char l_level;
signed char l_requested;
signed char l_blocking;
/* used from AST/BAST funcs. */
/* Data packed - enum type ocfs2_ast_action */
unsigned char l_action;
/* Data packed - enum type ocfs2_unlock_action */
unsigned char l_unlock_action;
unsigned int l_pending_gen;
spinlock_t l_lock;
struct ocfs2_dlm_lksb l_lksb;
wait_queue_head_t l_event;
struct list_head l_debug_list;
#ifdef CONFIG_OCFS2_FS_STATS
struct ocfs2_lock_stats l_lock_prmode; /* PR mode stats */
u32 l_lock_refresh; /* Disk refreshes */
struct ocfs2_lock_stats l_lock_exmode; /* EX mode stats */
struct ocfs2_lock_stats l_lock_cwmode; /* CW mode stats */
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map l_lockdep_map;
#endif
};
struct ocfs2_dlm_debug {
struct kref d_refcnt;
struct dentry *d_locking_state;
struct list_head d_lockres_tracking;
};
/* The cluster stack fields */
#define OCFS2_STACK_LABEL_LEN 4
#define OCFS2_CLUSTER_NAME_LEN 16
struct ocfs2_super
{
struct ocfs2_cluster_connection *cconn;
struct ocfs2_dlm_debug *osb_dlm_debug;
/* Downconvert thread */
spinlock_t dc_task_lock;
struct task_struct *dc_task;
wait_queue_head_t dc_event;
unsigned long dc_wake_sequence;
unsigned long dc_work_sequence;
/*
* Any thread can add locks to the list, but the downconvert
* thread is the only one allowed to remove locks. Any change
* to this rule requires updating
* ocfs2_downconvert_thread_do_work().
*/
struct list_head blocked_lock_list;
unsigned long blocked_lock_count;
/* refresh_gen needs to strictly increase as locks come and go */
atomic64_t refresh_gen;
unsigned long s_mount_opt;
/* sb is for use with scoutfs counter macros only. Eventually
* we'll roll our own counters code in dlmglue. */
struct super_block *sb;
};
/* 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. */
};
enum ocfs2_lock_events {
EVENT_DLM_LOCK = 0,
EVENT_DLM_UNLOCK,
EVENT_DLM_CONVERT,
EVENT_DLM_CANCEL_CONVERT,
EVENT_DLM_DOWNCONVERT_WORK,
};
/*
* 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);
/*
* Called before we free a lock from the system. This allows
* the filesystem to sync and invalidate caches before that
* happens. The concept is identical to ->downconvert_worker
* except for two exceptions:
* - The FS must do the full downconvert work - as if it were
* blocking an EX.
* - We do not return an ocfs2_unblock_action - this worker is not
* allowed to delay dropping of the lock.
*/
void (*drop_worker)(struct ocfs2_lock_res *);
/*
* Optional: pretty print the lockname into a buffer
*/
void (*print)(struct ocfs2_lock_res *, char *, unsigned int);
/*
* Optional: Lightweight event callback, intended for quick
* operations like collecting stats, etc.
*/
void (*notify_event)(struct ocfs2_lock_res *, enum ocfs2_lock_events);
/*
* 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
struct ocfs2_lock_holder {
struct list_head oh_list;
struct pid *oh_owner_pid;
};
/* ocfs2_inode_lock_full() 'arg_flags' flags */
/* don't wait on recovery. */
#define OCFS2_META_LOCK_RECOVERY (0x01)
/* Instruct the dlm not to queue ourselves on the other node. */
#define OCFS2_META_LOCK_NOQUEUE (0x02)
/* don't block waiting for the downconvert thread, instead return -EAGAIN */
#define OCFS2_LOCK_NONBLOCK (0x04)
/* just get back disk inode bh if we've got cluster lock. */
#define OCFS2_META_LOCK_GETBH (0x08)
/* Locking subclasses of inode cluster lock */
enum {
OI_LS_NORMAL = 0,
OI_LS_PARENT,
OI_LS_RENAME1,
OI_LS_RENAME2,
OI_LS_REFLINK_TARGET,
};
int ocfs2_cluster_lock(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres,
int level, u32 lkm_flags, int arg_flags);
void ocfs2_cluster_unlock(struct ocfs2_super *osb,
struct ocfs2_lock_res *lockres, int level);
int ocfs2_init_super(struct ocfs2_super *osb, int flags);
int ocfs2_dlm_init(struct ocfs2_super *osb, struct super_block *sb,
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);
void ocfs2_lock_res_free(struct ocfs2_lock_res *res);
u64 ocfs2_lock_refresh_gen(struct ocfs2_lock_res *lockres);
void ocfs2_mark_lockres_freeing(struct ocfs2_super *osb,
struct ocfs2_lock_res *lockres);
void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
struct ocfs2_lock_res *lockres);
/* for the downconvert thread */
void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb);
struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void);
void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug);
int ocfs2_levels_compat(struct ocfs2_lock_res *lockres, int wanted);
#if 0
/* To set the locking protocol on module initialization */
void ocfs2_set_locking_protocol(void);
/* The _tracker pair is used to avoid cluster recursive locking */
int ocfs2_inode_lock_tracker(struct inode *inode,
struct buffer_head **ret_bh,
int ex,
struct ocfs2_lock_holder *oh);
void ocfs2_inode_unlock_tracker(struct inode *inode,
int ex,
struct ocfs2_lock_holder *oh,
int had_lock);
#endif
#endif /* DLMGLUE_H */
kmod/src/inode.c
+2 -2
View File
@@ -258,7 +258,7 @@ int scoutfs_inode_refresh(struct inode *inode, struct scoutfs_lock *lock,
struct scoutfs_inode_key ikey;
struct scoutfs_inode sinode;
SCOUTFS_DECLARE_KVEC(val);
const u64 refresh_gen = scoutfs_lock_refresh_gen(lock);
const u64 refresh_gen = lock->refresh_gen;
int ret;
/*
@@ -1272,7 +1272,7 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir,
ci->data_version = 0;
ci->next_readdir_pos = SCOUTFS_DIRENT_FIRST_POS;
ci->have_item = false;
atomic64_set(&ci->last_refreshed, scoutfs_lock_refresh_gen(lock));
atomic64_set(&ci->last_refreshed, lock->refresh_gen);
ci->flags = 0;
ci->ino_alloc.ino = 0;
ci->ino_alloc.nr = 0;
kmod/src/item.c
+21 -28
View File
@@ -741,39 +741,23 @@ restart:
/*
* Return true if the lock protects the use of the key. Some locks not
* intended for item use don't have a key range and we wan't to safely
* detect that. We use the block 'rw' constants just because they're
* convenient. The level test is racey but it's a char.. how racy can
* it be? :).
* intended for item use don't have a key range and we want to safely
* detect that. The lock mode dereference is racy but the field always
* contains a single non-zero byte.
*/
static bool lock_coverage(struct scoutfs_lock *lock,
struct scoutfs_key_buf *key, int op_level)
struct scoutfs_key_buf *key, int op_mode)
{
signed char level;
signed char mode;
if (!lock || !lock->start || !lock->end)
return false;
level = ACCESS_ONCE(lock->lockres.l_level);
mode = ACCESS_ONCE(lock->granted_mode);
switch (op_level) {
case DLM_LOCK_CW:
if (level != DLM_LOCK_CW)
return false;
break;
case DLM_LOCK_PR:
if (level < DLM_LOCK_PR)
return false;
break;
case DLM_LOCK_EX:
if (level != DLM_LOCK_EX)
return false;
break;
default:
return false;
}
return scoutfs_key_compare_ranges(key, key,
return ((op_mode == mode) ||
(op_mode == DLM_LOCK_PR && mode == DLM_LOCK_EX)) &&
scoutfs_key_compare_ranges(key, key,
lock->start, lock->end) == 0;
}
@@ -1781,6 +1765,8 @@ int scoutfs_item_dirty_seg(struct super_block *sb, struct scoutfs_segment *seg)
* The caller wants us to write out any dirty items within the given
* range. We look for any dirty items within the range and if we find
* any we issue a sync which writes out all the dirty items.
*
* Returns a sync error or the number of dirty items written.
*/
int scoutfs_item_writeback(struct super_block *sb,
struct scoutfs_key_buf *start,
@@ -1791,6 +1777,7 @@ int scoutfs_item_writeback(struct super_block *sb,
struct cached_item *item;
unsigned long flags;
bool sync = false;
int count = 0;
int ret = 0;
/* XXX think about racing with trans write */
@@ -1801,8 +1788,10 @@ int scoutfs_item_writeback(struct super_block *sb,
item = next_item(&cac->items, start);
if (item && !(item->dirty & ITEM_DIRTY))
item = next_dirty(item);
if (item && scoutfs_key_compare(item->key, end) <= 0)
if (item && scoutfs_key_compare(item->key, end) <= 0) {
sync = true;
count = cac->nr_dirty_items;
}
}
spin_unlock_irqrestore(&cac->lock, flags);
@@ -1812,12 +1801,14 @@ int scoutfs_item_writeback(struct super_block *sb,
ret = scoutfs_trans_sync(sb, 1);
}
return ret;
return ret ?: count;
}
/*
* The caller wants us to drop any items within the range on the floor.
* They should have ensured that items in this range won't be dirty.
*
* Returns errors or the count of the items invalidated.
*/
int scoutfs_item_invalidate(struct super_block *sb,
struct scoutfs_key_buf *start,
@@ -1830,6 +1821,7 @@ int scoutfs_item_invalidate(struct super_block *sb,
struct cached_item *item;
struct rb_node *node;
unsigned long flags;
int count = 0;
int ret;
trace_scoutfs_item_invalidate_range(sb, start, end);
@@ -1866,6 +1858,7 @@ int scoutfs_item_invalidate(struct super_block *sb,
WARN_ON_ONCE(item->dirty & ITEM_DIRTY);
erase_item(sb, cac, item);
count++;
}
remove_range(sb, &cac->ranges, rng);
@@ -1874,7 +1867,7 @@ int scoutfs_item_invalidate(struct super_block *sb,
ret = 0;
out:
return ret;
return ret ?: count;
}
static struct cached_item *rb_next_item(struct cached_item *item)
kmod/src/lock.c
+822 -655
View File
File diff suppressed because it is too large Load Diff
kmod/src/lock.h
+23 -20
View File
@@ -3,40 +3,42 @@
#include <linux/dlm.h>
#include "key.h"
#include "dlmglue.h"
#define SCOUTFS_LKF_REFRESH_INODE 0x01 /* update stale inode from item */
#define SCOUTFS_LKF_TRYLOCK 0x02 /* EAGAIN if contention */
#define SCOUTFS_LKF_NO_TASK_REF 0x04 /* don't create a task ref */
#define SCOUTFS_LKF_NONBLOCK 0x02 /* only use already held locks */
/* flags for scoutfs_lock->flags */
enum {
SCOUTFS_LOCK_RECLAIM = 0, /* lock is queued for reclaim */
SCOUTFS_LOCK_DROPPED, /* lock is going away, drop reference */
};
#define SCOUTFS_LOCK_NR_MODES (DLM_LOCK_EX + 1)
/*
* A few fields (start, end, refresh_gen, granted_mode) are referenced
* by code outside lock.c.
*/
struct scoutfs_lock {
struct super_block *sb;
struct scoutfs_lock_name lock_name;
struct scoutfs_lock_name name;
struct scoutfs_key_buf *start;
struct scoutfs_key_buf *end;
struct dlm_lksb lksb;
unsigned int sequence; /* for debugging and sanity checks */
struct rb_node node;
struct rb_node range_node;
unsigned int refcnt;
unsigned int debug_locks_id;
struct ocfs2_lock_res lockres;
struct list_head lru_entry;
struct work_struct reclaim_work;
unsigned int users; /* Tracks active users of this lock */
unsigned long flags;
u64 refresh_gen;
struct list_head lru_head;
wait_queue_head_t waitq;
struct rb_root task_refs;
spinlock_t task_refs_lock;
struct work_struct work;
struct dlm_lksb lksb;
ktime_t grace_deadline;
struct delayed_work grace_work;
bool grace_pending;
int error;
int granted_mode;
int bast_mode;
int work_prev_mode;
int work_mode;
unsigned int waiters[SCOUTFS_LOCK_NR_MODES];
unsigned int users[SCOUTFS_LOCK_NR_MODES];
};
u64 scoutfs_lock_refresh_gen(struct scoutfs_lock *lock);
int scoutfs_lock_inode(struct super_block *sb, int mode, int flags,
struct inode *inode, struct scoutfs_lock **ret_lock);
int scoutfs_lock_ino(struct super_block *sb, int mode, int flags, u64 ino,
@@ -64,6 +66,7 @@ void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock,
void scoutfs_free_unused_locks(struct super_block *sb, unsigned long nr);
int scoutfs_lock_setup(struct super_block *sb);
void scoutfs_lock_shutdown(struct super_block *sb);
void scoutfs_lock_destroy(struct super_block *sb);
#endif
kmod/src/scoutfs_trace.h
+75 -203
View File
@@ -35,8 +35,6 @@
#include "ioctl.h"
#include "count.h"
#include "bio.h"
#include "dlmglue.h"
#include "stackglue.h"
#include "export.h"
struct lock_info;
@@ -1046,7 +1044,12 @@ DECLARE_EVENT_CLASS(scoutfs_lock_info_class,
TP_printk(FSID_FMT" linfo %p", __entry->fsid, __entry->linfo)
);
DEFINE_EVENT(scoutfs_lock_info_class, init_lock_info,
DEFINE_EVENT(scoutfs_lock_info_class, scoutfs_lock_setup,
TP_PROTO(struct super_block *sb, struct lock_info *linfo),
TP_ARGS(sb, linfo)
);
DEFINE_EVENT(scoutfs_lock_info_class, scoutfs_lock_shutdown,
TP_PROTO(struct super_block *sb, struct lock_info *linfo),
TP_ARGS(sb, linfo)
);
@@ -1594,94 +1597,85 @@ DECLARE_EVENT_CLASS(scoutfs_lock_class,
__field(u8, name_type)
__field(u64, name_first)
__field(u64, name_second)
__field(unsigned int, seq)
__field(unsigned int, refcnt)
__field(unsigned int, users)
__field(unsigned char, level)
__field(unsigned char, blocking)
__field(unsigned int, cw)
__field(unsigned int, pr)
__field(unsigned int, ex)
__field(u64, refresh_gen)
__field(int, error)
__field(int, granted_mode)
__field(int, bast_mode)
__field(int, work_prev_mode)
__field(int, work_mode)
__field(unsigned int, waiters_cw)
__field(unsigned int, waiters_pr)
__field(unsigned int, waiters_ex)
__field(unsigned int, users_cw)
__field(unsigned int, users_pr)
__field(unsigned int, users_ex)
),
TP_fast_assign(
__entry->fsid = FSID_ARG(sb);
__entry->name_scope = lck->lock_name.scope;
__entry->name_zone = lck->lock_name.zone;
__entry->name_type = lck->lock_name.type;
__entry->name_first = le64_to_cpu(lck->lock_name.first);
__entry->name_second = le64_to_cpu(lck->lock_name.second);
__entry->seq = lck->sequence;
__entry->refcnt = lck->refcnt;
__entry->users = lck->users;
/* racey, but safe refs of embedded struct */
__entry->level = lck->lockres.l_level;
__entry->blocking = lck->lockres.l_blocking;
__entry->cw = lck->lockres.l_cw_holders;
__entry->pr = lck->lockres.l_ro_holders;
__entry->ex = lck->lockres.l_ex_holders;
__entry->name_scope = lck->name.scope;
__entry->name_zone = lck->name.zone;
__entry->name_type = lck->name.type;
__entry->name_first = le64_to_cpu(lck->name.first);
__entry->name_second = le64_to_cpu(lck->name.second);
__entry->refresh_gen = lck->refresh_gen;
__entry->error = lck->error;
__entry->granted_mode = lck->granted_mode;
__entry->bast_mode = lck->bast_mode;
__entry->work_prev_mode = lck->work_prev_mode;
__entry->work_mode = lck->work_mode;
__entry->waiters_pr = lck->waiters[DLM_LOCK_PR];
__entry->waiters_ex = lck->waiters[DLM_LOCK_EX];
__entry->waiters_cw = lck->waiters[DLM_LOCK_CW];
__entry->users_pr = lck->users[DLM_LOCK_PR];
__entry->users_ex = lck->users[DLM_LOCK_EX];
__entry->users_cw = lck->users[DLM_LOCK_CW];
),
TP_printk("fsid "FSID_FMT" name %u.%u.%u.%llu.%llu seq %u refs %d users %d level %u blocking %u cw %u pr %u ex %u",
TP_printk("fsid "FSID_FMT" name %u.%u.%u.%llu.%llu refresh_gen %llu error %d granted %d bast %d prev %d work %d waiters: pr %u ex %u cw %u users: pr %u ex %u cw %u",
__entry->fsid, __entry->name_scope, __entry->name_zone,
__entry->name_type, __entry->name_first,
__entry->name_second, __entry->seq, __entry->refcnt,
__entry->users, __entry->level, __entry->blocking,
__entry->cw, __entry->pr, __entry->ex)
__entry->name_type, __entry->name_first, __entry->name_second,
__entry->refresh_gen, __entry->error, __entry->granted_mode,
__entry->bast_mode, __entry->work_prev_mode,
__entry->work_mode, __entry->waiters_pr,
__entry->waiters_ex, __entry->waiters_cw, __entry->users_pr,
__entry->users_ex, __entry->users_cw)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_resource,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_unlock,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_ast,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_bast,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_invalidate,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_invalidate_ret,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_reclaim,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, shrink_lock_tree,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_rb_insert,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_free,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_alloc,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_ast,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_bast,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_work,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_grace_work,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_wait,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_unlock,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_shrink,
TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
TP_ARGS(sb, lck)
);
DECLARE_EVENT_CLASS(scoutfs_seg_class,
TP_PROTO(struct scoutfs_segment *seg),
@@ -1954,128 +1948,6 @@ DEFINE_EVENT(scoutfs_super_lifecycle_class, scoutfs_kill_sb,
TP_ARGS(sb)
);
TRACE_EVENT(ocfs2_cluster_lock,
TP_PROTO(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres,
int requested, unsigned int lkm_flags, unsigned int arg_flags),
TP_ARGS(osb, lockres, requested, lkm_flags, arg_flags),
TP_STRUCT__entry(
__string(lockspace, osb->cconn->cc_name)
__string(lockname, lockres->l_pretty_name)
__field(int, requested)
__field(unsigned int, lkm_flags)
__field(unsigned int, arg_flags)
__field(unsigned int, lockres_flags)
__field(int, lockres_level)
__field(int, blocking)
__field(unsigned int, cw_holders)
__field(unsigned int, pr_holders)
__field(unsigned int, ex_holders)
),
TP_fast_assign(
__assign_str(lockspace, osb->cconn->cc_name);
__assign_str(lockname, lockres->l_pretty_name);
__entry->requested = requested;
__entry->lkm_flags = lkm_flags;
__entry->arg_flags = arg_flags;
__entry->lockres_flags = lockres->l_flags;
__entry->lockres_level = lockres->l_level;
__entry->blocking = lockres->l_blocking;
__entry->cw_holders = lockres->l_cw_holders;
__entry->pr_holders = lockres->l_ro_holders;
__entry->ex_holders = lockres->l_ex_holders;
),
TP_printk("lockspace %s lock %s requested %d lkm_flags 0x%x arg_flags 0x%x lockres->level %d lockres->flags 0x%x lockres->blocking %d holders cw/pr/ex %u/%u/%u",
__get_str(lockspace), __get_str(lockname), __entry->requested,
__entry->lkm_flags, __entry->arg_flags,
__entry->lockres_level, __entry->lockres_flags,
__entry->blocking, __entry->cw_holders, __entry->pr_holders,
__entry->ex_holders)
);
TRACE_EVENT(ocfs2_cluster_unlock,
TP_PROTO(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres,
int level),
TP_ARGS(osb, lockres, level),
TP_STRUCT__entry(
__string(lockspace, osb->cconn->cc_name)
__string(lockname, lockres->l_pretty_name)
__field(int, level)
__field(unsigned int, lockres_flags)
__field(int, lockres_level)
__field(int, blocking)
__field(unsigned int, cw_holders)
__field(unsigned int, pr_holders)
__field(unsigned int, ex_holders)
),
TP_fast_assign(
__assign_str(lockspace, osb->cconn->cc_name);
__assign_str(lockname, lockres->l_pretty_name);
__entry->level = level;
__entry->lockres_flags = lockres->l_flags;
__entry->lockres_level = lockres->l_level;
__entry->blocking = lockres->l_blocking;
__entry->cw_holders = lockres->l_cw_holders;
__entry->pr_holders = lockres->l_ro_holders;
__entry->ex_holders = lockres->l_ex_holders;
),
TP_printk("lockspace %s lock %s level %d lockres->level %d lockres->flags 0x%x lockres->blocking %d holders cw/pr/ex: %u/%u/%u",
__get_str(lockspace), __get_str(lockname), __entry->level,
__entry->lockres_level, __entry->lockres_flags,
__entry->blocking, __entry->cw_holders, __entry->pr_holders,
__entry->ex_holders)
);
DECLARE_EVENT_CLASS(ocfs2_lock_res_class,
TP_PROTO(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres),
TP_ARGS(osb, lockres),
TP_STRUCT__entry(
__string(lockspace, osb->cconn->cc_name)
__string(lockname, lockres->l_pretty_name)
__field(unsigned int, lockres_flags)
__field(int, lockres_level)
__field(int, blocking)
__field(unsigned int, cw_holders)
__field(unsigned int, pr_holders)
__field(unsigned int, ex_holders)
),
TP_fast_assign(
__assign_str(lockspace, osb->cconn->cc_name);
__assign_str(lockname, lockres->l_pretty_name);
__entry->lockres_flags = lockres->l_flags;
__entry->lockres_level = lockres->l_level;
__entry->blocking = lockres->l_blocking;
__entry->cw_holders = lockres->l_cw_holders;
__entry->pr_holders = lockres->l_ro_holders;
__entry->ex_holders = lockres->l_ex_holders;
),
TP_printk("lockspace %s lock %s lockres->level %d lockres->flags 0x%x lockres->blocking %d holders cw/pr/ex: %u/%u/%u",
__get_str(lockspace), __get_str(lockname),
__entry->lockres_level, __entry->lockres_flags,
__entry->blocking, __entry->cw_holders,
__entry->pr_holders, __entry->ex_holders)
);
DEFINE_EVENT(ocfs2_lock_res_class, ocfs2_simple_drop_lockres,
TP_PROTO(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres),
TP_ARGS(osb, lockres)
);
DEFINE_EVENT(ocfs2_lock_res_class, ocfs2_unblock_lock,
TP_PROTO(struct ocfs2_super *osb, struct ocfs2_lock_res *lockres),
TP_ARGS(osb, lockres)
);
DECLARE_EVENT_CLASS(scoutfs_fileid_class,
TP_PROTO(struct super_block *sb, int fh_type, struct scoutfs_fid *fid),
TP_ARGS(sb, fh_type, fid),
kmod/src/server.c
+2 -4
View File
@@ -932,10 +932,8 @@ static void scoutfs_server_func(struct work_struct *work)
init_waitqueue_head(&waitq);
ret = scoutfs_lock_global(sb, DLM_LOCK_EX,
SCOUTFS_LKF_TRYLOCK,
SCOUTFS_LOCK_TYPE_GLOBAL_SERVER,
&lock);
ret = scoutfs_lock_global(sb, DLM_LOCK_EX, 0,
SCOUTFS_LOCK_TYPE_GLOBAL_SERVER, &lock);
if (ret)
goto out;
kmod/src/stackglue.c
-412
View File
@@ -1,412 +0,0 @@
/* -*- mode: c; c-basic-offset: 8; -*-
* vim: noexpandtab sw=8 ts=8 sts=0:
*
* stackglue.c
*
* Code which implements an OCFS2 specific interface to underlying
* cluster stacks.
*
* Copyright (C) 2007, 2009 Oracle. All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, version 2.
*
* 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 for more details.
*/
#include <linux/list.h>
#include <linux/spinlock.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/kmod.h>
#include <linux/fs.h>
#include <linux/kobject.h>
#include <linux/sysfs.h>
#include <linux/sysctl.h>
#include "stackglue.h"
static void fsdlm_lock_ast_wrapper(void *astarg)
{
struct ocfs2_dlm_lksb *lksb = astarg;
int status = lksb->lksb_fsdlm.sb_status;
/*
* For now we're punting on the issue of other non-standard errors
* where we can't tell if the unlock_ast or lock_ast should be called.
* The main "other error" that's possible is EINVAL which means the
* function was called with invalid args, which shouldn't be possible
* since the caller here is under our control. Other non-standard
* errors probably fall into the same category, or otherwise are fatal
* which means we can't carry on anyway.
*/
if (status == -DLM_EUNLOCK || status == -DLM_ECANCEL)
lksb->lksb_conn->cc_proto->lp_unlock_ast(lksb, 0);
else
lksb->lksb_conn->cc_proto->lp_lock_ast(lksb);
}
static void fsdlm_blocking_ast_wrapper(void *astarg, int level)
{
struct ocfs2_dlm_lksb *lksb = astarg;
lksb->lksb_conn->cc_proto->lp_blocking_ast(lksb, level);
}
static int user_dlm_lock(struct ocfs2_cluster_connection *conn,
int mode,
struct ocfs2_dlm_lksb *lksb,
u32 flags,
void *name,
unsigned int namelen)
{
int ret;
if (!lksb->lksb_fsdlm.sb_lvbptr)
lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
sizeof(struct dlm_lksb);
ret = dlm_lock(conn->cc_lockspace, mode, &lksb->lksb_fsdlm,
flags|DLM_LKF_NODLCKWT, name, namelen, 0,
fsdlm_lock_ast_wrapper, lksb,
fsdlm_blocking_ast_wrapper);
return ret;
}
/*
* The ocfs2_dlm_lock() and ocfs2_dlm_unlock() functions take no argument
* for the ast and bast functions. They will pass the lksb to the ast
* and bast. The caller can wrap the lksb with their own structure to
* get more information.
*/
int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
int mode,
struct ocfs2_dlm_lksb *lksb,
u32 flags,
void *name,
unsigned int namelen)
{
if (!lksb->lksb_conn)
lksb->lksb_conn = conn;
else
BUG_ON(lksb->lksb_conn != conn);
return user_dlm_lock(conn, mode, lksb, flags, name, namelen);
}
static int user_dlm_unlock(struct ocfs2_cluster_connection *conn,
struct ocfs2_dlm_lksb *lksb,
u32 flags)
{
int ret;
ret = dlm_unlock(conn->cc_lockspace, lksb->lksb_fsdlm.sb_lkid,
flags, &lksb->lksb_fsdlm, lksb);
return ret;
}
int ocfs2_dlm_unlock(struct ocfs2_cluster_connection *conn,
struct ocfs2_dlm_lksb *lksb,
u32 flags)
{
BUG_ON(lksb->lksb_conn == NULL);
return user_dlm_unlock(conn, lksb, flags);
}
static int user_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
{
return lksb->lksb_fsdlm.sb_status;
}
int ocfs2_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
{
return user_dlm_lock_status(lksb);
}
static int user_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
{
int invalid = lksb->lksb_fsdlm.sb_flags & DLM_SBF_VALNOTVALID;
return !invalid;
}
int ocfs2_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
{
return user_dlm_lvb_valid(lksb);
}
static void *user_dlm_lvb(struct ocfs2_dlm_lksb *lksb)
{
if (!lksb->lksb_fsdlm.sb_lvbptr)
lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
sizeof(struct dlm_lksb);
return (void *)(lksb->lksb_fsdlm.sb_lvbptr);
}
void *ocfs2_dlm_lvb(struct ocfs2_dlm_lksb *lksb)
{
return user_dlm_lvb(lksb);
}
void ocfs2_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb)
{
}
#if 0
static int user_plock(struct ocfs2_cluster_connection *conn,
u64 ino,
struct file *file,
int cmd,
struct file_lock *fl)
{
/*
* This more or less just demuxes the plock request into any
* one of three dlm calls.
*
* Internally, fs/dlm will pass these to a misc device, which
* a userspace daemon will read and write to.
*
* For now, cancel requests (which happen internally only),
* are turned into unlocks. Most of this function taken from
* gfs2_lock.
*/
if (cmd == F_CANCELLK) {
cmd = F_SETLK;
fl->fl_type = F_UNLCK;
}
if (IS_GETLK(cmd))
return dlm_posix_get(conn->cc_lockspace, ino, file, fl);
else if (fl->fl_type == F_UNLCK)
return dlm_posix_unlock(conn->cc_lockspace, ino, file, fl);
else
return dlm_posix_lock(conn->cc_lockspace, ino, file, cmd, fl);
}
int ocfs2_plock(struct ocfs2_cluster_connection *conn, u64 ino,
struct file *file, int cmd, struct file_lock *fl)
{
return user_plock(conn, ino, file, cmd, fl);
}
#endif
static void user_recover_prep(void *arg)
{
/* XXX: Set FS in recovery here */
}
static void user_recover_slot(void *arg, struct dlm_slot *slot)
{
printk(KERN_INFO "scoutfs: Node %d/%d down. Initiating recovery.\n",
slot->nodeid, slot->slot);
}
static void user_recover_done(void *arg, struct dlm_slot *slots,
int num_slots, int our_slot,
uint32_t generation)
{
/* XXX: Do actual fs recovery here */
}
static const struct dlm_lockspace_ops ocfs2_ls_ops = {
.recover_prep = user_recover_prep,
.recover_slot = user_recover_slot,
.recover_done = user_recover_done,
};
static int user_cluster_connect(struct ocfs2_cluster_connection *conn)
{
dlm_lockspace_t *fsdlm;
// struct ocfs2_live_connection *lc;
int rc, ops_rv;
BUG_ON(conn == NULL);
#if 0
lc = kzalloc(sizeof(struct ocfs2_live_connection), GFP_KERNEL);
if (!lc)
return -ENOMEM;
init_waitqueue_head(&lc->oc_wait);
init_completion(&lc->oc_sync_wait);
atomic_set(&lc->oc_this_node, 0);
conn->cc_private = lc;
lc->oc_type = NO_CONTROLD;
#endif
rc = dlm_new_lockspace(conn->cc_name, conn->cc_cluster_name,
DLM_LSFL_FS | DLM_LSFL_NEWEXCL, DLM_LVB_LEN,
&ocfs2_ls_ops, conn, &ops_rv, &fsdlm);
if (rc) {
if (rc == -EEXIST || rc == -EPROTO)
printk(KERN_ERR "scoutfs: Unable to create the "
"lockspace %s (%d), because a scoutfs-utils "
"program is running on this file system "
"with the same name lockspace\n",
conn->cc_name, rc);
goto out;
}
if (ops_rv == -EOPNOTSUPP) {
/*
* If we get this return code, we're on a very old
* version of fs/dlm that doesn't have recovery
* callbacks enabled.
*/
// lc->oc_type = WITH_CONTROLD;
printk(KERN_NOTICE "scoutfs: You seem to be using an older "
"version of dlm_controld and/or scoutfs-utils."
" Please consider upgrading.\n");
} else if (ops_rv) {
rc = ops_rv;
goto out;
}
conn->cc_lockspace = fsdlm;
#if 0
rc = ocfs2_live_connection_attach(conn, lc);
if (rc)
goto out;
if (lc->oc_type == NO_CONTROLD) {
rc = get_protocol_version(conn);
if (rc) {
printk(KERN_ERR "ocfs2: Could not determine"
" locking version\n");
user_cluster_disconnect(conn);
goto out;
}
wait_event(lc->oc_wait, (atomic_read(&lc->oc_this_node) > 0));
}
/*
* running_proto must have been set before we allowed any mounts
* to proceed.
*/
if (fs_protocol_compare(&running_proto, &conn->cc_version)) {
printk(KERN_ERR
"Unable to mount with fs locking protocol version "
"%u.%u because negotiated protocol is %u.%u\n",
conn->cc_version.pv_major, conn->cc_version.pv_minor,
running_proto.pv_major, running_proto.pv_minor);
rc = -EPROTO;
ocfs2_live_connection_drop(lc);
lc = NULL;
}
#endif
out:
#if 0
if (rc)
kfree(lc);
#endif
return rc;
}
int ocfs2_cluster_connect(const char *stack_name,
const char *cluster_name,
int cluster_name_len,
const char *group,
int grouplen,
struct ocfs2_locking_protocol *lproto,
void (*recovery_handler)(int node_num,
void *recovery_data),
void *recovery_data,
struct ocfs2_cluster_connection **conn)
{
int rc = 0;
struct ocfs2_cluster_connection *new_conn;
BUG_ON(group == NULL);
BUG_ON(conn == NULL);
BUG_ON(recovery_handler == NULL);
if (grouplen > GROUP_NAME_MAX) {
rc = -EINVAL;
goto out;
}
#if 0
if (memcmp(&lproto->lp_max_version, &locking_max_version,
sizeof(struct ocfs2_protocol_version))) {
rc = -EINVAL;
goto out;
}
#endif
new_conn = kzalloc(sizeof(struct ocfs2_cluster_connection),
GFP_KERNEL);
if (!new_conn) {
rc = -ENOMEM;
goto out;
}
strlcpy(new_conn->cc_name, group, GROUP_NAME_MAX + 1);
new_conn->cc_namelen = grouplen;
if (cluster_name_len)
strlcpy(new_conn->cc_cluster_name, cluster_name,
CLUSTER_NAME_MAX + 1);
new_conn->cc_cluster_name_len = cluster_name_len;
new_conn->cc_recovery_handler = recovery_handler;
new_conn->cc_recovery_data = recovery_data;
new_conn->cc_proto = lproto;
/* Start the new connection at our maximum compatibility level */
new_conn->cc_version = lproto->lp_max_version;
#if 0
/* This will pin the stack driver if successful */
rc = ocfs2_stack_driver_get(stack_name);
if (rc)
goto out_free;
#endif
rc = user_cluster_connect(new_conn);
if (rc) {
// ocfs2_stack_driver_put();
goto out_free;
}
*conn = new_conn;
out_free:
if (rc)
kfree(new_conn);
out:
return rc;
}
static int user_cluster_disconnect(struct ocfs2_cluster_connection *conn)
{
dlm_release_lockspace(conn->cc_lockspace, 2);
conn->cc_lockspace = NULL;
conn->cc_private = NULL;
return 0;
}
/* If hangup_pending is 0, the stack driver will be dropped */
int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn,
int hangup_pending)
{
int ret;
BUG_ON(conn == NULL);
ret = user_cluster_disconnect(conn);
/* XXX Should we free it anyway? */
if (!ret) {
kfree(conn);
#if 0
if (!hangup_pending)
ocfs2_stack_driver_put();
#endif
}
return ret;
}
kmod/src/stackglue.h
-149
View File
@@ -1,149 +0,0 @@
/* -*- mode: c; c-basic-offset: 8; -*-
* vim: noexpandtab sw=8 ts=8 sts=0:
*
* stackglue.h
*
* Glue to the underlying cluster stack.
*
* Copyright (C) 2007 Oracle. All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, version 2.
*
* 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 for more details.
*/
#ifndef STACKGLUE_H
#define STACKGLUE_H
#include <linux/types.h>
#include <linux/list.h>
#include <linux/dlmconstants.h>
#include <linux/dlm.h>
#include <linux/dlm_plock.h>
#define DLM_LVB_LEN 64
/* Needed for plock-related prototypes */
struct file;
struct file_lock;
/* Scoutfs never uses this flag, we define it to zero to avoid errors */
#define DLM_LKF_LOCAL 0
/*
* This shadows DLM_LOCKSPACE_LEN in fs/dlm/dlm_internal.h. That probably
* wants to be in a public header.
*/
#define GROUP_NAME_MAX 64
/* This shadows OCFS2_CLUSTER_NAME_LEN */
#define CLUSTER_NAME_MAX 16
/*
* ocfs2_protocol_version changes when ocfs2 does something different in
* its inter-node behavior. See dlmglue.c for more information.
*/
struct ocfs2_protocol_version {
u8 pv_major;
u8 pv_minor;
};
/*
* The dlm_lockstatus struct includes lvb space, but the dlm_lksb struct only
* has a pointer to separately allocated lvb space. This struct exists only to
* include in the lksb union to make space for a combined dlm_lksb and lvb.
*/
struct fsdlm_lksb_plus_lvb {
struct dlm_lksb lksb;
char lvb[DLM_LVB_LEN];
};
/*
* A union of all lock status structures. We define it here so that the
* size of the union is known. Lock status structures are embedded in
* ocfs2 inodes.
*/
struct ocfs2_cluster_connection;
struct ocfs2_dlm_lksb {
union {
struct dlm_lksb lksb_fsdlm;
struct fsdlm_lksb_plus_lvb padding;
};
struct ocfs2_cluster_connection *lksb_conn;
};
/*
* The ocfs2_locking_protocol defines the handlers called on ocfs2's behalf.
*/
struct ocfs2_locking_protocol {
struct ocfs2_protocol_version lp_max_version;
void (*lp_lock_ast)(struct ocfs2_dlm_lksb *lksb);
void (*lp_blocking_ast)(struct ocfs2_dlm_lksb *lksb, int level);
void (*lp_unlock_ast)(struct ocfs2_dlm_lksb *lksb, int error);
};
/*
* A cluster connection. Mostly opaque to ocfs2, the connection holds
* state for the underlying stack. ocfs2 does use cc_version to determine
* locking compatibility.
*/
struct ocfs2_cluster_connection {
char cc_name[GROUP_NAME_MAX + 1];
int cc_namelen;
char cc_cluster_name[CLUSTER_NAME_MAX + 1];
int cc_cluster_name_len;
struct ocfs2_protocol_version cc_version;
struct ocfs2_locking_protocol *cc_proto;
void (*cc_recovery_handler)(int node_num, void *recovery_data);
void *cc_recovery_data;
void *cc_lockspace;
void *cc_private;
};
/* In ocfs2_downconvert_lock(), we need to know which stack we are using */
static inline int ocfs2_is_o2cb_active(void)
{
return 0;
}
/* Used by the filesystem */
int ocfs2_cluster_connect(const char *stack_name,
const char *cluster_name,
int cluster_name_len,
const char *group,
int grouplen,
struct ocfs2_locking_protocol *lproto,
void (*recovery_handler)(int node_num,
void *recovery_data),
void *recovery_data,
struct ocfs2_cluster_connection **conn);
int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn,
int hangup_pending);
struct ocfs2_lock_res;
int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
int mode,
struct ocfs2_dlm_lksb *lksb,
u32 flags,
void *name,
unsigned int namelen);
int ocfs2_dlm_unlock(struct ocfs2_cluster_connection *conn,
struct ocfs2_dlm_lksb *lksb,
u32 flags);
int ocfs2_dlm_lock_status(struct ocfs2_dlm_lksb *lksb);
int ocfs2_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb);
void *ocfs2_dlm_lvb(struct ocfs2_dlm_lksb *lksb);
void ocfs2_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb);
int ocfs2_plock(struct ocfs2_cluster_connection *conn, u64 ino,
struct file *file, int cmd, struct file_lock *fl);
#endif /* STACKGLUE_H */
kmod/src/super.c
+2 -2
View File
@@ -122,8 +122,7 @@ static void scoutfs_put_super(struct super_block *sb)
sbi->shutdown = true;
scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX,
SCOUTFS_LKF_NO_TASK_REF);
scoutfs_unlock(sb, sbi->node_id_lock, DLM_LOCK_EX);
sbi->node_id_lock = NULL;
scoutfs_shutdown_trans(sb);
@@ -133,6 +132,7 @@ static void scoutfs_put_super(struct super_block *sb)
scoutfs_item_destroy(sb);
/* the server locks the listen address and compacts */
scoutfs_lock_shutdown(sb);
scoutfs_server_destroy(sb);
scoutfs_seg_destroy(sb);
scoutfs_lock_destroy(sb);