lustre/ldlm/ldlm_resource.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2010, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/ldlm/ldlm_resource.c
  32  *
  33  * Author: Phil Schwan <phil@clusterfs.com>
  34  * Author: Peter Braam <braam@clusterfs.com>
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_LDLM
  38 #include <lustre_dlm.h>
  39 #include <lustre_fid.h>
  40 #include <obd_class.h>
  41 #include <libcfs/linux/linux-hash.h>
  42 #include "ldlm_internal.h"
  43
  44 struct kmem_cache *ldlm_resource_slab, *ldlm_lock_slab;
  45 struct kmem_cache *ldlm_interval_tree_slab;
  46 struct kmem_cache *ldlm_inodebits_slab;
  47
  48 int ldlm_srv_namespace_nr;
  49 int ldlm_cli_namespace_nr;
  50
  51 DEFINE_MUTEX(ldlm_srv_namespace_lock);
  52 LIST_HEAD(ldlm_srv_namespace_list);
  53
  54 DEFINE_MUTEX(ldlm_cli_namespace_lock);
  55 /* Client Namespaces that have active resources in them. Once all resources go
  56  * away, ldlm_poold moves such namespaces to the inactive list
  57  */
  58 LIST_HEAD(ldlm_cli_active_namespace_list);
  59 /* Client namespaces that don't have any locks in them */
  60 LIST_HEAD(ldlm_cli_inactive_namespace_list);
  61
  62 static struct dentry *ldlm_debugfs_dir;
  63 static struct dentry *ldlm_ns_debugfs_dir;
  64 struct dentry *ldlm_svc_debugfs_dir;
  65
  66 /* For debug dump, amount of granted locks for one resource to avoid DDOS. */
  67 static unsigned int ldlm_dump_granted_max = 256;
  68
  69 static ssize_t ldebugfs_dump_ns_seq_write(struct file *file,
  70                                           const char __user *buffer,
  71                                           size_t count, loff_t *off)
  72 {
  73         ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
  74         ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
  75         RETURN(count);
  76 }
  77
  78 LDEBUGFS_FOPS_WR_ONLY(ldlm, dump_ns);
  79
  80 static int ldlm_rw_uint_seq_show(struct seq_file *m, void *v)
  81 {
  82         seq_printf(m, "%u\n", *(unsigned int *)m->private);
  83         return 0;
  84 }
  85
  86 static ssize_t
  87 ldlm_rw_uint_seq_write(struct file *file, const char __user *buffer,
  88                        size_t count, loff_t *off)
  89 {
  90         struct seq_file *seq = file->private_data;
  91
  92         if (!count)
  93                 return 0;
  94
  95         return kstrtouint_from_user(buffer, count, 0,
  96                                     (unsigned int *)seq->private);
  97 }
  98
  99 LDEBUGFS_SEQ_FOPS(ldlm_rw_uint);
 100
 101 #ifdef HAVE_SERVER_SUPPORT
 102
 103 static int seq_watermark_show(struct seq_file *m, void *data)
 104 {
 105         seq_printf(m, "%llu\n", *(__u64 *)m->private);
 106         return 0;
 107 }
 108
 109 static ssize_t seq_watermark_write(struct file *file,
 110                                    const char __user *buffer, size_t count,
 111                                    loff_t *off)
 112 {
 113         struct seq_file *m = file->private_data;
 114         u64 value;
 115         __u64 watermark;
 116         __u64 *data = m->private;
 117         bool wm_low = (data == &ldlm_reclaim_threshold_mb) ? true : false;
 118         char kernbuf[22] = "";
 119         int rc;
 120
 121         if (count >= sizeof(kernbuf))
 122                 return -EINVAL;
 123
 124         if (copy_from_user(kernbuf, buffer, count))
 125                 return -EFAULT;
 126         kernbuf[count] = 0;
 127
 128         rc = sysfs_memparse(kernbuf, count, &value, "MiB");
 129         if (rc < 0) {
 130                 CERROR("Failed to set %s, rc = %d.\n",
 131                        wm_low ? "lock_reclaim_threshold_mb" : "lock_limit_mb",
 132                        rc);
 133                 return rc;
 134         } else if (value != 0 && value < (1 << 20)) {
 135                 CERROR("%s should be greater than 1MB.\n",
 136                        wm_low ? "lock_reclaim_threshold_mb" : "lock_limit_mb");
 137                 return -EINVAL;
 138         }
 139         watermark = value >> 20;
 140
 141         if (wm_low) {
 142                 if (ldlm_lock_limit_mb != 0 && watermark > ldlm_lock_limit_mb) {
 143                         CERROR("lock_reclaim_threshold_mb must be smaller than lock_limit_mb.\n");
 144                         return -EINVAL;
 145                 }
 146
 147                 *data = watermark;
 148                 if (watermark != 0) {
 149                         watermark <<= 20;
 150                         do_div(watermark, sizeof(struct ldlm_lock));
 151                 }
 152                 ldlm_reclaim_threshold = watermark;
 153         } else {
 154                 if (ldlm_reclaim_threshold_mb != 0 &&
 155                     watermark < ldlm_reclaim_threshold_mb) {
 156                         CERROR("lock_limit_mb must be greater than "
 157                                "lock_reclaim_threshold_mb.\n");
 158                         return -EINVAL;
 159                 }
 160
 161                 *data = watermark;
 162                 if (watermark != 0) {
 163                         watermark <<= 20;
 164                         do_div(watermark, sizeof(struct ldlm_lock));
 165                 }
 166                 ldlm_lock_limit = watermark;
 167         }
 168
 169         return count;
 170 }
 171
 172 static int seq_watermark_open(struct inode *inode, struct file *file)
 173 {
 174         return single_open(file, seq_watermark_show, inode->i_private);
 175 }
 176
 177 static const struct file_operations ldlm_watermark_fops = {
 178         .owner          = THIS_MODULE,
 179         .open           = seq_watermark_open,
 180         .read           = seq_read,
 181         .write          = seq_watermark_write,
 182         .llseek         = seq_lseek,
 183         .release        = lprocfs_single_release,
 184 };
 185
 186 static int seq_granted_show(struct seq_file *m, void *data)
 187 {
 188         seq_printf(m, "%llu\n", percpu_counter_sum_positive(
 189                    (struct percpu_counter *)m->private));
 190         return 0;
 191 }
 192
 193 static int seq_granted_open(struct inode *inode, struct file *file)
 194 {
 195         return single_open(file, seq_granted_show, inode->i_private);
 196 }
 197
 198 static const struct file_operations ldlm_granted_fops = {
 199         .owner  = THIS_MODULE,
 200         .open   = seq_granted_open,
 201         .read   = seq_read,
 202         .llseek = seq_lseek,
 203         .release = seq_release,
 204 };
 205
 206 #endif /* HAVE_SERVER_SUPPORT */
 207
 208 static struct ldebugfs_vars ldlm_debugfs_list[] = {
 209         { .name =       "dump_namespaces",
 210           .fops =       &ldlm_dump_ns_fops,
 211           .proc_mode =  0222 },
 212         { .name =       "dump_granted_max",
 213           .fops =       &ldlm_rw_uint_fops,
 214           .data =       &ldlm_dump_granted_max },
 215 #ifdef HAVE_SERVER_SUPPORT
 216         { .name =       "lock_reclaim_threshold_mb",
 217           .fops =       &ldlm_watermark_fops,
 218           .data =       &ldlm_reclaim_threshold_mb },
 219         { .name =       "lock_limit_mb",
 220           .fops =       &ldlm_watermark_fops,
 221           .data =       &ldlm_lock_limit_mb },
 222         { .name =       "lock_granted_count",
 223           .fops =       &ldlm_granted_fops,
 224           .data =       &ldlm_granted_total },
 225 #endif
 226         { NULL }
 227 };
 228
 229 int ldlm_debugfs_setup(void)
 230 {
 231         ENTRY;
 232         ldlm_debugfs_dir = debugfs_create_dir(OBD_LDLM_DEVICENAME,
 233                                              debugfs_lustre_root);
 234         ldlm_ns_debugfs_dir = debugfs_create_dir("namespaces",
 235                                                  ldlm_debugfs_dir);
 236         ldlm_svc_debugfs_dir = debugfs_create_dir("services",
 237                                                   ldlm_debugfs_dir);
 238
 239         ldebugfs_add_vars(ldlm_debugfs_dir, ldlm_debugfs_list, NULL);
 240
 241         RETURN(0);
 242 }
 243
 244 void ldlm_debugfs_cleanup(void)
 245 {
 246         debugfs_remove_recursive(ldlm_debugfs_dir);
 247
 248         ldlm_svc_debugfs_dir = NULL;
 249         ldlm_ns_debugfs_dir = NULL;
 250         ldlm_debugfs_dir = NULL;
 251 }
 252
 253 static ssize_t resource_count_show(struct kobject *kobj, struct attribute *attr,
 254                                    char *buf)
 255 {
 256         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 257                                                  ns_kobj);
 258         __u64                   res = 0;
 259         int                     i;
 260
 261         /* result is not strictly consistant */
 262         for (i = 0; i < (1 << ns->ns_bucket_bits); i++)
 263                 res += atomic_read(&ns->ns_rs_buckets[i].nsb_count);
 264         return sprintf(buf, "%lld\n", res);
 265 }
 266 LUSTRE_RO_ATTR(resource_count);
 267
 268 static ssize_t lock_count_show(struct kobject *kobj, struct attribute *attr,
 269                                char *buf)
 270 {
 271         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 272                                                  ns_kobj);
 273         __u64                   locks;
 274
 275         locks = lprocfs_stats_collector(ns->ns_stats, LDLM_NSS_LOCKS,
 276                                         LPROCFS_FIELDS_FLAGS_SUM);
 277         return sprintf(buf, "%lld\n", locks);
 278 }
 279 LUSTRE_RO_ATTR(lock_count);
 280
 281 static ssize_t lock_unused_count_show(struct kobject *kobj,
 282                                       struct attribute *attr,
 283                                       char *buf)
 284 {
 285         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 286                                                  ns_kobj);
 287
 288         return sprintf(buf, "%d\n", ns->ns_nr_unused);
 289 }
 290 LUSTRE_RO_ATTR(lock_unused_count);
 291
 292 static ssize_t lru_size_show(struct kobject *kobj, struct attribute *attr,
 293                              char *buf)
 294 {
 295         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 296                                                  ns_kobj);
 297         __u32 *nr = &ns->ns_max_unused;
 298
 299         if (ns_connect_lru_resize(ns))
 300                 nr = &ns->ns_nr_unused;
 301         return sprintf(buf, "%u\n", *nr);
 302 }
 303
 304 static ssize_t lru_size_store(struct kobject *kobj, struct attribute *attr,
 305                               const char *buffer, size_t count)
 306 {
 307         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 308                                                  ns_kobj);
 309         unsigned long tmp;
 310         int lru_resize;
 311         int err;
 312
 313         if (strncmp(buffer, "clear", 5) == 0) {
 314                 CDEBUG(D_DLMTRACE,
 315                        "dropping all unused locks from namespace %s\n",
 316                        ldlm_ns_name(ns));
 317                 /* Try to cancel all @ns_nr_unused locks. */
 318                 ldlm_cancel_lru(ns, INT_MAX, 0, LDLM_LRU_FLAG_CLEANUP);
 319                 return count;
 320         }
 321
 322         err = kstrtoul(buffer, 10, &tmp);
 323         if (err != 0) {
 324                 CERROR("lru_size: invalid value written\n");
 325                 return -EINVAL;
 326         }
 327         lru_resize = (tmp == 0);
 328
 329         if (ns_connect_lru_resize(ns)) {
 330                 if (!lru_resize)
 331                         ns->ns_max_unused = (unsigned int)tmp;
 332
 333                 if (tmp > ns->ns_nr_unused)
 334                         tmp = ns->ns_nr_unused;
 335                 tmp = ns->ns_nr_unused - tmp;
 336
 337                 CDEBUG(D_DLMTRACE,
 338                        "changing namespace %s unused locks from %u to %u\n",
 339                        ldlm_ns_name(ns), ns->ns_nr_unused,
 340                        (unsigned int)tmp);
 341
 342                 if (!lru_resize) {
 343                         CDEBUG(D_DLMTRACE,
 344                                "disable lru_resize for namespace %s\n",
 345                                ldlm_ns_name(ns));
 346                         ns->ns_connect_flags &= ~OBD_CONNECT_LRU_RESIZE;
 347                 }
 348                 ldlm_cancel_lru(ns, tmp, LCF_ASYNC, 0);
 349         } else {
 350                 CDEBUG(D_DLMTRACE,
 351                        "changing namespace %s max_unused from %u to %u\n",
 352                        ldlm_ns_name(ns), ns->ns_max_unused,
 353                        (unsigned int)tmp);
 354
 355                 /* Make sure that LRU resize was originally supported before
 356                  * turning it on here.
 357                  */
 358                 if (lru_resize &&
 359                     (ns->ns_orig_connect_flags & OBD_CONNECT_LRU_RESIZE)) {
 360                         CDEBUG(D_DLMTRACE,
 361                                "enable lru_resize for namespace %s\n",
 362                                ldlm_ns_name(ns));
 363                         ns->ns_connect_flags |= OBD_CONNECT_LRU_RESIZE;
 364                 }
 365                 ns->ns_max_unused = (unsigned int)tmp;
 366                 ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0);
 367         }
 368
 369         return count;
 370 }
 371 LUSTRE_RW_ATTR(lru_size);
 372
 373 static ssize_t lru_cancel_batch_show(struct kobject *kobj,
 374                                  struct attribute *attr, char *buf)
 375 {
 376         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 377                                                  ns_kobj);
 378
 379         return snprintf(buf, sizeof(buf) - 1, "%u\n", ns->ns_cancel_batch);
 380 }
 381
 382 static ssize_t lru_cancel_batch_store(struct kobject *kobj,
 383                                   struct attribute *attr,
 384                                   const char *buffer, size_t count)
 385 {
 386         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 387                                                  ns_kobj);
 388         unsigned long tmp;
 389
 390         if (kstrtoul(buffer, 10, &tmp))
 391                 return -EINVAL;
 392
 393         ns->ns_cancel_batch = (unsigned int)tmp;
 394
 395         return count;
 396 }
 397 LUSTRE_RW_ATTR(lru_cancel_batch);
 398
 399 static ssize_t ns_recalc_pct_show(struct kobject *kobj,
 400                                   struct attribute *attr, char *buf)
 401 {
 402         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 403                                                  ns_kobj);
 404
 405         return snprintf(buf, sizeof(buf) - 1, "%u\n", ns->ns_recalc_pct);
 406 }
 407
 408 static ssize_t ns_recalc_pct_store(struct kobject *kobj,
 409                                    struct attribute *attr,
 410                                    const char *buffer, size_t count)
 411 {
 412         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 413                                                  ns_kobj);
 414         unsigned long tmp;
 415
 416         if (kstrtoul(buffer, 10, &tmp))
 417                 return -EINVAL;
 418
 419         if (tmp > 100)
 420                 return -ERANGE;
 421
 422         ns->ns_recalc_pct = (unsigned int)tmp;
 423
 424         return count;
 425 }
 426 LUSTRE_RW_ATTR(ns_recalc_pct);
 427
 428 static ssize_t lru_max_age_show(struct kobject *kobj, struct attribute *attr,
 429                                 char *buf)
 430 {
 431         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 432                                                  ns_kobj);
 433
 434         return sprintf(buf, "%lld\n", ktime_to_ms(ns->ns_max_age));
 435 }
 436
 437 static ssize_t lru_max_age_store(struct kobject *kobj, struct attribute *attr,
 438                                  const char *buffer, size_t count)
 439 {
 440         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 441                                                  ns_kobj);
 442         int scale = NSEC_PER_MSEC;
 443         unsigned long long tmp;
 444         char *buf;
 445
 446         /* Did the user ask in seconds or milliseconds. Default is in ms */
 447         buf = strstr(buffer, "ms");
 448         if (!buf) {
 449                 buf = strchr(buffer, 's');
 450                 if (buf)
 451                         scale = NSEC_PER_SEC;
 452         }
 453
 454         if (buf)
 455                 *buf = '\0';
 456
 457         if (kstrtoull(buffer, 10, &tmp))
 458                 return -EINVAL;
 459
 460         ns->ns_max_age = ktime_set(0, tmp * scale);
 461
 462         return count;
 463 }
 464 LUSTRE_RW_ATTR(lru_max_age);
 465
 466 static ssize_t early_lock_cancel_show(struct kobject *kobj,
 467                                       struct attribute *attr,
 468                                       char *buf)
 469 {
 470         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 471                                                  ns_kobj);
 472
 473         return sprintf(buf, "%d\n", ns_connect_cancelset(ns));
 474 }
 475
 476 static ssize_t early_lock_cancel_store(struct kobject *kobj,
 477                                        struct attribute *attr,
 478                                        const char *buffer,
 479                                        size_t count)
 480 {
 481         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 482                                                  ns_kobj);
 483         unsigned long supp = -1;
 484         int rc;
 485
 486         rc = kstrtoul(buffer, 10, &supp);
 487         if (rc < 0)
 488                 return rc;
 489
 490         if (supp == 0)
 491                 ns->ns_connect_flags &= ~OBD_CONNECT_CANCELSET;
 492         else if (ns->ns_orig_connect_flags & OBD_CONNECT_CANCELSET)
 493                 ns->ns_connect_flags |= OBD_CONNECT_CANCELSET;
 494         return count;
 495 }
 496 LUSTRE_RW_ATTR(early_lock_cancel);
 497
 498 static ssize_t dirty_age_limit_show(struct kobject *kobj,
 499                                     struct attribute *attr, char *buf)
 500 {
 501         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 502                                                  ns_kobj);
 503
 504         return scnprintf(buf, PAGE_SIZE, "%llu\n",
 505                          ktime_divns(ns->ns_dirty_age_limit, NSEC_PER_SEC));
 506 }
 507
 508 static ssize_t dirty_age_limit_store(struct kobject *kobj,
 509                                      struct attribute *attr,
 510                                      const char *buffer, size_t count)
 511 {
 512         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 513                                                  ns_kobj);
 514         unsigned long long tmp;
 515
 516         if (kstrtoull(buffer, 10, &tmp))
 517                 return -EINVAL;
 518
 519         ns->ns_dirty_age_limit = ktime_set(tmp, 0);
 520
 521         return count;
 522 }
 523 LUSTRE_RW_ATTR(dirty_age_limit);
 524
 525 #ifdef HAVE_SERVER_SUPPORT
 526 static ssize_t ctime_age_limit_show(struct kobject *kobj,
 527                                     struct attribute *attr, char *buf)
 528 {
 529         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 530                                                  ns_kobj);
 531
 532         return scnprintf(buf, PAGE_SIZE, "%u\n", ns->ns_ctime_age_limit);
 533 }
 534
 535 static ssize_t ctime_age_limit_store(struct kobject *kobj,
 536                                      struct attribute *attr,
 537                                      const char *buffer, size_t count)
 538 {
 539         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 540                                                  ns_kobj);
 541         unsigned long tmp;
 542
 543         if (kstrtoul(buffer, 10, &tmp))
 544                 return -EINVAL;
 545
 546         ns->ns_ctime_age_limit = tmp;
 547
 548         return count;
 549 }
 550 LUSTRE_RW_ATTR(ctime_age_limit);
 551
 552 static ssize_t lock_timeouts_show(struct kobject *kobj, struct attribute *attr,
 553                                   char *buf)
 554 {
 555         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 556                                                  ns_kobj);
 557
 558         return sprintf(buf, "%d\n", ns->ns_timeouts);
 559 }
 560 LUSTRE_RO_ATTR(lock_timeouts);
 561
 562 static ssize_t max_nolock_bytes_show(struct kobject *kobj,
 563                                      struct attribute *attr, char *buf)
 564 {
 565         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 566                                                  ns_kobj);
 567
 568         return sprintf(buf, "%u\n", ns->ns_max_nolock_size);
 569 }
 570
 571 static ssize_t max_nolock_bytes_store(struct kobject *kobj,
 572                                       struct attribute *attr,
 573                                       const char *buffer, size_t count)
 574 {
 575         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 576                                                  ns_kobj);
 577         unsigned long tmp;
 578         int err;
 579
 580         err = kstrtoul(buffer, 10, &tmp);
 581         if (err != 0)
 582                 return -EINVAL;
 583
 584         ns->ns_max_nolock_size = tmp;
 585
 586         return count;
 587 }
 588 LUSTRE_RW_ATTR(max_nolock_bytes);
 589
 590 static ssize_t contention_seconds_show(struct kobject *kobj,
 591                                        struct attribute *attr, char *buf)
 592 {
 593         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 594                                                  ns_kobj);
 595
 596         return scnprintf(buf, PAGE_SIZE, "%d\n", ns->ns_contention_time);
 597 }
 598
 599 static ssize_t contention_seconds_store(struct kobject *kobj,
 600                                         struct attribute *attr,
 601                                         const char *buffer, size_t count)
 602 {
 603         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 604                                                  ns_kobj);
 605         unsigned int tmp;
 606
 607         if (kstrtouint(buffer, 10, &tmp))
 608                 return -EINVAL;
 609
 610         ns->ns_contention_time = tmp;
 611
 612         return count;
 613 }
 614 LUSTRE_RW_ATTR(contention_seconds);
 615
 616 static ssize_t contended_locks_show(struct kobject *kobj,
 617                                     struct attribute *attr, char *buf)
 618 {
 619         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 620                                                  ns_kobj);
 621
 622         return sprintf(buf, "%u\n", ns->ns_contended_locks);
 623 }
 624
 625 static ssize_t contended_locks_store(struct kobject *kobj,
 626                                      struct attribute *attr,
 627                                      const char *buffer, size_t count)
 628 {
 629         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 630                                                  ns_kobj);
 631         unsigned long tmp;
 632         int err;
 633
 634         err = kstrtoul(buffer, 10, &tmp);
 635         if (err != 0)
 636                 return -EINVAL;
 637
 638         ns->ns_contended_locks = tmp;
 639
 640         return count;
 641 }
 642 LUSTRE_RW_ATTR(contended_locks);
 643
 644 static ssize_t max_parallel_ast_show(struct kobject *kobj,
 645                                      struct attribute *attr, char *buf)
 646 {
 647         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 648                                                  ns_kobj);
 649
 650         return sprintf(buf, "%u\n", ns->ns_max_parallel_ast);
 651 }
 652
 653 static ssize_t max_parallel_ast_store(struct kobject *kobj,
 654                                       struct attribute *attr,
 655                                       const char *buffer, size_t count)
 656 {
 657         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 658                                                  ns_kobj);
 659         unsigned long tmp;
 660         int err;
 661
 662         err = kstrtoul(buffer, 10, &tmp);
 663         if (err != 0)
 664                 return -EINVAL;
 665
 666         ns->ns_max_parallel_ast = tmp;
 667
 668         return count;
 669 }
 670 LUSTRE_RW_ATTR(max_parallel_ast);
 671
 672 #endif /* HAVE_SERVER_SUPPORT */
 673
 674 /* These are for namespaces in /sys/fs/lustre/ldlm/namespaces/ */
 675 static struct attribute *ldlm_ns_attrs[] = {
 676         &lustre_attr_resource_count.attr,
 677         &lustre_attr_lock_count.attr,
 678         &lustre_attr_lock_unused_count.attr,
 679         &lustre_attr_ns_recalc_pct.attr,
 680         &lustre_attr_lru_size.attr,
 681         &lustre_attr_lru_cancel_batch.attr,
 682         &lustre_attr_lru_max_age.attr,
 683         &lustre_attr_early_lock_cancel.attr,
 684         &lustre_attr_dirty_age_limit.attr,
 685 #ifdef HAVE_SERVER_SUPPORT
 686         &lustre_attr_ctime_age_limit.attr,
 687         &lustre_attr_lock_timeouts.attr,
 688         &lustre_attr_max_nolock_bytes.attr,
 689         &lustre_attr_contention_seconds.attr,
 690         &lustre_attr_contended_locks.attr,
 691         &lustre_attr_max_parallel_ast.attr,
 692 #endif
 693         NULL,
 694 };
 695
 696 static void ldlm_ns_release(struct kobject *kobj)
 697 {
 698         struct ldlm_namespace *ns = container_of(kobj, struct ldlm_namespace,
 699                                                  ns_kobj);
 700         complete(&ns->ns_kobj_unregister);
 701 }
 702
 703 KOBJ_ATTRIBUTE_GROUPS(ldlm_ns);
 704
 705 static struct kobj_type ldlm_ns_ktype = {
 706         .default_groups = KOBJ_ATTR_GROUPS(ldlm_ns),
 707         .sysfs_ops      = &lustre_sysfs_ops,
 708         .release        = ldlm_ns_release,
 709 };
 710
 711 static void ldlm_namespace_debugfs_unregister(struct ldlm_namespace *ns)
 712 {
 713         if (IS_ERR_OR_NULL(ns->ns_debugfs_entry))
 714                 CERROR("dlm namespace %s has no procfs dir?\n",
 715                        ldlm_ns_name(ns));
 716         else
 717                 debugfs_remove_recursive(ns->ns_debugfs_entry);
 718
 719         if (ns->ns_stats != NULL)
 720                 lprocfs_stats_free(&ns->ns_stats);
 721 }
 722
 723 static void ldlm_namespace_sysfs_unregister(struct ldlm_namespace *ns)
 724 {
 725         kobject_put(&ns->ns_kobj);
 726         wait_for_completion(&ns->ns_kobj_unregister);
 727 }
 728
 729 static int ldlm_namespace_sysfs_register(struct ldlm_namespace *ns)
 730 {
 731         int err;
 732
 733         ns->ns_kobj.kset = ldlm_ns_kset;
 734         init_completion(&ns->ns_kobj_unregister);
 735         err = kobject_init_and_add(&ns->ns_kobj, &ldlm_ns_ktype, NULL,
 736                                    "%s", ldlm_ns_name(ns));
 737
 738         ns->ns_stats = lprocfs_stats_alloc(LDLM_NSS_LAST, 0);
 739         if (!ns->ns_stats) {
 740                 kobject_put(&ns->ns_kobj);
 741                 return -ENOMEM;
 742         }
 743
 744         lprocfs_counter_init(ns->ns_stats, LDLM_NSS_LOCKS,
 745                              LPROCFS_CNTR_AVGMINMAX | LPROCFS_TYPE_LOCKS,
 746                              "locks");
 747
 748         return err;
 749 }
 750
 751 static int ldlm_namespace_debugfs_register(struct ldlm_namespace *ns)
 752 {
 753         struct dentry *ns_entry;
 754
 755         if (!IS_ERR_OR_NULL(ns->ns_debugfs_entry)) {
 756                 ns_entry = ns->ns_debugfs_entry;
 757         } else {
 758                 ns_entry = debugfs_create_dir(ldlm_ns_name(ns),
 759                                               ldlm_ns_debugfs_dir);
 760                 if (!ns_entry)
 761                         return -ENOMEM;
 762                 ns->ns_debugfs_entry = ns_entry;
 763         }
 764
 765         return 0;
 766 }
 767 #undef MAX_STRING_SIZE
 768
 769 static unsigned int ldlm_res_hop_hash(struct cfs_hash *hs,
 770                                       const void *key, const unsigned int bits)
 771 {
 772         const struct ldlm_res_id *id = key;
 773         unsigned int val = 0;
 774         unsigned int i;
 775
 776         for (i = 0; i < RES_NAME_SIZE; i++)
 777                 val += id->name[i];
 778
 779         return val & ((1UL << bits) - 1);
 780 }
 781
 782 static unsigned int ldlm_res_hop_fid_hash(const struct ldlm_res_id *id,
 783                                           const unsigned int bits)
 784 {
 785         struct lu_fid       fid;
 786         __u32               hash;
 787         __u32               val;
 788
 789         fid.f_seq = id->name[LUSTRE_RES_ID_SEQ_OFF];
 790         fid.f_oid = (__u32)id->name[LUSTRE_RES_ID_VER_OID_OFF];
 791         fid.f_ver = (__u32)(id->name[LUSTRE_RES_ID_VER_OID_OFF] >> 32);
 792
 793         hash = fid_flatten32(&fid);
 794         hash += (hash >> 4) + (hash << 12); /* mixing oid and seq */
 795
 796         if (id->name[LUSTRE_RES_ID_HSH_OFF] != 0)
 797                 val = id->name[LUSTRE_RES_ID_HSH_OFF];
 798         else
 799                 val = fid_oid(&fid);
 800
 801         hash += (val >> 5) + (val << 11);
 802         return cfs_hash_32(hash, bits);
 803 }
 804
 805 static void *ldlm_res_hop_key(struct hlist_node *hnode)
 806 {
 807         struct ldlm_resource   *res;
 808
 809         res = hlist_entry(hnode, struct ldlm_resource, lr_hash);
 810         return &res->lr_name;
 811 }
 812
 813 static int ldlm_res_hop_keycmp(const void *key, struct hlist_node *hnode)
 814 {
 815         struct ldlm_resource   *res;
 816
 817         res = hlist_entry(hnode, struct ldlm_resource, lr_hash);
 818         return ldlm_res_eq((const struct ldlm_res_id *)key,
 819                            (const struct ldlm_res_id *)&res->lr_name);
 820 }
 821
 822 static void *ldlm_res_hop_object(struct hlist_node *hnode)
 823 {
 824         return hlist_entry(hnode, struct ldlm_resource, lr_hash);
 825 }
 826
 827 static void
 828 ldlm_res_hop_get_locked(struct cfs_hash *hs, struct hlist_node *hnode)
 829 {
 830         struct ldlm_resource *res;
 831
 832         res = hlist_entry(hnode, struct ldlm_resource, lr_hash);
 833         ldlm_resource_getref(res);
 834 }
 835
 836 static void ldlm_res_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
 837 {
 838         struct ldlm_resource *res;
 839
 840         res = hlist_entry(hnode, struct ldlm_resource, lr_hash);
 841         ldlm_resource_putref(res);
 842 }
 843
 844 static struct cfs_hash_ops ldlm_ns_hash_ops = {
 845         .hs_hash        = ldlm_res_hop_hash,
 846         .hs_key         = ldlm_res_hop_key,
 847         .hs_keycmp      = ldlm_res_hop_keycmp,
 848         .hs_keycpy      = NULL,
 849         .hs_object      = ldlm_res_hop_object,
 850         .hs_get         = ldlm_res_hop_get_locked,
 851         .hs_put         = ldlm_res_hop_put
 852 };
 853
 854 static struct {
 855         /** hash bucket bits */
 856         unsigned int            nsd_bkt_bits;
 857         /** hash bits */
 858         unsigned int            nsd_all_bits;
 859 } ldlm_ns_hash_defs[] = {
 860         [LDLM_NS_TYPE_MDC] = {
 861                 .nsd_bkt_bits   = 11,
 862                 .nsd_all_bits   = 16,
 863         },
 864         [LDLM_NS_TYPE_MDT] = {
 865                 .nsd_bkt_bits   = 14,
 866                 .nsd_all_bits   = 21,
 867         },
 868         [LDLM_NS_TYPE_OSC] = {
 869                 .nsd_bkt_bits   = 8,
 870                 .nsd_all_bits   = 12,
 871         },
 872         [LDLM_NS_TYPE_OST] = {
 873                 .nsd_bkt_bits   = 11,
 874                 .nsd_all_bits   = 17,
 875         },
 876         [LDLM_NS_TYPE_MGC] = {
 877                 .nsd_bkt_bits   = 3,
 878                 .nsd_all_bits   = 4,
 879         },
 880         [LDLM_NS_TYPE_MGT] = {
 881                 .nsd_bkt_bits   = 3,
 882                 .nsd_all_bits   = 4,
 883         },
 884 };
 885
 886 /**
 887  * Create and initialize new empty namespace.
 888  */
 889 struct ldlm_namespace *ldlm_namespace_new(struct obd_device *obd, char *name,
 890                                           enum ldlm_side client,
 891                                           enum ldlm_appetite apt,
 892                                           enum ldlm_ns_type ns_type)
 893 {
 894         struct ldlm_namespace *ns = NULL;
 895         int idx;
 896         int rc;
 897
 898         ENTRY;
 899         LASSERT(obd != NULL);
 900
 901         rc = ldlm_get_ref();
 902         if (rc) {
 903                 CERROR("%s: ldlm_get_ref failed: rc = %d\n", name, rc);
 904                 RETURN(ERR_PTR(rc));
 905         }
 906
 907         if (ns_type >= ARRAY_SIZE(ldlm_ns_hash_defs) ||
 908             ldlm_ns_hash_defs[ns_type].nsd_bkt_bits == 0) {
 909                 rc = -EINVAL;
 910                 CERROR("%s: unknown namespace type %d: rc = %d\n",
 911                        name, ns_type, rc);
 912                 GOTO(out_ref, rc);
 913         }
 914
 915         OBD_ALLOC_PTR(ns);
 916         if (!ns)
 917                 GOTO(out_ref, rc = -ENOMEM);
 918
 919         ns->ns_rs_hash = cfs_hash_create(name,
 920                                          ldlm_ns_hash_defs[ns_type].nsd_all_bits,
 921                                          ldlm_ns_hash_defs[ns_type].nsd_all_bits,
 922                                          ldlm_ns_hash_defs[ns_type].nsd_bkt_bits,
 923                                          0,
 924                                          CFS_HASH_MIN_THETA,
 925                                          CFS_HASH_MAX_THETA,
 926                                          &ldlm_ns_hash_ops,
 927                                          CFS_HASH_DEPTH |
 928                                          CFS_HASH_BIGNAME |
 929                                          CFS_HASH_SPIN_BKTLOCK |
 930                                          CFS_HASH_NO_ITEMREF);
 931         if (!ns->ns_rs_hash)
 932                 GOTO(out_ns, rc = -ENOMEM);
 933
 934         ns->ns_bucket_bits = ldlm_ns_hash_defs[ns_type].nsd_all_bits -
 935                              ldlm_ns_hash_defs[ns_type].nsd_bkt_bits;
 936
 937         OBD_ALLOC_PTR_ARRAY_LARGE(ns->ns_rs_buckets, 1 << ns->ns_bucket_bits);
 938         if (!ns->ns_rs_buckets)
 939                 GOTO(out_hash, rc = -ENOMEM);
 940
 941         for (idx = 0; idx < (1 << ns->ns_bucket_bits); idx++) {
 942                 struct ldlm_ns_bucket *nsb = &ns->ns_rs_buckets[idx];
 943
 944                 at_init(&nsb->nsb_at_estimate, obd_get_ldlm_enqueue_min(obd), 0);
 945                 nsb->nsb_namespace = ns;
 946                 nsb->nsb_reclaim_start = 0;
 947                 atomic_set(&nsb->nsb_count, 0);
 948         }
 949
 950         ns->ns_obd = obd;
 951         ns->ns_appetite = apt;
 952         ns->ns_client = client;
 953         ns->ns_name = kstrdup(name, GFP_KERNEL);
 954         if (!ns->ns_name)
 955                 GOTO(out_hash, rc = -ENOMEM);
 956
 957         INIT_LIST_HEAD(&ns->ns_list_chain);
 958         INIT_LIST_HEAD(&ns->ns_unused_list);
 959         spin_lock_init(&ns->ns_lock);
 960         atomic_set(&ns->ns_bref, 0);
 961         init_waitqueue_head(&ns->ns_waitq);
 962
 963         ns->ns_max_nolock_size    = NS_DEFAULT_MAX_NOLOCK_BYTES;
 964         ns->ns_contention_time    = NS_DEFAULT_CONTENTION_SECONDS;
 965         ns->ns_contended_locks    = NS_DEFAULT_CONTENDED_LOCKS;
 966
 967         ns->ns_max_parallel_ast   = LDLM_DEFAULT_PARALLEL_AST_LIMIT;
 968         ns->ns_nr_unused          = 0;
 969         ns->ns_max_unused         = LDLM_DEFAULT_LRU_SIZE;
 970         ns->ns_cancel_batch       = LDLM_DEFAULT_LRU_SHRINK_BATCH;
 971         ns->ns_recalc_pct         = LDLM_DEFAULT_SLV_RECALC_PCT;
 972         ns->ns_max_age            = ktime_set(LDLM_DEFAULT_MAX_ALIVE, 0);
 973         ns->ns_ctime_age_limit    = LDLM_CTIME_AGE_LIMIT;
 974         ns->ns_dirty_age_limit    = ktime_set(LDLM_DIRTY_AGE_LIMIT, 0);
 975         ns->ns_timeouts           = 0;
 976         ns->ns_orig_connect_flags = 0;
 977         ns->ns_connect_flags      = 0;
 978         ns->ns_stopping           = 0;
 979         ns->ns_reclaim_start      = 0;
 980         ns->ns_last_pos           = &ns->ns_unused_list;
 981         ns->ns_flags              = 0;
 982
 983         rc = ldlm_namespace_sysfs_register(ns);
 984         if (rc) {
 985                 CERROR("%s: cannot initialize ns sysfs: rc = %d\n", name, rc);
 986                 GOTO(out_hash, rc);
 987         }
 988
 989         rc = ldlm_namespace_debugfs_register(ns);
 990         if (rc) {
 991                 CERROR("%s: cannot initialize ns proc: rc = %d\n", name, rc);
 992                 GOTO(out_sysfs, rc);
 993         }
 994
 995         idx = ldlm_namespace_nr_read(client);
 996         rc = ldlm_pool_init(&ns->ns_pool, ns, idx, client);
 997         if (rc) {
 998                 CERROR("%s: cannot initialize lock pool, rc = %d\n", name, rc);
 999                 GOTO(out_proc, rc);
1000         }
1001
1002         ldlm_namespace_register(ns, client);
1003         RETURN(ns);
1004 out_proc:
1005         ldlm_namespace_debugfs_unregister(ns);
1006 out_sysfs:
1007         ldlm_namespace_sysfs_unregister(ns);
1008         ldlm_namespace_cleanup(ns, 0);
1009 out_hash:
1010         OBD_FREE_PTR_ARRAY_LARGE(ns->ns_rs_buckets, 1 << ns->ns_bucket_bits);
1011         kfree(ns->ns_name);
1012         cfs_hash_putref(ns->ns_rs_hash);
1013 out_ns:
1014         OBD_FREE_PTR(ns);
1015 out_ref:
1016         ldlm_put_ref();
1017         RETURN(ERR_PTR(rc));
1018 }
1019 EXPORT_SYMBOL(ldlm_namespace_new);
1020
1021 /**
1022  * Cancel and destroy all locks on a resource.
1023  *
1024  * If flags contains FL_LOCAL_ONLY, don't try to tell the server, just
1025  * clean up.  This is currently only used for recovery, and we make
1026  * certain assumptions as a result--notably, that we shouldn't cancel
1027  * locks with refs.
1028  */
1029 static void cleanup_resource(struct ldlm_resource *res, struct list_head *q,
1030                              u64 flags)
1031 {
1032         int rc = 0, client = ns_is_client(ldlm_res_to_ns(res));
1033         bool local_only = !!(flags & LDLM_FL_LOCAL_ONLY);
1034
1035         do {
1036                 struct ldlm_lock *lock = NULL, *tmp;
1037
1038                 /* First, we look for non-cleaned-yet lock. all cleaned locks
1039                  * are marked by CLEANED flag.
1040                  */
1041                 lock_res(res);
1042                 list_for_each_entry(tmp, q, l_res_link) {
1043                         if (ldlm_is_cleaned(tmp))
1044                                 continue;
1045
1046                         lock = tmp;
1047                         LDLM_LOCK_GET(lock);
1048                         ldlm_set_cleaned(lock);
1049                         break;
1050                 }
1051
1052                 if (lock == NULL) {
1053                         unlock_res(res);
1054                         break;
1055                 }
1056
1057                 /* Set CBPENDING so nothing is in the cancellation path
1058                  * can match this lock.
1059                  */
1060                 ldlm_set_cbpending(lock);
1061                 ldlm_set_failed(lock);
1062                 ldlm_clear_converting(lock);
1063                 lock->l_flags |= flags;
1064
1065                 /* ... without sending a CANCEL message for local_only. */
1066                 if (local_only)
1067                         ldlm_set_local_only(lock);
1068
1069                 if (local_only && (lock->l_readers || lock->l_writers)) {
1070                         /*
1071                          * This is a little bit gross, but much better than the
1072                          * alternative: pretend that we got a blocking AST from
1073                          * the server, so that when the lock is decref'd, it
1074                          * will go away ...
1075                          */
1076                         unlock_res(res);
1077                         LDLM_DEBUG(lock, "setting FL_LOCAL_ONLY");
1078                         if (lock->l_flags & LDLM_FL_FAIL_LOC)
1079                                 schedule_timeout_uninterruptible(
1080                                         cfs_time_seconds(4));
1081
1082                         if (lock->l_completion_ast)
1083                                 lock->l_completion_ast(lock,
1084                                                        LDLM_FL_FAILED, NULL);
1085                         LDLM_LOCK_RELEASE(lock);
1086                         continue;
1087                 }
1088
1089                 if (client) {
1090                         struct lustre_handle lockh;
1091
1092                         unlock_res(res);
1093                         ldlm_lock2handle(lock, &lockh);
1094                         rc = ldlm_cli_cancel(&lockh, LCF_LOCAL);
1095                         if (rc)
1096                                 CERROR("ldlm_cli_cancel: %d\n", rc);
1097                 } else {
1098                         unlock_res(res);
1099                         LDLM_DEBUG(lock,
1100                                    "Freeing a lock still held by a client node");
1101                         ldlm_lock_cancel(lock);
1102                 }
1103                 LDLM_LOCK_RELEASE(lock);
1104         } while (1);
1105 }
1106
1107 static int ldlm_resource_clean(struct cfs_hash *hs, struct cfs_hash_bd *bd,
1108                                struct hlist_node *hnode, void *arg)
1109 {
1110         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
1111         __u64 flags = *(__u64 *)arg;
1112
1113         cleanup_resource(res, &res->lr_granted, flags);
1114         cleanup_resource(res, &res->lr_waiting, flags);
1115
1116         return 0;
1117 }
1118
1119 static int ldlm_resource_complain(struct cfs_hash *hs, struct cfs_hash_bd *bd,
1120                                   struct hlist_node *hnode, void *arg)
1121 {
1122         struct ldlm_resource  *res = cfs_hash_object(hs, hnode);
1123
1124         lock_res(res);
1125         CERROR("%s: namespace resource "DLDLMRES" (%p) refcount nonzero "
1126                "(%d) after lock cleanup; forcing cleanup.\n",
1127                ldlm_ns_name(ldlm_res_to_ns(res)), PLDLMRES(res), res,
1128                refcount_read(&res->lr_refcount) - 1);
1129
1130         /* Use D_NETERROR since it is in the default mask */
1131         ldlm_resource_dump(D_NETERROR, res);
1132         unlock_res(res);
1133         return 0;
1134 }
1135
1136 /**
1137  * Cancel and destroy all locks in the namespace.
1138  *
1139  * Typically used during evictions when server notified client that it was
1140  * evicted and all of its state needs to be destroyed.
1141  * Also used during shutdown.
1142  */
1143 int ldlm_namespace_cleanup(struct ldlm_namespace *ns, __u64 flags)
1144 {
1145         if (ns == NULL) {
1146                 CDEBUG(D_INFO, "NULL ns, skipping cleanup\n");
1147                 return ELDLM_OK;
1148         }
1149
1150         cfs_hash_for_each_nolock(ns->ns_rs_hash, ldlm_resource_clean,
1151                                  &flags, 0);
1152         cfs_hash_for_each_nolock(ns->ns_rs_hash, ldlm_resource_complain,
1153                                  NULL, 0);
1154         return ELDLM_OK;
1155 }
1156 EXPORT_SYMBOL(ldlm_namespace_cleanup);
1157
1158 /**
1159  * Attempts to free namespace.
1160  *
1161  * Only used when namespace goes away, like during an unmount.
1162  */
1163 static int __ldlm_namespace_free(struct ldlm_namespace *ns, int force)
1164 {
1165         ENTRY;
1166
1167         /* At shutdown time, don't call the cancellation callback */
1168         ldlm_namespace_cleanup(ns, force ? LDLM_FL_LOCAL_ONLY : 0);
1169
1170         if (atomic_read(&ns->ns_bref) > 0) {
1171                 int rc;
1172
1173                 CDEBUG(D_DLMTRACE,
1174                        "dlm namespace %s free waiting on refcount %d\n",
1175                        ldlm_ns_name(ns), atomic_read(&ns->ns_bref));
1176 force_wait:
1177                 if (force)
1178                         rc = wait_event_idle_timeout(
1179                                 ns->ns_waitq,
1180                                 atomic_read(&ns->ns_bref) == 0,
1181                                 cfs_time_seconds(1) / 4);
1182                 else
1183                         rc = l_wait_event_abortable(
1184                                 ns->ns_waitq, atomic_read(&ns->ns_bref) == 0);
1185
1186                 /* Forced cleanups should be able to reclaim all references,
1187                  * so it's safe to wait forever... we can't leak locks...
1188                  */
1189                 if (force && rc == 0) {
1190                         rc = -ETIMEDOUT;
1191                         LCONSOLE_ERROR("Forced cleanup waiting for %s "
1192                                        "namespace with %d resources in use, "
1193                                        "(rc=%d)\n", ldlm_ns_name(ns),
1194                                        atomic_read(&ns->ns_bref), rc);
1195                         GOTO(force_wait, rc);
1196                 }
1197
1198                 if (atomic_read(&ns->ns_bref)) {
1199                         LCONSOLE_ERROR("Cleanup waiting for %s namespace "
1200                                        "with %d resources in use, (rc=%d)\n",
1201                                        ldlm_ns_name(ns),
1202                                        atomic_read(&ns->ns_bref), rc);
1203                         RETURN(ELDLM_NAMESPACE_EXISTS);
1204                 }
1205                 CDEBUG(D_DLMTRACE, "dlm namespace %s free done waiting\n",
1206                        ldlm_ns_name(ns));
1207         }
1208
1209         RETURN(ELDLM_OK);
1210 }
1211
1212 /**
1213  * Performs various cleanups for passed \a ns to make it drop refc and be
1214  * ready for freeing. Waits for refc == 0.
1215  *
1216  * The following is done:
1217  * (0) Unregister \a ns from its list to make inaccessible for potential
1218  * users like pools thread and others;
1219  * (1) Clear all locks in \a ns.
1220  */
1221 void ldlm_namespace_free_prior(struct ldlm_namespace *ns,
1222                                struct obd_import *imp, int force)
1223 {
1224         int rc;
1225
1226         ENTRY;
1227         if (!ns) {
1228                 EXIT;
1229                 return;
1230         }
1231
1232         spin_lock(&ns->ns_lock);
1233         ns->ns_stopping = 1;
1234         spin_unlock(&ns->ns_lock);
1235
1236         /* Can fail with -EINTR when force == 0 in which case try harder. */
1237         rc = __ldlm_namespace_free(ns, force);
1238         if (rc != ELDLM_OK) {
1239                 if (imp) {
1240                         ptlrpc_disconnect_import(imp, 0);
1241                         ptlrpc_invalidate_import(imp);
1242                 }
1243
1244                 /*
1245                  * With all requests dropped and the import inactive
1246                  * we are gaurenteed all reference will be dropped.
1247                  */
1248                 rc = __ldlm_namespace_free(ns, 1);
1249                 LASSERT(rc == 0);
1250         }
1251         EXIT;
1252 }
1253 EXPORT_SYMBOL(ldlm_namespace_free_prior);
1254
1255 /**
1256  * Performs freeing memory structures related to \a ns. This is only done
1257  * when ldlm_namespce_free_prior() successfully removed all resources
1258  * referencing \a ns and its refc == 0.
1259  */
1260 void ldlm_namespace_free_post(struct ldlm_namespace *ns)
1261 {
1262         ENTRY;
1263         if (!ns) {
1264                 EXIT;
1265                 return;
1266         }
1267
1268         /* Make sure that nobody can find this ns in its list. */
1269         ldlm_namespace_unregister(ns, ns->ns_client);
1270         /* Fini pool _before_ parent proc dir is removed. This is important as
1271          * ldlm_pool_fini() removes own proc dir which is child to @dir.
1272          * Removing it after @dir may cause oops.
1273          */
1274         ldlm_pool_fini(&ns->ns_pool);
1275
1276         ldlm_namespace_debugfs_unregister(ns);
1277         ldlm_namespace_sysfs_unregister(ns);
1278         cfs_hash_putref(ns->ns_rs_hash);
1279         OBD_FREE_PTR_ARRAY_LARGE(ns->ns_rs_buckets, 1 << ns->ns_bucket_bits);
1280         kfree(ns->ns_name);
1281         /* Namespace \a ns should be not on list at this time, otherwise
1282          * this will cause issues related to using freed \a ns in poold
1283          * thread.
1284          */
1285         LASSERT(list_empty(&ns->ns_list_chain));
1286         OBD_FREE_PTR(ns);
1287         ldlm_put_ref();
1288         EXIT;
1289 }
1290 EXPORT_SYMBOL(ldlm_namespace_free_post);
1291
1292 /**
1293  * Cleanup the resource, and free namespace.
1294  * bug 12864:
1295  * Deadlock issue:
1296  * proc1: destroy import
1297  *        class_disconnect_export(grab cl_sem) ->
1298  *              -> ldlm_namespace_free ->
1299  *              -> lprocfs_remove(grab _lprocfs_lock).
1300  * proc2: read proc info
1301  *        lprocfs_fops_read(grab _lprocfs_lock) ->
1302  *              -> osc_rd_active, etc(grab cl_sem).
1303  *
1304  * So that I have to split the ldlm_namespace_free into two parts - the first
1305  * part ldlm_namespace_free_prior is used to cleanup the resource which is
1306  * being used; the 2nd part ldlm_namespace_free_post is used to unregister the
1307  * lprocfs entries, and then free memory. It will be called w/o cli->cl_sem
1308  * held.
1309  */
1310 void ldlm_namespace_free(struct ldlm_namespace *ns,
1311                          struct obd_import *imp,
1312                          int force)
1313 {
1314         ldlm_namespace_free_prior(ns, imp, force);
1315         ldlm_namespace_free_post(ns);
1316 }
1317 EXPORT_SYMBOL(ldlm_namespace_free);
1318
1319 void ldlm_namespace_get(struct ldlm_namespace *ns)
1320 {
1321         atomic_inc(&ns->ns_bref);
1322 }
1323
1324 /* This is only for callers that care about refcount */
1325 static int ldlm_namespace_get_return(struct ldlm_namespace *ns)
1326 {
1327         return atomic_inc_return(&ns->ns_bref);
1328 }
1329
1330 void ldlm_namespace_put(struct ldlm_namespace *ns)
1331 {
1332         if (atomic_dec_and_lock(&ns->ns_bref, &ns->ns_lock)) {
1333                 wake_up(&ns->ns_waitq);
1334                 spin_unlock(&ns->ns_lock);
1335         }
1336 }
1337
1338 /** Register \a ns in the list of namespaces */
1339 void ldlm_namespace_register(struct ldlm_namespace *ns, enum ldlm_side client)
1340 {
1341         mutex_lock(ldlm_namespace_lock(client));
1342         LASSERT(list_empty(&ns->ns_list_chain));
1343         list_add(&ns->ns_list_chain, ldlm_namespace_inactive_list(client));
1344         ldlm_namespace_nr_inc(client);
1345         mutex_unlock(ldlm_namespace_lock(client));
1346 }
1347
1348 /** Unregister \a ns from the list of namespaces. */
1349 void ldlm_namespace_unregister(struct ldlm_namespace *ns, enum ldlm_side client)
1350 {
1351         mutex_lock(ldlm_namespace_lock(client));
1352         LASSERT(!list_empty(&ns->ns_list_chain));
1353         /* Some asserts and possibly other parts of the code are still
1354          * using list_empty(&ns->ns_list_chain). This is why it is
1355          * important to use list_del_init() here.
1356          */
1357         list_del_init(&ns->ns_list_chain);
1358         ldlm_namespace_nr_dec(client);
1359         mutex_unlock(ldlm_namespace_lock(client));
1360 }
1361
1362 /** Should be called with ldlm_namespace_lock(client) taken. */
1363 void ldlm_namespace_move_to_active_locked(struct ldlm_namespace *ns,
1364                                           enum ldlm_side client)
1365 {
1366         LASSERT(!list_empty(&ns->ns_list_chain));
1367         LASSERT(mutex_is_locked(ldlm_namespace_lock(client)));
1368         list_move_tail(&ns->ns_list_chain, ldlm_namespace_list(client));
1369 }
1370
1371 /** Should be called with ldlm_namespace_lock(client) taken. */
1372 void ldlm_namespace_move_to_inactive_locked(struct ldlm_namespace *ns,
1373                                             enum ldlm_side client)
1374 {
1375         LASSERT(!list_empty(&ns->ns_list_chain));
1376         LASSERT(mutex_is_locked(ldlm_namespace_lock(client)));
1377         list_move_tail(&ns->ns_list_chain,
1378                        ldlm_namespace_inactive_list(client));
1379 }
1380
1381 /** Should be called with ldlm_namespace_lock(client) taken. */
1382 struct ldlm_namespace *ldlm_namespace_first_locked(enum ldlm_side client)
1383 {
1384         LASSERT(mutex_is_locked(ldlm_namespace_lock(client)));
1385         LASSERT(!list_empty(ldlm_namespace_list(client)));
1386         return container_of(ldlm_namespace_list(client)->next,
1387                             struct ldlm_namespace, ns_list_chain);
1388 }
1389
1390 static bool ldlm_resource_extent_new(struct ldlm_resource *res)
1391 {
1392         int idx;
1393
1394         OBD_SLAB_ALLOC(res->lr_itree, ldlm_interval_tree_slab,
1395                        sizeof(*res->lr_itree) * LCK_MODE_NUM);
1396         if (res->lr_itree == NULL)
1397                 return false;
1398         /* Initialize interval trees for each lock mode. */
1399         for (idx = 0; idx < LCK_MODE_NUM; idx++) {
1400                 res->lr_itree[idx].lit_size = 0;
1401                 res->lr_itree[idx].lit_mode = BIT(idx);
1402                 res->lr_itree[idx].lit_root = NULL;
1403         }
1404         return true;
1405 }
1406
1407 static bool ldlm_resource_inodebits_new(struct ldlm_resource *res)
1408 {
1409         int i;
1410
1411         OBD_ALLOC_PTR(res->lr_ibits_queues);
1412         if (res->lr_ibits_queues == NULL)
1413                 return false;
1414         for (i = 0; i < MDS_INODELOCK_NUMBITS; i++)
1415                 INIT_LIST_HEAD(&res->lr_ibits_queues->liq_waiting[i]);
1416         return true;
1417 }
1418
1419 static bool ldlm_resource_flock_new(struct ldlm_resource *res)
1420 {
1421         res->lr_flock_node.lfn_needs_reprocess = false;
1422         res->lr_flock_node.lfn_root = NULL;
1423         atomic_set(&res->lr_flock_node.lfn_unlock_pending, 0);
1424
1425         return true;
1426 }
1427
1428 /** Create and initialize new resource. */
1429 static struct ldlm_resource *ldlm_resource_new(enum ldlm_type ldlm_type)
1430 {
1431         struct ldlm_resource *res;
1432         bool rc;
1433
1434         OBD_SLAB_ALLOC_PTR_GFP(res, ldlm_resource_slab, GFP_NOFS);
1435         if (res == NULL)
1436                 return NULL;
1437
1438         switch (ldlm_type) {
1439         case LDLM_EXTENT:
1440                 rc = ldlm_resource_extent_new(res);
1441                 break;
1442         case LDLM_IBITS:
1443                 rc = ldlm_resource_inodebits_new(res);
1444                 break;
1445         case LDLM_FLOCK:
1446                 rc = ldlm_resource_flock_new(res);
1447                 break;
1448         default:
1449                 rc = true;
1450                 break;
1451         }
1452         if (!rc) {
1453                 OBD_SLAB_FREE_PTR(res, ldlm_resource_slab);
1454                 return NULL;
1455         }
1456
1457         INIT_LIST_HEAD(&res->lr_granted);
1458         INIT_LIST_HEAD(&res->lr_waiting);
1459
1460         refcount_set(&res->lr_refcount, 1);
1461         spin_lock_init(&res->lr_lock);
1462         lu_ref_init(&res->lr_reference);
1463
1464         /* Since LVB init can be delayed now, there is no longer need to
1465          * immediatelly acquire mutex here.
1466          */
1467         mutex_init(&res->lr_lvb_mutex);
1468         res->lr_lvb_initialized = false;
1469
1470         return res;
1471 }
1472
1473 static void __ldlm_resource_free(struct rcu_head *head)
1474 {
1475         struct ldlm_resource *res = container_of(head, struct ldlm_resource,
1476                                                  lr_rcu);
1477
1478         OBD_SLAB_FREE_PTR(res, ldlm_resource_slab);
1479 }
1480
1481 static void ldlm_resource_free(struct ldlm_resource *res)
1482 {
1483         if (res->lr_type == LDLM_EXTENT) {
1484                 if (res->lr_itree != NULL)
1485                         OBD_SLAB_FREE(res->lr_itree, ldlm_interval_tree_slab,
1486                                       sizeof(*res->lr_itree) * LCK_MODE_NUM);
1487         } else if (res->lr_type == LDLM_IBITS) {
1488                 if (res->lr_ibits_queues != NULL)
1489                         OBD_FREE_PTR(res->lr_ibits_queues);
1490         }
1491
1492         call_rcu(&res->lr_rcu, __ldlm_resource_free);
1493 }
1494
1495 /**
1496  * Return a reference to resource with given name, creating it if necessary.
1497  * Args: namespace with ns_lock unlocked
1498  * Locks: takes and releases NS hash-lock and res->lr_lock
1499  * Returns: referenced, unlocked ldlm_resource or ERR_PTR
1500  */
1501 struct ldlm_resource *
1502 ldlm_resource_get(struct ldlm_namespace *ns, const struct ldlm_res_id *name,
1503                   enum ldlm_type type, int create)
1504 {
1505         struct hlist_node       *hnode;
1506         struct ldlm_resource    *res = NULL;
1507         struct cfs_hash_bd              bd;
1508         __u64                   version;
1509         int                     ns_refcount = 0;
1510         int hash;
1511
1512         LASSERT(ns != NULL);
1513         LASSERT(ns->ns_rs_hash != NULL);
1514         LASSERT(name->name[0] != 0);
1515
1516         cfs_hash_bd_get_and_lock(ns->ns_rs_hash, (void *)name, &bd, 0);
1517         hnode = cfs_hash_bd_lookup_locked(ns->ns_rs_hash, &bd, (void *)name);
1518         if (hnode != NULL) {
1519                 cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 0);
1520                 GOTO(found, res);
1521         }
1522
1523         version = cfs_hash_bd_version_get(&bd);
1524         cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 0);
1525
1526         if (create == 0)
1527                 return ERR_PTR(-ENOENT);
1528
1529         LASSERTF(type >= LDLM_MIN_TYPE && type < LDLM_MAX_TYPE,
1530                  "type: %d\n", type);
1531         res = ldlm_resource_new(type);
1532         if (res == NULL)
1533                 return ERR_PTR(-ENOMEM);
1534
1535         hash = ldlm_res_hop_fid_hash(name, ns->ns_bucket_bits);
1536         res->lr_ns_bucket = &ns->ns_rs_buckets[hash];
1537         res->lr_name = *name;
1538         res->lr_type = type;
1539
1540         cfs_hash_bd_lock(ns->ns_rs_hash, &bd, 1);
1541         hnode = (version == cfs_hash_bd_version_get(&bd)) ? NULL :
1542                 cfs_hash_bd_lookup_locked(ns->ns_rs_hash, &bd, (void *)name);
1543
1544         if (hnode != NULL) {
1545                 /* Someone won the race and already added the resource. */
1546                 cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
1547                 /* Clean lu_ref for failed resource. */
1548                 lu_ref_fini(&res->lr_reference);
1549                 ldlm_resource_free(res);
1550 found:
1551                 res = hlist_entry(hnode, struct ldlm_resource, lr_hash);
1552                 return res;
1553         }
1554         /* We won! Let's add the resource. */
1555         cfs_hash_bd_add_locked(ns->ns_rs_hash, &bd, &res->lr_hash);
1556         if (atomic_inc_return(&res->lr_ns_bucket->nsb_count) == 1)
1557                 ns_refcount = ldlm_namespace_get_return(ns);
1558
1559         cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
1560
1561         CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_CREATE_RESOURCE, 2);
1562
1563         /* Let's see if we happened to be the very first resource in this
1564          * namespace. If so, and this is a client namespace, we need to move
1565          * the namespace into the active namespaces list to be patrolled by
1566          * the ldlm_poold.
1567          */
1568         if (ns_is_client(ns) && ns_refcount == 1) {
1569                 mutex_lock(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
1570                 ldlm_namespace_move_to_active_locked(ns, LDLM_NAMESPACE_CLIENT);
1571                 mutex_unlock(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
1572         }
1573
1574         return res;
1575 }
1576 EXPORT_SYMBOL(ldlm_resource_get);
1577
1578 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
1579 {
1580         LASSERT(res != NULL);
1581         LASSERT(res != LP_POISON);
1582         refcount_inc(&res->lr_refcount);
1583         CDEBUG(D_INFO, "getref res: %p count: %d\n", res,
1584                refcount_read(&res->lr_refcount));
1585         return res;
1586 }
1587
1588 static void __ldlm_resource_putref_final(struct cfs_hash_bd *bd,
1589                                          struct ldlm_resource *res)
1590 {
1591         struct ldlm_ns_bucket *nsb = res->lr_ns_bucket;
1592
1593         if (!list_empty(&res->lr_granted)) {
1594                 ldlm_resource_dump(D_ERROR, res);
1595                 LBUG();
1596         }
1597
1598         if (!list_empty(&res->lr_waiting)) {
1599                 ldlm_resource_dump(D_ERROR, res);
1600                 LBUG();
1601         }
1602
1603         cfs_hash_bd_del_locked(nsb->nsb_namespace->ns_rs_hash,
1604                                bd, &res->lr_hash);
1605         lu_ref_fini(&res->lr_reference);
1606         if (atomic_dec_and_test(&nsb->nsb_count))
1607                 ldlm_namespace_put(nsb->nsb_namespace);
1608 }
1609
1610 /* Returns 1 if the resource was freed, 0 if it remains. */
1611 int ldlm_resource_putref(struct ldlm_resource *res)
1612 {
1613         struct ldlm_namespace *ns;
1614         struct cfs_hash_bd bd;
1615         int refcount;
1616
1617         if (refcount_dec_not_one(&res->lr_refcount))
1618                 return 0;
1619         ns = ldlm_res_to_ns(res);
1620         refcount = refcount_read(&res->lr_refcount);
1621         LASSERT(refcount < LI_POISON);
1622
1623         CDEBUG(D_INFO, "putref res: %p count: %d\n",
1624                res, refcount_read(&res->lr_refcount) - 1);
1625
1626         cfs_hash_bd_get(ns->ns_rs_hash, &res->lr_name, &bd);
1627         if (cfs_hash_bd_dec_and_lock(ns->ns_rs_hash, &bd, &res->lr_refcount)) {
1628                 __ldlm_resource_putref_final(&bd, res);
1629                 cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
1630                 if (ns->ns_lvbo && ns->ns_lvbo->lvbo_free)
1631                         ns->ns_lvbo->lvbo_free(res);
1632                 ldlm_resource_free(res);
1633                 return 1;
1634         }
1635         return 0;
1636 }
1637 EXPORT_SYMBOL(ldlm_resource_putref);
1638
1639 static void __ldlm_resource_add_lock(struct ldlm_resource *res,
1640                                      struct list_head *head,
1641                                      struct ldlm_lock *lock,
1642                                      bool tail)
1643 {
1644         check_res_locked(res);
1645
1646         if (ldlm_is_destroyed(lock)) {
1647                 CDEBUG(D_OTHER, "Lock destroyed, not adding to resource\n");
1648                 return;
1649         }
1650
1651         LASSERT(list_empty(&lock->l_res_link));
1652
1653         if (tail)
1654                 list_add_tail(&lock->l_res_link, head);
1655         else
1656                 list_add(&lock->l_res_link, head);
1657
1658         if (res->lr_type == LDLM_IBITS)
1659                 ldlm_inodebits_add_lock(res, head, lock, tail);
1660
1661         ldlm_resource_dump(D_INFO, res);
1662 }
1663
1664 /* Add a lock into a given resource into specified lock list. */
1665 void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
1666                             struct ldlm_lock *lock)
1667 {
1668         LDLM_DEBUG(lock, "About to add this lock");
1669
1670         __ldlm_resource_add_lock(res, head, lock, true);
1671 }
1672
1673 /* Insert a lock into resource after specified lock. */
1674 void ldlm_resource_insert_lock_after(struct ldlm_lock *original,
1675                                      struct ldlm_lock *new)
1676 {
1677         LASSERT(!list_empty(&original->l_res_link));
1678
1679         LDLM_DEBUG(new, "About to insert this lock after %p: ", original);
1680         __ldlm_resource_add_lock(original->l_resource,
1681                                  &original->l_res_link,
1682                                  new, false);
1683 }
1684
1685 /**
1686  * Insert a lock into resource before the specified lock.
1687  *
1688  * IBITS waiting locks are to be inserted to the ibit lists as well, and only
1689  * the insert-after operation is supported for them, because the set of bits
1690  * of the previous and the new locks must match. Therefore, get the previous
1691  * lock and insert after.
1692  */
1693 void ldlm_resource_insert_lock_before(struct ldlm_lock *original,
1694                                       struct ldlm_lock *new)
1695 {
1696         LASSERT(!list_empty(&original->l_res_link));
1697
1698         LDLM_DEBUG(new, "About to insert this lock before %p: ", original);
1699         __ldlm_resource_add_lock(original->l_resource,
1700                                  original->l_res_link.prev, new, false);
1701 }
1702
1703 void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
1704 {
1705         int type = lock->l_resource->lr_type;
1706
1707         check_res_locked(lock->l_resource);
1708         switch (type) {
1709         case LDLM_PLAIN:
1710                 ldlm_unlink_lock_skiplist(lock);
1711                 break;
1712         case LDLM_EXTENT:
1713                 ldlm_extent_unlink_lock(lock);
1714                 break;
1715         case LDLM_IBITS:
1716                 ldlm_inodebits_unlink_lock(lock);
1717                 break;
1718         case LDLM_FLOCK:
1719                 ldlm_flock_unlink_lock(lock);
1720                 break;
1721         }
1722         list_del_init(&lock->l_res_link);
1723 }
1724 EXPORT_SYMBOL(ldlm_resource_unlink_lock);
1725
1726 void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
1727 {
1728         desc->lr_type = res->lr_type;
1729         desc->lr_name = res->lr_name;
1730 }
1731
1732 /* Print info about all locks in all namespaces on this node to debug log. */
1733 void ldlm_dump_all_namespaces(enum ldlm_side client, int level)
1734 {
1735         struct ldlm_namespace *ns;
1736
1737         if (!((libcfs_debug | D_ERROR) & level))
1738                 return;
1739
1740         mutex_lock(ldlm_namespace_lock(client));
1741
1742         list_for_each_entry(ns, ldlm_namespace_list(client), ns_list_chain)
1743                 ldlm_namespace_dump(level, ns);
1744
1745         mutex_unlock(ldlm_namespace_lock(client));
1746 }
1747
1748 static int ldlm_res_hash_dump(struct cfs_hash *hs, struct cfs_hash_bd *bd,
1749                               struct hlist_node *hnode, void *arg)
1750 {
1751         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
1752         int    level = (int)(unsigned long)arg;
1753
1754         lock_res(res);
1755         ldlm_resource_dump(level, res);
1756         unlock_res(res);
1757
1758         return 0;
1759 }
1760
1761 /* Print info about all locks in this namespace on this node to debug log. */
1762 void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
1763 {
1764         if (!((libcfs_debug | D_ERROR) & level))
1765                 return;
1766
1767         CDEBUG(level, "--- Namespace: %s (rc: %d, side: %s)\n",
1768                ldlm_ns_name(ns), atomic_read(&ns->ns_bref),
1769                ns_is_client(ns) ? "client" : "server");
1770
1771         if (ktime_get_seconds() < ns->ns_next_dump)
1772                 return;
1773
1774         cfs_hash_for_each_nolock(ns->ns_rs_hash,
1775                                  ldlm_res_hash_dump,
1776                                  (void *)(unsigned long)level, 0);
1777         spin_lock(&ns->ns_lock);
1778         ns->ns_next_dump = ktime_get_seconds() + 10;
1779         spin_unlock(&ns->ns_lock);
1780 }
1781
1782 /* Print information about all locks in this resource to debug log. */
1783 void ldlm_resource_dump(int level, struct ldlm_resource *res)
1784 {
1785         struct ldlm_lock *lock;
1786         unsigned int granted = 0;
1787
1788         BUILD_BUG_ON(RES_NAME_SIZE != 4);
1789
1790         if (!((libcfs_debug | D_ERROR) & level))
1791                 return;
1792
1793         CDEBUG(level, "--- Resource: "DLDLMRES" (%p) refcount = %d\n",
1794                PLDLMRES(res), res, refcount_read(&res->lr_refcount));
1795
1796         if (!list_empty(&res->lr_granted)) {
1797                 CDEBUG(level, "Granted locks (in reverse order):\n");
1798                 list_for_each_entry_reverse(lock, &res->lr_granted,
1799                                                 l_res_link) {
1800                         LDLM_DEBUG_LIMIT(level, lock, "###");
1801                         if (!(level & D_CANTMASK) &&
1802                             ++granted > ldlm_dump_granted_max) {
1803                                 CDEBUG(level,
1804                                        "only dump %d granted locks to avoid DDOS.\n",
1805                                        granted);
1806                                 break;
1807                         }
1808                 }
1809         }
1810
1811         if (!list_empty(&res->lr_waiting)) {
1812                 CDEBUG(level, "Waiting locks:\n");
1813                 list_for_each_entry(lock, &res->lr_waiting, l_res_link)
1814                         LDLM_DEBUG_LIMIT(level, lock, "###");
1815         }
1816 }
1817 EXPORT_SYMBOL(ldlm_resource_dump);