lustre/ldlm/ldlm_lock.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  *   Author: Peter Braam <braam@clusterfs.com>
   6  *   Author: Phil Schwan <phil@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
  26 #include <linux/slab.h>
  27 #include <linux/module.h>
  28 #include <linux/random.h>
  29 #include <linux/lustre_dlm.h>
  30 #include <linux/lustre_mds.h>
  31
  32 /* lock types */
  33 char *ldlm_lockname[] = {
  34         [0] "--",
  35         [LCK_EX] "EX",
  36         [LCK_PW] "PW",
  37         [LCK_PR] "PR",
  38         [LCK_CW] "CW",
  39         [LCK_CR] "CR",
  40         [LCK_NL] "NL"
  41 };
  42 char *ldlm_typename[] = {
  43         [LDLM_PLAIN] "PLN",
  44         [LDLM_EXTENT] "EXT",
  45         [LDLM_MDSINTENT] "INT"
  46 };
  47
  48 char *ldlm_it2str(int it)
  49 {
  50         switch (it) {
  51         case IT_OPEN:
  52                 return "open";
  53         case IT_CREAT:
  54                 return "creat";
  55         case (IT_OPEN | IT_CREAT):
  56                 return "open|creat";
  57         case IT_MKDIR:
  58                 return "mkdir";
  59         case IT_LINK:
  60                 return "link";
  61         case IT_SYMLINK:
  62                 return "symlink";
  63         case IT_UNLINK:
  64                 return "unlink";
  65         case IT_RMDIR:
  66                 return "rmdir";
  67         case IT_RENAME:
  68                 return "rename";
  69         case IT_RENAME2:
  70                 return "rename2";
  71         case IT_READDIR:
  72                 return "readdir";
  73         case IT_GETATTR:
  74                 return "getattr";
  75         case IT_SETATTR:
  76                 return "setattr";
  77         case IT_READLINK:
  78                 return "readlink";
  79         case IT_MKNOD:
  80                 return "mknod";
  81         case IT_LOOKUP:
  82                 return "lookup";
  83         default:
  84                 CERROR("Unknown intent %d\n", it);
  85                 return "UNKNOWN";
  86         }
  87 }
  88
  89 extern kmem_cache_t *ldlm_lock_slab;
  90
  91 static int ldlm_plain_compat(struct ldlm_lock *a, struct ldlm_lock *b);
  92
  93 ldlm_res_compat ldlm_res_compat_table[] = {
  94         [LDLM_PLAIN] ldlm_plain_compat,
  95         [LDLM_EXTENT] ldlm_extent_compat,
  96         [LDLM_MDSINTENT] ldlm_plain_compat
  97 };
  98
  99 ldlm_res_policy ldlm_res_policy_table[] = {
 100         [LDLM_PLAIN] NULL,
 101         [LDLM_EXTENT] ldlm_extent_policy,
 102         [LDLM_MDSINTENT] NULL
 103 };
 104
 105 void ldlm_register_intent(int (*arg) (struct ldlm_lock * lock, void *req_cookie,
 106                                       ldlm_mode_t mode, void *data))
 107 {
 108         ldlm_res_policy_table[LDLM_MDSINTENT] = arg;
 109 }
 110
 111 void ldlm_unregister_intent(void)
 112 {
 113         ldlm_res_policy_table[LDLM_MDSINTENT] = NULL;
 114 }
 115
 116 /*
 117  * REFCOUNTED LOCK OBJECTS
 118  */
 119
 120
 121 /*
 122  * Lock refcounts, during creation:
 123  *   - one special one for allocation, dec'd only once in destroy
 124  *   - one for being a lock that's in-use
 125  *   - one for the addref associated with a new lock
 126  */
 127 struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock)
 128 {
 129         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 130         lock->l_refc++;
 131         ldlm_resource_getref(lock->l_resource);
 132         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 133         return lock;
 134 }
 135
 136 void ldlm_lock_put(struct ldlm_lock *lock)
 137 {
 138         struct lustre_lock *nslock = &lock->l_resource->lr_namespace->ns_lock;
 139         ENTRY;
 140
 141         l_lock(nslock);
 142         lock->l_refc--;
 143         //LDLM_DEBUG(lock, "after refc--");
 144         if (lock->l_refc < 0)
 145                 LBUG();
 146
 147         ldlm_resource_put(lock->l_resource);
 148         if (lock->l_parent)
 149                 LDLM_LOCK_PUT(lock->l_parent);
 150
 151         if (lock->l_refc == 0 && (lock->l_flags & LDLM_FL_DESTROYED)) {
 152                 lock->l_resource = NULL;
 153                 LDLM_DEBUG(lock, "final lock_put on destroyed lock, freeing");
 154                 if (lock->l_connection)
 155                         ptlrpc_put_connection(lock->l_connection);
 156                 CDEBUG(D_MALLOC, "kfreed 'lock': %d at %p (tot 1).\n",
 157                        sizeof(*lock), lock);
 158                 kmem_cache_free(ldlm_lock_slab, lock);
 159         }
 160         l_unlock(nslock);
 161         EXIT;
 162 }
 163
 164 void ldlm_lock_destroy(struct ldlm_lock *lock)
 165 {
 166         ENTRY;
 167         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 168
 169         if (!list_empty(&lock->l_children)) {
 170                 LDLM_DEBUG(lock, "still has children (%p)!",
 171                            lock->l_children.next);
 172                 ldlm_lock_dump(lock);
 173                 LBUG();
 174         }
 175         if (lock->l_readers || lock->l_writers) {
 176                 LDLM_DEBUG(lock, "lock still has references");
 177                 ldlm_lock_dump(lock);
 178                 LBUG();
 179         }
 180
 181         if (!list_empty(&lock->l_res_link)) {
 182                 ldlm_lock_dump(lock);
 183                 LBUG();
 184         }
 185
 186         if (lock->l_flags & LDLM_FL_DESTROYED) {
 187                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 188                 EXIT;
 189                 return;
 190         }
 191
 192         lock->l_flags = LDLM_FL_DESTROYED;
 193         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 194         LDLM_LOCK_PUT(lock);
 195         EXIT;
 196 }
 197
 198 /*
 199    usage: pass in a resource on which you have done get
 200           pass in a parent lock on which you have done a get
 201           do not put the resource or the parent
 202    returns: lock with refcount 1
 203 */
 204 static struct ldlm_lock *ldlm_lock_new(struct ldlm_lock *parent,
 205                                        struct ldlm_resource *resource)
 206 {
 207         struct ldlm_lock *lock;
 208         ENTRY;
 209
 210         if (resource == NULL)
 211                 LBUG();
 212
 213         lock = kmem_cache_alloc(ldlm_lock_slab, SLAB_KERNEL);
 214         if (lock == NULL)
 215                 RETURN(NULL);
 216         CDEBUG(D_MALLOC, "kmalloced 'lock': %d at "
 217                "%p (tot %d).\n", sizeof(*lock), lock, 1);
 218
 219         memset(lock, 0, sizeof(*lock));
 220         get_random_bytes(&lock->l_random, sizeof(__u64));
 221
 222         lock->l_resource = resource;
 223         /* this refcount matches the one of the resource passed
 224            in which is not being put away */
 225         lock->l_refc = 1;
 226         INIT_LIST_HEAD(&lock->l_children);
 227         INIT_LIST_HEAD(&lock->l_res_link);
 228         init_waitqueue_head(&lock->l_waitq);
 229
 230         if (parent != NULL) {
 231                 l_lock(&parent->l_resource->lr_namespace->ns_lock);
 232                 lock->l_parent = parent;
 233                 list_add(&lock->l_childof, &parent->l_children);
 234                 l_unlock(&parent->l_resource->lr_namespace->ns_lock);
 235         }
 236         /* this is the extra refcount, to prevent the lock
 237            evaporating */
 238         LDLM_LOCK_GET(lock);
 239         RETURN(lock);
 240 }
 241
 242 int ldlm_lock_change_resource(struct ldlm_lock *lock, __u64 new_resid[3])
 243 {
 244         struct ldlm_namespace *ns = lock->l_resource->lr_namespace;
 245         struct ldlm_resource *oldres = lock->l_resource;
 246         int type, i;
 247         ENTRY;
 248
 249         l_lock(&ns->ns_lock);
 250         if (memcmp(new_resid, lock->l_resource->lr_name,
 251                    sizeof(lock->l_resource->lr_name)) == 0) {
 252                 /* Nothing to do */
 253                 l_unlock(&ns->ns_lock);
 254                 RETURN(0);
 255         }
 256
 257         type = lock->l_resource->lr_type;
 258         if (new_resid[0] == 0)
 259                 LBUG();
 260         lock->l_resource = ldlm_resource_get(ns, NULL, new_resid, type, 1);
 261         if (lock->l_resource == NULL) {
 262                 LBUG();
 263                 RETURN(-ENOMEM);
 264         }
 265
 266         /* move references over */
 267         for (i = 0; i < lock->l_refc; i++) {
 268                 int rc;
 269                 ldlm_resource_getref(lock->l_resource);
 270                 rc = ldlm_resource_put(oldres);
 271                 if (rc == 1 && i != lock->l_refc - 1)
 272                         LBUG();
 273         }
 274         /* compensate for the initial get above.. */
 275         ldlm_resource_put(lock->l_resource);
 276
 277         l_unlock(&ns->ns_lock);
 278         RETURN(0);
 279 }
 280
 281 /*
 282  *  HANDLES
 283  */
 284
 285 void ldlm_lock2handle(struct ldlm_lock *lock, struct lustre_handle *lockh)
 286 {
 287         lockh->addr = (__u64) (unsigned long)lock;
 288         lockh->cookie = lock->l_random;
 289 }
 290
 291 struct ldlm_lock *ldlm_handle2lock(struct lustre_handle *handle)
 292 {
 293         struct ldlm_lock *lock = NULL, *retval = NULL;
 294         ENTRY;
 295
 296         if (!handle || !handle->addr)
 297                 RETURN(NULL);
 298
 299         lock = (struct ldlm_lock *)(unsigned long)(handle->addr);
 300         if (!kmem_cache_validate(ldlm_lock_slab, (void *)lock))
 301                 RETURN(NULL);
 302
 303         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 304         if (lock->l_random != handle->cookie)
 305                 GOTO(out, NULL);
 306
 307         if (lock->l_flags & LDLM_FL_DESTROYED)
 308                 GOTO(out, NULL);
 309
 310         retval = LDLM_LOCK_GET(lock);
 311         EXIT;
 312       out:
 313         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 314         return retval;
 315 }
 316
 317
 318
 319
 320 static int ldlm_plain_compat(struct ldlm_lock *a, struct ldlm_lock *b)
 321 {
 322         return lockmode_compat(a->l_req_mode, b->l_req_mode);
 323 }
 324
 325 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)
 326 {
 327         ldlm_res2desc(lock->l_resource, &desc->l_resource);
 328         desc->l_req_mode = lock->l_req_mode;
 329         desc->l_granted_mode = lock->l_granted_mode;
 330         memcpy(&desc->l_extent, &lock->l_extent, sizeof(desc->l_extent));
 331         memcpy(desc->l_version, lock->l_version, sizeof(desc->l_version));
 332 }
 333
 334 static void ldlm_add_ast_work_item(struct ldlm_lock *lock,
 335                                    struct ldlm_lock *new)
 336 {
 337         struct ldlm_ast_work *w;
 338         ENTRY;
 339
 340         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 341         if (new && (lock->l_flags & LDLM_FL_AST_SENT))
 342                 GOTO(out, 0);
 343
 344         OBD_ALLOC(w, sizeof(*w));
 345         if (!w) {
 346                 LBUG();
 347                 GOTO(out, 0);
 348         }
 349
 350         if (new) {
 351                 lock->l_flags |= LDLM_FL_AST_SENT;
 352                 w->w_blocking = 1;
 353                 ldlm_lock2desc(new, &w->w_desc);
 354         }
 355
 356         w->w_lock = LDLM_LOCK_GET(lock);
 357         list_add(&w->w_list, lock->l_resource->lr_tmp);
 358       out:
 359         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 360         return;
 361 }
 362
 363 void ldlm_lock_addref(struct lustre_handle *lockh, __u32 mode)
 364 {
 365         struct ldlm_lock *lock;
 366
 367         lock = ldlm_handle2lock(lockh);
 368         ldlm_lock_addref_internal(lock, mode);
 369         LDLM_LOCK_PUT(lock);
 370 }
 371
 372 /* only called for local locks */
 373 void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
 374 {
 375         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 376         if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR)
 377                 lock->l_readers++;
 378         else
 379                 lock->l_writers++;
 380         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 381         LDLM_LOCK_GET(lock);
 382         LDLM_DEBUG(lock, "ldlm_lock_addref(%s)", ldlm_lockname[mode]);
 383 }
 384
 385 /* Args: unlocked lock */
 386 void ldlm_lock_decref(struct lustre_handle *lockh, __u32 mode)
 387 {
 388         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
 389         ENTRY;
 390
 391         if (lock == NULL)
 392                 LBUG();
 393
 394         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
 395         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 396         if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR)
 397                 lock->l_readers--;
 398         else
 399                 lock->l_writers--;
 400
 401         /* If we received a blocked AST and this was the last reference,
 402          * run the callback. */
 403         if (!lock->l_readers && !lock->l_writers &&
 404             (lock->l_flags & LDLM_FL_CBPENDING)) {
 405                 if (!lock->l_resource->lr_namespace->ns_client) {
 406                         CERROR("LDLM_FL_CBPENDING set on non-local lock!\n");
 407                         LBUG();
 408                 }
 409
 410                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
 411                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 412
 413                 /* FIXME: need a real 'desc' here */
 414                 lock->l_blocking_ast(lock, NULL, lock->l_data,
 415                                      lock->l_data_len);
 416         } else
 417                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 418
 419         LDLM_LOCK_PUT(lock);    /* matches the ldlm_lock_get in addref */
 420         LDLM_LOCK_PUT(lock);    /* matches the handle2lock above */
 421
 422         EXIT;
 423 }
 424
 425 static int ldlm_lock_compat_list(struct ldlm_lock *lock, int send_cbs,
 426                                  struct list_head *queue)
 427 {
 428         struct list_head *tmp, *pos;
 429         int rc = 1;
 430
 431         list_for_each_safe(tmp, pos, queue) {
 432                 struct ldlm_lock *child;
 433                 ldlm_res_compat compat;
 434
 435                 child = list_entry(tmp, struct ldlm_lock, l_res_link);
 436                 if (lock == child)
 437                         continue;
 438
 439                 compat = ldlm_res_compat_table[child->l_resource->lr_type];
 440                 if (compat && compat(child, lock)) {
 441                         CDEBUG(D_OTHER, "compat function succeded, next.\n");
 442                         continue;
 443                 }
 444                 if (lockmode_compat(child->l_granted_mode, lock->l_req_mode)) {
 445                         CDEBUG(D_OTHER, "lock modes are compatible, next.\n");
 446                         continue;
 447                 }
 448
 449                 rc = 0;
 450
 451                 if (send_cbs && child->l_blocking_ast != NULL) {
 452                         CDEBUG(D_OTHER, "incompatible; sending blocking "
 453                                "AST.\n");
 454                         ldlm_add_ast_work_item(child, lock);
 455                 }
 456         }
 457
 458         return rc;
 459 }
 460
 461 static int ldlm_lock_compat(struct ldlm_lock *lock, int send_cbs)
 462 {
 463         int rc;
 464         ENTRY;
 465
 466         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 467         rc = ldlm_lock_compat_list(lock, send_cbs,
 468                                    &lock->l_resource->lr_granted);
 469         /* FIXME: should we be sending ASTs to converting? */
 470         if (rc)
 471                 rc = ldlm_lock_compat_list
 472                         (lock, send_cbs, &lock->l_resource->lr_converting);
 473
 474         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 475         RETURN(rc);
 476 }
 477
 478 /* NOTE: called by
 479    - ldlm_handle_enqueuque - resource
 480 */
 481 void ldlm_grant_lock(struct ldlm_lock *lock)
 482 {
 483         struct ldlm_resource *res = lock->l_resource;
 484         ENTRY;
 485
 486         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 487         ldlm_resource_add_lock(res, &res->lr_granted, lock);
 488         lock->l_granted_mode = lock->l_req_mode;
 489
 490         if (lock->l_granted_mode < res->lr_most_restr)
 491                 res->lr_most_restr = lock->l_granted_mode;
 492
 493         if (lock->l_completion_ast) {
 494                 ldlm_add_ast_work_item(lock, NULL);
 495         }
 496         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 497         EXIT;
 498 }
 499
 500 /* returns a referenced lock or NULL */
 501 static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
 502                                       struct ldlm_extent *extent)
 503 {
 504         struct ldlm_lock *lock;
 505         struct list_head *tmp;
 506
 507         list_for_each(tmp, queue) {
 508                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 509
 510                 if (lock->l_flags & LDLM_FL_CBPENDING)
 511                         continue;
 512
 513                 /* lock_convert() takes the resource lock, so we're sure that
 514                  * req_mode, lr_type, and l_cookie won't change beneath us */
 515                 if (lock->l_req_mode != mode)
 516                         continue;
 517
 518                 if (lock->l_resource->lr_type == LDLM_EXTENT &&
 519                     (lock->l_extent.start > extent->start ||
 520                      lock->l_extent.end < extent->end))
 521                         continue;
 522
 523                 ldlm_lock_addref_internal(lock, mode);
 524                 return lock;
 525         }
 526
 527         return NULL;
 528 }
 529
 530 /* Must be called with no resource or lock locks held.
 531  *
 532  * Returns 1 if it finds an already-existing lock that is compatible; in this
 533  * case, lockh is filled in with a addref()ed lock
 534 */
 535 int ldlm_lock_match(struct ldlm_namespace *ns, __u64 * res_id, __u32 type,
 536                     void *cookie, int cookielen, ldlm_mode_t mode,
 537                     struct lustre_handle *lockh)
 538 {
 539         struct ldlm_resource *res;
 540         struct ldlm_lock *lock;
 541         int rc = 0;
 542         ENTRY;
 543
 544         res = ldlm_resource_get(ns, NULL, res_id, type, 0);
 545         if (res == NULL)
 546                 RETURN(0);
 547
 548         ns = res->lr_namespace;
 549         l_lock(&ns->ns_lock);
 550
 551         if ((lock = search_queue(&res->lr_granted, mode, cookie)))
 552                 GOTO(out, rc = 1);
 553         if ((lock = search_queue(&res->lr_converting, mode, cookie)))
 554                 GOTO(out, rc = 1);
 555         if ((lock = search_queue(&res->lr_waiting, mode, cookie)))
 556                 GOTO(out, rc = 1);
 557
 558         EXIT;
 559       out:
 560         ldlm_resource_put(res);
 561         l_unlock(&ns->ns_lock);
 562
 563         if (lock) {
 564                 ldlm_lock2handle(lock, lockh);
 565                 if (lock->l_completion_ast)
 566                         lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC);
 567         }
 568         if (rc)
 569                 LDLM_DEBUG(lock, "matched");
 570         else
 571                 LDLM_DEBUG_NOLOCK("not matched");
 572         return rc;
 573 }
 574
 575 /* Returns a referenced lock */
 576 struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns,
 577                                    struct lustre_handle *parent_lock_handle,
 578                                    __u64 * res_id, __u32 type,
 579                                    ldlm_mode_t mode, void *data, __u32 data_len)
 580 {
 581         struct ldlm_resource *res, *parent_res = NULL;
 582         struct ldlm_lock *lock, *parent_lock;
 583
 584         parent_lock = ldlm_handle2lock(parent_lock_handle);
 585         if (parent_lock)
 586                 parent_res = parent_lock->l_resource;
 587
 588         res = ldlm_resource_get(ns, parent_res, res_id, type, 1);
 589         if (res == NULL)
 590                 RETURN(NULL);
 591
 592         lock = ldlm_lock_new(parent_lock, res);
 593         if (lock == NULL) {
 594                 ldlm_resource_put(res);
 595                 RETURN(NULL);
 596         }
 597
 598         lock->l_req_mode = mode;
 599         lock->l_data = data;
 600         lock->l_data_len = data_len;
 601
 602         return lock;
 603 }
 604
 605 /* Must be called with lock->l_lock and lock->l_resource->lr_lock not held */
 606 ldlm_error_t ldlm_lock_enqueue(struct ldlm_lock * lock,
 607                                void *cookie, int cookie_len,
 608                                int *flags,
 609                                ldlm_completion_callback completion,
 610                                ldlm_blocking_callback blocking)
 611 {
 612         struct ldlm_resource *res;
 613         int local;
 614         ldlm_res_policy policy;
 615         ENTRY;
 616
 617         res = lock->l_resource;
 618         lock->l_blocking_ast = blocking;
 619
 620         if (res->lr_type == LDLM_EXTENT)
 621                 memcpy(&lock->l_extent, cookie, sizeof(lock->l_extent));
 622
 623         /* policies are not executed on the client */
 624         local = res->lr_namespace->ns_client;
 625         if (!local && (policy = ldlm_res_policy_table[res->lr_type])) {
 626                 int rc;
 627                 rc = policy(lock, cookie, lock->l_req_mode, NULL);
 628
 629                 if (rc == ELDLM_LOCK_CHANGED) {
 630                         res = lock->l_resource;
 631                         *flags |= LDLM_FL_LOCK_CHANGED;
 632                 } else if (rc == ELDLM_LOCK_ABORTED) {
 633                         ldlm_lock_destroy(lock);
 634                         RETURN(rc);
 635                 }
 636         }
 637
 638         lock->l_cookie = cookie;
 639         lock->l_cookie_len = cookie_len;
 640
 641         if (local && lock->l_req_mode == lock->l_granted_mode) {
 642                 /* The server returned a blocked lock, but it was granted before
 643                  * we got a chance to actually enqueue it.  We don't need to do
 644                  * anything else. */
 645                 GOTO(out, ELDLM_OK);
 646         }
 647
 648         /* This distinction between local lock trees is very important; a client
 649          * namespace only has information about locks taken by that client, and
 650          * thus doesn't have enough information to decide for itself if it can
 651          * be granted (below).  In this case, we do exactly what the server
 652          * tells us to do, as dictated by the 'flags' */
 653         ldlm_resource_unlink_lock(lock);
 654         if (local) {
 655                 if (*flags & LDLM_FL_BLOCK_CONV)
 656                         ldlm_resource_add_lock(res, res->lr_converting.prev,
 657                                                lock);
 658                 else if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED))
 659                         ldlm_resource_add_lock(res, res->lr_waiting.prev, lock);
 660                 else
 661                         ldlm_grant_lock(lock);
 662                 GOTO(out, ELDLM_OK);
 663         }
 664
 665         /* FIXME: We may want to optimize by checking lr_most_restr */
 666         if (!list_empty(&res->lr_converting)) {
 667                 ldlm_resource_add_lock(res, res->lr_waiting.prev, lock);
 668                 *flags |= LDLM_FL_BLOCK_CONV;
 669                 GOTO(out, ELDLM_OK);
 670         }
 671         if (!list_empty(&res->lr_waiting)) {
 672                 ldlm_resource_add_lock(res, res->lr_waiting.prev, lock);
 673                 *flags |= LDLM_FL_BLOCK_WAIT;
 674                 GOTO(out, ELDLM_OK);
 675         }
 676         if (!ldlm_lock_compat(lock, 0)) {
 677                 ldlm_resource_add_lock(res, res->lr_waiting.prev, lock);
 678                 *flags |= LDLM_FL_BLOCK_GRANTED;
 679                 GOTO(out, ELDLM_OK);
 680         }
 681
 682         ldlm_grant_lock(lock);
 683         EXIT;
 684       out:
 685         /* Don't set 'completion_ast' until here so that if the lock is granted
 686          * immediately we don't do an unnecessary completion call. */
 687         lock->l_completion_ast = completion;
 688         return ELDLM_OK;
 689 }
 690
 691 /* Must be called with namespace taken: queue is waiting or converting. */
 692 static int ldlm_reprocess_queue(struct ldlm_resource *res,
 693                                 struct list_head *queue)
 694 {
 695         struct list_head *tmp, *pos;
 696         ENTRY;
 697
 698         list_for_each_safe(tmp, pos, queue) {
 699                 struct ldlm_lock *pending;
 700                 pending = list_entry(tmp, struct ldlm_lock, l_res_link);
 701
 702                 CDEBUG(D_INFO, "Reprocessing lock %p\n", pending);
 703
 704                 if (!ldlm_lock_compat(pending, 1))
 705                         RETURN(1);
 706
 707                 list_del_init(&pending->l_res_link);
 708                 ldlm_grant_lock(pending);
 709         }
 710
 711         RETURN(0);
 712 }
 713
 714 void ldlm_run_ast_work(struct list_head *rpc_list)
 715 {
 716         struct list_head *tmp, *pos;
 717         int rc;
 718         ENTRY;
 719
 720         list_for_each_safe(tmp, pos, rpc_list) {
 721                 struct ldlm_ast_work *w =
 722                         list_entry(tmp, struct ldlm_ast_work, w_list);
 723
 724                 if (w->w_blocking)
 725                         rc = w->w_lock->l_blocking_ast
 726                                 (w->w_lock, &w->w_desc, w->w_data,
 727                                  w->w_datalen);
 728                 else
 729                         rc = w->w_lock->l_completion_ast(w->w_lock, w->w_flags);
 730                 if (rc)
 731                         CERROR("Failed AST - should clean & disconnect "
 732                                "client\n");
 733                 LDLM_LOCK_PUT(w->w_lock);
 734                 list_del(&w->w_list);
 735                 OBD_FREE(w, sizeof(*w));
 736         }
 737         EXIT;
 738 }
 739
 740 /* Must be called with resource->lr_lock not taken. */
 741 void ldlm_reprocess_all(struct ldlm_resource *res)
 742 {
 743         struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
 744         ENTRY;
 745
 746         /* Local lock trees don't get reprocessed. */
 747         if (res->lr_namespace->ns_client) {
 748                 EXIT;
 749                 return;
 750         }
 751
 752         l_lock(&res->lr_namespace->ns_lock);
 753         res->lr_tmp = &rpc_list;
 754
 755         ldlm_reprocess_queue(res, &res->lr_converting);
 756         if (list_empty(&res->lr_converting))
 757                 ldlm_reprocess_queue(res, &res->lr_waiting);
 758
 759         res->lr_tmp = NULL;
 760         l_unlock(&res->lr_namespace->ns_lock);
 761
 762         ldlm_run_ast_work(&rpc_list);
 763         EXIT;
 764 }
 765
 766 void ldlm_lock_cancel(struct ldlm_lock *lock)
 767 {
 768         struct ldlm_resource *res;
 769         struct ldlm_namespace *ns;
 770         ENTRY;
 771
 772         res = lock->l_resource;
 773         ns = res->lr_namespace;
 774
 775         l_lock(&ns->ns_lock);
 776         if (lock->l_readers || lock->l_writers)
 777                 CDEBUG(D_INFO, "lock still has references (%d readers, %d "
 778                        "writers)\n", lock->l_readers, lock->l_writers);
 779
 780         ldlm_resource_unlink_lock(lock);
 781         ldlm_lock_destroy(lock);
 782         l_unlock(&ns->ns_lock);
 783         EXIT;
 784 }
 785
 786 struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode,
 787                                         int *flags)
 788 {
 789         struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
 790         struct ldlm_resource *res;
 791         struct ldlm_namespace *ns;
 792         int granted = 0;
 793         ENTRY;
 794
 795         res = lock->l_resource;
 796         ns = res->lr_namespace;
 797
 798         l_lock(&ns->ns_lock);
 799
 800         lock->l_req_mode = new_mode;
 801         ldlm_resource_unlink_lock(lock);
 802
 803         /* If this is a local resource, put it on the appropriate list. */
 804         if (res->lr_namespace->ns_client) {
 805                 if (*flags & (LDLM_FL_BLOCK_CONV | LDLM_FL_BLOCK_GRANTED))
 806                         ldlm_resource_add_lock(res, res->lr_converting.prev,
 807                                                lock);
 808                 else {
 809                         res->lr_tmp = &rpc_list;
 810                         ldlm_grant_lock(lock);
 811                         res->lr_tmp = NULL;
 812                         granted = 1;
 813                         /* FIXME: completion handling not with ns_lock held ! */
 814                         if (lock->l_completion_ast)
 815                                 lock->l_completion_ast(lock, 0);
 816                 }
 817         } else
 818                 list_add_tail(&lock->l_res_link, &res->lr_converting);
 819
 820         l_unlock(&ns->ns_lock);
 821
 822         if (granted)
 823                 ldlm_run_ast_work(&rpc_list);
 824         RETURN(res);
 825 }
 826
 827 void ldlm_lock_dump(struct ldlm_lock *lock)
 828 {
 829         char ver[128];
 830
 831         if (!(portal_debug & D_OTHER))
 832                 return;
 833
 834         if (RES_VERSION_SIZE != 4)
 835                 LBUG();
 836
 837         if (!lock) {
 838                 CDEBUG(D_OTHER, "  NULL LDLM lock\n");
 839                 return;
 840         }
 841
 842         snprintf(ver, sizeof(ver), "%x %x %x %x",
 843                  lock->l_version[0], lock->l_version[1],
 844                  lock->l_version[2], lock->l_version[3]);
 845
 846         CDEBUG(D_OTHER, "  -- Lock dump: %p (%s)\n", lock, ver);
 847         if (lock->l_connection)
 848                 CDEBUG(D_OTHER, "  Node: NID %x (rhandle: %Lx)\n",
 849                        lock->l_connection->c_peer.peer_nid,
 850                        lock->l_remote_handle.addr);
 851         else
 852                 CDEBUG(D_OTHER, "  Node: local\n");
 853         CDEBUG(D_OTHER, "  Parent: %p\n", lock->l_parent);
 854         CDEBUG(D_OTHER, "  Resource: %p (%Ld)\n", lock->l_resource,
 855                lock->l_resource->lr_name[0]);
 856         CDEBUG(D_OTHER, "  Requested mode: %d, granted mode: %d\n",
 857                (int)lock->l_req_mode, (int)lock->l_granted_mode);
 858         CDEBUG(D_OTHER, "  Readers: %u ; Writers; %u\n",
 859                lock->l_readers, lock->l_writers);
 860         if (lock->l_resource->lr_type == LDLM_EXTENT)
 861                 CDEBUG(D_OTHER, "  Extent: %Lu -> %Lu\n",
 862                        (unsigned long long)lock->l_extent.start,
 863                        (unsigned long long)lock->l_extent.end);
 864 }