GFS2集群文件系统分析——gfs2_glock 分析

接着上一片文章，看来不分析glock结构，代码是读下去了。

先看下代码

struct gfs2_glock {
struct hlist_node gl_list;//glock链表
unsigned long gl_flags; /* GLF_... */
struct lm_lockname gl_name;
atomic_t gl_ref;//被引用数

spinlock_t gl_spin;

/* State fields protected by gl_spin */
unsigned int gl_state:2, /* Current state */
gl_target:2, /* Target state */
gl_demote_state:2, /* State requested by remote node */
gl_req:2, /* State in last dlm request */
gl_reply:8; /* Last reply from the dlm */

unsigned int gl_hash;
unsigned long gl_demote_time; /* time of first demote request */
struct list_head gl_holders;

const struct gfs2_glock_operations *gl_ops;
char gl_strname[GDLM_STRNAME_BYTES];
struct dlm_lksb gl_lksb;/////注意这里有个dlm锁
char gl_lvb[32];
unsigned long gl_tchange;
void *gl_object;

struct list_head gl_lru;

struct gfs2_sbd *gl_sbd;//指向gfs2的超级快

struct list_head gl_ail_list;
atomic_t gl_ail_count;
struct delayed_work gl_work;
struct work_struct gl_delete;
};


那gfs2_glock是怎么初始化的呢？看一下 gfs2_glock_init（）函数（这个函数会被init_gfs2_fs（）函数调用，也就是说，在初始化文件系统时，这个函数就被调用了）
int __init gfs2_glock_init(void)
{
unsigned i;
for(i = 0; i < GFS2_GL_HASH_SIZE; i++) {
INIT_HLIST_HEAD(&gl_hash_table[i].hb_list);
}
#ifdef GL_HASH_LOCK_SZ
for(i = 0; i < GL_HASH_LOCK_SZ; i++) {
rwlock_init(&gl_hash_locks[i]);
}
#endif

glock_workqueue = alloc_workqueue("glock_workqueue", WQ_MEM_RECLAIM |
WQ_HIGHPRI | WQ_FREEZABLE, 0);
if (IS_ERR(glock_workqueue))
return PTR_ERR(glock_workqueue);
gfs2_delete_workqueue = alloc_workqueue("delete_workqueue",
WQ_MEM_RECLAIM | WQ_FREEZABLE,
0);
if (IS_ERR(gfs2_delete_workqueue)) {
destroy_workqueue(glock_workqueue);
return PTR_ERR(gfs2_delete_workqueue);
}

register_shrinker(&glock_shrinker);

return 0;
}

初始化哈希表，初始化gl_hash_locks，初始化了一个工作队列
关于shriker可以看一下http://www.ibm.com/developerworks/cn/linux/l-cn-pagerecycle/ 这篇文章

到这里为止，还是没看到glock怎么初始化，那么再回过头去看下 rwlock_init(&gl_hash_locks[i])到底是干什么的，gl_hash_locks放的都是rwlock类型的锁，似乎跟gfs2_lock也没什么关系。

这样还是没找到在哪里初始化了gfs2_lock结构，再来仔细找一下，发现是在函数 gfs2_glock_get（）里，回去获得一个glock锁或申请一个新的glock锁，那什么时候会调用这个get()函数呢？在gfs2_inode_lookup()里就会调用这个函数。

下面来仔细看下gfs2_glock_get（）这个函数

/**
* gfs2_glock_get() - Get a glock, or create one if one doesn't exist
* @sdp: The GFS2 superblock
* @number: the lock number
* @glops: The glock_operations to use
* @create: If 0, don't create the glock if it doesn't exist
* @glp: the glock is returned here
*
* This does not lock a glock, just finds/creates structures for one.
*
* Returns: errno
*/

