Whamcloud - gitweb
c04d948a6c5108e4eee12b13371c5d795d57c3e5
[fs/lustre-release.git] / lustre / ldlm / ldlm_resource.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
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/ldlm/ldlm_resource.c
37  *
38  * Author: Phil Schwan <phil@clusterfs.com>
39  * Author: Peter Braam <braam@clusterfs.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_LDLM
43 #ifdef __KERNEL__
44 # include <lustre_dlm.h>
45 #else
46 # include <liblustre.h>
47 #endif
48
49 #include <obd_class.h>
50 #include "ldlm_internal.h"
51
52 cfs_mem_cache_t *ldlm_resource_slab, *ldlm_lock_slab;
53
54 atomic_t ldlm_srv_namespace_nr = ATOMIC_INIT(0);
55 atomic_t ldlm_cli_namespace_nr = ATOMIC_INIT(0);
56
57 struct semaphore ldlm_srv_namespace_lock;
58 CFS_LIST_HEAD(ldlm_srv_namespace_list);
59
60 struct semaphore ldlm_cli_namespace_lock;
61 CFS_LIST_HEAD(ldlm_cli_namespace_list);
62
63 cfs_proc_dir_entry_t *ldlm_type_proc_dir = NULL;
64 cfs_proc_dir_entry_t *ldlm_ns_proc_dir = NULL;
65 cfs_proc_dir_entry_t *ldlm_svc_proc_dir = NULL;
66
67 #ifdef LPROCFS
68 static int ldlm_proc_dump_ns(struct file *file, const char *buffer,
69                              unsigned long count, void *data)
70 {
71         ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
72         ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
73         RETURN(count);
74 }
75
76 int ldlm_proc_setup(void)
77 {
78         int rc;
79         struct lprocfs_vars list[] = {
80                 { "dump_namespaces", NULL, ldlm_proc_dump_ns, NULL },
81                 { NULL }};
82         ENTRY;
83         LASSERT(ldlm_ns_proc_dir == NULL);
84
85         ldlm_type_proc_dir = lprocfs_register(OBD_LDLM_DEVICENAME,
86                                               proc_lustre_root,
87                                               NULL, NULL);
88         if (IS_ERR(ldlm_type_proc_dir)) {
89                 CERROR("LProcFS failed in ldlm-init\n");
90                 rc = PTR_ERR(ldlm_type_proc_dir);
91                 GOTO(err, rc);
92         }
93
94         ldlm_ns_proc_dir = lprocfs_register("namespaces",
95                                             ldlm_type_proc_dir,
96                                             NULL, NULL);
97         if (IS_ERR(ldlm_ns_proc_dir)) {
98                 CERROR("LProcFS failed in ldlm-init\n");
99                 rc = PTR_ERR(ldlm_ns_proc_dir);
100                 GOTO(err_type, rc);
101         }
102
103         ldlm_svc_proc_dir = lprocfs_register("services",
104                                             ldlm_type_proc_dir,
105                                             NULL, NULL);
106         if (IS_ERR(ldlm_svc_proc_dir)) {
107                 CERROR("LProcFS failed in ldlm-init\n");
108                 rc = PTR_ERR(ldlm_svc_proc_dir);
109                 GOTO(err_ns, rc);
110         }
111
112         rc = lprocfs_add_vars(ldlm_type_proc_dir, list, NULL);
113
114         RETURN(0);
115
116 err_ns:
117         lprocfs_remove(&ldlm_ns_proc_dir);
118 err_type:
119         lprocfs_remove(&ldlm_type_proc_dir);
120 err:
121         ldlm_svc_proc_dir = NULL;
122         RETURN(rc);
123 }
124
125 void ldlm_proc_cleanup(void)
126 {
127         if (ldlm_svc_proc_dir)
128                 lprocfs_remove(&ldlm_svc_proc_dir);
129
130         if (ldlm_ns_proc_dir)
131                 lprocfs_remove(&ldlm_ns_proc_dir);
132
133         if (ldlm_type_proc_dir)
134                 lprocfs_remove(&ldlm_type_proc_dir);
135 }
136
137 static int lprocfs_rd_lru_size(char *page, char **start, off_t off,
138                                int count, int *eof, void *data)
139 {
140         struct ldlm_namespace *ns = data;
141         __u32 *nr = &ns->ns_max_unused;
142
143         if (ns_connect_lru_resize(ns))
144                 nr = &ns->ns_nr_unused;
145         return lprocfs_rd_uint(page, start, off, count, eof, nr);
146 }
147
148 static int lprocfs_wr_lru_size(struct file *file, const char *buffer,
149                                unsigned long count, void *data)
150 {
151         struct ldlm_namespace *ns = data;
152         char dummy[MAX_STRING_SIZE + 1], *end;
153         unsigned long tmp;
154         int lru_resize;
155
156         dummy[MAX_STRING_SIZE] = '\0';
157         if (copy_from_user(dummy, buffer, MAX_STRING_SIZE))
158                 return -EFAULT;
159
160         if (strncmp(dummy, "clear", 5) == 0) {
161                 CDEBUG(D_DLMTRACE,
162                        "dropping all unused locks from namespace %s\n",
163                        ns->ns_name);
164                 if (ns_connect_lru_resize(ns)) {
165                         int canceled, unused  = ns->ns_nr_unused;
166
167                         /* Try to cancel all @ns_nr_unused locks. */
168                         canceled = ldlm_cancel_lru(ns, unused, LDLM_SYNC,
169                                                    LDLM_CANCEL_PASSED);
170                         if (canceled < unused) {
171                                 CERROR("not all requested locks are canceled, "
172                                        "requested: %d, canceled: %d\n", unused,
173                                        canceled);
174                                 return -EINVAL;
175                         }
176                 } else {
177                         tmp = ns->ns_max_unused;
178                         ns->ns_max_unused = 0;
179                         ldlm_cancel_lru(ns, 0, LDLM_SYNC, LDLM_CANCEL_PASSED);
180                         ns->ns_max_unused = tmp;
181                 }
182                 return count;
183         }
184
185         tmp = simple_strtoul(dummy, &end, 0);
186         if (dummy == end) {
187                 CERROR("invalid value written\n");
188                 return -EINVAL;
189         }
190         lru_resize = (tmp == 0);
191
192         if (ns_connect_lru_resize(ns)) {
193                 if (!lru_resize)
194                         ns->ns_max_unused = (unsigned int)tmp;
195
196                 if (tmp > ns->ns_nr_unused)
197                         tmp = ns->ns_nr_unused;
198                 tmp = ns->ns_nr_unused - tmp;
199
200                 CDEBUG(D_DLMTRACE,
201                        "changing namespace %s unused locks from %u to %u\n",
202                        ns->ns_name, ns->ns_nr_unused, (unsigned int)tmp);
203                 ldlm_cancel_lru(ns, tmp, LDLM_ASYNC, LDLM_CANCEL_PASSED);
204
205                 if (!lru_resize) {
206                         CDEBUG(D_DLMTRACE,
207                                "disable lru_resize for namespace %s\n",
208                                ns->ns_name);
209                         ns->ns_connect_flags &= ~OBD_CONNECT_LRU_RESIZE;
210                 }
211         } else {
212                 CDEBUG(D_DLMTRACE,
213                        "changing namespace %s max_unused from %u to %u\n",
214                        ns->ns_name, ns->ns_max_unused, (unsigned int)tmp);
215                 ns->ns_max_unused = (unsigned int)tmp;
216                 ldlm_cancel_lru(ns, 0, LDLM_ASYNC, LDLM_CANCEL_PASSED);
217
218                 /* Make sure that originally lru resize was supported before
219                  * turning it on here. */
220                 if (lru_resize &&
221                     (ns->ns_orig_connect_flags & OBD_CONNECT_LRU_RESIZE)) {
222                         CDEBUG(D_DLMTRACE,
223                                "enable lru_resize for namespace %s\n",
224                                ns->ns_name);
225                         ns->ns_connect_flags |= OBD_CONNECT_LRU_RESIZE;
226                 }
227         }
228
229         return count;
230 }
231
232 void ldlm_proc_namespace(struct ldlm_namespace *ns)
233 {
234         struct lprocfs_vars lock_vars[2];
235         char lock_name[MAX_STRING_SIZE + 1];
236
237         LASSERT(ns != NULL);
238         LASSERT(ns->ns_name != NULL);
239
240         lock_name[MAX_STRING_SIZE] = '\0';
241
242         memset(lock_vars, 0, sizeof(lock_vars));
243         lock_vars[0].name = lock_name;
244
245         snprintf(lock_name, MAX_STRING_SIZE, "%s/resource_count", ns->ns_name);
246         lock_vars[0].data = &ns->ns_refcount;
247         lock_vars[0].read_fptr = lprocfs_rd_atomic;
248         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
249
250         snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_count", ns->ns_name);
251         lock_vars[0].data = &ns->ns_locks;
252         lock_vars[0].read_fptr = lprocfs_rd_atomic;
253         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
254
255         if (ns_is_client(ns)) {
256                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_unused_count",
257                          ns->ns_name);
258                 lock_vars[0].data = &ns->ns_nr_unused;
259                 lock_vars[0].read_fptr = lprocfs_rd_uint;
260                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
261
262                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_size",
263                          ns->ns_name);
264                 lock_vars[0].data = ns;
265                 lock_vars[0].read_fptr = lprocfs_rd_lru_size;
266                 lock_vars[0].write_fptr = lprocfs_wr_lru_size;
267                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
268
269                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_max_age",
270                          ns->ns_name);
271                 lock_vars[0].data = &ns->ns_max_age;
272                 lock_vars[0].read_fptr = lprocfs_rd_uint;
273                 lock_vars[0].write_fptr = lprocfs_wr_uint;
274                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
275         } else {
276                 snprintf(lock_name, MAX_STRING_SIZE, "%s/ctime_age_limit",
277                          ns->ns_name);
278                 lock_vars[0].data = &ns->ns_ctime_age_limit;
279                 lock_vars[0].read_fptr = lprocfs_rd_uint;
280                 lock_vars[0].write_fptr = lprocfs_wr_uint;
281                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
282
283                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_timeouts",
284                          ns->ns_name);
285                 lock_vars[0].data = &ns->ns_timeouts;
286                 lock_vars[0].read_fptr = lprocfs_rd_uint;
287                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
288
289                 snprintf(lock_name, MAX_STRING_SIZE, "%s/max_nolock_bytes",
290                          ns->ns_name);
291                 lock_vars[0].data = &ns->ns_max_nolock_size;
292                 lock_vars[0].read_fptr = lprocfs_rd_uint;
293                 lock_vars[0].write_fptr = lprocfs_wr_uint;
294                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
295
296                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contention_seconds",
297                          ns->ns_name);
298                 lock_vars[0].data = &ns->ns_contention_time;
299                 lock_vars[0].read_fptr = lprocfs_rd_uint;
300                 lock_vars[0].write_fptr = lprocfs_wr_uint;
301                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
302
303                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contended_locks",
304                          ns->ns_name);
305                 lock_vars[0].data = &ns->ns_contended_locks;
306                 lock_vars[0].read_fptr = lprocfs_rd_uint;
307                 lock_vars[0].write_fptr = lprocfs_wr_uint;
308                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
309         }
310 }
311 #undef MAX_STRING_SIZE
312 #else
313 #define ldlm_proc_namespace(ns) do {} while (0)
314 #endif /* LPROCFS */
315
316 struct ldlm_namespace *ldlm_namespace_new(struct obd_device *obd, char *name,
317                                           ldlm_side_t client, ldlm_appetite_t apt)
318 {
319         struct ldlm_namespace *ns = NULL;
320         struct list_head *bucket;
321         int rc, idx, namelen;
322         ENTRY;
323
324         rc = ldlm_get_ref();
325         if (rc) {
326                 CERROR("ldlm_get_ref failed: %d\n", rc);
327                 RETURN(NULL);
328         }
329
330         OBD_ALLOC_PTR(ns);
331         if (!ns)
332                 GOTO(out_ref, NULL);
333
334         OBD_VMALLOC(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
335         if (!ns->ns_hash)
336                 GOTO(out_ns, NULL);
337
338         ns->ns_appetite = apt;
339
340         LASSERT(obd != NULL);
341         ns->ns_obd = obd;
342
343         namelen = strlen(name);
344         OBD_ALLOC(ns->ns_name, namelen + 1);
345         if (!ns->ns_name)
346                 GOTO(out_hash, NULL);
347
348         strcpy(ns->ns_name, name);
349
350         CFS_INIT_LIST_HEAD(&ns->ns_root_list);
351         CFS_INIT_LIST_HEAD(&ns->ns_list_chain);
352         ns->ns_refcount = 0;
353         ns->ns_client = client;
354         spin_lock_init(&ns->ns_hash_lock);
355         atomic_set(&ns->ns_locks, 0);
356         ns->ns_resources = 0;
357         cfs_waitq_init(&ns->ns_waitq);
358         ns->ns_max_nolock_size = NS_DEFAULT_MAX_NOLOCK_BYTES;
359         ns->ns_contention_time = NS_DEFAULT_CONTENTION_SECONDS;
360         ns->ns_contended_locks = NS_DEFAULT_CONTENDED_LOCKS;
361
362         for (bucket = ns->ns_hash + RES_HASH_SIZE - 1; bucket >= ns->ns_hash;
363              bucket--)
364                 CFS_INIT_LIST_HEAD(bucket);
365
366         CFS_INIT_LIST_HEAD(&ns->ns_unused_list);
367         ns->ns_nr_unused = 0;
368         ns->ns_max_unused = LDLM_DEFAULT_LRU_SIZE;
369         ns->ns_max_age = LDLM_DEFAULT_MAX_ALIVE;
370         ns->ns_ctime_age_limit = LDLM_CTIME_AGE_LIMIT;
371         ns->ns_timeouts = 0;
372         spin_lock_init(&ns->ns_unused_lock);
373         ns->ns_orig_connect_flags = 0;
374         ns->ns_connect_flags = 0;
375         ldlm_proc_namespace(ns);
376
377         idx = atomic_read(ldlm_namespace_nr(client));
378         rc = ldlm_pool_init(&ns->ns_pool, ns, idx, client);
379         if (rc) {
380                 CERROR("Can't initialize lock pool, rc %d\n", rc);
381                 GOTO(out_proc, rc);
382         }
383
384         at_init(&ns->ns_at_estimate, ldlm_enqueue_min, 0);
385
386         ldlm_namespace_register(ns, client);
387         RETURN(ns);
388 out_proc:
389         ldlm_namespace_cleanup(ns, 0);
390         OBD_FREE(ns->ns_name, namelen + 1);
391 out_hash:
392         OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
393 out_ns:
394         OBD_FREE_PTR(ns);
395 out_ref:
396         ldlm_put_ref();
397         RETURN(NULL);
398 }
399
400 extern struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock);
401
402 /* If flags contains FL_LOCAL_ONLY, don't try to tell the server, just cleanup.
403  * This is currently only used for recovery, and we make certain assumptions
404  * as a result--notably, that we shouldn't cancel locks with refs. -phil
405  *
406  * Called with the ns_lock held. */
407 static void cleanup_resource(struct ldlm_resource *res, struct list_head *q,
408                              int flags)
409 {
410         struct list_head *tmp;
411         int rc = 0, client = ns_is_client(res->lr_namespace);
412         int local_only = (flags & LDLM_FL_LOCAL_ONLY);
413         ENTRY;
414
415
416         do {
417                 struct ldlm_lock *lock = NULL;
418
419                 /* first, we look for non-cleaned-yet lock
420                  * all cleaned locks are marked by CLEANED flag */
421                 lock_res(res);
422                 list_for_each(tmp, q) {
423                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
424                         if (lock->l_flags & LDLM_FL_CLEANED) {
425                                 lock = NULL;
426                                 continue;
427                         }
428                         LDLM_LOCK_GET(lock);
429                         lock->l_flags |= LDLM_FL_CLEANED;
430                         break;
431                 }
432
433                 if (lock == NULL) {
434                         unlock_res(res);
435                         break;
436                 }
437
438                 /* Set CBPENDING so nothing in the cancellation path
439                  * can match this lock */
440                 lock->l_flags |= LDLM_FL_CBPENDING;
441                 lock->l_flags |= LDLM_FL_FAILED;
442                 lock->l_flags |= flags;
443
444                 /* ... without sending a CANCEL message for local_only. */
445                 if (local_only)
446                         lock->l_flags |= LDLM_FL_LOCAL_ONLY;
447
448                 if (local_only && (lock->l_readers || lock->l_writers)) {
449                         /* This is a little bit gross, but much better than the
450                          * alternative: pretend that we got a blocking AST from
451                          * the server, so that when the lock is decref'd, it
452                          * will go away ... */
453                         unlock_res(res);
454                         LDLM_DEBUG(lock, "setting FL_LOCAL_ONLY");
455                         if (lock->l_completion_ast)
456                                 lock->l_completion_ast(lock, 0, NULL);
457                         LDLM_LOCK_RELEASE(lock);
458                         continue;
459                 }
460
461                 if (client) {
462                         struct lustre_handle lockh;
463
464                         unlock_res(res);
465                         ldlm_lock2handle(lock, &lockh);
466                         rc = ldlm_cli_cancel(&lockh);
467                         if (rc)
468                                 CERROR("ldlm_cli_cancel: %d\n", rc);
469                 } else {
470                         ldlm_resource_unlink_lock(lock);
471                         unlock_res(res);
472                         LDLM_DEBUG(lock, "Freeing a lock still held by a "
473                                    "client node");
474                         ldlm_lock_destroy(lock);
475                 }
476                 LDLM_LOCK_RELEASE(lock);
477         } while (1);
478
479         EXIT;
480 }
481
482 int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags)
483 {
484         struct list_head *tmp;
485         int i;
486
487         if (ns == NULL) {
488                 CDEBUG(D_INFO, "NULL ns, skipping cleanup\n");
489                 return ELDLM_OK;
490         }
491
492         for (i = 0; i < RES_HASH_SIZE; i++) {
493                 spin_lock(&ns->ns_hash_lock);
494                 tmp = ns->ns_hash[i].next;
495                 while (tmp != &(ns->ns_hash[i])) {
496                         struct ldlm_resource *res;
497                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
498                         ldlm_resource_getref(res);
499                         spin_unlock(&ns->ns_hash_lock);
500                         LDLM_RESOURCE_ADDREF(res);
501
502                         cleanup_resource(res, &res->lr_granted, flags);
503                         cleanup_resource(res, &res->lr_converting, flags);
504                         cleanup_resource(res, &res->lr_waiting, flags);
505
506                         spin_lock(&ns->ns_hash_lock);
507                         tmp  = tmp->next;
508
509                         /* XXX: former stuff caused issues in case of race
510                          * between ldlm_namespace_cleanup() and lockd() when
511                          * client gets blocking ast when lock gets distracted by
512                          * server. This is 1_4 branch solution, let's see how
513                          * will it behave. */
514                         LDLM_RESOURCE_DELREF(res);
515                         if (!ldlm_resource_putref_locked(res))
516                                 CDEBUG(D_INFO,
517                                        "Namespace %s resource refcount nonzero "
518                                        "(%d) after lock cleanup; forcing cleanup.\n",
519                                        ns->ns_name, atomic_read(&res->lr_refcount));
520                 }
521                 spin_unlock(&ns->ns_hash_lock);
522         }
523
524         return ELDLM_OK;
525 }
526
527 static int __ldlm_namespace_free(struct ldlm_namespace *ns, int force)
528 {
529         ENTRY;
530
531         /* At shutdown time, don't call the cancellation callback */
532         ldlm_namespace_cleanup(ns, force ? LDLM_FL_LOCAL_ONLY : 0);
533
534         if (ns->ns_refcount > 0) {
535                 struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
536                 int rc;
537                 CDEBUG(D_DLMTRACE,
538                        "dlm namespace %s free waiting on refcount %d\n",
539                        ns->ns_name, ns->ns_refcount);
540 force_wait:
541                 if (force)
542                         lwi = LWI_TIMEOUT(obd_timeout * HZ / 4, NULL, NULL);
543
544                 rc = l_wait_event(ns->ns_waitq,
545                                   ns->ns_refcount == 0, &lwi);
546
547                 /* Forced cleanups should be able to reclaim all references,
548                  * so it's safe to wait forever... we can't leak locks... */
549                 if (force && rc == -ETIMEDOUT) {
550                         LCONSOLE_ERROR("Forced cleanup waiting for %s "
551                                        "namespace with %d resources in use, "
552                                        "(rc=%d)\n", ns->ns_name,
553                                        ns->ns_refcount, rc);
554                         GOTO(force_wait, rc);
555                 }
556
557                 if (ns->ns_refcount) {
558                         LCONSOLE_ERROR("Cleanup waiting for %s namespace "
559                                        "with %d resources in use, (rc=%d)\n",
560                                        ns->ns_name,
561                                        ns->ns_refcount, rc);
562                         RETURN(ELDLM_NAMESPACE_EXISTS);
563                 }
564                 CDEBUG(D_DLMTRACE,
565                        "dlm namespace %s free done waiting\n", ns->ns_name);
566         }
567
568         RETURN(ELDLM_OK);
569 }
570
571 /**
572  * Performs various cleanups for passed \a ns to make it drop refc and be ready
573  * for freeing. Waits for refc == 0.
574  *
575  * The following is done:
576  * (0) Unregister \a ns from its list to make inaccessible for potential users
577  * like pools thread and others;
578  * (1) Clear all locks in \a ns.
579  */
580 void ldlm_namespace_free_prior(struct ldlm_namespace *ns,
581                                struct obd_import *imp,
582                                int force)
583 {
584         int rc;
585         ENTRY;
586         if (!ns) {
587                 EXIT;
588                 return;
589         }
590
591         /*
592          * Make sure that nobody can find this ns in its list.
593          */
594         ldlm_namespace_unregister(ns, ns->ns_client);
595
596         /*
597          * Can fail with -EINTR when force == 0 in which case try harder.
598          */
599         rc = __ldlm_namespace_free(ns, force);
600         if (rc != ELDLM_OK) {
601                 if (imp) {
602                         ptlrpc_disconnect_import(imp, 0);
603                         ptlrpc_invalidate_import(imp);
604                 }
605
606                 /*
607                  * With all requests dropped and the import inactive
608                  * we are gaurenteed all reference will be dropped.
609                  */
610                 rc = __ldlm_namespace_free(ns, 1);
611                 LASSERT(rc == 0);
612         }
613         EXIT;
614 }
615
616 /**
617  * Performs freeing memory structures related to \a ns. This is only done when
618  * ldlm_namespce_free_prior() successfully removed all resources referencing
619  * \a ns and its refc == 0.
620  */
621 void ldlm_namespace_free_post(struct ldlm_namespace *ns)
622 {
623         ENTRY;
624         if (!ns) {
625                 EXIT;
626                 return;
627         }
628
629         /*
630          * Fini pool _before_ parent proc dir is removed. This is important as
631          * ldlm_pool_fini() removes own proc dir which is child to @dir. Removing
632          * it after @dir may cause oops.
633          */
634         ldlm_pool_fini(&ns->ns_pool);
635
636 #ifdef LPROCFS
637         {
638                 struct proc_dir_entry *dir;
639                 dir = lprocfs_srch(ldlm_ns_proc_dir, ns->ns_name);
640                 if (dir == NULL) {
641                         CERROR("dlm namespace %s has no procfs dir?\n",
642                                ns->ns_name);
643                 } else {
644                         lprocfs_remove(&dir);
645                 }
646         }
647 #endif
648
649         OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
650         OBD_FREE(ns->ns_name, strlen(ns->ns_name) + 1);
651
652         /*
653          * Namespace \a ns should be not on list in this time, otherwise this
654          * will cause issues realted to using freed \a ns in pools thread.
655          */
656         LASSERT(list_empty(&ns->ns_list_chain));
657         OBD_FREE_PTR(ns);
658         ldlm_put_ref();
659         EXIT;
660 }
661
662
663 /* Cleanup the resource, and free namespace.
664  * bug 12864:
665  * Deadlock issue:
666  * proc1: destroy import
667  *        class_disconnect_export(grab cl_sem) ->
668  *              -> ldlm_namespace_free ->
669  *              -> lprocfs_remove(grab _lprocfs_lock).
670  * proc2: read proc info
671  *        lprocfs_fops_read(grab _lprocfs_lock) ->
672  *              -> osc_rd_active, etc(grab cl_sem).
673  *
674  * So that I have to split the ldlm_namespace_free into two parts - the first
675  * part ldlm_namespace_free_prior is used to cleanup the resource which is
676  * being used; the 2nd part ldlm_namespace_free_post is used to unregister the
677  * lprocfs entries, and then free memory. It will be called w/o cli->cl_sem
678  * held.
679  */
680 void ldlm_namespace_free(struct ldlm_namespace *ns,
681                          struct obd_import *imp,
682                          int force)
683 {
684         ldlm_namespace_free_prior(ns, imp, force);
685         ldlm_namespace_free_post(ns);
686 }
687
688
689 void ldlm_namespace_get_locked(struct ldlm_namespace *ns)
690 {
691         LASSERT(ns->ns_refcount >= 0);
692         ns->ns_refcount++;
693 }
694
695 void ldlm_namespace_get(struct ldlm_namespace *ns)
696 {
697         spin_lock(&ns->ns_hash_lock);
698         ldlm_namespace_get_locked(ns);
699         spin_unlock(&ns->ns_hash_lock);
700 }
701
702 void ldlm_namespace_put_locked(struct ldlm_namespace *ns, int wakeup)
703 {
704         LASSERT(ns->ns_refcount > 0);
705         ns->ns_refcount--;
706         if (ns->ns_refcount == 0 && wakeup)
707                 wake_up(&ns->ns_waitq);
708 }
709
710 void ldlm_namespace_put(struct ldlm_namespace *ns, int wakeup)
711 {
712         spin_lock(&ns->ns_hash_lock);
713         ldlm_namespace_put_locked(ns, wakeup);
714         spin_unlock(&ns->ns_hash_lock);
715 }
716
717 /* Register @ns in the list of namespaces */
718 void ldlm_namespace_register(struct ldlm_namespace *ns, ldlm_side_t client)
719 {
720         mutex_down(ldlm_namespace_lock(client));
721         LASSERT(list_empty(&ns->ns_list_chain));
722         list_add(&ns->ns_list_chain, ldlm_namespace_list(client));
723         atomic_inc(ldlm_namespace_nr(client));
724         mutex_up(ldlm_namespace_lock(client));
725 }
726
727 /* Unregister @ns from the list of namespaces */
728 void ldlm_namespace_unregister(struct ldlm_namespace *ns, ldlm_side_t client)
729 {
730         mutex_down(ldlm_namespace_lock(client));
731         LASSERT(!list_empty(&ns->ns_list_chain));
732         /*
733          * Some asserts and possibly other parts of code still using
734          * list_empty(&ns->ns_list_chain). This is why it is important
735          * to use list_del_init() here.
736          */
737         list_del_init(&ns->ns_list_chain);
738         atomic_dec(ldlm_namespace_nr(client));
739         mutex_up(ldlm_namespace_lock(client));
740 }
741
742 /* Should be called under ldlm_namespace_lock(client) taken */
743 void ldlm_namespace_move_locked(struct ldlm_namespace *ns, ldlm_side_t client)
744 {
745         LASSERT(!list_empty(&ns->ns_list_chain));
746         LASSERT_SEM_LOCKED(ldlm_namespace_lock(client));
747         list_move_tail(&ns->ns_list_chain, ldlm_namespace_list(client));
748 }
749
750 /* Should be called under ldlm_namespace_lock(client) taken */
751 struct ldlm_namespace *ldlm_namespace_first_locked(ldlm_side_t client)
752 {
753         LASSERT_SEM_LOCKED(ldlm_namespace_lock(client));
754         LASSERT(!list_empty(ldlm_namespace_list(client)));
755         return container_of(ldlm_namespace_list(client)->next,
756                 struct ldlm_namespace, ns_list_chain);
757 }
758 static __u32 ldlm_hash_fn(struct ldlm_resource *parent,
759                           const struct ldlm_res_id *name)
760 {
761         __u32 hash = 0;
762         int i;
763
764         for (i = 0; i < RES_NAME_SIZE; i++)
765                 hash += name->name[i];
766
767         hash += (__u32)((unsigned long)parent >> 4);
768
769         return (hash & RES_HASH_MASK);
770 }
771
772 static struct ldlm_resource *ldlm_resource_new(void)
773 {
774         struct ldlm_resource *res;
775         int idx;
776
777         OBD_SLAB_ALLOC(res, ldlm_resource_slab, CFS_ALLOC_IO, sizeof *res);
778         if (res == NULL)
779                 return NULL;
780
781         memset(res, 0, sizeof(*res));
782
783         CFS_INIT_LIST_HEAD(&res->lr_children);
784         CFS_INIT_LIST_HEAD(&res->lr_childof);
785         CFS_INIT_LIST_HEAD(&res->lr_granted);
786         CFS_INIT_LIST_HEAD(&res->lr_converting);
787         CFS_INIT_LIST_HEAD(&res->lr_waiting);
788
789         /* initialize interval trees for each lock mode*/
790         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
791                 res->lr_itree[idx].lit_size = 0;
792                 res->lr_itree[idx].lit_mode = 1 << idx;
793                 res->lr_itree[idx].lit_root = NULL;
794         }
795
796         atomic_set(&res->lr_refcount, 1);
797         spin_lock_init(&res->lr_lock);
798         lu_ref_init(&res->lr_reference);
799
800         /* one who creates the resource must unlock
801          * the semaphore after lvb initialization */
802         init_MUTEX_LOCKED(&res->lr_lvb_sem);
803
804         return res;
805 }
806
807 /* must be called with hash lock held */
808 static struct ldlm_resource *
809 ldlm_resource_find(struct ldlm_namespace *ns, const struct ldlm_res_id *name,
810                    __u32 hash)
811 {
812         struct list_head *bucket, *tmp;
813         struct ldlm_resource *res;
814
815         LASSERT_SPIN_LOCKED(&ns->ns_hash_lock);
816         bucket = ns->ns_hash + hash;
817
818         list_for_each(tmp, bucket) {
819                 res = list_entry(tmp, struct ldlm_resource, lr_hash);
820                 if (memcmp(&res->lr_name, name, sizeof(res->lr_name)) == 0)
821                         return res;
822         }
823
824         return NULL;
825 }
826
827 /* Args: locked namespace
828  * Returns: newly-allocated, referenced, unlocked resource */
829 static struct ldlm_resource *
830 ldlm_resource_add(struct ldlm_namespace *ns, struct ldlm_resource *parent,
831                   const struct ldlm_res_id *name, __u32 hash, ldlm_type_t type)
832 {
833         struct list_head *bucket;
834         struct ldlm_resource *res, *old_res;
835         ENTRY;
836
837         LASSERTF(type >= LDLM_MIN_TYPE && type < LDLM_MAX_TYPE,
838                  "type: %d\n", type);
839
840         res = ldlm_resource_new();
841         if (!res)
842                 RETURN(NULL);
843
844         res->lr_name = *name;
845         res->lr_namespace = ns;
846         res->lr_type = type;
847         res->lr_most_restr = LCK_NL;
848
849         spin_lock(&ns->ns_hash_lock);
850         old_res = ldlm_resource_find(ns, name, hash);
851         if (old_res) {
852                 /* someone won the race and added the resource before */
853                 ldlm_resource_getref(old_res);
854                 spin_unlock(&ns->ns_hash_lock);
855                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
856                 /* synchronize WRT resource creation */
857                 if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
858                         down(&old_res->lr_lvb_sem);
859                         up(&old_res->lr_lvb_sem);
860                 }
861                 RETURN(old_res);
862         }
863
864         /* we won! let's add the resource */
865         bucket = ns->ns_hash + hash;
866         list_add(&res->lr_hash, bucket);
867         ns->ns_resources++;
868         ldlm_namespace_get_locked(ns);
869
870         if (parent == NULL) {
871                 list_add(&res->lr_childof, &ns->ns_root_list);
872         } else {
873                 res->lr_parent = parent;
874                 list_add(&res->lr_childof, &parent->lr_children);
875         }
876         spin_unlock(&ns->ns_hash_lock);
877
878         if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
879                 int rc;
880
881                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CREATE_RESOURCE, 2);
882                 rc = ns->ns_lvbo->lvbo_init(res);
883                 if (rc)
884                         CERROR("lvbo_init failed for resource "
885                                LPU64": rc %d\n", name->name[0], rc);
886                 /* we create resource with locked lr_lvb_sem */
887                 up(&res->lr_lvb_sem);
888         }
889
890         RETURN(res);
891 }
892
893 /* Args: unlocked namespace
894  * Locks: takes and releases ns->ns_lock and res->lr_lock
895  * Returns: referenced, unlocked ldlm_resource or NULL */
896 struct ldlm_resource *
897 ldlm_resource_get(struct ldlm_namespace *ns, struct ldlm_resource *parent,
898                   const struct ldlm_res_id *name, ldlm_type_t type, int create)
899 {
900         __u32 hash = ldlm_hash_fn(parent, name);
901         struct ldlm_resource *res = NULL;
902         ENTRY;
903
904         LASSERT(ns != NULL);
905         LASSERT(ns->ns_hash != NULL);
906         LASSERT(name->name[0] != 0);
907
908         spin_lock(&ns->ns_hash_lock);
909         res = ldlm_resource_find(ns, name, hash);
910         if (res) {
911                 ldlm_resource_getref(res);
912                 spin_unlock(&ns->ns_hash_lock);
913                 /* synchronize WRT resource creation */
914                 if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
915                         down(&res->lr_lvb_sem);
916                         up(&res->lr_lvb_sem);
917                 }
918                 RETURN(res);
919         }
920         spin_unlock(&ns->ns_hash_lock);
921
922         if (create == 0)
923                 RETURN(NULL);
924
925         res = ldlm_resource_add(ns, parent, name, hash, type);
926         RETURN(res);
927 }
928
929 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
930 {
931         LASSERT(res != NULL);
932         LASSERT(res != LP_POISON);
933         atomic_inc(&res->lr_refcount);
934         CDEBUG(D_INFO, "getref res: %p count: %d\n", res,
935                atomic_read(&res->lr_refcount));
936         return res;
937 }
938
939 void __ldlm_resource_putref_final(struct ldlm_resource *res)
940 {
941         struct ldlm_namespace *ns = res->lr_namespace;
942
943         LASSERT_SPIN_LOCKED(&ns->ns_hash_lock);
944
945         if (!list_empty(&res->lr_granted)) {
946                 ldlm_resource_dump(D_ERROR, res);
947                 LBUG();
948         }
949
950         if (!list_empty(&res->lr_converting)) {
951                 ldlm_resource_dump(D_ERROR, res);
952                 LBUG();
953         }
954
955         if (!list_empty(&res->lr_waiting)) {
956                 ldlm_resource_dump(D_ERROR, res);
957                 LBUG();
958         }
959
960         if (!list_empty(&res->lr_children)) {
961                 ldlm_resource_dump(D_ERROR, res);
962                 LBUG();
963         }
964
965         /* Pass 0 here to not wake ->ns_waitq up yet, we will do it few
966          * lines below when all children are freed. */
967         ldlm_namespace_put_locked(ns, 0);
968         list_del_init(&res->lr_hash);
969         list_del_init(&res->lr_childof);
970         lu_ref_fini(&res->lr_reference);
971
972         ns->ns_resources--;
973         if (ns->ns_resources == 0)
974                 wake_up(&ns->ns_waitq);
975 }
976
977 /* Returns 1 if the resource was freed, 0 if it remains. */
978 int ldlm_resource_putref(struct ldlm_resource *res)
979 {
980         struct ldlm_namespace *ns = res->lr_namespace;
981         int rc = 0;
982         ENTRY;
983
984         CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
985                atomic_read(&res->lr_refcount) - 1);
986         LASSERTF(atomic_read(&res->lr_refcount) > 0, "%d",
987                  atomic_read(&res->lr_refcount));
988         LASSERTF(atomic_read(&res->lr_refcount) < LI_POISON, "%d",
989                  atomic_read(&res->lr_refcount));
990
991         if (atomic_dec_and_lock(&res->lr_refcount, &ns->ns_hash_lock)) {
992                 __ldlm_resource_putref_final(res);
993                 spin_unlock(&ns->ns_hash_lock);
994                 if (res->lr_lvb_data)
995                         OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
996                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
997                 rc = 1;
998         }
999
1000         RETURN(rc);
1001 }
1002
1003 /* Returns 1 if the resource was freed, 0 if it remains. */
1004 int ldlm_resource_putref_locked(struct ldlm_resource *res)
1005 {
1006         int rc = 0;
1007         ENTRY;
1008
1009         CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
1010                atomic_read(&res->lr_refcount) - 1);
1011         LASSERT(atomic_read(&res->lr_refcount) > 0);
1012         LASSERT(atomic_read(&res->lr_refcount) < LI_POISON);
1013
1014         LASSERT(atomic_read(&res->lr_refcount) >= 0);
1015         if (atomic_dec_and_test(&res->lr_refcount)) {
1016                 __ldlm_resource_putref_final(res);
1017                 if (res->lr_lvb_data)
1018                         OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
1019                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
1020                 rc = 1;
1021         }
1022
1023         RETURN(rc);
1024 }
1025
1026 void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
1027                             struct ldlm_lock *lock)
1028 {
1029         check_res_locked(res);
1030
1031         ldlm_resource_dump(D_INFO, res);
1032         CDEBUG(D_OTHER, "About to add this lock:\n");
1033         ldlm_lock_dump(D_OTHER, lock, 0);
1034
1035         if (lock->l_destroyed) {
1036                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
1037                 return;
1038         }
1039
1040         LASSERT(list_empty(&lock->l_res_link));
1041
1042         list_add_tail(&lock->l_res_link, head);
1043 }
1044
1045 void ldlm_resource_insert_lock_after(struct ldlm_lock *original,
1046                                      struct ldlm_lock *new)
1047 {
1048         struct ldlm_resource *res = original->l_resource;
1049
1050         check_res_locked(res);
1051
1052         ldlm_resource_dump(D_INFO, res);
1053         CDEBUG(D_OTHER, "About to insert this lock after %p:\n", original);
1054         ldlm_lock_dump(D_OTHER, new, 0);
1055
1056         if (new->l_destroyed) {
1057                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
1058                 goto out;
1059         }
1060
1061         LASSERT(list_empty(&new->l_res_link));
1062
1063         list_add(&new->l_res_link, &original->l_res_link);
1064  out:;
1065 }
1066
1067 void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
1068 {
1069         int type = lock->l_resource->lr_type;
1070
1071         check_res_locked(lock->l_resource);
1072         if (type == LDLM_IBITS || type == LDLM_PLAIN)
1073                 ldlm_unlink_lock_skiplist(lock);
1074         else if (type == LDLM_EXTENT)
1075                 ldlm_extent_unlink_lock(lock);
1076         list_del_init(&lock->l_res_link);
1077 }
1078
1079 void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
1080 {
1081         desc->lr_type = res->lr_type;
1082         desc->lr_name = res->lr_name;
1083 }
1084
1085 void ldlm_dump_all_namespaces(ldlm_side_t client, int level)
1086 {
1087         struct list_head *tmp;
1088
1089         if (!((libcfs_debug | D_ERROR) & level))
1090                 return;
1091
1092         mutex_down(ldlm_namespace_lock(client));
1093
1094         list_for_each(tmp, ldlm_namespace_list(client)) {
1095                 struct ldlm_namespace *ns;
1096                 ns = list_entry(tmp, struct ldlm_namespace, ns_list_chain);
1097                 ldlm_namespace_dump(level, ns);
1098         }
1099
1100         mutex_up(ldlm_namespace_lock(client));
1101 }
1102
1103 void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
1104 {
1105         struct list_head *tmp;
1106
1107         if (!((libcfs_debug | D_ERROR) & level))
1108                 return;
1109
1110         CDEBUG(level, "--- Namespace: %s (rc: %d, side: %s)\n",
1111                ns->ns_name, ns->ns_refcount,
1112                ns_is_client(ns) ? "client" : "server");
1113
1114         if (cfs_time_before(cfs_time_current(), ns->ns_next_dump))
1115                 return;
1116
1117         spin_lock(&ns->ns_hash_lock);
1118         tmp = ns->ns_root_list.next;
1119         while (tmp != &ns->ns_root_list) {
1120                 struct ldlm_resource *res;
1121                 res = list_entry(tmp, struct ldlm_resource, lr_childof);
1122
1123                 ldlm_resource_getref(res);
1124                 spin_unlock(&ns->ns_hash_lock);
1125                 LDLM_RESOURCE_ADDREF(res);
1126
1127                 lock_res(res);
1128                 ldlm_resource_dump(level, res);
1129                 unlock_res(res);
1130
1131                 LDLM_RESOURCE_DELREF(res);
1132                 spin_lock(&ns->ns_hash_lock);
1133                 tmp = tmp->next;
1134                 ldlm_resource_putref_locked(res);
1135         }
1136         ns->ns_next_dump = cfs_time_shift(10);
1137         spin_unlock(&ns->ns_hash_lock);
1138 }
1139
1140 void ldlm_resource_dump(int level, struct ldlm_resource *res)
1141 {
1142         struct list_head *tmp;
1143         int pos;
1144
1145         CLASSERT(RES_NAME_SIZE == 4);
1146
1147         if (!((libcfs_debug | D_ERROR) & level))
1148                 return;
1149
1150         CDEBUG(level, "--- Resource: %p ("LPU64"/"LPU64"/"LPU64"/"LPU64
1151                ") (rc: %d)\n", res, res->lr_name.name[0], res->lr_name.name[1],
1152                res->lr_name.name[2], res->lr_name.name[3],
1153                atomic_read(&res->lr_refcount));
1154
1155         if (!list_empty(&res->lr_granted)) {
1156                 pos = 0;
1157                 CDEBUG(level, "Granted locks:\n");
1158                 list_for_each(tmp, &res->lr_granted) {
1159                         struct ldlm_lock *lock;
1160                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1161                         ldlm_lock_dump(level, lock, ++pos);
1162                 }
1163         }
1164         if (!list_empty(&res->lr_converting)) {
1165                 pos = 0;
1166                 CDEBUG(level, "Converting locks:\n");
1167                 list_for_each(tmp, &res->lr_converting) {
1168                         struct ldlm_lock *lock;
1169                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1170                         ldlm_lock_dump(level, lock, ++pos);
1171                 }
1172         }
1173         if (!list_empty(&res->lr_waiting)) {
1174                 pos = 0;
1175                 CDEBUG(level, "Waiting locks:\n");
1176                 list_for_each(tmp, &res->lr_waiting) {
1177                         struct ldlm_lock *lock;
1178                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
1179                         ldlm_lock_dump(level, lock, ++pos);
1180                 }
1181         }
1182 }