ldlm_resource.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,062 行 · 第 1/3 页

/* Args: locked namespace
 * Returns: newly-allocated, referenced, unlocked resource */
static struct ldlm_resource *
ldlm_resource_add(struct ldlm_namespace *ns, struct ldlm_resource *parent,
                  struct ldlm_res_id name, __u32 hash, ldlm_type_t type)
{
        struct list_head *bucket;
        struct ldlm_resource *res, *old_res;
        ENTRY;

        LASSERTF(type >= LDLM_MIN_TYPE && type < LDLM_MAX_TYPE,
                 "type: %d\n", type);

        res = ldlm_resource_new();
        if (!res)
                RETURN(NULL);

        res->lr_name = name;
        res->lr_namespace = ns;
        res->lr_type = type;
        res->lr_most_restr = LCK_NL;

        spin_lock(&ns->ns_hash_lock);
        old_res = ldlm_resource_find(ns, name, hash);
        if (old_res) {
                /* someone won the race and added the resource before */
                ldlm_resource_getref(old_res);
                spin_unlock(&ns->ns_hash_lock);
                OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
                /* synchronize WRT resource creation */
                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                        down(&old_res->lr_lvb_sem);
                        up(&old_res->lr_lvb_sem);
                }
                RETURN(old_res);
        }

        /* we won! let's add the resource */
        bucket = ns->ns_hash + hash;
        list_add(&res->lr_hash, bucket);
        ns->ns_resources++;
        ldlm_namespace_get_nolock(ns);

        if (parent == NULL) {
                list_add(&res->lr_childof, &ns->ns_root_list);
        } else {
                res->lr_parent = parent;
                list_add(&res->lr_childof, &parent->lr_children);
        }
        spin_unlock(&ns->ns_hash_lock);

        if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                int rc;

                OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CREATE_RESOURCE, 2);
                rc = ns->ns_lvbo->lvbo_init(res);
                if (rc)
                        CERROR("lvbo_init failed for resource "
                               LPU64": rc %d\n", name.name[0], rc);
                /* we create resource with locked lr_lvb_sem */
                up(&res->lr_lvb_sem);
        }

        RETURN(res);
}

/* Args: unlocked namespace
 * Locks: takes and releases ns->ns_lock and res->lr_lock
 * Returns: referenced, unlocked ldlm_resource or NULL */
struct ldlm_resource *
ldlm_resource_get(struct ldlm_namespace *ns, struct ldlm_resource *parent,
                  struct ldlm_res_id name, ldlm_type_t type, int create)
{
        __u32 hash = ldlm_hash_fn(parent, name);
        struct ldlm_resource *res = NULL;
        ENTRY;

        LASSERT(ns != NULL);
        LASSERT(ns->ns_hash != NULL);
        LASSERT(name.name[0] != 0);

        spin_lock(&ns->ns_hash_lock);
        res = ldlm_resource_find(ns, name, hash);
        if (res) {
                ldlm_resource_getref(res);
                spin_unlock(&ns->ns_hash_lock);
                /* synchronize WRT resource creation */
                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                        down(&res->lr_lvb_sem);
                        up(&res->lr_lvb_sem);
                }
                RETURN(res);
        }
        spin_unlock(&ns->ns_hash_lock);

        if (create == 0)
                RETURN(NULL);

        res = ldlm_resource_add(ns, parent, name, hash, type);
        RETURN(res);
}

struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
{
        LASSERT(res != NULL);
        LASSERT(res != LP_POISON);
        atomic_inc(&res->lr_refcount);
        CDEBUG(D_INFO, "getref res: %p count: %d\n", res,
               atomic_read(&res->lr_refcount));
        return res;
}

void __ldlm_resource_putref_final(struct ldlm_resource *res)
{
        struct ldlm_namespace *ns = res->lr_namespace;

        LASSERT_SPIN_LOCKED(&ns->ns_hash_lock);

        if (!list_empty(&res->lr_granted)) {
                ldlm_resource_dump(D_ERROR, res);
                LBUG();
        }

        if (!list_empty(&res->lr_converting)) {
                ldlm_resource_dump(D_ERROR, res);
                LBUG();
        }

        if (!list_empty(&res->lr_waiting)) {
                ldlm_resource_dump(D_ERROR, res);
                LBUG();
        }

        if (!list_empty(&res->lr_children)) {
                ldlm_resource_dump(D_ERROR, res);
                LBUG();
        }

        /* Pass 0 as second argument to not wake up ->ns_waitq yet, will do it
         * later. */
        ldlm_namespace_put_nolock(ns, 0);
        list_del_init(&res->lr_hash);
        list_del_init(&res->lr_childof);

        ns->ns_resources--;
        if (ns->ns_resources == 0)
                wake_up(&ns->ns_waitq);
}

/* Returns 1 if the resource was freed, 0 if it remains. */
int ldlm_resource_putref(struct ldlm_resource *res)
{
        struct ldlm_namespace *ns = res->lr_namespace;
        int rc = 0;
        ENTRY;

        CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
               atomic_read(&res->lr_refcount) - 1);
        LASSERTF(atomic_read(&res->lr_refcount) > 0, "%d",
                 atomic_read(&res->lr_refcount));
        LASSERTF(atomic_read(&res->lr_refcount) < LI_POISON, "%d",
                 atomic_read(&res->lr_refcount));

        if (atomic_dec_and_lock(&res->lr_refcount, &ns->ns_hash_lock)) {
                __ldlm_resource_putref_final(res);
                spin_unlock(&ns->ns_hash_lock);
                if (res->lr_lvb_data)
                        OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
                OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
                rc = 1;
        }

        RETURN(rc);
}

