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  * Copyright (C) 2002, 2003 Cluster File Systems, Inc.
   5  *   Author: Phil Schwan <phil@clusterfs.com>
   6  *   Author: Peter Braam <braam@clusterfs.com>
   7  *
   8  *   This file is part of Lustre, http://www.lustre.org.
   9  *
  10  *   Lustre is free software; you can redistribute it and/or
  11  *   modify it under the terms of version 2 of the GNU General Public
  12  *   License as published by the Free Software Foundation.
  13  *
  14  *   Lustre is distributed in the hope that it will be useful,
  15  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  *   GNU General Public License for more details.
  18  *
  19  *   You should have received a copy of the GNU General Public License
  20  *   along with Lustre; if not, write to the Free Software
  21  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  22  */
  23
  24 #define DEBUG_SUBSYSTEM S_LDLM
  25 #ifdef __KERNEL__
  26 # include <linux/lustre_dlm.h>
  27 #else
  28 # include <liblustre.h>
  29 #endif
  30
  31 #include <linux/obd_class.h>
  32 #include "ldlm_internal.h"
  33
  34 kmem_cache_t *ldlm_resource_slab, *ldlm_lock_slab;
  35
  36 DECLARE_MUTEX(ldlm_namespace_lock);
  37 struct list_head ldlm_namespace_list = LIST_HEAD_INIT(ldlm_namespace_list);
  38 struct proc_dir_entry *ldlm_type_proc_dir = NULL;
  39 struct proc_dir_entry *ldlm_ns_proc_dir = NULL;
  40 struct proc_dir_entry *ldlm_svc_proc_dir = NULL;
  41
  42 #ifdef __KERNEL__
  43 static int ldlm_proc_dump_ns(struct file *file, const char *buffer,
  44                              unsigned long count, void *data)
  45 {
  46         ldlm_dump_all_namespaces(D_DLMTRACE);
  47         RETURN(count);
  48 }
  49
  50 int ldlm_proc_setup(void)
  51 {
  52         int rc;
  53         struct lprocfs_vars list[] = {
  54                 { "dump_namespaces", NULL, ldlm_proc_dump_ns, NULL },
  55                 { NULL }};
  56         ENTRY;
  57         LASSERT(ldlm_ns_proc_dir == NULL);
  58
  59         ldlm_type_proc_dir = lprocfs_register(OBD_LDLM_DEVICENAME,
  60                                               proc_lustre_root,
  61                                               NULL, NULL);
  62         if (IS_ERR(ldlm_type_proc_dir)) {
  63                 CERROR("LProcFS failed in ldlm-init\n");
  64                 rc = PTR_ERR(ldlm_type_proc_dir);
  65                 GOTO(err, rc);
  66         }
  67
  68         ldlm_ns_proc_dir = lprocfs_register("namespaces",
  69                                             ldlm_type_proc_dir,
  70                                             NULL, NULL);
  71         if (IS_ERR(ldlm_ns_proc_dir)) {
  72                 CERROR("LProcFS failed in ldlm-init\n");
  73                 rc = PTR_ERR(ldlm_ns_proc_dir);
  74                 GOTO(err_type, rc);
  75         }
  76
  77         ldlm_svc_proc_dir = lprocfs_register("services",
  78                                             ldlm_type_proc_dir,
  79                                             NULL, NULL);
  80         if (IS_ERR(ldlm_svc_proc_dir)) {
  81                 CERROR("LProcFS failed in ldlm-init\n");
  82                 rc = PTR_ERR(ldlm_svc_proc_dir);
  83                 GOTO(err_ns, rc);
  84         }
  85
  86         rc = lprocfs_add_vars(ldlm_type_proc_dir, list, NULL);
  87
  88         RETURN(0);
  89
  90 err_ns:
  91         lprocfs_remove(ldlm_ns_proc_dir);
  92 err_type:
  93         lprocfs_remove(ldlm_type_proc_dir);
  94 err:
  95         ldlm_type_proc_dir = NULL;
  96         ldlm_ns_proc_dir = NULL;
  97         ldlm_svc_proc_dir = NULL;
  98         RETURN(rc);
  99 }
 100
 101 void ldlm_proc_cleanup(void)
 102 {
 103         if (ldlm_svc_proc_dir) {
 104                 lprocfs_remove(ldlm_svc_proc_dir);
 105                 ldlm_svc_proc_dir = NULL;
 106         }
 107
 108         if (ldlm_ns_proc_dir) {
 109                 lprocfs_remove(ldlm_ns_proc_dir);
 110                 ldlm_ns_proc_dir = NULL;
 111         }
 112
 113         if (ldlm_type_proc_dir) {
 114                 lprocfs_remove(ldlm_type_proc_dir);
 115                 ldlm_type_proc_dir = NULL;
 116         }
 117 }
 118
 119 static int lprocfs_uint_rd(char *page, char **start, off_t off,
 120                            int count, int *eof, void *data)
 121 {
 122         unsigned int *temp = (unsigned int *)data;
 123         return snprintf(page, count, "%u\n", *temp);
 124 }
 125
 126 static int lprocfs_read_lru_size(char *page, char **start, off_t off,
 127                                  int count, int *eof, void *data)
 128 {
 129         struct ldlm_namespace *ns = data;
 130         return lprocfs_uint_rd(page, start, off, count, eof,
 131                                &ns->ns_max_unused);
 132 }
 133
 134 #define MAX_STRING_SIZE 128
 135 static int lprocfs_write_lru_size(struct file *file, const char *buffer,
 136                                   unsigned long count, void *data)
 137 {
 138         struct ldlm_namespace *ns = data;
 139         char dummy[MAX_STRING_SIZE + 1], *end;
 140         unsigned long tmp;
 141
 142         dummy[MAX_STRING_SIZE] = '\0';
 143         copy_from_user(dummy, buffer, MAX_STRING_SIZE);
 144
 145         if (count == 6 && memcmp(dummy, "clear", 5) == 0) {
 146                 CDEBUG(D_DLMTRACE,
 147                        "dropping all unused locks from namespace %s\n",
 148                        ns->ns_name);
 149                 tmp = ns->ns_max_unused;
 150                 ns->ns_max_unused = 0;
 151                 ldlm_cancel_lru(ns, LDLM_SYNC);
 152                 ns->ns_max_unused = tmp;
 153                 return count;
 154         }
 155
 156         tmp = simple_strtoul(dummy, &end, 0);
 157         if (tmp == 0 && *end) {
 158                 CERROR("invalid value written\n");
 159                 return -EINVAL;
 160         }
 161
 162         CDEBUG(D_DLMTRACE, "changing namespace %s max_unused from %u to %u\n",
 163                ns->ns_name, ns->ns_max_unused, (unsigned int)tmp);
 164
 165         ns->ns_max_unused = (unsigned int)tmp;
 166         ldlm_cancel_lru(ns, LDLM_ASYNC);
 167         return count;
 168 }
 169
 170 void ldlm_proc_namespace(struct ldlm_namespace *ns)
 171 {
 172         struct lprocfs_vars lock_vars[2];
 173         char lock_name[MAX_STRING_SIZE + 1];
 174
 175         LASSERT(ns != NULL);
 176         LASSERT(ns->ns_name != NULL);
 177
 178         lock_name[MAX_STRING_SIZE] = '\0';
 179
 180         memset(lock_vars, 0, sizeof(lock_vars));
 181         lock_vars[0].read_fptr = lprocfs_rd_u64;
 182         lock_vars[0].name = lock_name;
 183
 184         snprintf(lock_name, MAX_STRING_SIZE, "%s/resource_count", ns->ns_name);
 185         lock_vars[0].data = &ns->ns_resources;
 186         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 187
 188         snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_count", ns->ns_name);
 189         lock_vars[0].data = &ns->ns_locks;
 190         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 191
 192         if (ns->ns_client) {
 193                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_unused_count",
 194                          ns->ns_name);
 195                 lock_vars[0].data = &ns->ns_nr_unused;
 196                 lock_vars[0].read_fptr = lprocfs_uint_rd;
 197                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 198
 199                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_size",
 200                          ns->ns_name);
 201                 lock_vars[0].data = ns;
 202                 lock_vars[0].read_fptr = lprocfs_read_lru_size;
 203                 lock_vars[0].write_fptr = lprocfs_write_lru_size;
 204                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 205         }
 206 }
 207 #endif
 208 #undef MAX_STRING_SIZE
 209
 210 struct ldlm_namespace *ldlm_namespace_new(char *name, __u32 client)
 211 {
 212         struct ldlm_namespace *ns = NULL;
 213         struct list_head *bucket;
 214         int rc;
 215         ENTRY;
 216
 217         rc = ldlm_get_ref();
 218         if (rc) {
 219                 CERROR("ldlm_get_ref failed: %d\n", rc);
 220                 RETURN(NULL);
 221         }
 222
 223         OBD_ALLOC(ns, sizeof(*ns));
 224         if (!ns)
 225                 GOTO(out_ref, NULL);
 226
 227         OBD_VMALLOC(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
 228         if (!ns->ns_hash)
 229                 GOTO(out_ns, NULL);
 230
 231         OBD_ALLOC(ns->ns_name, strlen(name) + 1);
 232         if (!ns->ns_name)
 233                 GOTO(out_hash, NULL);
 234
 235         strcpy(ns->ns_name, name);
 236
 237         INIT_LIST_HEAD(&ns->ns_root_list);
 238         l_lock_init(&ns->ns_lock);
 239         ns->ns_refcount = 0;
 240         ns->ns_client = client;
 241         spin_lock_init(&ns->ns_counter_lock);
 242         ns->ns_locks = 0;
 243         ns->ns_resources = 0;
 244         init_waitqueue_head(&ns->ns_waitq);
 245
 246         for (bucket = ns->ns_hash + RES_HASH_SIZE - 1; bucket >= ns->ns_hash;
 247              bucket--)
 248                 INIT_LIST_HEAD(bucket);
 249
 250         INIT_LIST_HEAD(&ns->ns_unused_list);
 251         ns->ns_nr_unused = 0;
 252         ns->ns_max_unused = LDLM_DEFAULT_LRU_SIZE;
 253
 254         down(&ldlm_namespace_lock);
 255         list_add(&ns->ns_list_chain, &ldlm_namespace_list);
 256         up(&ldlm_namespace_lock);
 257 #ifdef __KERNEL__
 258         ldlm_proc_namespace(ns);
 259 #endif
 260         RETURN(ns);
 261
 262 out_hash:
 263         POISON(ns->ns_hash, 0x5a, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
 264         OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
 265 out_ns:
 266         OBD_FREE(ns, sizeof(*ns));
 267 out_ref:
 268         ldlm_put_ref(0);
 269         RETURN(NULL);
 270 }
 271
 272 extern struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock);
 273
 274 /* If flags contains FL_LOCAL_ONLY, don't try to tell the server, just cleanup.
 275  * This is currently only used for recovery, and we make certain assumptions
 276  * as a result--notably, that we shouldn't cancel locks with refs. -phil
 277  *
 278  * Called with the ns_lock held. */
 279 static void cleanup_resource(struct ldlm_resource *res, struct list_head *q,
 280                              int flags)
 281 {
 282         struct list_head *tmp, *pos;
 283         int rc = 0, client = res->lr_namespace->ns_client;
 284         int local_only = (flags & LDLM_FL_LOCAL_ONLY);
 285         ENTRY;
 286
 287         list_for_each_safe(tmp, pos, q) {
 288                 struct ldlm_lock *lock;
 289                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 290                 LDLM_LOCK_GET(lock);
 291
 292                 /* Set CBPENDING so nothing in the cancellation path
 293                  * can match this lock */
 294                 lock->l_flags |= LDLM_FL_CBPENDING;
 295                 lock->l_flags |= LDLM_FL_FAILED;
 296                 lock->l_flags |= flags;
 297
 298                 if (local_only && (lock->l_readers || lock->l_writers)) {
 299                         /* This is a little bit gross, but much better than the
 300                          * alternative: pretend that we got a blocking AST from
 301                          * the server, so that when the lock is decref'd, it
 302                          * will go away ... */
 303                         /* ... without sending a CANCEL message. */
 304                         lock->l_flags |= LDLM_FL_LOCAL_ONLY;
 305                         LDLM_DEBUG(lock, "setting FL_LOCAL_ONLY");
 306                         if (lock->l_completion_ast)
 307                                 lock->l_completion_ast(lock, 0, NULL);
 308                         LDLM_LOCK_PUT(lock);
 309                         continue;
 310                 }
 311
 312                 if (client) {
 313                         struct lustre_handle lockh;
 314                         ldlm_lock2handle(lock, &lockh);
 315                         if (!local_only) {
 316                                 rc = ldlm_cli_cancel(&lockh);
 317                                 if (rc)
 318                                         CERROR("ldlm_cli_cancel: %d\n", rc);
 319                         }
 320                         /* Force local cleanup on errors, too. */
 321                         if (local_only || rc != ELDLM_OK)
 322                                 ldlm_lock_cancel(lock);
 323                 } else {
 324                         LDLM_DEBUG(lock, "Freeing a lock still held by a "
 325                                    "client node");
 326
 327                         ldlm_resource_unlink_lock(lock);
 328                         ldlm_lock_destroy(lock);
 329                 }
 330                 LDLM_LOCK_PUT(lock);
 331         }
 332         EXIT;
 333 }
 334
 335 int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags)
 336 {
 337         int i;
 338
 339         if (ns == NULL) {
 340                 CDEBUG(D_INFO, "NULL ns, skipping cleanup\n");
 341                 return ELDLM_OK;
 342         }
 343
 344         l_lock(&ns->ns_lock);
 345         for (i = 0; i < RES_HASH_SIZE; i++) {
 346                 struct list_head *tmp, *pos;
 347                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
 348                         struct ldlm_resource *res;
 349                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
 350                         ldlm_resource_getref(res);
 351
 352                         cleanup_resource(res, &res->lr_granted, flags);
 353                         cleanup_resource(res, &res->lr_converting, flags);
 354                         cleanup_resource(res, &res->lr_waiting, flags);
 355
 356                         /* XXX what a mess: don't force cleanup if we're
 357                          * local_only (which is only used by recovery).  In that
 358                          * case, we probably still have outstanding lock refs
 359                          * which reference these resources. -phil */
 360                         if (!ldlm_resource_putref(res) &&
 361                             !(flags & LDLM_FL_LOCAL_ONLY)) {
 362                                 CERROR("Resource refcount nonzero (%d) after "
 363                                        "lock cleanup; forcing cleanup.\n",
 364                                        atomic_read(&res->lr_refcount));
 365                                 ldlm_resource_dump(D_ERROR, res);
 366                                 atomic_set(&res->lr_refcount, 1);
 367                                 ldlm_resource_putref(res);
 368                         }
 369                 }
 370         }
 371         l_unlock(&ns->ns_lock);
 372
 373         return ELDLM_OK;
 374 }
 375
 376 /* Cleanup, but also free, the namespace */
 377 int ldlm_namespace_free(struct ldlm_namespace *ns, int force)
 378 {
 379         if (!ns)
 380                 RETURN(ELDLM_OK);
 381
 382         down(&ldlm_namespace_lock);
 383         list_del(&ns->ns_list_chain);
 384         up(&ldlm_namespace_lock);
 385
 386         /* At shutdown time, don't call the cancellation callback */
 387         ldlm_namespace_cleanup(ns, 0);
 388
 389 #ifdef __KERNEL__
 390         {
 391                 struct proc_dir_entry *dir;
 392                 dir = lprocfs_srch(ldlm_ns_proc_dir, ns->ns_name);
 393                 if (dir == NULL) {
 394                         CERROR("dlm namespace %s has no procfs dir?\n",
 395                                ns->ns_name);
 396                 } else {
 397                         lprocfs_remove(dir);
 398                 }
 399         }
 400 #endif
 401
 402         POISON(ns->ns_hash, 0x5a, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
 403         OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
 404         OBD_FREE(ns->ns_name, strlen(ns->ns_name) + 1);
 405         OBD_FREE(ns, sizeof(*ns));
 406
 407         ldlm_put_ref(force);
 408
 409         return ELDLM_OK;
 410 }
 411
 412 static __u32 ldlm_hash_fn(struct ldlm_resource *parent, struct ldlm_res_id name)
 413 {
 414         __u32 hash = 0;
 415         int i;
 416
 417         for (i = 0; i < RES_NAME_SIZE; i++)
 418                 hash += name.name[i];
 419
 420         hash += (__u32)((unsigned long)parent >> 4);
 421
 422         return (hash & RES_HASH_MASK);
 423 }
 424
 425 static struct ldlm_resource *ldlm_resource_new(void)
 426 {
 427         struct ldlm_resource *res;
 428
 429         OBD_SLAB_ALLOC(res, ldlm_resource_slab, SLAB_NOFS, sizeof *res);
 430         if (res == NULL)
 431                 return NULL;
 432
 433         memset(res, 0, sizeof(*res));
 434
 435         INIT_LIST_HEAD(&res->lr_children);
 436         INIT_LIST_HEAD(&res->lr_childof);
 437         INIT_LIST_HEAD(&res->lr_granted);
 438         INIT_LIST_HEAD(&res->lr_converting);
 439         INIT_LIST_HEAD(&res->lr_waiting);
 440         sema_init(&res->lr_lvb_sem, 1);
 441         atomic_set(&res->lr_refcount, 1);
 442
 443         return res;
 444 }
 445
 446 /* Args: locked namespace
 447  * Returns: newly-allocated, referenced, unlocked resource */
 448 static struct ldlm_resource *
 449 ldlm_resource_add(struct ldlm_namespace *ns, struct ldlm_resource *parent,
 450                   struct ldlm_res_id name, __u32 type)
 451 {
 452         struct list_head *bucket;
 453         struct ldlm_resource *res;
 454         ENTRY;
 455
 456         LASSERTF(type >= LDLM_MIN_TYPE && type <= LDLM_MAX_TYPE,
 457                  "type: %d", type);
 458
 459         res = ldlm_resource_new();
 460         if (!res)
 461                 RETURN(NULL);
 462
 463         spin_lock(&ns->ns_counter_lock);
 464         ns->ns_resources++;
 465         spin_unlock(&ns->ns_counter_lock);
 466
 467         l_lock(&ns->ns_lock);
 468         memcpy(&res->lr_name, &name, sizeof(res->lr_name));
 469         res->lr_namespace = ns;
 470         ns->ns_refcount++;
 471
 472         res->lr_type = type;
 473         res->lr_most_restr = LCK_NL;
 474
 475         bucket = ns->ns_hash + ldlm_hash_fn(parent, name);
 476         list_add(&res->lr_hash, bucket);
 477
 478         if (parent == NULL) {
 479                 list_add(&res->lr_childof, &ns->ns_root_list);
 480         } else {
 481                 res->lr_parent = parent;
 482                 list_add(&res->lr_childof, &parent->lr_children);
 483         }
 484         l_unlock(&ns->ns_lock);
 485
 486
 487         RETURN(res);
 488 }
 489
 490 /* Args: unlocked namespace
 491  * Locks: takes and releases ns->ns_lock and res->lr_lock
 492  * Returns: referenced, unlocked ldlm_resource or NULL */
 493 struct ldlm_resource *
 494 ldlm_resource_get(struct ldlm_namespace *ns, struct ldlm_resource *parent,
 495                   struct ldlm_res_id name, __u32 type, int create)
 496 {
 497         struct list_head *bucket, *tmp;
 498         struct ldlm_resource *res = NULL;
 499         ENTRY;
 500
 501         LASSERT(ns != NULL);
 502         LASSERT(ns->ns_hash != NULL);
 503         LASSERT(name.name[0] != 0);
 504
 505         l_lock(&ns->ns_lock);
 506         bucket = ns->ns_hash + ldlm_hash_fn(parent, name);
 507
 508         list_for_each(tmp, bucket) {
 509                 res = list_entry(tmp, struct ldlm_resource, lr_hash);
 510
 511                 if (memcmp(&res->lr_name, &name, sizeof(res->lr_name)) == 0) {
 512                         ldlm_resource_getref(res);
 513                         l_unlock(&ns->ns_lock);
 514                         RETURN(res);
 515                 }
 516         }
 517
 518         if (create) {
 519                 res = ldlm_resource_add(ns, parent, name, type);
 520                 if (res == NULL)
 521                         GOTO(out, NULL);
 522         } else {
 523                 res = NULL;
 524         }
 525
 526         if (create && ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
 527                 int rc;
 528
 529                 /* Although this is technically a lock inversion risk (lvb_sem
 530                  * should be taken before DLM lock), this resource was just
 531                  * created, so nobody else can take the lvb_sem yet. -p */
 532                 down(&res->lr_lvb_sem);
 533                 /* Drop the dlm lock, because lvbo_init can touch the disk */
 534                 l_unlock(&ns->ns_lock);
 535                 OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CREATE_RESOURCE, 2);
 536                 rc = ns->ns_lvbo->lvbo_init(res);
 537                 up(&res->lr_lvb_sem);
 538                 if (rc)
 539                         CERROR("lvbo_init failed for resource "LPU64": rc %d\n",
 540                                name.name[0], rc);
 541         } else {
 542 out:
 543                 l_unlock(&ns->ns_lock);
 544         }
 545
 546         RETURN(res);
 547 }
 548
 549 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
 550 {
 551         LASSERT(res != NULL);
 552         LASSERT(res != LP_POISON);
 553         atomic_inc(&res->lr_refcount);
 554         CDEBUG(D_INFO, "getref res: %p count: %d\n", res,
 555                atomic_read(&res->lr_refcount));
 556         return res;
 557 }
 558
 559 /* Returns 1 if the resource was freed, 0 if it remains. */
 560 int ldlm_resource_putref(struct ldlm_resource *res)
 561 {
 562         int rc = 0;
 563         ENTRY;
 564
 565         CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
 566                atomic_read(&res->lr_refcount) - 1);
 567         LASSERT(atomic_read(&res->lr_refcount) > 0);
 568         LASSERT(atomic_read(&res->lr_refcount) < LI_POISON);
 569
 570         if (atomic_dec_and_test(&res->lr_refcount)) {
 571                 struct ldlm_namespace *ns = res->lr_namespace;
 572                 ENTRY;
 573
 574                 l_lock(&ns->ns_lock);
 575
 576                 if (atomic_read(&res->lr_refcount) != 0) {
 577                         /* We lost the race. */
 578                         l_unlock(&ns->ns_lock);
 579                         RETURN(rc);
 580                 }
 581
 582                 if (!list_empty(&res->lr_granted)) {
 583                         ldlm_resource_dump(D_ERROR, res);
 584                         LBUG();
 585                 }
 586
 587                 if (!list_empty(&res->lr_converting)) {
 588                         ldlm_resource_dump(D_ERROR, res);
 589                         LBUG();
 590                 }
 591
 592                 if (!list_empty(&res->lr_waiting)) {
 593                         ldlm_resource_dump(D_ERROR, res);
 594                         LBUG();
 595                 }
 596
 597                 if (!list_empty(&res->lr_children)) {
 598                         ldlm_resource_dump(D_ERROR, res);
 599                         LBUG();
 600                 }
 601
 602                 ns->ns_refcount--;
 603                 list_del_init(&res->lr_hash);
 604                 list_del_init(&res->lr_childof);
 605                 if (res->lr_lvb_data)
 606                         OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
 607                 l_unlock(&ns->ns_lock);
 608
 609                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
 610
 611                 spin_lock(&ns->ns_counter_lock);
 612                 ns->ns_resources--;
 613                 if (ns->ns_resources == 0)
 614                         wake_up(&ns->ns_waitq);
 615                 spin_unlock(&ns->ns_counter_lock);
 616
 617                 rc = 1;
 618                 EXIT;
 619         }
 620
 621         RETURN(rc);
 622 }
 623
 624 void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
 625                             struct ldlm_lock *lock)
 626 {
 627         l_lock(&res->lr_namespace->ns_lock);
 628
 629         ldlm_resource_dump(D_OTHER, res);
 630         CDEBUG(D_OTHER, "About to add this lock:\n");
 631         ldlm_lock_dump(D_OTHER, lock, 0);
 632
 633         if (lock->l_destroyed) {
 634                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
 635                 goto out;
 636         }
 637
 638         LASSERT(list_empty(&lock->l_res_link));
 639
 640         list_add_tail(&lock->l_res_link, head);
 641  out:
 642         l_unlock(&res->lr_namespace->ns_lock);
 643 }
 644
 645 void ldlm_resource_insert_lock_after(struct ldlm_lock *original,
 646                                      struct ldlm_lock *new)
 647 {
 648         struct ldlm_resource *res = original->l_resource;
 649
 650         l_lock(&res->lr_namespace->ns_lock);
 651
 652         ldlm_resource_dump(D_OTHER, res);
 653         CDEBUG(D_OTHER, "About to insert this lock after %p:\n", original);
 654         ldlm_lock_dump(D_OTHER, new, 0);
 655
 656         if (new->l_destroyed) {
 657                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
 658                 goto out;
 659         }
 660
 661         LASSERT(list_empty(&new->l_res_link));
 662
 663         list_add(&new->l_res_link, &original->l_res_link);
 664  out:
 665         l_unlock(&res->lr_namespace->ns_lock);
 666 }
 667
 668 void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
 669 {
 670         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 671         list_del_init(&lock->l_res_link);
 672         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 673 }
 674
 675 void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
 676 {
 677         desc->lr_type = res->lr_type;
 678         memcpy(&desc->lr_name, &res->lr_name, sizeof(desc->lr_name));
 679 }
 680
 681 void ldlm_dump_all_namespaces(int level)
 682 {
 683         struct list_head *tmp;
 684
 685         down(&ldlm_namespace_lock);
 686
 687         list_for_each(tmp, &ldlm_namespace_list) {
 688                 struct ldlm_namespace *ns;
 689                 ns = list_entry(tmp, struct ldlm_namespace, ns_list_chain);
 690                 ldlm_namespace_dump(level, ns);
 691         }
 692
 693         up(&ldlm_namespace_lock);
 694 }
 695
 696 void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
 697 {
 698         struct list_head *tmp;
 699
 700         CDEBUG(level, "--- Namespace: %s (rc: %d, client: %d)\n",
 701                ns->ns_name, ns->ns_refcount, ns->ns_client);
 702
 703         l_lock(&ns->ns_lock);
 704         if (time_after(jiffies, ns->ns_next_dump)) {
 705                 list_for_each(tmp, &ns->ns_root_list) {
 706                         struct ldlm_resource *res;
 707                         res = list_entry(tmp, struct ldlm_resource, lr_childof);
 708
 709                         /* Once we have resources with children, this should
 710                          * really dump them recursively. */
 711                         ldlm_resource_dump(level, res);
 712                 }
 713                 ns->ns_next_dump = jiffies + 10 * HZ;
 714         }
 715         l_unlock(&ns->ns_lock);
 716 }
 717
 718 void ldlm_resource_dump(int level, struct ldlm_resource *res)
 719 {
 720         struct list_head *tmp;
 721         int pos;
 722
 723         if (RES_NAME_SIZE != 4)
 724                 LBUG();
 725
 726         CDEBUG(level, "--- Resource: %p ("LPU64"/"LPU64"/"LPU64"/"LPU64
 727                ") (rc: %d)\n", res, res->lr_name.name[0], res->lr_name.name[1],
 728                res->lr_name.name[2], res->lr_name.name[3],
 729                atomic_read(&res->lr_refcount));
 730
 731         if (!list_empty(&res->lr_granted)) {
 732                 pos = 0;
 733                 CDEBUG(level, "Granted locks:\n");
 734                 list_for_each(tmp, &res->lr_granted) {
 735                         struct ldlm_lock *lock;
 736                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 737                         ldlm_lock_dump(level, lock, ++pos);
 738                 }
 739         }
 740         if (!list_empty(&res->lr_converting)) {
 741                 pos = 0;
 742                 CDEBUG(level, "Converting locks:\n");
 743                 list_for_each(tmp, &res->lr_converting) {
 744                         struct ldlm_lock *lock;
 745                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 746                         ldlm_lock_dump(level, lock, ++pos);
 747                 }
 748         }
 749         if (!list_empty(&res->lr_waiting)) {
 750                 pos = 0;
 751                 CDEBUG(level, "Waiting locks:\n");
 752                 list_for_each(tmp, &res->lr_waiting) {
 753                         struct ldlm_lock *lock;
 754                         lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 755                         ldlm_lock_dump(level, lock, ++pos);
 756                 }
 757         }
 758 }