Logo Search packages:      
Sourcecode: redhat-cluster-suite version File versions  Download package

glock.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2004 Red Hat, Inc.  All rights reserved.
**
**  This copyrighted material is made available to anyone wishing to use,
**  modify, copy, or redistribute it subject to the terms and conditions
**  of the GNU General Public License v.2.
**
*******************************************************************************
******************************************************************************/

#include <linux/sched.h>
#include <linux/slab.h>
#include <linux/smp_lock.h>
#include <linux/spinlock.h>
#include <asm/semaphore.h>
#include <linux/completion.h>
#include <linux/buffer_head.h>
#include <asm/uaccess.h>

#include "gfs.h"
#include "dio.h"
#include "glock.h"
#include "glops.h"
#include "inode.h"
#include "lm.h"
#include "lops.h"
#include "quota.h"
#include "recovery.h"

/*  Must be kept in sync with the beginning of struct gfs_glock  */
struct glock_plug {
      struct list_head gl_list;
      unsigned long gl_flags;
};

struct greedy {
      struct gfs_holder gr_gh;
      struct work_struct gr_work;
};

typedef void (*glock_examiner) (struct gfs_glock * gl);

/**
 * relaxed_state_ok - is a requested lock compatible with the current lock mode?
 * @actual: the current state of the lock
 * @requested: the lock state that was requested by the caller
 * @flags: the modifier flags passed in by the caller
 *
 * Returns: TRUE if the locks are compatible, FALSE otherwise
 *
 * It's often possible that a holder B may request the lock in SHARED mode,
 * while another holder A (on this same node) has the lock in EXCLUSIVE mode
 * (node must hold the glock in EXCLUSIVE mode for this situation, of course).
 * This is okay to grant, in some cases, since both holders would have access
 * to the in-core up-to-date cached data that the EX holder would write to disk.
 * This is the default behavior.
 *
 * The EXACT flag disallows this behavior, though.  A SHARED request would
 * compatible only with a SHARED lock with this flag.
 *
 * The ANY flag provides broader permission to grant the lock to a holder,
 * whatever the requested state is, as long as the lock is locked in any mode.
 */

static __inline__ int
relaxed_state_ok(unsigned int actual, unsigned requested, int flags)
{
      if (actual == requested)
            return TRUE;

      if (flags & GL_EXACT)
            return FALSE;

      if (actual == LM_ST_EXCLUSIVE && requested == LM_ST_SHARED)
            return TRUE;

      if (actual != LM_ST_UNLOCKED && (flags & LM_FLAG_ANY))
            return TRUE;

      return FALSE;
}

/**
 * gl_hash() - Turn glock number into hash bucket number
 * @lock: The glock number
 *
 * Returns: The number of the corresponding hash bucket
 */

static unsigned int
gl_hash(struct lm_lockname *name)
{
      unsigned int h;

      h = gfs_hash(&name->ln_number, sizeof(uint64_t));
      h = gfs_hash_more(&name->ln_type, sizeof(unsigned int), h);
      h &= GFS_GL_HASH_MASK;

      return h;
}

/**
 * glock_hold() - increment reference count on glock
 * @gl: The glock to hold
 *
 */

static __inline__ void
glock_hold(struct gfs_glock *gl)
{
      gfs_assert(gl->gl_sbd, atomic_read(&gl->gl_count) > 0,);
      atomic_inc(&gl->gl_count);
}

/**
 * glock_put() - Decrement reference count on glock
 * @gl: The glock to put
 *
 */

static __inline__ void
glock_put(struct gfs_glock *gl)
{
      if (atomic_read(&gl->gl_count) == 1)
            gfs_glock_schedule_for_reclaim(gl);
      gfs_assert(gl->gl_sbd, atomic_read(&gl->gl_count) > 0,);
      atomic_dec(&gl->gl_count);
}

/**
 * queue_empty - check to see if a glock's queue is empty
 * @gl: the glock
 * @head: the head of the queue to check
 *
 * Returns: TRUE if the queue is empty
 */

static __inline__ int
queue_empty(struct gfs_glock *gl, struct list_head *head)
{
      int empty;
      spin_lock(&gl->gl_spin);
      empty = list_empty(head);
      spin_unlock(&gl->gl_spin);
      return empty;
}

/**
 * search_bucket() - Find struct gfs_glock by lock number
 * @bucket: the bucket to search
 * @name: The lock name
 *
 * Returns: NULL, or the struct gfs_glock with the requested number
 */

static struct gfs_glock *
search_bucket(struct gfs_gl_hash_bucket *bucket, struct lm_lockname *name)
{
      struct list_head *tmp, *head;
      struct gfs_glock *gl;

      for (head = &bucket->hb_list, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gl = list_entry(tmp, struct gfs_glock, gl_list);

            if (test_bit(GLF_PLUG, &gl->gl_flags))
                  continue;
            if (!lm_name_equal(&gl->gl_name, name))
                  continue;

            atomic_inc(&gl->gl_count);

            return gl;
      }

      return NULL;
}

/**
 * gfs_glock_find() - Find glock by lock number
 * @sdp: The GFS superblock
 * @name: The lock name
 *
 * Figure out what bucket the lock is in, acquire the read lock on
 * it and call search_bucket().
 *
 * Returns: NULL, or the struct gfs_glock with the requested number
 */

struct gfs_glock *
gfs_glock_find(struct gfs_sbd *sdp, struct lm_lockname *name)
{
      struct gfs_gl_hash_bucket *bucket = &sdp->sd_gl_hash[gl_hash(name)];
      struct gfs_glock *gl;

      read_lock(&bucket->hb_lock);
      gl = search_bucket(bucket, name);
      read_unlock(&bucket->hb_lock);

      return gl;
}

/**
 * glock_free() - Perform a few checks and then release struct gfs_glock
 * @gl: The glock to release
 *
 * Also calls lock module to release its internal structure for this glock.
 *
 */

static void
glock_free(struct gfs_glock *gl)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct inode *aspace = gl->gl_aspace;

      gfs_assert_warn(sdp, list_empty(&gl->gl_list));
      gfs_assert_warn(sdp, atomic_read(&gl->gl_count) == 1);
      gfs_assert_warn(sdp, list_empty(&gl->gl_holders));
      gfs_assert_warn(sdp, list_empty(&gl->gl_waiters1));
        gfs_assert_warn(sdp, list_empty(&gl->gl_waiters2));
      gfs_assert_warn(sdp, list_empty(&gl->gl_waiters3));
      gfs_assert_warn(sdp, gl->gl_state == LM_ST_UNLOCKED);
      gfs_assert_warn(sdp, !gl->gl_object);
      gfs_assert_warn(sdp, !gl->gl_lvb);
      gfs_assert_warn(sdp, list_empty(&gl->gl_reclaim));

      gfs_lm_put_lock(sdp, gl->gl_lock);

      if (aspace)
            gfs_aspace_put(aspace);

      kmem_cache_free(gfs_glock_cachep, gl);

      atomic_dec(&sdp->sd_glock_count);
}

/**
 * gfs_glock_get() - Get a glock, or create one if one doesn't exist
 * @sdp: The GFS superblock
 * @number: the lock number
 * @glops: The glock_operations to use
 * @create: If FALSE, 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
gfs_glock_get(struct gfs_sbd *sdp,
            uint64_t number, struct gfs_glock_operations *glops,
            int create, struct gfs_glock **glp)
{
      struct lm_lockname name;
      struct gfs_glock *gl, *tmp;
      struct gfs_gl_hash_bucket *bucket;
      int error;

      /* Look for pre-existing glock in hash table */
      name.ln_number = number;
      name.ln_type = glops->go_type;
      bucket = &sdp->sd_gl_hash[gl_hash(&name)];

      read_lock(&bucket->hb_lock);
      gl = search_bucket(bucket, &name);
      read_unlock(&bucket->hb_lock);

      if (gl || !create) {
            *glp = gl;
            return 0;
      }

      /* None found; create a new one */
      gl = kmem_cache_alloc(gfs_glock_cachep, GFP_KERNEL);
      if (!gl)
            return -ENOMEM;

      memset(gl, 0, sizeof(struct gfs_glock));

      INIT_LIST_HEAD(&gl->gl_list);
      gl->gl_name = name;
      atomic_set(&gl->gl_count, 1);

      spin_lock_init(&gl->gl_spin);

      gl->gl_state = LM_ST_UNLOCKED;
      INIT_LIST_HEAD(&gl->gl_holders);
      INIT_LIST_HEAD(&gl->gl_waiters1);
      INIT_LIST_HEAD(&gl->gl_waiters2);
      INIT_LIST_HEAD(&gl->gl_waiters3);

      gl->gl_ops = glops;

      INIT_LE(&gl->gl_new_le, &gfs_glock_lops);
      INIT_LE(&gl->gl_incore_le, &gfs_glock_lops);

      gl->gl_bucket = bucket;
      INIT_LIST_HEAD(&gl->gl_reclaim);

      gl->gl_sbd = sdp;

      INIT_LIST_HEAD(&gl->gl_ail_bufs);

      /* If this glock protects actual on-disk data or metadata blocks,
         create a VFS inode to manage the pages/buffers holding them. */
      if (glops == &gfs_inode_glops ||
          glops == &gfs_rgrp_glops ||
          glops == &gfs_meta_glops) {
            gl->gl_aspace = gfs_aspace_get(sdp);
            if (!gl->gl_aspace) {
                  error = -ENOMEM;
                  goto fail;
            }
      }

      /* Ask lock module to find/create its structure for this lock
         (but this doesn't lock the inter-node lock yet) */
      error = gfs_lm_get_lock(sdp, &name, &gl->gl_lock);
      if (error)
            goto fail_aspace;

      atomic_inc(&sdp->sd_glock_count);

      /* Double-check, in case another process created the glock, and has
         put it in the hash table while we were preparing this one */
      write_lock(&bucket->hb_lock);
      tmp = search_bucket(bucket, &name);
      if (tmp) {
            /* Somebody beat us to it; forget the one we prepared */
            write_unlock(&bucket->hb_lock);
            glock_free(gl);
            gl = tmp;
      } else {
            /* Add our glock to hash table */
            list_add_tail(&gl->gl_list, &bucket->hb_list);
            write_unlock(&bucket->hb_lock);
      }

      *glp = gl;

      return 0;

 fail_aspace:
      if (gl->gl_aspace)
            gfs_aspace_put(gl->gl_aspace);

 fail:
      kmem_cache_free(gfs_glock_cachep, gl);    

      return error;
}

/**
 * gfs_glock_hold() - As glock_hold(), but suitable for exporting
 * @gl: The glock to hold
 *
 */

void
gfs_glock_hold(struct gfs_glock *gl)
{
      glock_hold(gl);
}

/**
 * gfs_glock_put() - As glock_put(), but suitable for exporting
 * @gl: The glock to put
 *
 */

void
gfs_glock_put(struct gfs_glock *gl)
{
      glock_put(gl);
}

/**
 * gfs_holder_init - initialize a struct gfs_holder in the default way
 * @gl: the glock 
 * @state: the state we're requesting
 * @flags: the modifier flags
 * @gh: the holder structure
 *
 */

void
gfs_holder_init(struct gfs_glock *gl, unsigned int state, int flags,
            struct gfs_holder *gh)
{
      memset(gh, 0, sizeof(struct gfs_holder));

