ocfs2/dlm: add DEREF_DONE message

[karo-tx-linux.git] / fs / ocfs2 / dlm / dlmmaster.c

diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 9477d6e1de372427c36849b6eb3dc5a041089f64..8913e7d443da7db48575f5f7219d8e22204aedf8 100644 (file)
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -2375,6 +2375,122 @@ done:
        return ret;
 }
 
+int dlm_deref_lockres_done_handler(struct o2net_msg *msg, u32 len, void *data,
+                             void **ret_data)
+{
+       struct dlm_ctxt *dlm = data;
+       struct dlm_deref_lockres_done *deref
+                       = (struct dlm_deref_lockres_done *)msg->buf;
+       struct dlm_lock_resource *res = NULL;
+       char *name;
+       unsigned int namelen;
+       int ret = -EINVAL;
+       u8 node;
+       unsigned int hash;
+
+       if (!dlm_grab(dlm))
+               return 0;
+
+       name = deref->name;
+       namelen = deref->namelen;
+       node = deref->node_idx;
+
+       if (namelen > DLM_LOCKID_NAME_MAX) {
+               mlog(ML_ERROR, "Invalid name length!");
+               goto done;
+       }
+       if (deref->node_idx >= O2NM_MAX_NODES) {
+               mlog(ML_ERROR, "Invalid node number: %u\n", node);
+               goto done;
+       }
+
+       hash = dlm_lockid_hash(name, namelen);
+
+       spin_lock(&dlm->spinlock);
+       res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);
+       if (!res) {
+               spin_unlock(&dlm->spinlock);
+               mlog(ML_ERROR, "%s:%.*s: bad lockres name\n",
+                    dlm->name, namelen, name);
+               goto done;
+       }
+
+       spin_lock(&res->spinlock);
+       BUG_ON(!(res->state & DLM_LOCK_RES_DROPPING_REF));
+       if (!list_empty(&res->purge)) {
+               mlog(0, "%s: Removing res %.*s from purgelist\n",
+                       dlm->name, res->lockname.len, res->lockname.name);
+               list_del_init(&res->purge);
+               dlm_lockres_put(res);
+               dlm->purge_count--;
+       }
+
+       if (!__dlm_lockres_unused(res)) {
+               mlog(ML_ERROR, "%s: res %.*s in use after deref\n",
+                       dlm->name, res->lockname.len, res->lockname.name);
+               __dlm_print_one_lock_resource(res);
+               BUG();
+       }
+
+       __dlm_unhash_lockres(dlm, res);
+
+       spin_lock(&dlm->track_lock);
+       if (!list_empty(&res->tracking))
+               list_del_init(&res->tracking);
+       else {
+               mlog(ML_ERROR, "%s: Resource %.*s not on the Tracking list\n",
+                    dlm->name, res->lockname.len, res->lockname.name);
+               __dlm_print_one_lock_resource(res);
+       }
+       spin_unlock(&dlm->track_lock);
+
+       /* lockres is not in the hash now. drop the flag and wake up
+        * any processes waiting in dlm_get_lock_resource.
+        */
+       res->state &= ~DLM_LOCK_RES_DROPPING_REF;
+       spin_unlock(&res->spinlock);
+       wake_up(&res->wq);
+
+       dlm_lockres_put(res);
+
+       spin_unlock(&dlm->spinlock);
+
+done:
+       dlm_put(dlm);
+       return ret;
+}
+
+static void dlm_drop_lockres_ref_done(struct dlm_ctxt *dlm,
+               struct dlm_lock_resource *res, u8 node)
+{
+       struct dlm_deref_lockres_done deref;
+       int ret = 0, r;
+       const char *lockname;
+       unsigned int namelen;
+
+       lockname = res->lockname.name;
+       namelen = res->lockname.len;
+       BUG_ON(namelen > O2NM_MAX_NAME_LEN);
+
+       memset(&deref, 0, sizeof(deref));
+       deref.node_idx = dlm->node_num;
+       deref.namelen = namelen;
+       memcpy(deref.name, lockname, namelen);
+
+       ret = o2net_send_message(DLM_DEREF_LOCKRES_DONE, dlm->key,
+                                &deref, sizeof(deref), node, &r);
+       if (ret < 0) {
+               mlog(ML_ERROR, "%s: res %.*s, error %d send DEREF DONE "
+                               " to node %u\n", dlm->name, namelen,
+                               lockname, ret, node);
+       } else if (r < 0) {
+               /* ignore the error */
+               mlog(ML_ERROR, "%s: res %.*s, DEREF to node %u got %d\n",
+                    dlm->name, namelen, lockname, node, r);
+               dlm_print_one_lock_resource(res);
+       }
+}
+
 static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
 {
        struct dlm_ctxt *dlm;