8 files changed, 144 insertions, 82 deletions

diff --git a/fs/ocfs2/aops.c b/fs/ocfs2/aops.c
index 93628b02ef5d..875c11443817 100644
--- a/fs/ocfs2/aops.c
+++ b/fs/ocfs2/aops.c
@@ -614,6 +614,27 @@ static void ocfs2_dio_end_io(struct kiocb *iocb,
         ocfs2_rw_unlock(inode, 0);
 }
 
+/*
+ * ocfs2_invalidatepage() and ocfs2_releasepage() are shamelessly stolen
+ * from ext3.  PageChecked() bits have been removed as OCFS2 does not
+ * do journalled data.
+ */
+static void ocfs2_invalidatepage(struct page *page, unsigned long offset)
+{
+        journal_t *journal = OCFS2_SB(page->mapping->host->i_sb)->journal->j_journal;
+
+        journal_invalidatepage(journal, page, offset);
+}
+
+static int ocfs2_releasepage(struct page *page, gfp_t wait)
+{
+        journal_t *journal = OCFS2_SB(page->mapping->host->i_sb)->journal->j_journal;
+
+        if (!page_has_buffers(page))
+                return 0;
+        return journal_try_to_free_buffers(journal, page, wait);
+}
+
 static ssize_t ocfs2_direct_IO(int rw,
                                struct kiocb *iocb,
                                const struct iovec *iov,
@@ -661,5 +682,8 @@ const struct address_space_operations ocfs2_aops = {
         .commit_write   = ocfs2_commit_write,
         .bmap           = ocfs2_bmap,
         .sync_page      = block_sync_page,
-        .direct_IO      = ocfs2_direct_IO
+        .direct_IO      = ocfs2_direct_IO,
+        .invalidatepage = ocfs2_invalidatepage,
+        .releasepage    = ocfs2_releasepage,
+        .migratepage    = buffer_migrate_page,
 };
diff --git a/fs/ocfs2/cluster/heartbeat.c b/fs/ocfs2/cluster/heartbeat.c
index 5a9779bb9236..eba282da500e 100644
--- a/fs/ocfs2/cluster/heartbeat.c
+++ b/fs/ocfs2/cluster/heartbeat.c
@@ -1234,6 +1234,7 @@ static ssize_t o2hb_region_dev_write(struct o2hb_region *reg,
                                      const char *page,
                                      size_t count)
 {
+        struct task_struct *hb_task;
         long fd;
         int sectsize;
         char *p = (char *)page;
@@ -1319,20 +1320,28 @@ static ssize_t o2hb_region_dev_write(struct o2hb_region *reg,
          */
         atomic_set(&reg->hr_steady_iterations, O2HB_LIVE_THRESHOLD + 1);
 
-        reg->hr_task = kthread_run(o2hb_thread, reg, "o2hb-%s",
-                                   reg->hr_item.ci_name);
-        if (IS_ERR(reg->hr_task)) {
-                ret = PTR_ERR(reg->hr_task);
+        hb_task = kthread_run(o2hb_thread, reg, "o2hb-%s",
+                              reg->hr_item.ci_name);
+        if (IS_ERR(hb_task)) {
+                ret = PTR_ERR(hb_task);
                 mlog_errno(ret);
-                reg->hr_task = NULL;
                 goto out;
         }
 
+        spin_lock(&o2hb_live_lock);
+        reg->hr_task = hb_task;
+        spin_unlock(&o2hb_live_lock);
+
         ret = wait_event_interruptible(o2hb_steady_queue,
                                 atomic_read(&reg->hr_steady_iterations) == 0);
         if (ret) {
-                kthread_stop(reg->hr_task);
+                spin_lock(&o2hb_live_lock);
+                hb_task = reg->hr_task;
                 reg->hr_task = NULL;
+                spin_unlock(&o2hb_live_lock);
+
+                if (hb_task)
+                        kthread_stop(hb_task);
                 goto out;
         }
 
@@ -1354,10 +1363,17 @@ out:
 static ssize_t o2hb_region_pid_read(struct o2hb_region *reg,
                                       char *page)
 {
-        if (!reg->hr_task)
+        pid_t pid = 0;
+
+        spin_lock(&o2hb_live_lock);
+        if (reg->hr_task)
+                pid = reg->hr_task->pid;
+        spin_unlock(&o2hb_live_lock);
+
+        if (!pid)
                 return 0;
 
-        return sprintf(page, "%u\n", reg->hr_task->pid);
+        return sprintf(page, "%u\n", pid);
 }
 
 struct o2hb_region_attribute {
@@ -1495,13 +1511,17 @@ out:
 static void o2hb_heartbeat_group_drop_item(struct config_group *group,
                                            struct config_item *item)
 {
+        struct task_struct *hb_task;
         struct o2hb_region *reg = to_o2hb_region(item);
 
         /* stop the thread when the user removes the region dir */
-        if (reg->hr_task) {
-                kthread_stop(reg->hr_task);
-                reg->hr_task = NULL;
-        }
+        spin_lock(&o2hb_live_lock);
+        hb_task = reg->hr_task;
+        reg->hr_task = NULL;
+        spin_unlock(&o2hb_live_lock);
+
+        if (hb_task)
+                kthread_stop(hb_task);
 
         config_item_put(item);
 }
@@ -1682,7 +1702,7 @@ out:
 }
 EXPORT_SYMBOL_GPL(o2hb_register_callback);
 
-int o2hb_unregister_callback(struct o2hb_callback_func *hc)
+void o2hb_unregister_callback(struct o2hb_callback_func *hc)
 {
         BUG_ON(hc->hc_magic != O2HB_CB_MAGIC);
 
@@ -1690,15 +1710,13 @@ int o2hb_unregister_callback(struct o2hb_callback_func *hc)
              __builtin_return_address(0), hc);
 
         if (list_empty(&hc->hc_item))
-                return 0;
+                return;
 
         down_write(&o2hb_callback_sem);
 
         list_del_init(&hc->hc_item);
 
         up_write(&o2hb_callback_sem);
-
-        return 0;
 }
 EXPORT_SYMBOL_GPL(o2hb_unregister_callback);
 
diff --git a/fs/ocfs2/cluster/heartbeat.h b/fs/ocfs2/cluster/heartbeat.h
index cac6223206a9..cc6d40b39771 100644
--- a/fs/ocfs2/cluster/heartbeat.h
+++ b/fs/ocfs2/cluster/heartbeat.h
@@ -70,7 +70,7 @@ void o2hb_setup_callback(struct o2hb_callback_func *hc,
                          void *data,
                          int priority);
 int o2hb_register_callback(struct o2hb_callback_func *hc);
-int o2hb_unregister_callback(struct o2hb_callback_func *hc);
+void o2hb_unregister_callback(struct o2hb_callback_func *hc);
 void o2hb_fill_node_map(unsigned long *map,
                         unsigned bytes);
 void o2hb_init(void);
diff --git a/fs/ocfs2/cluster/tcp.c b/fs/ocfs2/cluster/tcp.c
index 1718215fc018..69caf3e12fea 100644
--- a/fs/ocfs2/cluster/tcp.c
+++ b/fs/ocfs2/cluster/tcp.c
@@ -1638,17 +1638,8 @@ static void o2net_hb_node_up_cb(struct o2nm_node *node, int node_num,
 
 void o2net_unregister_hb_callbacks(void)
 {
-        int ret;
-
-        ret = o2hb_unregister_callback(&o2net_hb_up);
-        if (ret < 0)
-                mlog(ML_ERROR, "Status return %d unregistering heartbeat up "
-                     "callback!\n", ret);
-
-        ret = o2hb_unregister_callback(&o2net_hb_down);
-        if (ret < 0)
-                mlog(ML_ERROR, "Status return %d unregistering heartbeat down "
-                     "callback!\n", ret);
+        o2hb_unregister_callback(&o2net_hb_up);
+        o2hb_unregister_callback(&o2net_hb_down);
 }
 
 int o2net_register_hb_callbacks(void)
diff --git a/fs/ocfs2/dlm/dlmdomain.c b/fs/ocfs2/dlm/dlmdomain.c
index 6087c4749fee..c558442a0b44 100644
--- a/fs/ocfs2/dlm/dlmdomain.c
+++ b/fs/ocfs2/dlm/dlmdomain.c
@@ -138,8 +138,10 @@ static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);
 
 void __dlm_unhash_lockres(struct dlm_lock_resource *lockres)
 {
-        hlist_del_init(&lockres->hash_node);
-        dlm_lockres_put(lockres);
+        if (!hlist_unhashed(&lockres->hash_node)) {
+                hlist_del_init(&lockres->hash_node);
+                dlm_lockres_put(lockres);
+        }
 }
 
 void __dlm_insert_lockres(struct dlm_ctxt *dlm,
@@ -655,6 +657,8 @@ void dlm_unregister_domain(struct dlm_ctxt *dlm)
                 dlm_kick_thread(dlm, NULL);
 
                 while (dlm_migrate_all_locks(dlm)) {
+                        /* Give dlm_thread time to purge the lockres' */
+                        msleep(500);
                         mlog(0, "%s: more migration to do\n", dlm->name);
                 }
                 dlm_mark_domain_leaving(dlm);
diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c
index 77e4e6169a0d..6edffca99d98 100644
--- a/fs/ocfs2/dlm/dlmmaster.c
+++ b/fs/ocfs2/dlm/dlmmaster.c
@@ -2424,6 +2424,57 @@ static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
         dlm_lockres_put(res);
 }
 
+/* Checks whether the lockres can be migrated. Returns 0 if yes, < 0
+ * if not. If 0, numlocks is set to the number of locks in the lockres.
+ */
+static int dlm_is_lockres_migrateable(struct dlm_ctxt *dlm,
+                                      struct dlm_lock_resource *res,
+                                      int *numlocks)
+{
+        int ret;
+        int i;
+        int count = 0;
+        struct list_head *queue, *iter;
+        struct dlm_lock *lock;
+
+        assert_spin_locked(&res->spinlock);
+
+        ret = -EINVAL;
+        if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
+                mlog(0, "cannot migrate lockres with unknown owner!\n");
+                goto leave;
+        }
+
+        if (res->owner != dlm->node_num) {
+                mlog(0, "cannot migrate lockres this node doesn't own!\n");
+                goto leave;
+        }
+
+        ret = 0;
+        queue = &res->granted;
+        for (i = 0; i < 3; i++) {
+                list_for_each(iter, queue) {
+                        lock = list_entry(iter, struct dlm_lock, list);
+                        ++count;
+                        if (lock->ml.node == dlm->node_num) {
+                                mlog(0, "found a lock owned by this node still "
+                                     "on the %s queue!  will not migrate this "
+                                     "lockres\n", (i == 0 ? "granted" :
+                                                   (i == 1 ? "converting" :
+                                                    "blocked")));
+                                ret = -ENOTEMPTY;
+                                goto leave;
+                        }
+                }
+                queue++;
+        }
+
+        *numlocks = count;
+        mlog(0, "migrateable lockres having %d locks\n", *numlocks);
+
+leave:
+        return ret;
+}
 
 /*
  * DLM_MIGRATE_LOCKRES
@@ -2437,14 +2488,12 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
         struct dlm_master_list_entry *mle = NULL;
         struct dlm_master_list_entry *oldmle = NULL;
         struct dlm_migratable_lockres *mres = NULL;
-        int ret = -EINVAL;
+        int ret = 0;
         const char *name;
         unsigned int namelen;
         int mle_added = 0;
-        struct list_head *queue, *iter;
-        int i;
-        struct dlm_lock *lock;
-        int empty = 1, wake = 0;
+        int numlocks;
+        int wake = 0;
 
         if (!dlm_grab(dlm))
                 return -EINVAL;
@@ -2458,42 +2507,16 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
          * ensure this lockres is a proper candidate for migration
          */
         spin_lock(&res->spinlock);
-        if (res->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
-                mlog(0, "cannot migrate lockres with unknown owner!\n");
-                spin_unlock(&res->spinlock);
-                goto leave;
-        }
-        if (res->owner != dlm->node_num) {
-                mlog(0, "cannot migrate lockres this node doesn't own!\n");
+        ret = dlm_is_lockres_migrateable(dlm, res, &numlocks);
+        if (ret < 0) {
                 spin_unlock(&res->spinlock);
                 goto leave;
         }
-        mlog(0, "checking queues...\n");
-        queue = &res->granted;
-        for (i=0; i<3; i++) {
-                list_for_each(iter, queue) {
-                        lock = list_entry (iter, struct dlm_lock, list);
-                        empty = 0;
-                        if (lock->ml.node == dlm->node_num) {
-                                mlog(0, "found a lock owned by this node "
-                                     "still on the %s queue!  will not "
-                                     "migrate this lockres\n",
-                                     i==0 ? "granted" :
-                                     (i==1 ? "converting" : "blocked"));
-                                spin_unlock(&res->spinlock);
-                                ret = -ENOTEMPTY;
-                                goto leave;
-                        }
-                }
-                queue++;
-        }
-        mlog(0, "all locks on this lockres are nonlocal.  continuing\n");
         spin_unlock(&res->spinlock);
 
         /* no work to do */
-        if (empty) {
+        if (numlocks == 0) {
                 mlog(0, "no locks were found on this lockres! done!\n");
-                ret = 0;
                 goto leave;
         }
 
@@ -2729,15 +2752,26 @@ int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
 {
         int ret;
         int lock_dropped = 0;
+        int numlocks;
 
+        spin_lock(&res->spinlock);
         if (res->owner != dlm->node_num) {
                 if (!__dlm_lockres_unused(res)) {
                         mlog(ML_ERROR, "%s:%.*s: this node is not master, "
                              "trying to free this but locks remain\n",
                              dlm->name, res->lockname.len, res->lockname.name);
                 }
+                spin_unlock(&res->spinlock);
+                goto leave;
+        }
+
+        /* No need to migrate a lockres having no locks */
+        ret = dlm_is_lockres_migrateable(dlm, res, &numlocks);
+        if (ret >= 0 && numlocks == 0) {
+                spin_unlock(&res->spinlock);
                 goto leave;
         }
+        spin_unlock(&res->spinlock);
 
         /* Wheee! Migrate lockres here! Will sleep so drop spinlock. */
         spin_unlock(&dlm->spinlock);
diff --git a/fs/ocfs2/dlm/dlmthread.c b/fs/ocfs2/dlm/dlmthread.c
index 8ffa0916eb86..2b264c6ba039 100644
--- a/fs/ocfs2/dlm/dlmthread.c
+++ b/fs/ocfs2/dlm/dlmthread.c
@@ -256,18 +256,14 @@ static void dlm_run_purge_list(struct dlm_ctxt *dlm,
                         break;
                 }
 
-                mlog(0, "removing lockres %.*s:%p from purgelist\n",
-                     lockres->lockname.len, lockres->lockname.name, lockres);
-                list_del_init(&lockres->purge);
-                dlm_lockres_put(lockres);
-                dlm->purge_count--;
+                dlm_lockres_get(lockres);
 
                 /* This may drop and reacquire the dlm spinlock if it
                  * has to do migration. */
-                mlog(0, "calling dlm_purge_lockres!\n");
                 if (dlm_purge_lockres(dlm, lockres))
                         BUG();
-                mlog(0, "DONE calling dlm_purge_lockres!\n");
+
+                dlm_lockres_put(lockres);
 
                 /* Avoid adding any scheduling latencies */
                 cond_resched_lock(&dlm->spinlock);
diff --git a/fs/ocfs2/heartbeat.c b/fs/ocfs2/heartbeat.c
index 8fc52d6d0ce7..b25ef63781ba 100644
--- a/fs/ocfs2/heartbeat.c
+++ b/fs/ocfs2/heartbeat.c
@@ -164,8 +164,10 @@ int ocfs2_register_hb_callbacks(struct ocfs2_super *osb)
         }
 
         status = o2hb_register_callback(&osb->osb_hb_up);
-        if (status < 0)
+        if (status < 0) {
                 mlog_errno(status);
+                o2hb_unregister_callback(&osb->osb_hb_down);
+        }
 
 bail:
         return status;
@@ -173,18 +175,11 @@ bail:
 
 void ocfs2_clear_hb_callbacks(struct ocfs2_super *osb)
 {
-        int status;
-
         if (ocfs2_mount_local(osb))
                 return;
 
-        status = o2hb_unregister_callback(&osb->osb_hb_down);
-        if (status < 0)
-                mlog_errno(status);
-
-        status = o2hb_unregister_callback(&osb->osb_hb_up);
-        if (status < 0)
-                mlog_errno(status);
+        o2hb_unregister_callback(&osb->osb_hb_down);
+        o2hb_unregister_callback(&osb->osb_hb_up);
 }
 
 void ocfs2_stop_heartbeat(struct ocfs2_super *osb)