      INIT_LIST_HEAD(&gh->gh_list);
      gh->gh_gl = gl;
      gh->gh_owner = current;
      gh->gh_state = state;
      gh->gh_flags = flags;

      if (gh->gh_state == LM_ST_EXCLUSIVE)
            gh->gh_flags |= GL_LOCAL_EXCL;

      init_completion(&gh->gh_wait);

      glock_hold(gl);
}

/**
 * gfs_holder_reinit - reinitialize a struct gfs_holder so we can requeue it
 * @state: the state we're requesting
 * @flags: the modifier flags
 * @gh: the holder structure
 *
 * Preserve holder's associated glock and owning process.
 * Reset all holder state flags (we're starting a new request from scratch),
 *   except for HIF_ALLOCED.
 * Don't do glock_hold() again (it was done in gfs_holder_init()).
 * Don't mess with the glock.
 *
 * Rules:
 *   Holder must have been gfs_holder_init()d already
 *   Holder must *not* be in glock's holder list or wait queues now
 */

void
gfs_holder_reinit(unsigned int state, int flags, struct gfs_holder *gh)
{
      int alloced;

      gfs_assert_warn(gh->gh_gl->gl_sbd,
                  list_empty(&gh->gh_list));

      gh->gh_state = state;
      gh->gh_flags = flags;

      if (gh->gh_state == LM_ST_EXCLUSIVE)
            gh->gh_flags |= GL_LOCAL_EXCL;

      alloced = test_bit(HIF_ALLOCED, &gh->gh_iflags);
      memset(&gh->gh_iflags, 0, sizeof(unsigned long));
      if (alloced)
            set_bit(HIF_ALLOCED, &gh->gh_iflags);
}

/**
 * gfs_holder_uninit - uninitialize a holder structure (drop reference on glock)
 * @gh: the holder structure
 *
 */

void
gfs_holder_uninit(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;

      gfs_assert_warn(gl->gl_sbd, list_empty(&gh->gh_list));
      gh->gh_gl = NULL;

      glock_put(gl);
}

/**
 * gfs_holder_get - get a struct gfs_holder structure
 * @gl: the glock 
 * @state: the state we're requesting
 * @flags: the modifier flags
 *
 * Figure out how big an impact this function has.  Either:
 * 1) Replace it with a cache of structures hanging off the struct gfs_sbd
 * 2) Leave it like it is
 *
 * Returns: the holder structure, NULL on ENOMEM
 */

struct gfs_holder *
gfs_holder_get(struct gfs_glock *gl, unsigned int state, int flags)
{
      struct gfs_holder *gh;

      gh = kmalloc(sizeof(struct gfs_holder), GFP_KERNEL);
      if (!gh)
            return NULL;

      gfs_holder_init(gl, state, flags, gh);
      set_bit(HIF_ALLOCED, &gh->gh_iflags);

      return gh;
}

/**
 * gfs_holder_put - get rid of a struct gfs_holder structure
 * @gh: the holder structure
 *
 */

void
gfs_holder_put(struct gfs_holder *gh)
{
      if (gfs_assert_warn(gh->gh_gl->gl_sbd,
                      test_bit(HIF_ALLOCED, &gh->gh_iflags)))
            return;
      gfs_holder_uninit(gh);
      kfree(gh);
}

/**
 * handle_recurse - put other holder structures (marked recursive) into the holders list
 * @gh: the holder structure
 *
 */

static void
handle_recurse(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct list_head *tmp, *head, *next;
      struct gfs_holder *tmp_gh;
      int found = FALSE;

      if (gfs_assert_warn(sdp, gh->gh_owner))
            return;

      for (head = &gl->gl_waiters3, tmp = head->next, next = tmp->next;
           tmp != head;
           tmp = next, next = tmp->next) {
            tmp_gh = list_entry(tmp, struct gfs_holder, gh_list);
            if (tmp_gh->gh_owner != gh->gh_owner)
                  continue;

            gfs_assert_warn(sdp, test_bit(HIF_RECURSE, &tmp_gh->gh_iflags));

            list_move_tail(&tmp_gh->gh_list, &gl->gl_holders);
            tmp_gh->gh_error = 0;
            set_bit(HIF_HOLDER, &tmp_gh->gh_iflags);

            complete(&tmp_gh->gh_wait);

            found = TRUE;
      }

      gfs_assert_warn(sdp, found);
}

/**
 * do_unrecurse - a recursive holder was just dropped of the waiters3 list
 * @gh: the holder
 *
 * If there is only one other recursive holder, clear its HIF_RECURSE bit
 *   (it's no longer a recursive request).
 * If there is more than one, leave them alone (they're recursive!).
 *
 */

static void
do_unrecurse(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct list_head *tmp, *head;
      struct gfs_holder *tmp_gh, *last_gh = NULL;
      int found = FALSE;

      if (gfs_assert_warn(sdp, gh->gh_owner))
            return;

      for (head = &gl->gl_waiters3, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            tmp_gh = list_entry(tmp, struct gfs_holder, gh_list);
            if (tmp_gh->gh_owner != gh->gh_owner)
                  continue;

            gfs_assert_warn(sdp, test_bit(HIF_RECURSE, &tmp_gh->gh_iflags));

            /* found more than one */
            if (found)
                  return;

            found = TRUE;
            last_gh = tmp_gh;
      }

      /* found just one */
      if (!gfs_assert_warn(sdp, found))
            clear_bit(HIF_RECURSE, &last_gh->gh_iflags);
}

/**
 * rq_mutex - process a mutex request in the queue
 * @gh: the glock holder
 *
 * Returns: TRUE if the queue is blocked (always, since there can be only one
 *      holder of the mutex).
 *
 * See lock_on_glock()
 */

static int
rq_mutex(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;

      list_del_init(&gh->gh_list);
      /*  gh->gh_error never examined.  */
      set_bit(GLF_LOCK, &gl->gl_flags);
      complete(&gh->gh_wait);

      return TRUE;
}

/**
 * rq_promote - process a promote request in the queue
 * @gh: the glock holder
 *
 * Acquire a new inter-node lock, or change a lock state to more restrictive.
 *
 * Returns: TRUE if the queue is blocked
 */

static int
rq_promote(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      int recurse;

      if (!relaxed_state_ok(gl->gl_state, gh->gh_state, gh->gh_flags)) {
            if (list_empty(&gl->gl_holders)) {
                  gl->gl_req_gh = gh;
                  set_bit(GLF_LOCK, &gl->gl_flags);
                  spin_unlock(&gl->gl_spin);

                  /* If we notice a lot of glocks in reclaim list, free
                     up memory for 2 of them before locking a new one */
                  if (atomic_read(&sdp->sd_reclaim_count) >
                      gfs_tune_get(sdp, gt_reclaim_limit) &&
                      !(gh->gh_flags & LM_FLAG_PRIORITY)) {
                        gfs_reclaim_glock(sdp);
                        gfs_reclaim_glock(sdp);
                  }

                  glops->go_xmote_th(gl, gh->gh_state,
                                 gh->gh_flags);

                  spin_lock(&gl->gl_spin);
            }
            return TRUE;
      }

      if (list_empty(&gl->gl_holders)) {
            set_bit(HIF_FIRST, &gh->gh_iflags);
            set_bit(GLF_LOCK, &gl->gl_flags);
            recurse = FALSE;
      } else {
            struct gfs_holder *next_gh;
            if (gh->gh_flags & GL_LOCAL_EXCL)
                  return TRUE;
            next_gh = list_entry(gl->gl_holders.next, struct gfs_holder, gh_list);
            if (next_gh->gh_flags & GL_LOCAL_EXCL)
                   return TRUE;
            recurse = test_bit(HIF_RECURSE, &gh->gh_iflags);
      }

      list_move_tail(&gh->gh_list, &gl->gl_holders);
      gh->gh_error = 0;
      set_bit(HIF_HOLDER, &gh->gh_iflags);

      if (recurse)
            handle_recurse(gh);

      complete(&gh->gh_wait);

      return FALSE;
}

/**
 * rq_demote - process a demote request in the queue
 * @gh: the glock holder
 *
 * Returns: TRUE if the queue is blocked
 *
 * Unlock an inter-node lock, or change a lock state to less restrictive.
 * If the glock is already the same as the holder's requested state, or is
 *   UNLOCKED, no lock module request is required.
 * Otherwise, we need to ask lock module to unlock or change locked state
 *   of the glock.
 * If requested state is UNLOCKED, or current glock state is SHARED or
 *   DEFERRED (neither of which have a less restrictive state other than
 *   UNLOCK), we call go_drop_th() to unlock the lock.
 * Otherwise (i.e. requested is SHARED or DEFERRED, and current is EXCLUSIVE),
 *   we can continue to hold the lock, and just ask for a new state;
 *   we call go_xmote_th() to change state.
 *
 * Must be called with glock's gl->gl_spin locked.
 */

static int
rq_demote(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_glock_operations *glops = gl->gl_ops;

      if (!list_empty(&gl->gl_holders))
            return TRUE;

      if (gl->gl_state == gh->gh_state || gl->gl_state == LM_ST_UNLOCKED) {
            list_del_init(&gh->gh_list);
            gh->gh_error = 0;
            spin_unlock(&gl->gl_spin);
            if (test_bit(HIF_DEALLOC, &gh->gh_iflags))
                  gfs_holder_put(gh);
            else
                  complete(&gh->gh_wait);
            spin_lock(&gl->gl_spin);
      } else {
            gl->gl_req_gh = gh;
            set_bit(GLF_LOCK, &gl->gl_flags);
            spin_unlock(&gl->gl_spin);

            if (gh->gh_state == LM_ST_UNLOCKED ||
                gl->gl_state != LM_ST_EXCLUSIVE)
                  /* Unlock */
                  glops->go_drop_th(gl);
            else
                  /* Change state while holding lock */
                  glops->go_xmote_th(gl, gh->gh_state, gh->gh_flags);

            spin_lock(&gl->gl_spin);
      }

      return FALSE;
}

/**
 * rq_greedy - process a queued request to drop greedy status
 * @gh: the glock holder
 *
 * Returns: TRUE if the queue is blocked
 */

static int
rq_greedy(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;

      list_del_init(&gh->gh_list);
      /*  gh->gh_error never examined.  */
      clear_bit(GLF_GREEDY, &gl->gl_flags);
      spin_unlock(&gl->gl_spin);

      gfs_holder_uninit(gh);
      kfree(container_of(gh, struct greedy, gr_gh));

      spin_lock(&gl->gl_spin);            

      return FALSE;
}

/**
 * run_queue - process holder structures on the glock's wait queues
 * @gl: the glock
 *
 * Rules:
 *   Caller must hold gl->gl_spin.
 */

