add recount protection for osc callbacks, so avoid panic on shutdown

Branch HEAD
b=15210
i=green
i=umka

lustre/ChangeLog

@@ -12,6 +12,10 @@ tbd  Sun Microsystems, Inc.
        * RHEL 4 and RHEL 5/SLES 10 clients behaves differently on 'cd' to a
         removed cwd "./" (refer to Bugzilla 14399).
 
+Severity   : normal
+Bugzilla   : 15210
+Description: add recount protection for osc callbacks, so avoid panic on shutdown
+
 Severity   : normal
 Bugzilla   : 12653
 Description: sanity test 65a fails if stripecount of -1 is set

lustre/include/lustre_cache.h

@@ -13,6 +13,7 @@ struct osc_async_page;
 struct page_removal_cb_element {
         struct list_head        prce_list;
         obd_page_removal_cb_t   prce_callback;
+        atomic_t                prce_refcnt;
 };
 
 typedef int (*cache_iterate_extents_cb_t)(struct lustre_cache *,
@@ -27,6 +28,7 @@ struct lustre_cache {
         struct list_head         lc_locks_list;
         spinlock_t               lc_locks_list_lock;
         struct list_head         lc_page_removal_callback_list;
+        rwlock_t                 lc_page_removal_cb_lock; /* iterate vs modify list */
         struct obd_device       *lc_obd;
         obd_pin_extent_cb        lc_pin_extent_cb;
 };

lustre/osc/cache.c

@@ -138,16 +138,39 @@ int cache_add_extent(struct lustre_cache *cache, struct ldlm_res_id *res,
         RETURN(0);
 }
 
+static void cache_extent_removal_get(struct page_removal_cb_element *element)
+{
+        atomic_inc(&element->prce_refcnt);
+}
+
+static void cache_extent_removal_put(struct page_removal_cb_element *element)
+{
+        if(atomic_dec_and_test(&element->prce_refcnt))
+                OBD_FREE_PTR(element);
+}
+
 static int cache_extent_removal_event(struct lustre_cache *cache,
                                       void *data, int discard)
 {
         struct page *page = data;
+        struct list_head *iter;
         struct page_removal_cb_element *element;
 
-        list_for_each_entry(element, &cache->lc_page_removal_callback_list,
-                            prce_list) {
+        read_lock(&cache->lc_page_removal_cb_lock);
+        iter = cache->lc_page_removal_callback_list.next;
+        while(iter != &cache->lc_page_removal_callback_list) {
+                element = list_entry(iter, struct page_removal_cb_element, prce_list);
+                cache_extent_removal_get(element);
+                read_unlock(&cache->lc_page_removal_cb_lock);
+
                 element->prce_callback(page, discard);
+
+                read_lock(&cache->lc_page_removal_cb_lock);
+                iter = iter->next;
+                cache_extent_removal_put(element);
         }
+        read_unlock(&cache->lc_page_removal_cb_lock);
+
         return 0;
 }
 
@@ -167,12 +190,17 @@ int cache_add_extent_removal_cb(struct lustre_cache *cache,
 
         if (!func_cb)
                 return 0;
-        OBD_ALLOC(element, sizeof(*element));
+
+        OBD_ALLOC_PTR(element);
         if (!element)
                 return -ENOMEM;
         element->prce_callback = func_cb;
+        atomic_set(&element->prce_refcnt, 1);
+
+        write_lock(&cache->lc_page_removal_cb_lock);
         list_add_tail(&element->prce_list,
                       &cache->lc_page_removal_callback_list);
+        write_unlock(&cache->lc_page_removal_cb_lock);
 
         cache->lc_pin_extent_cb = pin_cb;
         return 0;
@@ -188,17 +216,21 @@ int cache_del_extent_removal_cb(struct lustre_cache *cache,
         int found = 0;
         struct page_removal_cb_element *element, *t;
 
+        write_lock(&cache->lc_page_removal_cb_lock);
         list_for_each_entry_safe(element, t,
                                  &cache->lc_page_removal_callback_list,
                                  prce_list) {
                 if (element->prce_callback == func_cb) {
                         list_del(&element->prce_list);
-                        OBD_FREE(element, sizeof(*element));
+                        write_unlock(&cache->lc_page_removal_cb_lock);
                         found = 1;
+                        cache_extent_removal_put(element);
+                        write_lock(&cache->lc_page_removal_cb_lock);
                         /* We continue iterating the list in case this function
                            was registered more than once */
                 }
         }
+        write_unlock(&cache->lc_page_removal_cb_lock);
 
         if (list_empty(&cache->lc_page_removal_callback_list))
                 cache->lc_pin_extent_cb = NULL;
@@ -358,6 +390,7 @@ struct lustre_cache *cache_create(struct obd_device *obd)
         spin_lock_init(&cache->lc_locks_list_lock);
         CFS_INIT_LIST_HEAD(&cache->lc_locks_list);
         CFS_INIT_LIST_HEAD(&cache->lc_page_removal_callback_list);
+        rwlock_init(&cache->lc_page_removal_cb_lock);
         cache->lc_obd = obd;
 
       out: