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 Cluster File Systems, Inc.
   5  *
   6  * This code is issued under the GNU General Public License.
   7  * See the file COPYING in this distribution
   8  *
   9  * by Cluster File Systems, Inc.
  10  */
  11
  12 #define DEBUG_SUBSYSTEM S_LDLM
  13
  14 #include <linux/lustre_dlm.h>
  15
  16 kmem_cache_t *ldlm_resource_slab, *ldlm_lock_slab;
  17
  18 spinlock_t ldlm_namespace_lock = SPIN_LOCK_UNLOCKED;
  19 struct list_head ldlm_namespace_list = LIST_HEAD_INIT(ldlm_namespace_list);
  20 static struct proc_dir_entry *ldlm_ns_proc_dir = NULL;
  21
  22 int ldlm_proc_setup(struct obd_device *obd)
  23 {
  24         ENTRY;
  25
  26         if (obd->obd_proc_entry == NULL)
  27                 RETURN(-EINVAL);
  28
  29         ldlm_ns_proc_dir = proc_mkdir("namespaces", obd->obd_proc_entry);
  30         if (ldlm_ns_proc_dir == NULL) {
  31                 CERROR("Couldn't create /proc/lustre/ldlm/namespaces\n");
  32                 RETURN(-EPERM);
  33         }
  34         RETURN(0);
  35 }
  36
  37 void ldlm_proc_cleanup(struct obd_device *obd)
  38 {
  39         proc_lustre_remove_obd_entry("namespaces", obd);
  40 }
  41
  42 struct ldlm_namespace *ldlm_namespace_new(char *name, __u32 client)
  43 {
  44         struct ldlm_namespace *ns = NULL;
  45         struct list_head *bucket;
  46
  47         OBD_ALLOC(ns, sizeof(*ns));
  48         if (!ns) {
  49                 LBUG();
  50                 GOTO(out, NULL);
  51         }
  52
  53         ns->ns_hash = vmalloc(sizeof(*ns->ns_hash) * RES_HASH_SIZE);
  54         if (!ns->ns_hash) {
  55                 LBUG();
  56                 GOTO(out, ns);
  57         }
  58         obd_memory += sizeof(*ns->ns_hash) * RES_HASH_SIZE;
  59
  60         OBD_ALLOC(ns->ns_name, strlen(name) + 1);
  61         if (!ns->ns_name) {
  62                 LBUG();
  63                 GOTO(out, ns);
  64         }
  65         strcpy(ns->ns_name, name);
  66
  67         ptlrpc_init_client(NULL, NULL, LDLM_REQUEST_PORTAL, LDLM_REPLY_PORTAL,
  68                            &ns->ns_rpc_client);
  69
  70         INIT_LIST_HEAD(&ns->ns_root_list);
  71         l_lock_init(&ns->ns_lock);
  72         ns->ns_refcount = 0;
  73         ns->ns_client = client;
  74
  75         for (bucket = ns->ns_hash + RES_HASH_SIZE - 1; bucket >= ns->ns_hash;
  76              bucket--)
  77                 INIT_LIST_HEAD(bucket);
  78
  79         spin_lock(&ldlm_namespace_lock);
  80         list_add(&ns->ns_list_chain, &ldlm_namespace_list);
  81         ns->ns_proc_dir = proc_mkdir(ns->ns_name, ldlm_ns_proc_dir);
  82         if (ns->ns_proc_dir == NULL)
  83                 CERROR("Unable to create proc directory for namespace.\n");
  84         spin_unlock(&ldlm_namespace_lock);
  85
  86         RETURN(ns);
  87
  88  out:
  89         if (ns && ns->ns_hash) {
  90                 vfree(ns->ns_hash);
  91                 obd_memory -= sizeof(*ns->ns_hash) * RES_HASH_SIZE;
  92         }
  93         if (ns && ns->ns_name)
  94                 OBD_FREE(ns->ns_name, strlen(name) + 1);
  95         if (ns)
  96                 OBD_FREE(ns, sizeof(*ns));
  97         return NULL;
  98 }
  99
 100 extern struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock);
 101
 102 static void cleanup_resource(struct ldlm_resource *res, struct list_head *q)
 103 {
 104         struct list_head *tmp, *pos;
 105         int rc = 0, client = res->lr_namespace->ns_client;
 106         ENTRY;
 107
 108         list_for_each_safe(tmp, pos, q) {
 109                 struct ldlm_lock *lock;
 110                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 111                 LDLM_LOCK_GET(lock);
 112
 113                 if (client) {
 114                         struct lustre_handle lockh;
 115                         ldlm_lock2handle(lock, &lockh);
 116                         /* can we get away without a connh here? */
 117                         rc = ldlm_cli_cancel(&lockh);
 118                         if (rc != ELDLM_OK) {
 119                                 /* It failed remotely, but we'll force it to
 120                                  * cleanup locally. */
 121                                 CERROR("ldlm_cli_cancel: %d\n", rc);
 122                                 ldlm_lock_cancel(lock);
 123                         }
 124                 } else {
 125                         LDLM_DEBUG(lock, "Freeing a lock still held by a "
 126                                    "client node.\n");
 127
 128                         ldlm_resource_unlink_lock(lock);
 129                         ldlm_lock_destroy(lock);
 130                 }
 131                 LDLM_LOCK_PUT(lock);
 132         }
 133 }
 134
 135 int ldlm_namespace_free(struct ldlm_namespace *ns)
 136 {
 137         struct list_head *tmp, *pos;
 138         int i;
 139
 140         if (!ns)
 141                 RETURN(ELDLM_OK);
 142
 143         spin_lock(&ldlm_namespace_lock);
 144         list_del(&ns->ns_list_chain);
 145         remove_proc_entry(ns->ns_name, ldlm_ns_proc_dir);
 146         spin_unlock(&ldlm_namespace_lock);
 147
 148         l_lock(&ns->ns_lock);
 149
 150         for (i = 0; i < RES_HASH_SIZE; i++) {
 151                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
 152                         struct ldlm_resource *res;
 153                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
 154                         ldlm_resource_getref(res);
 155
 156                         cleanup_resource(res, &res->lr_granted);
 157                         cleanup_resource(res, &res->lr_converting);
 158                         cleanup_resource(res, &res->lr_waiting);
 159
 160                         if (!ldlm_resource_put(res)) {
 161                                 CERROR("Resource refcount nonzero (%d) after "
 162                                        "lock cleanup; forcing cleanup.\n",
 163                                        atomic_read(&res->lr_refcount));
 164                                 ldlm_resource_dump(res);
 165                                 atomic_set(&res->lr_refcount, 1);
 166                                 ldlm_resource_put(res);
 167                         }
 168                 }
 169         }
 170
 171         vfree(ns->ns_hash /* , sizeof(*ns->ns_hash) * RES_HASH_SIZE */);
 172         obd_memory -= sizeof(*ns->ns_hash) * RES_HASH_SIZE;
 173         ptlrpc_cleanup_client(&ns->ns_rpc_client);
 174         OBD_FREE(ns->ns_name, strlen(ns->ns_name) + 1);
 175         OBD_FREE(ns, sizeof(*ns));
 176
 177         return ELDLM_OK;
 178 }
 179
 180 static __u32 ldlm_hash_fn(struct ldlm_resource *parent, __u64 *name)
 181 {
 182         __u32 hash = 0;
 183         int i;
 184
 185         for (i = 0; i < RES_NAME_SIZE; i++)
 186                 hash += name[i];
 187
 188         hash += (__u32)((unsigned long)parent >> 4);
 189
 190         return (hash & RES_HASH_MASK);
 191 }
 192
 193 static struct ldlm_resource *ldlm_resource_new(void)
 194 {
 195         struct ldlm_resource *res;
 196
 197         res = kmem_cache_alloc(ldlm_resource_slab, SLAB_KERNEL);
 198         if (res == NULL) {
 199                 LBUG();
 200                 return NULL;
 201         }
 202         memset(res, 0, sizeof(*res));
 203
 204         INIT_LIST_HEAD(&res->lr_children);
 205         INIT_LIST_HEAD(&res->lr_childof);
 206         INIT_LIST_HEAD(&res->lr_granted);
 207         INIT_LIST_HEAD(&res->lr_converting);
 208         INIT_LIST_HEAD(&res->lr_waiting);
 209
 210         atomic_set(&res->lr_refcount, 1);
 211
 212         return res;
 213 }
 214
 215 /* Args: locked namespace
 216  * Returns: newly-allocated, referenced, unlocked resource */
 217 static struct ldlm_resource *ldlm_resource_add(struct ldlm_namespace *ns,
 218                                                struct ldlm_resource *parent,
 219                                                __u64 *name, __u32 type)
 220 {
 221         struct list_head *bucket;
 222         struct ldlm_resource *res;
 223         ENTRY;
 224
 225         if (type < LDLM_MIN_TYPE || type > LDLM_MAX_TYPE) {
 226                 LBUG();
 227                 RETURN(NULL);
 228         }
 229
 230         res = ldlm_resource_new();
 231         if (!res) {
 232                 LBUG();
 233                 RETURN(NULL);
 234         }
 235
 236         memcpy(res->lr_name, name, sizeof(res->lr_name));
 237         res->lr_namespace = ns;
 238         ns->ns_refcount++;
 239
 240         res->lr_type = type;
 241         res->lr_most_restr = LCK_NL;
 242
 243         bucket = ns->ns_hash + ldlm_hash_fn(parent, name);
 244         list_add(&res->lr_hash, bucket);
 245
 246         if (parent == NULL)
 247                 list_add(&res->lr_childof, &ns->ns_root_list);
 248         else {
 249                 res->lr_parent = parent;
 250                 list_add(&res->lr_childof, &parent->lr_children);
 251         }
 252
 253         RETURN(res);
 254 }
 255
 256 /* Args: unlocked namespace
 257  * Locks: takes and releases ns->ns_lock and res->lr_lock
 258  * Returns: referenced, unlocked ldlm_resource or NULL */
 259 struct ldlm_resource *ldlm_resource_get(struct ldlm_namespace *ns,
 260                                         struct ldlm_resource *parent,
 261                                         __u64 *name, __u32 type, int create)
 262 {
 263         struct list_head *bucket;
 264         struct list_head *tmp = bucket;
 265         struct ldlm_resource *res = NULL;
 266         ENTRY;
 267
 268         if (ns == NULL || ns->ns_hash == NULL) {
 269                 LBUG();
 270                 RETURN(NULL);
 271         }
 272
 273         l_lock(&ns->ns_lock);
 274         bucket = ns->ns_hash + ldlm_hash_fn(parent, name);
 275
 276         list_for_each(tmp, bucket) {
 277                 struct ldlm_resource *chk;
 278                 chk = list_entry(tmp, struct ldlm_resource, lr_hash);
 279
 280                 if (memcmp(chk->lr_name, name, sizeof(chk->lr_name)) == 0) {
 281                         res = chk;
 282                         atomic_inc(&res->lr_refcount);
 283                         EXIT;
 284                         break;
 285                 }
 286         }
 287
 288         if (res == NULL && create)
 289                 res = ldlm_resource_add(ns, parent, name, type);
 290         l_unlock(&ns->ns_lock);
 291
 292         RETURN(res);
 293 }
 294
 295 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
 296 {
 297         atomic_inc(&res->lr_refcount);
 298         return res;
 299 }
 300
 301 /* Returns 1 if the resource was freed, 0 if it remains. */
 302 int ldlm_resource_put(struct ldlm_resource *res)
 303 {
 304         int rc = 0;
 305
 306         if (atomic_dec_and_test(&res->lr_refcount)) {
 307                 struct ldlm_namespace *ns = res->lr_namespace;
 308                 ENTRY;
 309
 310                 l_lock(&ns->ns_lock);
 311
 312                 if (atomic_read(&res->lr_refcount) != 0) {
 313                         /* We lost the race. */
 314                         l_unlock(&ns->ns_lock);
 315                         goto out;
 316                 }
 317
 318                 if (!list_empty(&res->lr_granted))
 319                         LBUG();
 320
 321                 if (!list_empty(&res->lr_converting))
 322                         LBUG();
 323
 324                 if (!list_empty(&res->lr_waiting))
 325                         LBUG();
 326
 327                 if (!list_empty(&res->lr_children))
 328                         LBUG();
 329
 330                 ns->ns_refcount--;
 331                 list_del(&res->lr_hash);
 332                 list_del(&res->lr_childof);
 333
 334                 kmem_cache_free(ldlm_resource_slab, res);
 335                 l_unlock(&ns->ns_lock);
 336                 rc = 1;
 337         } else {
 338                 ENTRY;
 339         out:
 340                 if (atomic_read(&res->lr_refcount) < 0)
 341                         LBUG();
 342         }
 343
 344         RETURN(rc);
 345 }
 346
 347 void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
 348                             struct ldlm_lock *lock)
 349 {
 350         l_lock(&res->lr_namespace->ns_lock);
 351
 352         ldlm_resource_dump(res);
 353         ldlm_lock_dump(lock);
 354
 355         if (!list_empty(&lock->l_res_link))
 356                 LBUG();
 357
 358         list_add(&lock->l_res_link, head);
 359         l_unlock(&res->lr_namespace->ns_lock);
 360 }
 361
 362 void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
 363 {
 364         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 365         list_del_init(&lock->l_res_link);
 366         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 367 }
 368
 369 void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
 370 {
 371         desc->lr_type = res->lr_type;
 372         memcpy(desc->lr_name, res->lr_name, sizeof(desc->lr_name));
 373         memcpy(desc->lr_version, res->lr_version, sizeof(desc->lr_version));
 374 }
 375
 376 void ldlm_dump_all_namespaces(void)
 377 {
 378         struct list_head *tmp;
 379
 380         spin_lock(&ldlm_namespace_lock);
 381
 382         list_for_each(tmp, &ldlm_namespace_list) {
 383                 struct ldlm_namespace *ns;
 384                 ns = list_entry(tmp, struct ldlm_namespace, ns_list_chain);
 385                 ldlm_namespace_dump(ns);
 386         }
 387
 388         spin_unlock(&ldlm_namespace_lock);
 389 }
 390
 391 void ldlm_namespace_dump(struct ldlm_namespace *ns)
 392 {
 393         struct list_head *tmp;
 394
 395         l_lock(&ns->ns_lock);
 396         CDEBUG(D_OTHER, "--- Namespace: %s (rc: %d, client: %d)\n", ns->ns_name,
 397                ns->ns_refcount, ns->ns_client);
 398
 399         list_for_each(tmp, &ns->ns_root_list) {
 400                 struct ldlm_resource *res;
 401                 res = list_entry(tmp, struct ldlm_resource, lr_childof);
 402
 403                 /* Once we have resources with children, this should really dump
 404                  * them recursively. */
 405                 ldlm_resource_dump(res);
 406         }
 407         l_unlock(&ns->ns_lock);
 408 }
 409
 410 void ldlm_resource_dump(struct ldlm_resource *res)
 411 {
 412         struct list_head *tmp;
 413         char name[256];
 414
 415         if (RES_NAME_SIZE != 3)
 416                 LBUG();
 417
 418         snprintf(name, sizeof(name), "%Lx %Lx %Lx",
 419                  (unsigned long long)res->lr_name[0],
 420                  (unsigned long long)res->lr_name[1],
 421                  (unsigned long long)res->lr_name[2]);
 422
 423         CDEBUG(D_OTHER, "--- Resource: %p (%s) (rc: %d)\n", res, name,
 424                atomic_read(&res->lr_refcount));
 425         CDEBUG(D_OTHER, "Namespace: %p (%s)\n", res->lr_namespace,
 426                res->lr_namespace->ns_name);
 427         CDEBUG(D_OTHER, "Parent: %p, root: %p\n", res->lr_parent, res->lr_root);
 428
 429         CDEBUG(D_OTHER, "Granted locks:\n");
 430         list_for_each(tmp, &res->lr_granted) {
 431                 struct ldlm_lock *lock;
 432                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 433                 ldlm_lock_dump(lock);
 434         }
 435
 436         CDEBUG(D_OTHER, "Converting locks:\n");
 437         list_for_each(tmp, &res->lr_converting) {
 438                 struct ldlm_lock *lock;
 439                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 440                 ldlm_lock_dump(lock);
 441         }
 442
 443         CDEBUG(D_OTHER, "Waiting locks:\n");
 444         list_for_each(tmp, &res->lr_waiting) {
 445                 struct ldlm_lock *lock;
 446                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 447                 ldlm_lock_dump(lock);
 448         }
 449 }