static void
run_queue(struct gfs_glock *gl)
{
      struct gfs_holder *gh;
      int blocked = TRUE;

      for (;;) {
            /* Another process is manipulating the glock structure;
               we can't do anything now */
            if (test_bit(GLF_LOCK, &gl->gl_flags))
                  break;

            /* Waiting to manipulate the glock structure */
            if (!list_empty(&gl->gl_waiters1)) {
                  gh = list_entry(gl->gl_waiters1.next,
                              struct gfs_holder, gh_list);

                  if (test_bit(HIF_MUTEX, &gh->gh_iflags))
                        blocked = rq_mutex(gh);
                  else
                        gfs_assert_warn(gl->gl_sbd, FALSE);

            /* Waiting to demote the lock, or drop greedy status */
            } else if (!list_empty(&gl->gl_waiters2) &&
                     !test_bit(GLF_SKIP_WAITERS2, &gl->gl_flags)) {
                  gh = list_entry(gl->gl_waiters2.next,
                              struct gfs_holder, gh_list);

                  if (test_bit(HIF_DEMOTE, &gh->gh_iflags))
                        blocked = rq_demote(gh);
                  else if (test_bit(HIF_GREEDY, &gh->gh_iflags))
                        blocked = rq_greedy(gh);
                  else
                        gfs_assert_warn(gl->gl_sbd, FALSE);

            /* Waiting to promote the lock */
            } else if (!list_empty(&gl->gl_waiters3)) {
                  gh = list_entry(gl->gl_waiters3.next,
                              struct gfs_holder, gh_list);

                  if (test_bit(HIF_PROMOTE, &gh->gh_iflags))
                        blocked = rq_promote(gh);
                  else
                        gfs_assert_warn(gl->gl_sbd, FALSE);

            } else
                  break;

            if (blocked)
                  break;
      }
}

/**
 * lock_on_glock - acquire a local lock on a glock (structure)
 * @gl: the glock
 *
 * Gives caller exclusive access to manipulate a glock structure.
 * Has nothing to do with inter-node lock state or GL_LOCAL_EXCL!
 *
 * If structure already locked, places temporary holder structure on glock's
 * wait-for-exclusive-access queue, and blocks until exclusive access granted.
 */

static void
lock_on_glock(struct gfs_glock *gl)
{
      struct gfs_holder gh;

      gfs_holder_init(gl, 0, 0, &gh);
      set_bit(HIF_MUTEX, &gh.gh_iflags);

      spin_lock(&gl->gl_spin);
      if (test_and_set_bit(GLF_LOCK, &gl->gl_flags))
            list_add_tail(&gh.gh_list, &gl->gl_waiters1);
      else
            complete(&gh.gh_wait);
      spin_unlock(&gl->gl_spin);

      wait_for_completion(&gh.gh_wait);
      gfs_holder_uninit(&gh);
}

/**
 * trylock_on_glock - try to acquire a local lock on a glock (structure)
 * @gl: the glock
 *
 * Returns: TRUE if the glock is acquired
 *
 * Tries to give caller exclusive access to manipulate a glock structure.
 * Has nothing to do with inter-node lock state or LOCAL_EXCL!
 *
 * If structure already locked, does not block to wait; returns FALSE.
 */

static int
trylock_on_glock(struct gfs_glock *gl)
{
      int acquired = TRUE;

      spin_lock(&gl->gl_spin);
      if (test_and_set_bit(GLF_LOCK, &gl->gl_flags))
            acquired = FALSE;
      spin_unlock(&gl->gl_spin);

      return acquired;
}

/**
 * unlock_on_glock - release a local lock on a glock (structure)
 * @gl: the glock
 *
 * Caller is done manipulating glock structure.
 * Service any others waiting for exclusive access.
 */

static void
unlock_on_glock(struct gfs_glock *gl)
{
      spin_lock(&gl->gl_spin);
      clear_bit(GLF_LOCK, &gl->gl_flags);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);
}

/**
 * handle_callback - add a demote request to a lock's queue
 * @gl: the glock
 * @state: the state the caller wants us to change to
 *
 * Called when we learn that another node needs a lock held by this node,
 *   or when this node simply wants to drop a lock as soon as it's done with
 *   it (NOCACHE flag), or dump a glock out of glock cache (reclaim it).
 *
 * We are told the @state that will satisfy the needs of the caller, so
 *   we can ask for a demote to that state.
 *
 * If another demote request is already on the queue for a different state, just
 *   set its request to UNLOCK (and don't bother queueing a request for us).
 *   This consolidates LM requests and moves the lock to the least restrictive
 *   state, so it will be compatible with whatever reason we were called.
 *   No need to be too smart here.  Demotes between the shared and deferred
 *   states will often fail, so don't even try.
 *
 * Otherwise, queue a demote request to the requested state.
 */

static void
handle_callback(struct gfs_glock *gl, unsigned int state)
{
      struct list_head *tmp, *head;
      struct gfs_holder *gh, *new_gh = NULL;

      if (gfs_assert_warn(gl->gl_sbd, state != LM_ST_EXCLUSIVE))
            return;

 restart:
      spin_lock(&gl->gl_spin);

      /* If another queued demote request is for a different state,
         set its request to UNLOCKED */
      for (head = &gl->gl_waiters2, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gh = list_entry(tmp, struct gfs_holder, gh_list);
            if (test_bit(HIF_DEMOTE, &gh->gh_iflags) &&
                gl->gl_req_gh != gh) {
                  if (gh->gh_state != state)
                        gh->gh_state = LM_ST_UNLOCKED;
                  goto out;
            }
      }

      /* pass 2; add new holder to glock's demote request queue */
      if (new_gh) {
            list_add_tail(&new_gh->gh_list, &gl->gl_waiters2);
            new_gh = NULL;

      /* pass 1; set up a new holder struct for a demote request, then
         check again to see if another process added a demote request
         while we were preparing this one. */
      } else {
            spin_unlock(&gl->gl_spin);

            RETRY_MALLOC(new_gh = gfs_holder_get(gl, state, LM_FLAG_TRY),
                       new_gh);
            set_bit(HIF_DEMOTE, &new_gh->gh_iflags);
            set_bit(HIF_DEALLOC, &new_gh->gh_iflags);
            new_gh->gh_owner = NULL;

            goto restart;
      }

 out:
      spin_unlock(&gl->gl_spin);

      if (new_gh)
            gfs_holder_put(new_gh);
}

/**
 * state_change - record that the glock is now in a different state
 * @gl: the glock
 * @new_state the new state
 *
 */

static void
state_change(struct gfs_glock *gl, unsigned int new_state)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      int held1, held2;

      held1 = (gl->gl_state != LM_ST_UNLOCKED);
      held2 = (new_state != LM_ST_UNLOCKED);

      if (held1 != held2) {
            if (held2) {
                  atomic_inc(&sdp->sd_glock_held_count);
                  glock_hold(gl);
            } else {
                  atomic_dec(&sdp->sd_glock_held_count);
                  glock_put(gl);
            }
      }

      gl->gl_state = new_state;
}

/**
 * xmote_bh - Called after the lock module is done acquiring a lock
 * @gl: The glock in question
 * @ret: the int returned from the lock module
 *
 */

static void
xmote_bh(struct gfs_glock *gl, unsigned int ret)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      struct gfs_holder *gh = gl->gl_req_gh;
      int prev_state = gl->gl_state;
      int op_done = TRUE;

      gfs_assert_warn(sdp, test_bit(GLF_LOCK, &gl->gl_flags));
      gfs_assert_warn(sdp, queue_empty(gl, &gl->gl_holders));
      gfs_assert_warn(sdp, !(ret & LM_OUT_ASYNC));

      state_change(gl, ret & LM_OUT_ST_MASK);

      if (prev_state != LM_ST_UNLOCKED && !(ret & LM_OUT_CACHEABLE)) {
            if (glops->go_inval)
                  glops->go_inval(gl, DIO_METADATA | DIO_DATA);
      } else if (gl->gl_state == LM_ST_DEFERRED) {
            /* We might not want to do this here.
               Look at moving to the inode glops. */
            if (glops->go_inval)
                  glops->go_inval(gl, DIO_DATA);
      }

      /*  Deal with each possible exit condition  */

      if (!gh)
            gl->gl_stamp = jiffies;

      else if (unlikely(test_bit(SDF_SHUTDOWN, &sdp->sd_flags))) {
            spin_lock(&gl->gl_spin);
            list_del_init(&gh->gh_list);
            gh->gh_error = -EIO;
            if (test_bit(HIF_RECURSE, &gh->gh_iflags))
                  do_unrecurse(gh);
            spin_unlock(&gl->gl_spin);

      } else if (test_bit(HIF_DEMOTE, &gh->gh_iflags)) {
            spin_lock(&gl->gl_spin);
            list_del_init(&gh->gh_list);
            if (gl->gl_state == gh->gh_state ||
                gl->gl_state == LM_ST_UNLOCKED)
                  gh->gh_error = 0;
            else {
                  if (gfs_assert_warn(sdp, gh->gh_flags &
                                  (LM_FLAG_TRY | LM_FLAG_TRY_1CB)) == -1)
                        printk("GFS: fsid=%s: ret = 0x%.8X\n",
                               sdp->sd_fsname, ret);
                  gh->gh_error = GLR_TRYFAILED;
            }
            spin_unlock(&gl->gl_spin);

            if (ret & LM_OUT_CANCELED)
                  handle_callback(gl, LM_ST_UNLOCKED); /* Lame */

      } else if (ret & LM_OUT_CANCELED) {
            spin_lock(&gl->gl_spin);
            list_del_init(&gh->gh_list);
            gh->gh_error = GLR_CANCELED;
            if (test_bit(HIF_RECURSE, &gh->gh_iflags))
                  do_unrecurse(gh);
            spin_unlock(&gl->gl_spin);

      } else if (relaxed_state_ok(gl->gl_state, gh->gh_state, gh->gh_flags)) {
            spin_lock(&gl->gl_spin);
            list_move_tail(&gh->gh_list, &gl->gl_holders);
            gh->gh_error = 0;
            set_bit(HIF_HOLDER, &gh->gh_iflags);
            spin_unlock(&gl->gl_spin);

            set_bit(HIF_FIRST, &gh->gh_iflags);

            op_done = FALSE;

      } else if (gh->gh_flags & (LM_FLAG_TRY | LM_FLAG_TRY_1CB)) {
            spin_lock(&gl->gl_spin);
            list_del_init(&gh->gh_list);
            gh->gh_error = GLR_TRYFAILED;
            if (test_bit(HIF_RECURSE, &gh->gh_iflags))
                  do_unrecurse(gh);
            spin_unlock(&gl->gl_spin);

      } else {
            if (gfs_assert_withdraw(sdp, FALSE) == -1)
                  printk("GFS: fsid=%s: ret = 0x%.8X\n",
                         sdp->sd_fsname, ret);
      }

      if (glops->go_xmote_bh)
            glops->go_xmote_bh(gl);

      if (op_done) {
            spin_lock(&gl->gl_spin);
            gl->gl_req_gh = NULL;
            gl->gl_req_bh = NULL;
            clear_bit(GLF_LOCK, &gl->gl_flags);
            run_queue(gl);
            spin_unlock(&gl->gl_spin);
      }

      glock_put(gl);

      if (gh) {
            if (test_bit(HIF_DEALLOC, &gh->gh_iflags))
                  gfs_holder_put(gh);
            else
                  complete(&gh->gh_wait);
      }
}

/**
 * gfs_glock_xmote_th - Call into the lock module to acquire or change a glock
 * @gl: The glock in question
 * @state: the requested state
 * @flags: modifier flags to the lock call
 *
 * Used to acquire a new glock, or to change an already-acquired glock to
 *   more/less restrictive state (other than LM_ST_UNLOCKED).
 *
 * *Not* used to unlock a glock; use gfs_glock_drop_th() for that.
 */