int gfs2_glock_get(struct gfs2_sbd *sdp, u64 number,
const struct gfs2_glock_operations *glops, int create,
struct gfs2_glock **glp)
{
struct super_block *s = sdp->sd_vfs;
struct lm_lockname name = { .ln_number = number, .ln_type = glops->go_type };
struct gfs2_glock *gl, *tmp;
unsigned int hash = gl_hash(sdp, &name);
struct address_space *mapping;

read_lock(gl_lock_addr(hash));
gl = search_bucket(hash, sdp, &name);
read_unlock(gl_lock_addr(hash));

*glp = gl;
if (gl)
return 0;
if (!create)
return -ENOENT;

if (glops->go_flags & GLOF_ASPACE)
gl = kmem_cache_alloc(gfs2_glock_aspace_cachep, GFP_KERNEL);
else
gl = kmem_cache_alloc(gfs2_glock_cachep, GFP_KERNEL);
if (!gl)
return -ENOMEM;

atomic_inc(&sdp->sd_glock_disposal);
gl->gl_flags = 0;
gl->gl_name = name;
atomic_set(&gl->gl_ref, 1);
gl->gl_state = LM_ST_UNLOCKED;
gl->gl_target = LM_ST_UNLOCKED;
gl->gl_demote_state = LM_ST_EXCLUSIVE;
gl->gl_hash = hash;
gl->gl_ops = glops;
snprintf(gl->gl_strname, GDLM_STRNAME_BYTES, "%8x%16llx", name.ln_type, (unsigned long long)number);
memset(&gl->gl_lksb, 0, sizeof(struct dlm_lksb));
gl->gl_lksb.sb_lvbptr = gl->gl_lvb;
gl->gl_tchange = jiffies;
gl->gl_object = NULL;
gl->gl_sbd = sdp;
INIT_DELAYED_WORK(&gl->gl_work, glock_work_func);
INIT_WORK(&gl->gl_delete, delete_work_func);

mapping = gfs2_glock2aspace(gl);
if (mapping) {
mapping->a_ops = &gfs2_meta_aops;
mapping->host = s->s_bdev->bd_inode;
mapping->flags = 0;
mapping_set_gfp_mask(mapping, GFP_NOFS);
mapping->assoc_mapping = NULL;
mapping->backing_dev_info = s->s_bdi;
mapping->writeback_index = 0;
}

write_lock(gl_lock_addr(hash));
tmp = search_bucket(hash, sdp, &name);
if (tmp) {
write_unlock(gl_lock_addr(hash));
glock_free(gl);
gl = tmp;
} else {
hlist_add_head(&gl->gl_list, &gl_hash_table[hash].hb_list);
write_unlock(gl_lock_addr(hash));
}

*glp = gl;

return 0;
}

//真tm复杂，吃个饭回来再写
//好了，现在继续。。。。 = =

首先看传见来的参数  glops，就是操作glock的一组函数，这组函数的定义是这样的

const struct gfs2_glock_operations gfs2_inode_glops = {
.go_xmote_th = inode_go_sync,
.go_inval = inode_go_inval,
.go_demote_ok = inode_go_demote_ok,
.go_lock = inode_go_lock,
.go_dump = inode_go_dump,
.go_type = LM_TYPE_INODE,
.go_min_hold_time = HZ / 5,
.go_flags = GLOF_ASPACE,
};

然后现在哈希表里找这个glock，找到就返回，找不到就创建一个新的glock对象

首先分配内存空间，然后初始化其数据成员，锁状态初始化为未锁定。

接着有两行代码

INIT_DELAYED_WORK(&gl->gl_work, glock_work_func);
INIT_WORK(&gl->gl_delete, delete_work_func);

这两句话是初始化了两个个工作队列，处理函数分别是 glock_work_func和delete_work_func，
关于工作队列，可以参考 http://blog.csdn.net/zchill/article/details/7076561 这篇博文

下面来看下glock_work_func（）这个函数，也就是在这个glock上等待的任务是怎么被处理的

来看源码

static void glock_work_func(struct work_struct *work)
{
unsigned long delay = 0;
struct gfs2_glock *gl = container_of(work, struct gfs2_glock, gl_work.work);
int drop_ref = 0;

if (test_and_clear_bit(GLF_REPLY_PENDING, &gl->gl_flags)) {
finish_xmote(gl, gl->gl_reply);
drop_ref = 1;
}
spin_lock(&gl->gl_spin);
if (test_and_clear_bit(GLF_PENDING_DEMOTE, &gl->gl_flags) &&
gl->gl_state != LM_ST_UNLOCKED &&
gl->gl_demote_state != LM_ST_EXCLUSIVE) {
unsigned long holdtime, now = jiffies;
holdtime = gl->gl_tchange + gl->gl_ops->go_min_hold_time;
if (time_before(now, holdtime))
delay = holdtime - now;
set_bit(delay ? GLF_PENDING_DEMOTE : GLF_DEMOTE, &gl->gl_flags);
}
run_queue(gl, 0);
spin_unlock(&gl->gl_spin);
if (!delay ||
queue_delayed_work(glock_workqueue, &gl->gl_work, delay) == 0)
gfs2_glock_put(gl);
if (drop_ref)
gfs2_glock_put(gl);
}

