land b_hd_remote_acl: support get/set ACL from remote client.

[fs/lustre-release.git] / lustre / sec / upcall_cache.c

diff --git a/lustre/sec/upcall_cache.c b/lustre/sec/upcall_cache.c
index 49e9522..ad4686a 100644 (file)
--- a/lustre/sec/upcall_cache.c
+++ b/lustre/sec/upcall_cache.c
@@ -68,7 +68,7 @@ static void free_entry(struct upcall_cache_entry *entry)
         LASSERT(cache->free_entry);
         LASSERT(atomic_read(&entry->ue_refcount) == 0);
 
-        CDEBUG(D_OTHER, "destroy %s entry %p for key "LPU64"\n",
+        CDEBUG(D_SEC, "destroy %s entry %p for key "LPU64"\n",
                cache->uc_name, entry, entry->ue_key);
 
         list_del(&entry->ue_hash);
@@ -100,14 +100,22 @@ static inline int refresh_entry(struct upcall_cache_entry *entry)
 
 static int check_unlink_entry(struct upcall_cache_entry *entry)
 {
+        /* upcall will be issued upon new entries immediately
+         * after they are created
+         */
+        LASSERT(!UC_CACHE_IS_NEW(entry));
+
         if (UC_CACHE_IS_VALID(entry) &&
             time_before(get_seconds(), entry->ue_expire))
                 return 0;
 
-        if (UC_CACHE_IS_ACQUIRING(entry) &&
-            time_after(get_seconds(), entry->ue_acquire_expire)) {
-                UC_CACHE_SET_EXPIRED(entry);
-                wake_up_all(&entry->ue_waitq);
+        if (UC_CACHE_IS_ACQUIRING(entry)) {
+                if (time_before(get_seconds(), entry->ue_acquire_expire))
+                        return 0;
+                else {
+                        UC_CACHE_SET_EXPIRED(entry);
+                        wake_up_all(&entry->ue_waitq);
+                }
         } else if (!UC_CACHE_IS_INVALID(entry)) {
                 UC_CACHE_SET_EXPIRED(entry);
         }
@@ -206,7 +214,7 @@ find_again:
                 set_current_state(TASK_INTERRUPTIBLE);
                 write_unlock(&cache->uc_hashlock);
 
-                schedule_timeout(cache->uc_acquire_expire);
+                schedule_timeout(cache->uc_acquire_expire * HZ);
 
                 write_lock(&cache->uc_hashlock);
                 remove_wait_queue(&entry->ue_waitq, &wait);
@@ -222,6 +230,8 @@ find_again:
                                entry->ue_expire);
                         put_entry(entry);
                         write_unlock(&cache->uc_hashlock);
+                        CERROR("Interrupted? Or check whether %s is in place\n",
+                               cache->uc_upcall);
                         RETURN(NULL);
                 }
                 /* fall through */
@@ -253,7 +263,6 @@ find_again:
         }
         
         /* Now we know it's good */
-        LASSERT(UC_CACHE_IS_VALID(entry));
         write_unlock(&cache->uc_hashlock);
 
         RETURN(entry);
@@ -284,8 +293,7 @@ void upcall_cache_put_entry(struct upcall_cache_entry *entry)
 }
 EXPORT_SYMBOL(upcall_cache_put_entry);
 
-int upcall_cache_downcall(struct upcall_cache *cache, __u64 key,
-                          int err, void *args)
+int upcall_cache_downcall(struct upcall_cache *cache, __u64 key, void *args)
 {
         struct list_head *head;
         struct upcall_cache_entry *entry;
@@ -312,9 +320,23 @@ int upcall_cache_downcall(struct upcall_cache *cache, __u64 key,
                 RETURN(-EINVAL);
         }
 
-        if (err < 0) {
-                UC_CACHE_SET_INVALID(entry);
-                GOTO(out, rc = err);
+        if (!UC_CACHE_IS_ACQUIRING(entry)) {
+                if (UC_CACHE_IS_VALID(entry)) {
+                        /* This should not happen, just give a warning
+                         * at this moment.
+                         */
+                        CWARN("entry %p(key "LPU64", ac %ld, ex %ld): "
+                              "already valid???\n", entry, entry->ue_key,
+                              entry->ue_acquire_expire, entry->ue_expire);
+                        GOTO(out, rc = 0);
+                }
+
+                CWARN("stale entry %p: cur %lu, key "LPU64", ref %d, "
+                      "fl %u, ac %ld, ex %ld\n",
+                       entry, get_seconds(), entry->ue_key,
+                       atomic_read(&entry->ue_refcount), entry->ue_flags,
+                       entry->ue_acquire_expire, entry->ue_expire);
+                GOTO(out, rc = -EINVAL);
         }
 
         if (!UC_CACHE_IS_ACQUIRING(entry) ||
@@ -333,14 +355,19 @@ int upcall_cache_downcall(struct upcall_cache *cache, __u64 key,
         rc = cache->parse_downcall(cache, entry, args);
         write_lock(&cache->uc_hashlock);
         atomic_dec(&entry->ue_refcount);
-        if (rc) {
+
+        if (rc < 0) {
                 UC_CACHE_SET_INVALID(entry);
                 list_del_init(&entry->ue_hash);
                 GOTO(out, rc);
+        } else if (rc == 0) {
+                entry->ue_expire = get_seconds() + cache->uc_entry_expire;
+        } else {
+                entry->ue_expire = get_seconds() + cache->uc_err_entry_expire;
         }
-        entry->ue_expire = get_seconds() + cache->uc_entry_expire;
+
         UC_CACHE_SET_VALID(entry);
-        CDEBUG(D_OTHER, "create ucache entry %p(key "LPU64")\n",
+        CDEBUG(D_SEC, "create ucache entry %p(key "LPU64")\n",
                entry, entry->ue_key);
 out:
         wake_up_all(&entry->ue_waitq);