void
gfs_glock_xmote_th(struct gfs_glock *gl, unsigned int state, int flags)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      int lck_flags = flags & (LM_FLAG_TRY | LM_FLAG_TRY_1CB |
                         LM_FLAG_NOEXP | LM_FLAG_ANY |
                         LM_FLAG_PRIORITY);
      unsigned int lck_ret;

      gfs_assert_warn(sdp, test_bit(GLF_LOCK, &gl->gl_flags));
      gfs_assert_warn(sdp, queue_empty(gl, &gl->gl_holders));
      gfs_assert_warn(sdp, state != LM_ST_UNLOCKED);
      gfs_assert_warn(sdp, state != gl->gl_state);

      /* Current state EX, may need to sync log/data/metadata to disk */
      if (gl->gl_state == LM_ST_EXCLUSIVE) {
            if (glops->go_sync)
                  glops->go_sync(gl, DIO_METADATA | DIO_DATA);
      }

      glock_hold(gl);
      gl->gl_req_bh = xmote_bh;

      atomic_inc(&sdp->sd_lm_lock_calls);

      lck_ret = gfs_lm_lock(sdp, gl->gl_lock,
                        gl->gl_state, state,
                        lck_flags);

      if (lck_ret & LM_OUT_ASYNC)
            gfs_assert_warn(sdp, lck_ret == LM_OUT_ASYNC);
      else
            xmote_bh(gl, lck_ret);
}

/**
 * drop_bh - Called after a lock module unlock completes
 * @gl: the glock
 * @ret: the return status
 *
 * Doesn't wake up the process waiting on the struct gfs_holder (if any)
 * Doesn't drop the reference on the glock the top half took out
 *
 */

static void
drop_bh(struct gfs_glock *gl, unsigned int ret)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      struct gfs_holder *gh = gl->gl_req_gh;

      clear_bit(GLF_PREFETCH, &gl->gl_flags);

      gfs_assert_warn(sdp, test_bit(GLF_LOCK, &gl->gl_flags));
      gfs_assert_warn(sdp, queue_empty(gl, &gl->gl_holders));
      gfs_assert_warn(sdp, !ret);

      state_change(gl, LM_ST_UNLOCKED);

      if (glops->go_inval)
            glops->go_inval(gl, DIO_METADATA | DIO_DATA);

      if (gh) {
            spin_lock(&gl->gl_spin);
            list_del_init(&gh->gh_list);
            gh->gh_error = 0;
            spin_unlock(&gl->gl_spin);
      }

      if (glops->go_drop_bh)
            glops->go_drop_bh(gl);

      spin_lock(&gl->gl_spin);
      gl->gl_req_gh = NULL;
      gl->gl_req_bh = NULL;
      clear_bit(GLF_LOCK, &gl->gl_flags);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);

      glock_put(gl);

      if (gh) {
            if (test_bit(HIF_DEALLOC, &gh->gh_iflags))
                  gfs_holder_put(gh);
            else
                  complete(&gh->gh_wait);
      }
}

/**
 * gfs_glock_drop_th - call into the lock module to unlock a lock 
 * @gl: the glock
 *
 */

void
gfs_glock_drop_th(struct gfs_glock *gl)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      unsigned int ret;

      gfs_assert_warn(sdp, test_bit(GLF_LOCK, &gl->gl_flags));
      gfs_assert_warn(sdp, queue_empty(gl, &gl->gl_holders));
      gfs_assert_warn(sdp, gl->gl_state != LM_ST_UNLOCKED);

      /* Leaving state EX, may need to sync log/data/metadata to disk */
      if (gl->gl_state == LM_ST_EXCLUSIVE) {
            if (glops->go_sync)
                  glops->go_sync(gl, DIO_METADATA | DIO_DATA);
      }

      glock_hold(gl);
      gl->gl_req_bh = drop_bh;

      atomic_inc(&sdp->sd_lm_unlock_calls);

      ret = gfs_lm_unlock(sdp, gl->gl_lock, gl->gl_state);

      if (!ret)
            drop_bh(gl, ret);
      else
            gfs_assert_warn(sdp, ret == LM_OUT_ASYNC);
}

/**
 * do_cancels - cancel requests for locks stuck waiting on an expire flag
 * @gh: the LM_FLAG_PRIORITY holder waiting to acquire the lock
 *
 * Don't cancel GL_NOCANCEL requests.
 */

static void
do_cancels(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;

      spin_lock(&gl->gl_spin);

      while (gl->gl_req_gh != gh &&
             !test_bit(HIF_HOLDER, &gh->gh_iflags) &&
             !list_empty(&gh->gh_list)) {
            if (gl->gl_req_bh &&
                !(gl->gl_req_gh &&
                  (gl->gl_req_gh->gh_flags & GL_NOCANCEL))) {
                  spin_unlock(&gl->gl_spin);
                  gfs_lm_cancel(gl->gl_sbd, gl->gl_lock);
                  set_current_state(TASK_UNINTERRUPTIBLE);
                  schedule_timeout(HZ / 10);
                  spin_lock(&gl->gl_spin);
            } else {
                  spin_unlock(&gl->gl_spin);
                  set_current_state(TASK_UNINTERRUPTIBLE);
                  schedule_timeout(HZ / 10);
                  spin_lock(&gl->gl_spin);
            }
      }

      spin_unlock(&gl->gl_spin);
}

/**
 * glock_wait_internal - wait on a glock acquisition
 * @gh: the glock holder
 *
 * Returns: 0 on success
 */

static int
glock_wait_internal(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      int error = 0;

      if (test_bit(HIF_ABORTED, &gh->gh_iflags))
            return -EIO;

      if (gh->gh_flags & (LM_FLAG_TRY | LM_FLAG_TRY_1CB)) {
            spin_lock(&gl->gl_spin);
            if (gl->gl_req_gh != gh &&
                !test_bit(HIF_HOLDER, &gh->gh_iflags) &&
                !list_empty(&gh->gh_list)) {
                  list_del_init(&gh->gh_list);
                  gh->gh_error = GLR_TRYFAILED;
                  if (test_bit(HIF_RECURSE, &gh->gh_iflags))
                        do_unrecurse(gh);
                  run_queue(gl);
                  spin_unlock(&gl->gl_spin);
                  return GLR_TRYFAILED;
            }
            spin_unlock(&gl->gl_spin);
      }

      if (gh->gh_flags & LM_FLAG_PRIORITY)
            do_cancels(gh);

      wait_for_completion(&gh->gh_wait);

      if (gh->gh_error)
            return gh->gh_error;

      gfs_assert_withdraw(sdp, test_bit(HIF_HOLDER, &gh->gh_iflags));
      gfs_assert_withdraw(sdp, relaxed_state_ok(gl->gl_state,
                                      gh->gh_state,
                                      gh->gh_flags));

      if (test_bit(HIF_FIRST, &gh->gh_iflags)) {
            gfs_assert_warn(sdp, test_bit(GLF_LOCK, &gl->gl_flags));

            if (glops->go_lock) {
                  error = glops->go_lock(gl, gh->gh_flags);
                  if (error) {
                        spin_lock(&gl->gl_spin);
                        list_del_init(&gh->gh_list);
                        gh->gh_error = error;
                        if (test_and_clear_bit(HIF_RECURSE, &gh->gh_iflags))
                              do_unrecurse(gh);
                        spin_unlock(&gl->gl_spin);
                  }
            }

            spin_lock(&gl->gl_spin);
            gl->gl_req_gh = NULL;
            gl->gl_req_bh = NULL;
            clear_bit(GLF_LOCK, &gl->gl_flags);
            if (test_bit(HIF_RECURSE, &gh->gh_iflags))
                  handle_recurse(gh);
            run_queue(gl);
            spin_unlock(&gl->gl_spin);
      }

      return error;
}

/**
 * add_to_queue - Add a holder to the wait-for-promotion queue or holder list
 *       (according to recursion)
 * @gh: the holder structure to add
 *
 * If the hold requestor's process already has a granted lock (on holder list),
 *   and this new request is compatible, go ahead and grant it, adding this
 *   new holder to the glock's holder list. 
 *
 * If the hold requestor's process has earlier requested a lock, and is still
 *   waiting for it to be granted, and this new request is compatible with
 *   the earlier one, they can be handled at the same time when the request
 *   is finally granted.  Mark both (all) with RECURSE flags, and add new
 *   holder to wait-for-promotion queue.
 *
 * If there is no previous holder from this process (on holder list or wait-
 *   for-promotion queue), simply add new holder to wait-for-promotion queue.
 */

static void
add_to_queue(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct list_head *tmp, *head;
      struct gfs_holder *tmp_gh;

      if (gh->gh_owner) {
            /* Search through glock's holders list to see if this process
                 already holds a granted lock. */
            for (head = &gl->gl_holders, tmp = head->next;
                 tmp != head;
                 tmp = tmp->next) {
                  tmp_gh = list_entry(tmp, struct gfs_holder, gh_list);
                  if (tmp_gh->gh_owner == gh->gh_owner) {
                        /* Make sure pre-existing holder is compatible
                           with this new one. */
                        if (gfs_assert_warn(sdp, (gh->gh_flags & LM_FLAG_ANY) ||
                                        !(tmp_gh->gh_flags & LM_FLAG_ANY)) ||
                            gfs_assert_warn(sdp, (tmp_gh->gh_flags & GL_LOCAL_EXCL) ||
                                        !(gh->gh_flags & GL_LOCAL_EXCL)) ||
                            gfs_assert_warn(sdp, relaxed_state_ok(gl->gl_state,
                                                        gh->gh_state,
                                                        gh->gh_flags)))
                              goto fail;

                        /* We're good!  Grant the hold. */
                        list_add_tail(&gh->gh_list, &gl->gl_holders);
                        set_bit(HIF_HOLDER, &gh->gh_iflags);

                        gh->gh_error = 0;
                        complete(&gh->gh_wait);

                        return;
                  }
            }

            /* If not, Search through glock's wait-for-promotion list to
               see if this process already is waiting for a grant. */
            for (head = &gl->gl_waiters3, tmp = head->next;
                 tmp != head;
                 tmp = tmp->next) {
                  tmp_gh = list_entry(tmp, struct gfs_holder, gh_list);
                  if (tmp_gh->gh_owner == gh->gh_owner) {
                        /* Yes, make sure it is compatible with new */
                        if (gfs_assert_warn(sdp, test_bit(HIF_PROMOTE,
                                                  &tmp_gh->gh_iflags)) ||
                            gfs_assert_warn(sdp, (gh->gh_flags & LM_FLAG_ANY) ||
                                        !(tmp_gh->gh_flags & LM_FLAG_ANY)) ||
                            gfs_assert_warn(sdp, (tmp_gh->gh_flags & GL_LOCAL_EXCL) ||
                                        !(gh->gh_flags & GL_LOCAL_EXCL)) ||
                            gfs_assert_warn(sdp, relaxed_state_ok(tmp_gh->gh_state,
                                                        gh->gh_state,
                                                        gh->gh_flags)))
                              goto fail;

                        /* OK, make sure they're marked, so
                         * when one gets granted, the other will too. */
                        set_bit(HIF_RECURSE, &gh->gh_iflags);
                        set_bit(HIF_RECURSE, &tmp_gh->gh_iflags);

                        list_add_tail(&gh->gh_list, &gl->gl_waiters3);

                        return;
                  }
            }
      }

      /* Else, no recursion ...
         If high priority request, add to head of promote queue, else tail */
      if (gh->gh_flags & LM_FLAG_PRIORITY)
            list_add(&gh->gh_list, &gl->gl_waiters3);
      else
            list_add_tail(&gh->gh_list, &gl->gl_waiters3);

      return;

 fail:
      set_bit(HIF_ABORTED, &gh->gh_iflags);
}