函数对一些标志位进行判断

总共有这些标志（目前我也不是很清楚这些标志位表示什么意思，也找不到文档，只能继续看下去再说）

enum {
GLF_LOCK = 1,
GLF_DEMOTE = 3,
GLF_PENDING_DEMOTE = 4,
GLF_DEMOTE_IN_PROGRESS = 5,
GLF_DIRTY = 6,
GLF_LFLUSH = 7,
GLF_INVALIDATE_IN_PROGRESS = 8,
GLF_REPLY_PENDING = 9,
GLF_INITIAL = 10,
GLF_FROZEN = 11,
GLF_QUEUED = 12,
};
清除了pending标志（干嘛的？待处理标志？）后调用finish_xmote(），来看下这个函数又是干嘛的

/**
* finish_xmote - The DLM has replied to one of our lock requests
* @gl: The glock
* @ret: The status from the DLM
*
*/

static void finish_xmote(struct gfs2_glock *gl, unsigned int ret)
{
const struct gfs2_glock_operations *glops = gl->gl_ops;
struct gfs2_holder *gh;
unsigned state = ret & LM_OUT_ST_MASK;
int rv;

spin_lock(&gl->gl_spin);
trace_gfs2_glock_state_change(gl, state);
state_change(gl, state);
gh = find_first_waiter(gl);

/* Demote to UN request arrived during demote to SH or DF */
if (test_bit(GLF_DEMOTE_IN_PROGRESS, &gl->gl_flags) &&
state != LM_ST_UNLOCKED && gl->gl_demote_state == LM_ST_UNLOCKED)
gl->gl_target = LM_ST_UNLOCKED;

/* Check for state != intended state */
if (unlikely(state != gl->gl_target)) {
if (gh && !test_bit(GLF_DEMOTE_IN_PROGRESS, &gl->gl_flags)) {
/* move to back of queue and try next entry */
if (ret & LM_OUT_CANCELED) {
if ((gh->gh_flags & LM_FLAG_PRIORITY) == 0)
list_move_tail(&gh->gh_list, &gl->gl_holders);
gh = find_first_waiter(gl);
gl->gl_target = gh->gh_state;
goto retry;
}
/* Some error or failed "try lock" - report it */
if ((ret & LM_OUT_ERROR) ||
(gh->gh_flags & (LM_FLAG_TRY | LM_FLAG_TRY_1CB))) {
gl->gl_target = gl->gl_state;
do_error(gl, ret);
goto out;
}
}
switch(state) {
/* Unlocked due to conversion deadlock, try again */
case LM_ST_UNLOCKED:
retry:
do_xmote(gl, gh, gl->gl_target);
break;
/* Conversion fails, unlock and try again */
case LM_ST_SHARED:
case LM_ST_DEFERRED:
do_xmote(gl, gh, LM_ST_UNLOCKED);
break;
default: /* Everything else */
printk(KERN_ERR "GFS2: wanted %u got %u\n", gl->gl_target, state);
GLOCK_BUG_ON(gl, 1);
}
spin_unlock(&gl->gl_spin);
return;
}

/* Fast path - we got what we asked for */
if (test_and_clear_bit(GLF_DEMOTE_IN_PROGRESS, &gl->gl_flags))
gfs2_demote_wake(gl);
if (state != LM_ST_UNLOCKED) {
if (glops->go_xmote_bh) {
spin_unlo
9cf2
ck(&gl->gl_spin);
rv = glops->go_xmote_bh(gl, gh);
spin_lock(&gl->gl_spin);
if (rv) {
do_error(gl, rv);
goto out;
}
}
rv = do_promote(gl);
if (rv == 2)
goto out_locked;
}
out:
clear_bit(GLF_LOCK, &gl->gl_flags);
out_locked:
spin_unlock(&gl->gl_spin);
}


//又是一大口血。。涉及到 dlm锁了(dlm是一个经典的分布式锁管理器，本身就非常复杂，这里就不深入了解这东西了）
trace_gfs2_glock_state_change（）//尼马啊，。。这是什么啊。。注释上说是“DLM lock request completes”， 就是请求到dlm锁了？

state_change(gl, state);//