Whamcloud - gitweb
LU-6447 mdt: mdt_identity_upcall to not block with rwlock held
[fs/lustre-release.git] / lustre / obdclass / upcall_cache.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, 2014, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/obdclass/upcall_cache.c
37  *
38  * Supplementary groups cache.
39  */
40 #define DEBUG_SUBSYSTEM S_SEC
41
42 #include <libcfs/libcfs.h>
43 #include <lnet/types.h>
44 #include <upcall_cache.h>
45
46 static struct upcall_cache_entry *alloc_entry(struct upcall_cache *cache,
47                                               __u64 key, void *args)
48 {
49         struct upcall_cache_entry *entry;
50
51         LIBCFS_ALLOC(entry, sizeof(*entry));
52         if (!entry)
53                 return NULL;
54
55         UC_CACHE_SET_NEW(entry);
56         INIT_LIST_HEAD(&entry->ue_hash);
57         entry->ue_key = key;
58         atomic_set(&entry->ue_refcount, 0);
59         init_waitqueue_head(&entry->ue_waitq);
60         if (cache->uc_ops->init_entry)
61                 cache->uc_ops->init_entry(entry, args);
62         return entry;
63 }
64
65 /* protected by cache lock */
66 static void free_entry(struct upcall_cache *cache,
67                        struct upcall_cache_entry *entry)
68 {
69         if (cache->uc_ops->free_entry)
70                 cache->uc_ops->free_entry(cache, entry);
71
72         list_del(&entry->ue_hash);
73         CDEBUG(D_OTHER, "destroy cache entry %p for key "LPU64"\n",
74                 entry, entry->ue_key);
75         LIBCFS_FREE(entry, sizeof(*entry));
76 }
77
78 static inline int upcall_compare(struct upcall_cache *cache,
79                                  struct upcall_cache_entry *entry,
80                                  __u64 key, void *args)
81 {
82         if (entry->ue_key != key)
83                 return -1;
84
85         if (cache->uc_ops->upcall_compare)
86                 return cache->uc_ops->upcall_compare(cache, entry, key, args);
87
88         return 0;
89 }
90
91 static inline int downcall_compare(struct upcall_cache *cache,
92                                    struct upcall_cache_entry *entry,
93                                    __u64 key, void *args)
94 {
95         if (entry->ue_key != key)
96                 return -1;
97
98         if (cache->uc_ops->downcall_compare)
99                 return cache->uc_ops->downcall_compare(cache, entry, key, args);
100
101         return 0;
102 }
103
104 static inline void get_entry(struct upcall_cache_entry *entry)
105 {
106         atomic_inc(&entry->ue_refcount);
107 }
108
109 static inline void put_entry(struct upcall_cache *cache,
110                              struct upcall_cache_entry *entry)
111 {
112         if (atomic_dec_and_test(&entry->ue_refcount) &&
113             (UC_CACHE_IS_INVALID(entry) || UC_CACHE_IS_EXPIRED(entry))) {
114                 free_entry(cache, entry);
115         }
116 }
117
118 static int check_unlink_entry(struct upcall_cache *cache,
119                               struct upcall_cache_entry *entry)
120 {
121         if (UC_CACHE_IS_VALID(entry) &&
122             cfs_time_before(cfs_time_current(), entry->ue_expire))
123                 return 0;
124
125         if (UC_CACHE_IS_ACQUIRING(entry)) {
126                 if (entry->ue_acquire_expire == 0 ||
127                     cfs_time_before(cfs_time_current(),
128                                     entry->ue_acquire_expire))
129                         return 0;
130
131                 UC_CACHE_SET_EXPIRED(entry);
132                 wake_up_all(&entry->ue_waitq);
133         } else if (!UC_CACHE_IS_INVALID(entry)) {
134                 UC_CACHE_SET_EXPIRED(entry);
135         }
136
137         list_del_init(&entry->ue_hash);
138         if (!atomic_read(&entry->ue_refcount))
139                 free_entry(cache, entry);
140         return 1;
141 }
142
143 static inline int refresh_entry(struct upcall_cache *cache,
144                          struct upcall_cache_entry *entry)
145 {
146         LASSERT(cache->uc_ops->do_upcall);
147         return cache->uc_ops->do_upcall(cache, entry);
148 }
149
150 struct upcall_cache_entry *upcall_cache_get_entry(struct upcall_cache *cache,
151                                                   __u64 key, void *args)
152 {
153         struct upcall_cache_entry *entry = NULL, *new = NULL, *next;
154         struct list_head *head;
155         wait_queue_t wait;
156         int rc, found;
157         ENTRY;
158
159         LASSERT(cache);
160
161         head = &cache->uc_hashtable[UC_CACHE_HASH_INDEX(key)];
162 find_again:
163         found = 0;
164         spin_lock(&cache->uc_lock);
165         list_for_each_entry_safe(entry, next, head, ue_hash) {
166                 /* check invalid & expired items */
167                 if (check_unlink_entry(cache, entry))
168                         continue;
169                 if (upcall_compare(cache, entry, key, args) == 0) {
170                         found = 1;
171                         break;
172                 }
173         }
174
175         if (!found) {
176                 if (!new) {
177                         spin_unlock(&cache->uc_lock);
178                         new = alloc_entry(cache, key, args);
179                         if (!new) {
180                                 CERROR("fail to alloc entry\n");
181                                 RETURN(ERR_PTR(-ENOMEM));
182                         }
183                         goto find_again;
184                 } else {
185                         list_add(&new->ue_hash, head);
186                         entry = new;
187                 }
188         } else {
189                 if (new) {
190                         free_entry(cache, new);
191                         new = NULL;
192                 }
193                 list_move(&entry->ue_hash, head);
194         }
195         get_entry(entry);
196
197         /* acquire for new one */
198         if (UC_CACHE_IS_NEW(entry)) {
199                 UC_CACHE_SET_ACQUIRING(entry);
200                 UC_CACHE_CLEAR_NEW(entry);
201                 spin_unlock(&cache->uc_lock);
202                 rc = refresh_entry(cache, entry);
203                 spin_lock(&cache->uc_lock);
204                 entry->ue_acquire_expire =
205                         cfs_time_shift(cache->uc_acquire_expire);
206                 if (rc < 0) {
207                         UC_CACHE_CLEAR_ACQUIRING(entry);
208                         UC_CACHE_SET_INVALID(entry);
209                         wake_up_all(&entry->ue_waitq);
210                         if (unlikely(rc == -EREMCHG)) {
211                                 put_entry(cache, entry);
212                                 GOTO(out, entry = ERR_PTR(rc));
213                         }
214                 }
215         }
216         /* someone (and only one) is doing upcall upon this item,
217          * wait it to complete */
218         if (UC_CACHE_IS_ACQUIRING(entry)) {
219                 long expiry = (entry == new) ?
220                               cfs_time_seconds(cache->uc_acquire_expire) :
221                               MAX_SCHEDULE_TIMEOUT;
222                 long left;
223
224                 init_waitqueue_entry(&wait, current);
225                 add_wait_queue(&entry->ue_waitq, &wait);
226                 set_current_state(TASK_INTERRUPTIBLE);
227                 spin_unlock(&cache->uc_lock);
228
229                 left = schedule_timeout(expiry);
230
231                 spin_lock(&cache->uc_lock);
232                 remove_wait_queue(&entry->ue_waitq, &wait);
233                 if (UC_CACHE_IS_ACQUIRING(entry)) {
234                         /* we're interrupted or upcall failed in the middle */
235                         rc = left > 0 ? -EINTR : -ETIMEDOUT;
236                         CERROR("acquire for key "LPU64": error %d\n",
237                                entry->ue_key, rc);
238                         put_entry(cache, entry);
239                         GOTO(out, entry = ERR_PTR(rc));
240                 }
241         }
242
243         /* invalid means error, don't need to try again */
244         if (UC_CACHE_IS_INVALID(entry)) {
245                 put_entry(cache, entry);
246                 GOTO(out, entry = ERR_PTR(-EIDRM));
247         }
248
249         /* check expired
250          * We can't refresh the existing one because some
251          * memory might be shared by multiple processes.
252          */
253         if (check_unlink_entry(cache, entry)) {
254                 /* if expired, try again. but if this entry is
255                  * created by me but too quickly turn to expired
256                  * without any error, should at least give a
257                  * chance to use it once.
258                  */
259                 if (entry != new) {
260                         put_entry(cache, entry);
261                         spin_unlock(&cache->uc_lock);
262                         new = NULL;
263                         goto find_again;
264                 }
265         }
266
267         /* Now we know it's good */
268 out:
269         spin_unlock(&cache->uc_lock);
270         RETURN(entry);
271 }
272 EXPORT_SYMBOL(upcall_cache_get_entry);
273
274 void upcall_cache_put_entry(struct upcall_cache *cache,
275                             struct upcall_cache_entry *entry)
276 {
277         ENTRY;
278
279         if (!entry) {
280                 EXIT;
281                 return;
282         }
283
284         LASSERT(atomic_read(&entry->ue_refcount) > 0);
285         spin_lock(&cache->uc_lock);
286         put_entry(cache, entry);
287         spin_unlock(&cache->uc_lock);
288         EXIT;
289 }
290 EXPORT_SYMBOL(upcall_cache_put_entry);
291
292 int upcall_cache_downcall(struct upcall_cache *cache, __u32 err, __u64 key,
293                           void *args)
294 {
295         struct upcall_cache_entry *entry = NULL;
296         struct list_head *head;
297         int found = 0, rc = 0;
298         ENTRY;
299
300         LASSERT(cache);
301
302         head = &cache->uc_hashtable[UC_CACHE_HASH_INDEX(key)];
303
304         spin_lock(&cache->uc_lock);
305         list_for_each_entry(entry, head, ue_hash) {
306                 if (downcall_compare(cache, entry, key, args) == 0) {
307                         found = 1;
308                         get_entry(entry);
309                         break;
310                 }
311         }
312
313         if (!found) {
314                 CDEBUG(D_OTHER, "%s: upcall for key "LPU64" not expected\n",
315                        cache->uc_name, key);
316                 /* haven't found, it's possible */
317                 spin_unlock(&cache->uc_lock);
318                 RETURN(-EINVAL);
319         }
320
321         if (err) {
322                 CDEBUG(D_OTHER, "%s: upcall for key "LPU64" returned %d\n",
323                        cache->uc_name, entry->ue_key, err);
324                 GOTO(out, rc = -EINVAL);
325         }
326
327         if (!UC_CACHE_IS_ACQUIRING(entry)) {
328                 CDEBUG(D_RPCTRACE, "%s: found uptodate entry %p (key "LPU64")"
329                        "\n", cache->uc_name, entry, entry->ue_key);
330                 GOTO(out, rc = 0);
331         }
332
333         if (UC_CACHE_IS_INVALID(entry) || UC_CACHE_IS_EXPIRED(entry)) {
334                 CERROR("%s: found a stale entry %p (key "LPU64") in ioctl\n",
335                        cache->uc_name, entry, entry->ue_key);
336                 GOTO(out, rc = -EINVAL);
337         }
338
339         spin_unlock(&cache->uc_lock);
340         if (cache->uc_ops->parse_downcall)
341                 rc = cache->uc_ops->parse_downcall(cache, entry, args);
342         spin_lock(&cache->uc_lock);
343         if (rc)
344                 GOTO(out, rc);
345
346         entry->ue_expire = cfs_time_shift(cache->uc_entry_expire);
347         UC_CACHE_SET_VALID(entry);
348         CDEBUG(D_OTHER, "%s: created upcall cache entry %p for key "LPU64"\n",
349                cache->uc_name, entry, entry->ue_key);
350 out:
351         if (rc) {
352                 UC_CACHE_SET_INVALID(entry);
353                 list_del_init(&entry->ue_hash);
354         }
355         UC_CACHE_CLEAR_ACQUIRING(entry);
356         spin_unlock(&cache->uc_lock);
357         wake_up_all(&entry->ue_waitq);
358         put_entry(cache, entry);
359
360         RETURN(rc);
361 }
362 EXPORT_SYMBOL(upcall_cache_downcall);
363
364 void upcall_cache_flush(struct upcall_cache *cache, int force)
365 {
366         struct upcall_cache_entry *entry, *next;
367         int i;
368         ENTRY;
369
370         spin_lock(&cache->uc_lock);
371         for (i = 0; i < UC_CACHE_HASH_SIZE; i++) {
372                 list_for_each_entry_safe(entry, next,
373                                          &cache->uc_hashtable[i], ue_hash) {
374                         if (!force && atomic_read(&entry->ue_refcount)) {
375                                 UC_CACHE_SET_EXPIRED(entry);
376                                 continue;
377                         }
378                         LASSERT(!atomic_read(&entry->ue_refcount));
379                         free_entry(cache, entry);
380                 }
381         }
382         spin_unlock(&cache->uc_lock);
383         EXIT;
384 }
385 EXPORT_SYMBOL(upcall_cache_flush);
386
387 void upcall_cache_flush_one(struct upcall_cache *cache, __u64 key, void *args)
388 {
389         struct list_head *head;
390         struct upcall_cache_entry *entry;
391         int found = 0;
392         ENTRY;
393
394         head = &cache->uc_hashtable[UC_CACHE_HASH_INDEX(key)];
395
396         spin_lock(&cache->uc_lock);
397         list_for_each_entry(entry, head, ue_hash) {
398                 if (upcall_compare(cache, entry, key, args) == 0) {
399                         found = 1;
400                         break;
401                 }
402         }
403
404         if (found) {
405                 CWARN("%s: flush entry %p: key "LPU64", ref %d, fl %x, "
406                       "cur %lu, ex %ld/%ld\n",
407                       cache->uc_name, entry, entry->ue_key,
408                       atomic_read(&entry->ue_refcount), entry->ue_flags,
409                       cfs_time_current_sec(), entry->ue_acquire_expire,
410                       entry->ue_expire);
411                 UC_CACHE_SET_EXPIRED(entry);
412                 if (!atomic_read(&entry->ue_refcount))
413                         free_entry(cache, entry);
414         }
415         spin_unlock(&cache->uc_lock);
416 }
417 EXPORT_SYMBOL(upcall_cache_flush_one);
418
419 struct upcall_cache *upcall_cache_init(const char *name, const char *upcall,
420                                        struct upcall_cache_ops *ops)
421 {
422         struct upcall_cache *cache;
423         int i;
424         ENTRY;
425
426         LIBCFS_ALLOC(cache, sizeof(*cache));
427         if (!cache)
428                 RETURN(ERR_PTR(-ENOMEM));
429
430         spin_lock_init(&cache->uc_lock);
431         init_rwsem(&cache->uc_upcall_rwsem);
432         for (i = 0; i < UC_CACHE_HASH_SIZE; i++)
433                 INIT_LIST_HEAD(&cache->uc_hashtable[i]);
434         strlcpy(cache->uc_name, name, sizeof(cache->uc_name));
435         /* upcall pathname proc tunable */
436         strlcpy(cache->uc_upcall, upcall, sizeof(cache->uc_upcall));
437         cache->uc_entry_expire = 20 * 60;
438         cache->uc_acquire_expire = 30;
439         cache->uc_ops = ops;
440
441         RETURN(cache);
442 }
443 EXPORT_SYMBOL(upcall_cache_init);
444
445 void upcall_cache_cleanup(struct upcall_cache *cache)
446 {
447         if (!cache)
448                 return;
449         upcall_cache_flush_all(cache);
450         LIBCFS_FREE(cache, sizeof(*cache));
451 }
452 EXPORT_SYMBOL(upcall_cache_cleanup);