/**
 * gfs_glock_nq - enqueue a struct gfs_holder onto a glock (acquire a glock)
 * @gh: the holder structure
 *
 * if (gh->gh_flags & GL_ASYNC), this never returns an error
 *
 * Returns: 0, GLR_TRYFAILED, or errno on failure
 *
 * Rules:
 *   @gh must not be already attached to a glock.
 *   Don't ask for UNLOCKED state (use gfs_glock_dq() for that).
 *   LM_FLAG_ANY (liberal) and GL_EXACT (restrictive) are mutually exclusive.
 */

int
gfs_glock_nq(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      int error = 0;

      atomic_inc(&sdp->sd_glock_nq_calls);

 restart:
      if (unlikely(test_bit(SDF_SHUTDOWN, &sdp->sd_flags)) ||
          gfs_assert_warn(sdp, list_empty(&gh->gh_list)) ||
          gfs_assert_warn(sdp, gh->gh_state != LM_ST_UNLOCKED) ||
          gfs_assert_warn(sdp, (gh->gh_flags & (LM_FLAG_ANY | GL_EXACT)) !=
                      (LM_FLAG_ANY | GL_EXACT))) {
            set_bit(HIF_ABORTED, &gh->gh_iflags);
            return -EIO;
      }

      set_bit(HIF_PROMOTE, &gh->gh_iflags);

      spin_lock(&gl->gl_spin);
      add_to_queue(gh);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);

      if (!(gh->gh_flags & GL_ASYNC)) {
            error = glock_wait_internal(gh);
            if (error == GLR_CANCELED) {
                  set_current_state(TASK_UNINTERRUPTIBLE);
                  schedule_timeout(HZ);
                  goto restart;
            }
      }

      clear_bit(GLF_PREFETCH, &gl->gl_flags);

      return error;
}

/**
 * gfs_glock_poll - poll to see if an async request has been completed
 * @gh: the holder
 *
 * Returns: TRUE if the request is ready to be gfs_glock_wait()ed on
 */

int
gfs_glock_poll(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      int ready = FALSE;

      gfs_assert_warn(gl->gl_sbd, gh->gh_flags & GL_ASYNC);

      spin_lock(&gl->gl_spin);

      if (test_bit(HIF_HOLDER, &gh->gh_iflags))
            ready = TRUE;
      else if (list_empty(&gh->gh_list)) {
            if (gh->gh_error == GLR_CANCELED) {
                  spin_unlock(&gl->gl_spin);
                  set_current_state(TASK_UNINTERRUPTIBLE);
                  schedule_timeout(HZ);
                  if (gfs_glock_nq(gh))
                        return TRUE;
                  return FALSE;
            } else
                  ready = TRUE;
      }

      spin_unlock(&gl->gl_spin);

      return ready;
}

/**
 * gfs_glock_wait - wait for a lock acquisition that ended in a GLR_ASYNC
 * @gh: the holder structure
 *
 * Returns: 0, GLR_TRYFAILED, or errno on failure
 */

int
gfs_glock_wait(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      int error;

        gfs_assert_warn(gl->gl_sbd, gh->gh_flags & GL_ASYNC);

      error = glock_wait_internal(gh);
      if (error == GLR_CANCELED) {
            set_current_state(TASK_UNINTERRUPTIBLE);
            schedule_timeout(HZ);
            gh->gh_flags &= ~GL_ASYNC;
            error = gfs_glock_nq(gh);
      }

      return error;
}

/**
 * gfs_glock_dq - dequeue a struct gfs_holder from a glock (release a glock)
 * @gh: the glock holder
 *
 * This releases a local process' hold on a glock, and services other waiters.
 * If this is the last holder on this node, calls glock operation go_unlock(),
 *    and go_sync() if requested by glock's GL_SYNC flag.
 * If glock's GL_NOCACHE flag is set, requests demotion to unlock the inter-
 *    node lock now, rather than caching the glock for later use.
 * Otherwise, this function does *not* release the glock at inter-node scope.
 *   The glock will stay in glock cache until:
 *   --  This node uses it again (extending residence in glock cache), or
 *   --  Another node asks (via callback) for the lock, or
 *   --  The glock sits unused in glock cache for a while, and the cleanup
 *         daemons (gfs_scand and gfs_glockd) reclaim it.
 */

void
gfs_glock_dq(struct gfs_holder *gh)
{
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;

      atomic_inc(&gl->gl_sbd->sd_glock_dq_calls);

      gfs_assert_withdraw(sdp, !queue_empty(gl, &gh->gh_list));
      gfs_assert_withdraw(sdp, test_bit(HIF_HOLDER, &gh->gh_iflags));

      if (gh->gh_flags & GL_SYNC)
            set_bit(GLF_SYNC, &gl->gl_flags);

      /* Don't cache glock; request demote to unlock at inter-node scope */
      if (gh->gh_flags & GL_NOCACHE)
            handle_callback(gl, LM_ST_UNLOCKED);

      lock_on_glock(gl);

      spin_lock(&gl->gl_spin);
      list_del_init(&gh->gh_list);

      /* If last holder, do appropriate glock operations, set cache timer */
      if (list_empty(&gl->gl_holders)) {
            spin_unlock(&gl->gl_spin);

            if (glops->go_unlock)
                  glops->go_unlock(gl, gh->gh_flags);

            /* Do "early" sync, if requested by holder */
            if (test_bit(GLF_SYNC, &gl->gl_flags)) {
                  if (glops->go_sync)
                        glops->go_sync(gl,
                                     DIO_METADATA |
                                     DIO_DATA |
                                     DIO_INVISIBLE);
            }

            gl->gl_stamp = jiffies;

            spin_lock(&gl->gl_spin);
      }

      clear_bit(GLF_LOCK, &gl->gl_flags);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);
}

/**
 * gfs_glock_prefetch - Try to prefetch a glock
 * @gl: the glock
 * @state: the state to prefetch in 
 * @flags: flags passed to go_xmote_th()
 *
 * Bypass request queues of glock (i.e. no holder involved), and directly call
 *   go_xmote_th() to ask lock module for lock, to put in glock cache for
 *   later use.
 *
 * Will not prefetch the lock (no need to) if a process on this node is already
 *   interested in the lock, or if it's sitting in glock cache in a compatible
 *   state.
 *
 * Rules:
 *   Don't ask for UNLOCKED state (use gfs_glock_dq() for that).
 *   LM_FLAG_ANY (liberal) and GL_EXACT (restrictive) are mutually exclusive.
 */

void
gfs_glock_prefetch(struct gfs_glock *gl, unsigned int state, int flags)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;

      if (gfs_assert_warn(sdp, state != LM_ST_UNLOCKED) ||
          gfs_assert_warn(sdp, (flags & (LM_FLAG_ANY | GL_EXACT)) !=
                      (LM_FLAG_ANY | GL_EXACT)))
            return;

      spin_lock(&gl->gl_spin);

      /* Should we prefetch? */
      if (test_bit(GLF_LOCK, &gl->gl_flags) ||
          !list_empty(&gl->gl_holders) ||
          !list_empty(&gl->gl_waiters1) ||
          !list_empty(&gl->gl_waiters2) ||
          !list_empty(&gl->gl_waiters3) ||
          relaxed_state_ok(gl->gl_state, state, flags)) {
            spin_unlock(&gl->gl_spin);
            return;
      }

      /* Let bottom half know we're prefetching, ask lock module for lock */
      set_bit(GLF_PREFETCH, &gl->gl_flags);

      if (gfs_assert_warn(sdp, !gl->gl_req_gh))
            gl->gl_req_gh = NULL;
      set_bit(GLF_LOCK, &gl->gl_flags);
      spin_unlock(&gl->gl_spin);

      glops->go_xmote_th(gl, state, flags);

      atomic_inc(&gl->gl_sbd->sd_glock_prefetch_calls);
}

/**
 * gfs_glock_force_drop - Force a glock to be uncached
 * @gl: the glock
 *
 */

void
gfs_glock_force_drop(struct gfs_glock *gl)
{
      struct gfs_holder gh;

      gfs_holder_init(gl, LM_ST_UNLOCKED, 0, &gh);
      set_bit(HIF_DEMOTE, &gh.gh_iflags);
      gh.gh_owner = NULL;

      spin_lock(&gl->gl_spin);
      list_add_tail(&gh.gh_list, &gl->gl_waiters2);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);

      wait_for_completion(&gh.gh_wait);
      gfs_holder_uninit(&gh);
}

/**
 * greedy_work -
 * @data:
 *
 */

static void
greedy_work(void *data)
{
      struct greedy *gr = (struct greedy *)data;
      struct gfs_holder *gh = &gr->gr_gh;
      struct gfs_glock *gl = gh->gh_gl;
      struct gfs_glock_operations *glops = gl->gl_ops;

      clear_bit(GLF_SKIP_WAITERS2, &gl->gl_flags);

      if (glops->go_greedy)
            glops->go_greedy(gl);

      spin_lock(&gl->gl_spin);

      if (list_empty(&gl->gl_waiters2)) {
            clear_bit(GLF_GREEDY, &gl->gl_flags);
            spin_unlock(&gl->gl_spin);
            gfs_holder_uninit(gh);
            kfree(gr);
      } else {
            glock_hold(gl);
            list_add_tail(&gh->gh_list, &gl->gl_waiters2);
            run_queue(gl);
            spin_unlock(&gl->gl_spin);
            glock_put(gl);
      }
}

/**
 * gfs_glock_be_greedy -
 * @gl:
 * @time:
 *
 * Returns: 0 if go_greedy will be called, 1 otherwise
 */

int
gfs_glock_be_greedy(struct gfs_glock *gl, unsigned int time)
{
      struct greedy *gr;
      struct gfs_holder *gh;

      if (!time ||
          gl->gl_sbd->sd_args.ar_localcaching ||
          test_and_set_bit(GLF_GREEDY, &gl->gl_flags))
            return 1;

      gr = kmalloc(sizeof(struct greedy), GFP_KERNEL);
      if (!gr) {
            clear_bit(GLF_GREEDY, &gl->gl_flags);
            return 1;
      }
      gh = &gr->gr_gh;

      gfs_holder_init(gl, 0, 0, gh);
      set_bit(HIF_GREEDY, &gh->gh_iflags);
      gh->gh_owner = NULL;
      INIT_WORK(&gr->gr_work, greedy_work, gr);

      set_bit(GLF_SKIP_WAITERS2, &gl->gl_flags);
      schedule_delayed_work(&gr->gr_work, time);

      return 0;
}

/**
 * gfs_glock_nq_init - intialize a holder and enqueue it on a glock
 * @gl: the glock 
 * @state: the state we're requesting
 * @flags: the modifier flags
 * @gh: the holder structure
 *
 * Returns: 0, GLR_*, or errno
 */

int
gfs_glock_nq_init(struct gfs_glock *gl, unsigned int state, int flags,
              struct gfs_holder *gh)
{
      int error;

      gfs_holder_init(gl, state, flags, gh);

      error = gfs_glock_nq(gh);
      if (error)
            gfs_holder_uninit(gh);

      return error;
}

/**
 * gfs_glock_dq_uninit - dequeue a holder from a glock and initialize it
 * @gh: the holder structure
 *
 */