/* Returns 1 if the resource was freed, 0 if it remains. */
int ldlm_resource_putref_locked(struct ldlm_resource *res)
{
        int rc = 0;
        ENTRY;

        CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
               atomic_read(&res->lr_refcount) - 1);
        LASSERT(atomic_read(&res->lr_refcount) > 0);
        LASSERT(atomic_read(&res->lr_refcount) < LI_POISON);

        LASSERT(atomic_read(&res->lr_refcount) >= 0);
        if (atomic_dec_and_test(&res->lr_refcount)) {
                __ldlm_resource_putref_final(res);
                if (res->lr_lvb_data)
                        OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
                OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
                rc = 1;
        }

        RETURN(rc);
}

void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
                            struct ldlm_lock *lock)
{
        check_res_locked(res);

        ldlm_resource_dump(D_OTHER, res);
        CDEBUG(D_OTHER, "About to add this lock:\n");
        ldlm_lock_dump(D_OTHER, lock, 0);

        if (lock->l_destroyed) {
                CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
                return;
        }

        LASSERT(list_empty(&lock->l_res_link));

        list_add_tail(&lock->l_res_link, head);
}

void ldlm_resource_insert_lock_after(struct ldlm_lock *original,
                                     struct ldlm_lock *new)
{
        struct ldlm_resource *res = original->l_resource;

        check_res_locked(res);

        ldlm_resource_dump(D_OTHER, res);
        CDEBUG(D_OTHER, "About to insert this lock after %p:\n", original);
        ldlm_lock_dump(D_OTHER, new, 0);

        if (new->l_destroyed) {
                CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
                goto out;
        }

        LASSERT(list_empty(&new->l_res_link));

        list_add(&new->l_res_link, &original->l_res_link);
 out:;
}

void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
{
        int type = lock->l_resource->lr_type;

        check_res_locked(lock->l_resource);
        if (type == LDLM_IBITS || type == LDLM_PLAIN)
                ldlm_unlink_lock_skiplist(lock);
        else if (type == LDLM_EXTENT)
                ldlm_extent_unlink_lock(lock);
        list_del_init(&lock->l_res_link);
}

void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
{
        desc->lr_type = res->lr_type;
        desc->lr_name = res->lr_name;
}

void ldlm_dump_all_namespaces(ldlm_side_t client, int level)
{
        struct list_head *tmp;

        if (!((libcfs_debug | D_ERROR) & level))
                return;

        mutex_down(ldlm_namespace_lock(client));

        list_for_each(tmp, ldlm_namespace_list(client)) {
                struct ldlm_namespace *ns;
                ns = list_entry(tmp, struct ldlm_namespace, ns_list_chain);
                ldlm_namespace_dump(level, ns);
        }

        mutex_up(ldlm_namespace_lock(client));
}

void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
{
        struct list_head *tmp;

        if (!((libcfs_debug | D_ERROR) & level))
                return;

        CDEBUG(level, "--- Namespace: %s (rc: %d, side: %s)\n", 
               ns->ns_name, ns->ns_refcount, 
               ns_is_client(ns) ? "client" : "server");

        if (cfs_time_before(cfs_time_current(), ns->ns_next_dump))
                return;

        spin_lock(&ns->ns_hash_lock);
        tmp = ns->ns_root_list.next;
        while (tmp != &ns->ns_root_list) {
                struct ldlm_resource *res;
                res = list_entry(tmp, struct ldlm_resource, lr_childof);

                ldlm_resource_getref(res);
                spin_unlock(&ns->ns_hash_lock);

                lock_res(res);
                ldlm_resource_dump(level, res);
                unlock_res(res);
                
                spin_lock(&ns->ns_hash_lock);
                tmp = tmp->next;
                ldlm_resource_putref_locked(res);
        }
        ns->ns_next_dump = cfs_time_shift(10);
        spin_unlock(&ns->ns_hash_lock);
}

void ldlm_resource_dump(int level, struct ldlm_resource *res)
{
        struct list_head *tmp;
        int pos;

        CLASSERT(RES_NAME_SIZE == 4);

        if (!((libcfs_debug | D_ERROR) & level))
                return;

        CDEBUG(level, "--- Resource: %p ("LPU64"/"LPU64"/"LPU64"/"LPU64
               ") (rc: %d)\n", res, res->lr_name.name[0], res->lr_name.name[1],
               res->lr_name.name[2], res->lr_name.name[3],
               atomic_read(&res->lr_refcount));

        if (!list_empty(&res->lr_granted)) {
                pos = 0;
                CDEBUG(level, "Granted locks:\n");
                list_for_each(tmp, &res->lr_granted) {
                        struct ldlm_lock *lock;
                        lock = list_entry(tmp, struct ldlm_lock, l_res_link);
                        ldlm_lock_dump(level, lock, ++pos);
                }
        }
        if (!list_empty(&res->lr_converting)) {
                pos = 0;
                CDEBUG(level, "Converting locks:\n");
                list_for_each(tmp, &res->lr_converting) {
                        struct ldlm_lock *lock;
                        lock = list_entry(tmp, struct ldlm_lock, l_res_link);
                        ldlm_lock_dump(level, lock, ++pos);
                }
        }
        if (!list_empty(&res->lr_waiting)) {
                pos = 0;
                CDEBUG(level, "Waiting locks:\n");
                list_for_each(tmp, &res->lr_waiting) {
                        struct ldlm_lock *lock;
                        lock = list_entry(tmp, struct ldlm_lock, l_res_link);
                        ldlm_lock_dump(level, lock, ++pos);
                }
        }
}