void
gfs_glock_dq_uninit(struct gfs_holder *gh)
{
      gfs_glock_dq(gh);
      gfs_holder_uninit(gh);
}

/**
 * gfs_glock_nq_num - acquire a glock based on lock number
 * @sdp: the filesystem
 * @number: the lock number
 * @glops: the glock operations for the type of glock
 * @state: the state to acquire the glock in
 * @flags: modifier flags for the aquisition
 * @gh: the struct gfs_holder
 *
 * Returns: errno
 */

int
gfs_glock_nq_num(struct gfs_sbd *sdp,
             uint64_t number, struct gfs_glock_operations *glops,
             unsigned int state, int flags, struct gfs_holder *gh)
{
      struct gfs_glock *gl;
      int error;

      error = gfs_glock_get(sdp, number, glops, CREATE, &gl);
      if (!error) {
            error = gfs_glock_nq_init(gl, state, flags, gh);
            glock_put(gl);
      }

      return error;
}

/**
 * glock_compare - Compare two struct gfs_glock structures for sorting
 * @arg_a: the first structure
 * @arg_b: the second structure
 *
 */

static int
glock_compare(const void *arg_a, const void *arg_b)
{
      struct gfs_holder *gh_a = *(struct gfs_holder **)arg_a;
      struct gfs_holder *gh_b = *(struct gfs_holder **)arg_b;
      struct lm_lockname *a = &gh_a->gh_gl->gl_name;
      struct lm_lockname *b = &gh_b->gh_gl->gl_name;
      int ret = 0;

      if (a->ln_number > b->ln_number)
            ret = 1;
      else if (a->ln_number < b->ln_number)
            ret = -1;
      else {
            if (gh_a->gh_state == LM_ST_SHARED &&
                gh_b->gh_state == LM_ST_EXCLUSIVE)
                  ret = 1;
            else if (!(gh_a->gh_flags & GL_LOCAL_EXCL) &&
                   (gh_b->gh_flags & GL_LOCAL_EXCL))
                  ret = 1;
      }

      return ret;
}

/**
 * nq_m_sync - synchonously acquire more than one glock in deadlock free order
 * @num_gh: the number of structures
 * @ghs: an array of struct gfs_holder structures
 *
 * Returns: 0 on success (all glocks acquired), errno on failure (no glocks acquired)
 */

static int
nq_m_sync(unsigned int num_gh, struct gfs_holder *ghs, struct gfs_holder **p)
{
      unsigned int x;
      int error = 0;

      for (x = 0; x < num_gh; x++)
            p[x] = &ghs[x];

      gfs_sort(p, num_gh, sizeof(struct gfs_holder *), glock_compare);

      for (x = 0; x < num_gh; x++) {
            p[x]->gh_flags &= ~(LM_FLAG_TRY | GL_ASYNC);

            error = gfs_glock_nq(p[x]);
            if (error) {
                  while (x--)
                        gfs_glock_dq(p[x]);
                  break;
            }
      }

      return error;
}

/**
 * gfs_glock_nq_m - acquire multiple glocks
 * @num_gh: the number of structures
 * @ghs: an array of struct gfs_holder structures
 *
 * Figure out how big an impact this function has.  Either:
 * 1) Replace this code with code that calls gfs_glock_prefetch()
 * 2) Forget async stuff and just call nq_m_sync()
 * 3) Leave it like it is
 *
 * Returns: 0 on success (all glocks acquired), errno on failure (no glocks acquired)
 */

int
gfs_glock_nq_m(unsigned int num_gh, struct gfs_holder *ghs)
{
      int *e;
      unsigned int x;
      int borked = FALSE, serious = 0;
      int error = 0;

      if (!num_gh)
            return 0;

      /* For just one gh, do request synchronously */
      if (num_gh == 1) {
            ghs->gh_flags &= ~(LM_FLAG_TRY | GL_ASYNC);
            return gfs_glock_nq(ghs);
      }

      /* using sizeof(struct gfs_holder *) instead of sizeof(int), because
       * we're also using this memory for nq_m_sync and ints should never be
       * larger than pointers.... I hope
       */
      e = kmalloc(num_gh * sizeof(struct gfs_holder *), GFP_KERNEL);
      if (!e)
            return -ENOMEM;

      /* Send off asynchronous requests */
      for (x = 0; x < num_gh; x++) {
            ghs[x].gh_flags |= LM_FLAG_TRY | GL_ASYNC;
            error = gfs_glock_nq(&ghs[x]);
            if (error) {
                  borked = TRUE;
                  serious = error;
                  num_gh = x;
                  break;
            }
      }

      /* Wait for all to complete */
      for (x = 0; x < num_gh; x++) {
            error = e[x] = glock_wait_internal(&ghs[x]);
            if (error) {
                  borked = TRUE;
                  if (error != GLR_TRYFAILED && error != GLR_CANCELED)
                        serious = error;
            }
      }

      /* If all good, done! */
      if (!borked) {
            kfree(e);
            return 0;
      }

      for (x = 0; x < num_gh; x++)
            if (!e[x])
                  gfs_glock_dq(&ghs[x]);

      if (serious)
            error = serious;
      else {
            for (x = 0; x < num_gh; x++)
                  gfs_holder_reinit(ghs[x].gh_state, ghs[x].gh_flags,
                                &ghs[x]);
            error = nq_m_sync(num_gh, ghs, (struct gfs_holder **)e);
      }

      kfree(e);
      return error;
}

/**
 * gfs_glock_dq_m - release multiple glocks
 * @num_gh: the number of structures
 * @ghs: an array of struct gfs_holder structures
 *
 */

void
gfs_glock_dq_m(unsigned int num_gh, struct gfs_holder *ghs)
{
      unsigned int x;

      for (x = 0; x < num_gh; x++)
            gfs_glock_dq(&ghs[x]);
}

/**
 * gfs_glock_prefetch_num - prefetch a glock based on lock number
 * @sdp: the filesystem
 * @number: the lock number
 * @glops: the glock operations for the type of glock
 * @state: the state to acquire the glock in
 * @flags: modifier flags for the aquisition
 *
 * Returns: errno
 */

void
gfs_glock_prefetch_num(struct gfs_sbd *sdp,
                   uint64_t number, struct gfs_glock_operations *glops,
                   unsigned int state, int flags)
{
      struct gfs_glock *gl;
      int error;

      if (atomic_read(&sdp->sd_reclaim_count) < gfs_tune_get(sdp, gt_reclaim_limit)) {
            error = gfs_glock_get(sdp, number, glops, CREATE, &gl);
            if (!error) {
                  gfs_glock_prefetch(gl, state, flags);
                  glock_put(gl);
            }
      }
}

/**
 * gfs_lvb_hold - attach a LVB from a glock
 * @gl: The glock in question
 *
 */

int
gfs_lvb_hold(struct gfs_glock *gl)
{
      int error;

      lock_on_glock(gl);

      if (!atomic_read(&gl->gl_lvb_count)) {
            gfs_assert_warn(gl->gl_sbd, !gl->gl_lvb);
            error = gfs_lm_hold_lvb(gl->gl_sbd, gl->gl_lock, &gl->gl_lvb);
            if (error) {
                  unlock_on_glock(gl);
                  return error;
            }
            glock_hold(gl);
      }
      atomic_inc(&gl->gl_lvb_count);

      unlock_on_glock(gl);

      return 0;
}

/**
 * gfs_lvb_unhold - detach a LVB from a glock
 * @gl: The glock in question
 * 
 */

void
gfs_lvb_unhold(struct gfs_glock *gl)
{
      glock_hold(gl);

      lock_on_glock(gl);

      if (!gfs_assert_warn(gl->gl_sbd, atomic_read(&gl->gl_lvb_count) > 0) &&
          atomic_dec_and_test(&gl->gl_lvb_count)) {
            gfs_assert_warn(gl->gl_sbd, gl->gl_lvb);
            gfs_lm_unhold_lvb(gl->gl_sbd, gl->gl_lock, gl->gl_lvb);
            gl->gl_lvb = NULL;
            glock_put(gl);
      }

      unlock_on_glock(gl);

      glock_put(gl);
}

/**
 * gfs_lvb_sync - sync a LVB
 * @gl: The glock in question
 * 
 */

void
gfs_lvb_sync(struct gfs_glock *gl)
{
      if (gfs_assert_warn(gl->gl_sbd, atomic_read(&gl->gl_lvb_count)))
            return;

      lock_on_glock(gl);

      if (!gfs_assert_warn(gl->gl_sbd, gfs_glock_is_held_excl(gl)))
            gfs_lm_sync_lvb(gl->gl_sbd, gl->gl_lock, gl->gl_lvb);

      unlock_on_glock(gl);
}

/**
 * blocking_cb -
 * @sdp:
 * @name:
 * @state:
 *
 */

void
blocking_cb(struct gfs_sbd *sdp, struct lm_lockname *name, unsigned int state)
{
      struct gfs_glock *gl;

      gl = gfs_glock_find(sdp, name);
      if (!gl)
            return;

      if (gl->gl_ops->go_callback)
            gl->gl_ops->go_callback(gl, state);
      handle_callback(gl, state);

      spin_lock(&gl->gl_spin);
      run_queue(gl);
      spin_unlock(&gl->gl_spin);

      glock_put(gl);
}

/**
 * gfs_glock_cb - Callback used by locking module
 * @fsdata: Pointer to the superblock
 * @type: Type of callback
 * @data: Type dependent data pointer
 *
 * Called by the locking module when it wants to tell us something.
 * Either we need to drop a lock, one of our ASYNC requests completed, or
 *   another client expired (crashed/died) and we need to recover its journal.
 * If another node needs a lock held by this node, we queue a request to demote
 *   our lock to a state compatible with that needed by the other node.  
 *   For example, if the other node needs EXCLUSIVE, we request UNLOCKED.
 *   SHARED and DEFERRED modes can be shared with other nodes, so we request
 *   accordingly.
 * Once all incompatible holders on this node are done with the lock, the
 *   queued request will cause run_queue() to call the lock module to demote
 *   our lock to a compatible state, allowing the other node to grab the lock.
 */

void
gfs_glock_cb(lm_fsdata_t *fsdata, unsigned int type, void *data)
{
      struct gfs_sbd *sdp = (struct gfs_sbd *)fsdata;

      atomic_inc(&sdp->sd_lm_callbacks);

      switch (type) {
      case LM_CB_NEED_E:
            blocking_cb(sdp, (struct lm_lockname *)data, LM_ST_UNLOCKED);
            return;

      case LM_CB_NEED_D:
            blocking_cb(sdp, (struct lm_lockname *)data, LM_ST_DEFERRED);
            return;

      case LM_CB_NEED_S:
            blocking_cb(sdp, (struct lm_lockname *)data, LM_ST_SHARED);
            return;

      case LM_CB_ASYNC: {
            struct lm_async_cb *async = (struct lm_async_cb *)data;
            struct gfs_glock *gl;

            gl = gfs_glock_find(sdp, &async->lc_name);
            if (gfs_assert_warn(sdp, gl))
                  return;
            if (!gfs_assert_warn(sdp, gl->gl_req_bh))
                  gl->gl_req_bh(gl, async->lc_ret);
            glock_put(gl);

            return;
      }

      case LM_CB_NEED_RECOVERY:
            gfs_add_dirty_j(sdp, *(unsigned int *)data);
            if (test_bit(SDF_RECOVERD_RUN, &sdp->sd_flags))
                  wake_up_process(sdp->sd_recoverd_process);
            return;

      case LM_CB_DROPLOCKS:
            gfs_gl_hash_clear(sdp, FALSE);
            gfs_quota_scan(sdp);
            return;

      default:
            gfs_assert_warn(sdp, FALSE);
            return;
      }
}

/**
 * gfs_try_toss_inode - try to remove a particular GFS inode struct from cache
 * sdp: the filesystem
 * inum: the inode number
 *
 * Look for the glock protecting the inode of interest.
 * If no process is manipulating or holding the glock, see if the glock
 *   has a gfs_inode attached.
 * If gfs_inode has no references, unhold its iopen glock, release any
 *   indirect addressing buffers, and destroy the gfs_inode.
 */

void
gfs_try_toss_inode(struct gfs_sbd *sdp, struct gfs_inum *inum)
{
      struct gfs_glock *gl;
      struct gfs_inode *ip;
      int error;

      error = gfs_glock_get(sdp,
                        inum->no_formal_ino, &gfs_inode_glops,
                        NO_CREATE, &gl);
      if (error || !gl)
            return;

      if (!trylock_on_glock(gl))
            goto out;

      if (!queue_empty(gl, &gl->gl_holders))
            goto out_unlock;

      ip = get_gl2ip(gl);
      if (!ip)
            goto out_unlock;

      if (atomic_read(&ip->i_count))
            goto out_unlock;

      gfs_inode_destroy(ip);

 out_unlock:
      unlock_on_glock(gl);

 out:
      glock_put(gl);
}

/**
 * gfs_iopen_go_callback - Try to kick the inode/vnode associated with an iopen glock from memory
 * @io_gl: the iopen glock
 * @state: the state into which the glock should be put
 *
 */

void
gfs_iopen_go_callback(struct gfs_glock *io_gl, unsigned int state)
{
      struct gfs_glock *i_gl;
      struct gfs_inode *ip;

      if (state != LM_ST_UNLOCKED)
            return;

      spin_lock(&io_gl->gl_spin);
      i_gl = get_gl2gl(io_gl);
      if (i_gl) {
            glock_hold(i_gl);
            spin_unlock(&io_gl->gl_spin);
      } else {
            spin_unlock(&io_gl->gl_spin);
            return;
      }

      if (trylock_on_glock(i_gl)) {
            if (queue_empty(i_gl, &i_gl->gl_holders)) {
                  ip = get_gl2ip(i_gl);
                  if (ip) {
                        gfs_try_toss_vnode(ip);
                        unlock_on_glock(i_gl);
                        gfs_glock_schedule_for_reclaim(i_gl);
                        goto out;
                  }
            }
            unlock_on_glock(i_gl);
      }

 out:
      glock_put(i_gl);
}

/**
 * demote_ok - Check to see if it's ok to unlock a glock (to remove it
 *       from glock cache)
 * @gl: the glock
 *
 * Called when trying to reclaim glocks, once it's determined that the glock
 *   has no holders on this node.
 *
 * Returns: TRUE if it's ok
 *
 * It's not okay if:
 * --  glock is STICKY
 * --  PREFETCHed glock has not been given enough chance to be used
 * --  glock-type-specific test says "no"
 */

static int
demote_ok(struct gfs_glock *gl)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_glock_operations *glops = gl->gl_ops;
      int demote = TRUE;

      if (test_bit(GLF_STICKY, &gl->gl_flags))
            demote = FALSE;
      else if (test_bit(GLF_PREFETCH, &gl->gl_flags))
            demote = time_after_eq(jiffies,
                               gl->gl_stamp +
                               gfs_tune_get(sdp, gt_prefetch_secs) * HZ);
      else if (glops->go_demote_ok)
            demote = glops->go_demote_ok(gl);

      return demote;
}

/**
 * gfs_glock_schedule_for_reclaim - Add a glock to the reclaim list
 * @gl: the glock
 *
 */

void
gfs_glock_schedule_for_reclaim(struct gfs_glock *gl)
{
      struct gfs_sbd *sdp = gl->gl_sbd;

      spin_lock(&sdp->sd_reclaim_lock);
      if (list_empty(&gl->gl_reclaim)) {
            glock_hold(gl);
            list_add(&gl->gl_reclaim, &sdp->sd_reclaim_list);
            atomic_inc(&sdp->sd_reclaim_count);
      }
      spin_unlock(&sdp->sd_reclaim_lock);

      wake_up(&sdp->sd_reclaim_wchan);
}

/**
 * gfs_reclaim_glock - process the next glock on the filesystem's reclaim list
 * @sdp: the filesystem
 *
 * Called from gfs_glockd() glock reclaim daemon, or when promoting a
 *   (different) glock and we notice that there are a lot of glocks in the
 *   reclaim list.
 *
 * Remove glock from filesystem's reclaim list, update reclaim statistics.
 * If no holders (might have gotten added since glock was placed on reclaim
 *   list):
 *   --  Destroy any now-unused inode protected by glock
 *         (and release hold on iopen glock).
 *   --  Ask for demote to UNLOCKED to enable removal of glock from glock cache.
 *
 * If no further interest in glock struct, remove it from glock cache, and
 *   free it from memory.  (During normal operation, this is the only place
 *   that this is done).
 *
 * Glock-type-specific considerations for permission to demote are handled
 *   in demote_ok().  This includes how long to retain a glock in cache after it
 *   is no longer held by any process.
 */

void
gfs_reclaim_glock(struct gfs_sbd *sdp)
{
      struct gfs_glock *gl;
      struct gfs_gl_hash_bucket *bucket;

      spin_lock(&sdp->sd_reclaim_lock);

      /* Nothing to reclaim?  Done! */
      if (list_empty(&sdp->sd_reclaim_list)) {
            spin_unlock(&sdp->sd_reclaim_lock);
            return;
      }

      /* Remove next victim from reclaim list */
      gl = list_entry(sdp->sd_reclaim_list.next,
                  struct gfs_glock, gl_reclaim);
      list_del_init(&gl->gl_reclaim);

      spin_unlock(&sdp->sd_reclaim_lock);

      atomic_dec(&sdp->sd_reclaim_count);
      atomic_inc(&sdp->sd_reclaimed);

      if (trylock_on_glock(gl)) {
            if (queue_empty(gl, &gl->gl_holders)) {
                  /* Inode glock-type-specific; free unused gfs inode,
                     and release hold on iopen glock */
                  if (gl->gl_ops == &gfs_inode_glops) {
                        struct gfs_inode *ip = get_gl2ip(gl);
                        if (ip && !atomic_read(&ip->i_count))
                              gfs_inode_destroy(ip);
                  }
                  /* Generic (including inodes); try to unlock glock */
                  if (gl->gl_state != LM_ST_UNLOCKED &&
                      demote_ok(gl))
                        handle_callback(gl, LM_ST_UNLOCKED);
            }
            unlock_on_glock(gl);
      }

      bucket = gl->gl_bucket;

      /* If glock struct's only remaining reference is from being put on
         the reclaim list, remove glock from hash table (sd_gl_hash),
         and free the glock's memory */
      write_lock(&bucket->hb_lock);
      if (atomic_read(&gl->gl_count) == 1) {
            list_del_init(&gl->gl_list);
            write_unlock(&bucket->hb_lock);
            glock_free(gl);
      } else {
            write_unlock(&bucket->hb_lock);
            glock_put(gl);  /* see gfs_glock_schedule_for_reclaim() */
      }
}

/**
 * examine_bucket - Call a function for glock in a hash bucket
 * @examiner: the function 
 * @sdp: the filesystem
 * @bucket: the bucket
 *
 * Returns: TRUE if the bucket has entries
 */

static int
examine_bucket(glock_examiner examiner,
             struct gfs_sbd *sdp, struct gfs_gl_hash_bucket *bucket)
{
      struct glock_plug plug;
      struct list_head *tmp;
      struct gfs_glock *gl;
      int entries;

      /* Add "plug" to end of bucket list, work back up list from there */
      memset(&plug.gl_flags, 0, sizeof(unsigned long));
      set_bit(GLF_PLUG, &plug.gl_flags);

      write_lock(&bucket->hb_lock);
      list_add(&plug.gl_list, &bucket->hb_list);
      write_unlock(&bucket->hb_lock);

      /* Look at each bucket entry */
      for (;;) {
            write_lock(&bucket->hb_lock);

            /* Work back up list from plug */
            for (;;) {
                  tmp = plug.gl_list.next;

                  /* Top of list; we're done */
                  if (tmp == &bucket->hb_list) {
                        list_del(&plug.gl_list);
                        entries = !list_empty(&bucket->hb_list);
                        write_unlock(&bucket->hb_lock);
                        return entries;
                  }
                  gl = list_entry(tmp, struct gfs_glock, gl_list);

                  /* Move plug up list */
                  list_move(&plug.gl_list, &gl->gl_list);

                  if (test_bit(GLF_PLUG, &gl->gl_flags))
                        continue;

                  /* glock_hold; examiner must glock_put() */
                  atomic_inc(&gl->gl_count);

                  break;
            }

            write_unlock(&bucket->hb_lock);

            examiner(gl);
      }
}

/**
 * scan_glock - look at a glock and see if we can reclaim it
 * @gl: the glock to look at
 *
 * Called via examine_bucket() when trying to release glocks from glock cache,
 *   during normal operation (i.e. not unmount time).
 * 
 * Place glock on filesystem's reclaim list if, on this node:
 * --  No process is manipulating glock struct, and
 * --  No current holders, and either:
 *     --  GFS incore inode, protected by glock, is no longer in use, or
 *     --  Glock-type-specific demote_ok glops gives permission
 */

static void
scan_glock(struct gfs_glock *gl)
{
      if (trylock_on_glock(gl)) {
            if (queue_empty(gl, &gl->gl_holders)) {
                  /* Inode glock-type-specific; reclaim glock if gfs inode
                     no longer in use. */
                  if (gl->gl_ops == &gfs_inode_glops) {
                        struct gfs_inode *ip = get_gl2ip(gl);
                        if (ip && !atomic_read(&ip->i_count)) {
                              unlock_on_glock(gl);
                              gfs_glock_schedule_for_reclaim(gl);
                              goto out;
                        }
                  }
                  /* Generic (including inodes not scheduled above) */
                  if (gl->gl_state != LM_ST_UNLOCKED &&
                      demote_ok(gl)) {
                        unlock_on_glock(gl);
                        gfs_glock_schedule_for_reclaim(gl);
                        goto out;
                  }
            }

            unlock_on_glock(gl);
      }

 out:
      glock_put(gl);  /* see examine_bucket() */
}

/**
 * gfs_scand_internal - Look for glocks and inodes to toss from memory
 * @sdp: the filesystem
 *
 * Invokes scan_glock() for each glock in each cache bucket.
 *
 * Steps of reclaiming a glock:
 * --  scan_glock() places eligible glocks on filesystem's reclaim list.
 * --  gfs_reclaim_glock() processes list members, attaches demotion requests
 *     to wait queues of glocks still locked at inter-node scope.
 * --  Demote to UNLOCKED state (if not already unlocked).
 * --  gfs_reclaim_lock() cleans up glock structure.
 */

void
gfs_scand_internal(struct gfs_sbd *sdp)
{
      unsigned int x;

      for (x = 0; x < GFS_GL_HASH_SIZE; x++) {
            examine_bucket(scan_glock, sdp, &sdp->sd_gl_hash[x]);
            cond_resched();
      }
}

/**
 * clear_glock - look at a glock and see if we can free it from glock cache
 * @gl: the glock to look at
 *
 * Called via examine_bucket() when unmounting the filesystem, or
 *   when inter-node lock manager requests DROPLOCKS because it is running
 *   out of capacity.
 *
 * Similar to gfs_reclaim_glock(), except does *not*:
 *   --  Consult demote_ok() for permission
 *   --  Increment sdp->sd_reclaimed statistic
 *
 */

static void
clear_glock(struct gfs_glock *gl)
{
      struct gfs_sbd *sdp = gl->gl_sbd;
      struct gfs_gl_hash_bucket *bucket = gl->gl_bucket;

      spin_lock(&sdp->sd_reclaim_lock);
      if (!list_empty(&gl->gl_reclaim)) {
            list_del_init(&gl->gl_reclaim);
            atomic_dec(&sdp->sd_reclaim_count);
            glock_put(gl);  /* see gfs_glock_schedule_for_reclaim() */
      }
      spin_unlock(&sdp->sd_reclaim_lock);

      if (trylock_on_glock(gl)) {
            if (queue_empty(gl, &gl->gl_holders)) {
                  /* Inode glock-type-specific; free unused gfs inode,
                     and release hold on iopen glock */
                  if (gl->gl_ops == &gfs_inode_glops) {
                        struct gfs_inode *ip = get_gl2ip(gl);
                        if (ip && !atomic_read(&ip->i_count))
                              gfs_inode_destroy(ip);
                  }
                  /* Generic (including inodes); unlock glock */
                  if (gl->gl_state != LM_ST_UNLOCKED)
                        handle_callback(gl, LM_ST_UNLOCKED);
            }

            unlock_on_glock(gl);
      }

      /* If glock struct's only remaining reference is from examine_bucket(),
         remove glock from hash table (sd_gl_hash), and free glock's memory */
      write_lock(&bucket->hb_lock);
      if (atomic_read(&gl->gl_count) == 1) {
            list_del_init(&gl->gl_list);
            write_unlock(&bucket->hb_lock);
            glock_free(gl);
      } else {
            write_unlock(&bucket->hb_lock);
            glock_put(gl);   /* see examine_bucket() */
      }
}

/**
 * gfs_gl_hash_clear - Empty out the glock hash table
 * @sdp: the filesystem
 * @wait: wait until it's all gone
 *
 * Called when unmounting the filesystem, or when inter-node lock manager
 *   requests DROPLOCKS because it is running out of capacity.
 */

void
gfs_gl_hash_clear(struct gfs_sbd *sdp, int wait)
{
      unsigned long t;
      unsigned int x;
      int cont;

      t = jiffies;

      for (;;) {
            cont = FALSE;

            for (x = 0; x < GFS_GL_HASH_SIZE; x++)
                  if (examine_bucket(clear_glock, sdp, &sdp->sd_gl_hash[x]))
                        cont = TRUE;

            if (!wait || !cont)
                  break;

            if (time_after_eq(jiffies, t + gfs_tune_get(sdp, gt_stall_secs) * HZ)) {
                  printk("GFS: fsid=%s: Unmount seems to be stalled. Dumping lock state...\n",
                         sdp->sd_fsname);
                  gfs_dump_lockstate(sdp, NULL);
                  t = jiffies;
            }

            invalidate_inodes(sdp->sd_vfs);
            yield();
      }
}

/*
 *  Diagnostic routines to help debug distributed deadlock
 */

/**
 * dump_holder - print information about a glock holder
 * @str: a string naming the type of holder
 * @gh: the glock holder
 * @buf: the buffer
 * @size: the size of the buffer
 * @count: where we are in the buffer
 *
 * Returns: 0 on success, -ENOBUFS when we run out of space
 */

static int
dump_holder(char *str, struct gfs_holder *gh,
          char *buf, unsigned int size, unsigned int *count)
{
      unsigned int x;
      int error = -ENOBUFS;

      gfs_printf("  %s\n", str);
      gfs_printf("    owner = %ld\n",
               (gh->gh_owner) ? (long)gh->gh_owner->pid : -1);
      gfs_printf("    gh_state = %u\n", gh->gh_state);
      gfs_printf("    gh_flags =");
      for (x = 0; x < 32; x++)
            if (gh->gh_flags & (1 << x))
                  gfs_printf(" %u", x);
      gfs_printf(" \n");
      gfs_printf("    error = %d\n", gh->gh_error);
      gfs_printf("    gh_iflags =");
      for (x = 0; x < 32; x++)
            if (test_bit(x, &gh->gh_iflags))
                  gfs_printf(" %u", x);
      gfs_printf(" \n");

      error = 0;

 out:
      return error;
}

/**
 * dump_inode - print information about an inode
 * @ip: the inode
 * @buf: the buffer
 * @size: the size of the buffer
 * @count: where we are in the buffer
 *
 * Returns: 0 on success, -ENOBUFS when we run out of space
 */

static int
dump_inode(struct gfs_inode *ip,
         char *buf, unsigned int size, unsigned int *count)
{
      unsigned int x;
      int error = -ENOBUFS;

      gfs_printf("  Inode:\n");
      gfs_printf("    num = %" PRIu64 "/%" PRIu64 "\n",
                ip->i_num.no_formal_ino, ip->i_num.no_addr);
      gfs_printf("    type = %u\n", ip->i_di.di_type);
      gfs_printf("    i_count = %d\n", atomic_read(&ip->i_count));
      gfs_printf("    i_flags =");
      for (x = 0; x < 32; x++)
            if (test_bit(x, &ip->i_flags))
                  gfs_printf(" %u", x);
      gfs_printf(" \n");
      gfs_printf("    vnode = %s\n", (ip->i_vnode) ? "yes" : "no");

      error = 0;

 out:
      return error;
}

/**
 * dump_glock - print information about a glock
 * @gl: the glock
 * @buf: the buffer
 * @size: the size of the buffer
 * @count: where we are in the buffer
 *
 * Returns: 0 on success, -ENOBUFS when we run out of space
 */

static int
dump_glock(struct gfs_glock *gl,
         char *buf, unsigned int size, unsigned int *count)
{
      struct list_head *head, *tmp;
      struct gfs_holder *gh;
      unsigned int x;
      int error = -ENOBUFS;

      spin_lock(&gl->gl_spin);

      gfs_printf("Glock (%u, %" PRIu64 ")\n",
                gl->gl_name.ln_type,
                gl->gl_name.ln_number);
      gfs_printf("  gl_flags =");
      for (x = 0; x < 32; x++)
            if (test_bit(x, &gl->gl_flags))
                  gfs_printf(" %u", x);
      gfs_printf(" \n");
      gfs_printf("  gl_count = %d\n", atomic_read(&gl->gl_count));
      gfs_printf("  gl_state = %u\n", gl->gl_state);
      gfs_printf("  req_gh = %s\n", (gl->gl_req_gh) ? "yes" : "no");
      gfs_printf("  req_bh = %s\n", (gl->gl_req_bh) ? "yes" : "no");
      gfs_printf("  lvb_count = %d\n", atomic_read(&gl->gl_lvb_count));
      gfs_printf("  object = %s\n", (gl->gl_object) ? "yes" : "no");
      gfs_printf("  new_le = %s\n", (gl->gl_new_le.le_trans) ? "yes" : "no");
      gfs_printf("  incore_le = %s\n", (gl->gl_incore_le.le_trans) ? "yes" : "no");
      gfs_printf("  reclaim = %s\n",
                (list_empty(&gl->gl_reclaim)) ? "no" : "yes");
      if (gl->gl_aspace)
            gfs_printf("  aspace = %lu\n",
                      gl->gl_aspace->i_mapping->nrpages);
      else
            gfs_printf("  aspace = no\n");
      gfs_printf("  ail_bufs = %s\n",
               (list_empty(&gl->gl_ail_bufs)) ? "no" : "yes");
      if (gl->gl_req_gh) {
            error = dump_holder("Request", gl->gl_req_gh, buf, size, count);
            if (error)
                  goto out;
      }
      for (head = &gl->gl_holders, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gh = list_entry(tmp, struct gfs_holder, gh_list);
            error = dump_holder("Holder", gh, buf, size, count);
            if (error)
                  goto out;
      }
      for (head = &gl->gl_waiters1, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gh = list_entry(tmp, struct gfs_holder, gh_list);
            error = dump_holder("Waiter1", gh, buf, size, count);
            if (error)
                  goto out;
      }
      for (head = &gl->gl_waiters2, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gh = list_entry(tmp, struct gfs_holder, gh_list);
            error = dump_holder("Waiter2", gh, buf, size, count);
            if (error)
                  goto out;
      }
      for (head = &gl->gl_waiters3, tmp = head->next;
           tmp != head;
           tmp = tmp->next) {
            gh = list_entry(tmp, struct gfs_holder, gh_list);
            error = dump_holder("Waiter3", gh, buf, size, count);
            if (error)
                  goto out;
      }
      if (gl->gl_ops == &gfs_inode_glops && get_gl2ip(gl)) {
            if (!test_bit(GLF_LOCK, &gl->gl_flags) &&
                list_empty(&gl->gl_holders)) {
                  error = dump_inode(get_gl2ip(gl), buf, size, count);
                  if (error)
                        goto out;
            } else {
                  error = -ENOBUFS;
                  gfs_printf("  Inode: busy\n");
            }
      }

      error = 0;

 out:
      spin_unlock(&gl->gl_spin);

      return error;
}

/**
 * gfs_dump_lockstate - print out the current lockstate
 * @sdp: the filesystem
 * @ub: the buffer to copy the information into
 *
 * If @ub is NULL, dump the lockstate to the console.
 *
 */

int
gfs_dump_lockstate(struct gfs_sbd *sdp, struct gfs_user_buffer *ub)
{
      struct gfs_gl_hash_bucket *bucket;
      struct list_head *tmp, *head;
      struct gfs_glock *gl;
      char *buf = NULL;
      unsigned int size = gfs_tune_get(sdp, gt_lockdump_size);
      unsigned int x, count;
      int error = 0;

      if (ub) {
            buf = kmalloc(size, GFP_KERNEL);
            if (!buf)
                  return -ENOMEM;
      }

      for (x = 0; x < GFS_GL_HASH_SIZE; x++) {
            bucket = &sdp->sd_gl_hash[x];
            count = 0;

            read_lock(&bucket->hb_lock);

            for (head = &bucket->hb_list, tmp = head->next;
                 tmp != head;
                 tmp = tmp->next) {
                  gl = list_entry(tmp, struct gfs_glock, gl_list);

                  if (test_bit(GLF_PLUG, &gl->gl_flags))
                        continue;

                  error = dump_glock(gl, buf, size, &count);
                  if (error)
                        break;
            }

            read_unlock(&bucket->hb_lock);

            if (error)
                  break;

            if (ub) {
                  if (ub->ub_count + count > ub->ub_size) {
                        error = -ENOMEM;
                        break;
                  }
                  if (copy_to_user(ub->ub_data + ub->ub_count, buf, count)) {
                        error = -EFAULT;
                        break;
                  }
                  ub->ub_count += count;
            }
      }

      if (ub)
            kfree(buf);

      return error;
}

Generated by  Doxygen 1.6.0   Back to index