lustre/lov/lovsub_lock.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * Implementation of cl_lock for LOVSUB layer.
  37  *
  38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_LOV
  42
  43 #include "lov_cl_internal.h"
  44
  45 /** \addtogroup lov lov @{ */
  46
  47 /*****************************************************************************
  48  *
  49  * Lovsub lock operations.
  50  *
  51  */
  52
  53 static void lovsub_lock_fini(const struct lu_env *env,
  54                              struct cl_lock_slice *slice)
  55 {
  56         struct lovsub_lock   *lsl;
  57
  58         ENTRY;
  59         lsl = cl2lovsub_lock(slice);
  60         LASSERT(list_empty(&lsl->lss_parents));
  61         OBD_SLAB_FREE_PTR(lsl, lovsub_lock_kmem);
  62         EXIT;
  63 }
  64
  65 static void lovsub_parent_lock(const struct lu_env *env, struct lov_lock *lov)
  66 {
  67         struct cl_lock *parent;
  68
  69         ENTRY;
  70         parent = lov->lls_cl.cls_lock;
  71         cl_lock_get(parent);
  72         lu_ref_add(&parent->cll_reference, "lovsub-parent", cfs_current());
  73         cl_lock_mutex_get(env, parent);
  74         EXIT;
  75 }
  76
  77 static void lovsub_parent_unlock(const struct lu_env *env, struct lov_lock *lov)
  78 {
  79         struct cl_lock *parent;
  80
  81         ENTRY;
  82         parent = lov->lls_cl.cls_lock;
  83         cl_lock_mutex_put(env, lov->lls_cl.cls_lock);
  84         lu_ref_del(&parent->cll_reference, "lovsub-parent", cfs_current());
  85         cl_lock_put(env, parent);
  86         EXIT;
  87 }
  88
  89 static int lovsub_lock_state_one(const struct lu_env *env,
  90                                  const struct lovsub_lock *lovsub,
  91                                  struct lov_lock *lov)
  92 {
  93         struct cl_lock *parent;
  94         struct cl_lock *child;
  95         int             restart = 0;
  96
  97         ENTRY;
  98         parent = lov->lls_cl.cls_lock;
  99         child  = lovsub->lss_cl.cls_lock;
 100
 101         if (lovsub->lss_active != parent) {
 102                 lovsub_parent_lock(env, lov);
 103                 if (child->cll_error != 0 && parent->cll_error == 0) {
 104                         /*
 105                          * This is a deadlock case:
 106                          * cl_lock_error(for the parent lock)
 107                          *   -> cl_lock_delete
 108                          *     -> lov_lock_delete
 109                          *       -> cl_lock_enclosure
 110                          *         -> cl_lock_mutex_try(for the child lock)
 111                          */
 112                         cl_lock_mutex_put(env, child);
 113                         cl_lock_error(env, parent, child->cll_error);
 114                         restart = 1;
 115                 } else {
 116                         cl_lock_signal(env, parent);
 117                 }
 118                 lovsub_parent_unlock(env, lov);
 119         }
 120         RETURN(restart);
 121 }
 122
 123 /**
 124  * Implements cl_lock_operations::clo_state() method for lovsub layer, which
 125  * method is called whenever sub-lock state changes. Propagates state change
 126  * to the top-locks.
 127  */
 128 static void lovsub_lock_state(const struct lu_env *env,
 129                               const struct cl_lock_slice *slice,
 130                               enum cl_lock_state state)
 131 {
 132         struct lovsub_lock   *sub = cl2lovsub_lock(slice);
 133         struct lov_lock_link *scan;
 134         int                   restart = 0;
 135
 136         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
 137         ENTRY;
 138
 139         do {
 140                 restart = 0;
 141                 list_for_each_entry(scan, &sub->lss_parents, lll_list) {
 142                         restart = lovsub_lock_state_one(env, sub,
 143                                                         scan->lll_super);
 144                         if (restart) {
 145                                 cl_lock_mutex_get(env, slice->cls_lock);
 146                                 break;
 147                         }
 148                 }
 149         } while(restart);
 150         EXIT;
 151 }
 152
 153 /**
 154  * Implementation of cl_lock_operation::clo_weigh() estimating lock weight by
 155  * asking parent lock.
 156  */
 157 static unsigned long lovsub_lock_weigh(const struct lu_env *env,
 158                                        const struct cl_lock_slice *slice)
 159 {
 160         struct lovsub_lock *lock = cl2lovsub_lock(slice);
 161         struct lov_lock    *lov;
 162         unsigned long       dumbbell;
 163
 164         ENTRY;
 165
 166         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
 167
 168         if (!list_empty(&lock->lss_parents)) {
 169                 /*
 170                  * It is not clear whether all parents have to be asked and
 171                  * their estimations summed, or it is enough to ask one. For
 172                  * the current usages, one is always enough.
 173                  */
 174                 lov = container_of(lock->lss_parents.next,
 175                                    struct lov_lock_link, lll_list)->lll_super;
 176
 177                 lovsub_parent_lock(env, lov);
 178                 dumbbell = cl_lock_weigh(env, lov->lls_cl.cls_lock);
 179                 lovsub_parent_unlock(env, lov);
 180         } else
 181                 dumbbell = 0;
 182
 183         RETURN(dumbbell);
 184 }
 185
 186 /**
 187  * Maps start/end offsets within a stripe, to offsets within a file.
 188  */
 189 static void lovsub_lock_descr_map(const struct cl_lock_descr *in,
 190                                   struct lov_object *obj,
 191                                   int stripe, struct cl_lock_descr *out)
 192 {
 193         struct lov_stripe_md *lsm = lov_r0(obj)->lo_lsm;
 194         pgoff_t size; /* stripe size in pages */
 195         pgoff_t skip; /* how many pages in every stripe are occupied by
 196                        * "other" stripes */
 197         pgoff_t start;
 198         pgoff_t end;
 199
 200         ENTRY;
 201         start = in->cld_start;
 202         end   = in->cld_end;
 203
 204         /*
 205          * XXX join file support.
 206          */
 207         if (lsm->lsm_stripe_count > 1) {
 208                 size = cl_index(lov2cl(obj), lsm->lsm_stripe_size);
 209                 skip = (lsm->lsm_stripe_count - 1) * size;
 210
 211                 /* XXX overflow check here? */
 212                 start += start/size * skip + stripe * size;
 213
 214                 if (end != CL_PAGE_EOF) {
 215                         end += end/size * skip + stripe * size;
 216                         /*
 217                          * And check for overflow...
 218                          */
 219                         if (end < in->cld_end)
 220                                 end = CL_PAGE_EOF;
 221                 }
 222         }
 223         out->cld_start = start;
 224         out->cld_end   = end;
 225         EXIT;
 226 }
 227
 228 /**
 229  * Adjusts parent lock extent when a sub-lock is attached to a parent. This is
 230  * called in two ways:
 231  *
 232  *     - as part of receive call-back, when server returns granted extent to
 233  *       the client, and
 234  *
 235  *     - when top-lock finds existing sub-lock in the cache.
 236  *
 237  * Note, that lock mode is not propagated to the parent: i.e., if CLM_READ
 238  * top-lock matches CLM_WRITE sub-lock, top-lock is still CLM_READ.
 239  */
 240 int lov_sublock_modify(const struct lu_env *env, struct lov_lock *lov,
 241                        struct lovsub_lock *sublock,
 242                        const struct cl_lock_descr *d, int idx)
 243 {
 244         struct cl_lock       *parent;
 245         struct cl_lock       *child;
 246         struct lovsub_object *subobj;
 247         struct cl_lock_descr *pd;
 248         struct cl_lock_descr *parent_descr;
 249         int                   result;
 250
 251         parent       = lov->lls_cl.cls_lock;
 252         parent_descr = &parent->cll_descr;
 253         LASSERT(cl_lock_mode_match(d->cld_mode, parent_descr->cld_mode));
 254
 255         child  = sublock->lss_cl.cls_lock;
 256         subobj = cl2lovsub(sublock->lss_cl.cls_obj);
 257         pd     = &lov_env_info(env)->lti_ldescr;
 258
 259         pd->cld_obj  = parent_descr->cld_obj;
 260         pd->cld_mode = parent_descr->cld_mode;
 261         pd->cld_gid  = parent_descr->cld_gid;
 262         lovsub_lock_descr_map(d, subobj->lso_super, subobj->lso_index, pd);
 263         lov->lls_sub[idx].sub_got = *d;
 264         /*
 265          * Notify top-lock about modification, if lock description changes
 266          * materially.
 267          */
 268         if (!cl_lock_ext_match(parent_descr, pd))
 269                 result = cl_lock_modify(env, parent, pd);
 270         else
 271                 result = 0;
 272         return result;
 273 }
 274
 275 static int lovsub_lock_modify(const struct lu_env *env,
 276                               const struct cl_lock_slice *s,
 277                               const struct cl_lock_descr *d)
 278 {
 279         struct lovsub_lock   *lock   = cl2lovsub_lock(s);
 280         struct lov_lock_link *scan;
 281         struct lov_lock      *lov;
 282         int result                   = 0;
 283
 284         ENTRY;
 285
 286         LASSERT(cl_lock_mode_match(d->cld_mode,
 287                                    s->cls_lock->cll_descr.cld_mode));
 288         list_for_each_entry(scan, &lock->lss_parents, lll_list) {
 289                 int rc;
 290
 291                 lov = scan->lll_super;
 292                 lovsub_parent_lock(env, lov);
 293                 rc = lov_sublock_modify(env, lov, lock, d, scan->lll_idx);
 294                 lovsub_parent_unlock(env, lov);
 295                 result = result ?: rc;
 296         }
 297         RETURN(result);
 298 }
 299
 300 static int lovsub_lock_closure(const struct lu_env *env,
 301                                const struct cl_lock_slice *slice,
 302                                struct cl_lock_closure *closure)
 303 {
 304         struct lovsub_lock   *sub;
 305         struct cl_lock       *parent;
 306         struct lov_lock_link *scan;
 307         int                   result;
 308
 309         LASSERT(cl_lock_is_mutexed(slice->cls_lock));
 310         ENTRY;
 311
 312         sub    = cl2lovsub_lock(slice);
 313         result = 0;
 314
 315         list_for_each_entry(scan, &sub->lss_parents, lll_list) {
 316                 parent = scan->lll_super->lls_cl.cls_lock;
 317                 result = cl_lock_closure_build(env, parent, closure);
 318                 if (result != 0)
 319                         break;
 320         }
 321         RETURN(result);
 322 }
 323
 324 /**
 325  * A helper function for lovsub_lock_delete() that deals with a given parent
 326  * top-lock.
 327  */
 328 static int lovsub_lock_delete_one(const struct lu_env *env,
 329                                   struct cl_lock *child, struct lov_lock *lov)
 330 {
 331         struct cl_lock       *parent;
 332         int             result;
 333         ENTRY;
 334
 335         parent  = lov->lls_cl.cls_lock;
 336         result = 0;
 337
 338         switch (parent->cll_state) {
 339         case CLS_NEW:
 340         case CLS_QUEUING:
 341         case CLS_ENQUEUED:
 342         case CLS_FREEING:
 343                 cl_lock_signal(env, parent);
 344                 break;
 345         case CLS_UNLOCKING:
 346                 /*
 347                  * Here lies a problem: a sub-lock is canceled while top-lock
 348                  * is being unlocked. Top-lock cannot be moved into CLS_NEW
 349                  * state, because unlocking has to succeed eventually by
 350                  * placing lock into CLS_CACHED (or failing it), see
 351                  * cl_unuse_try(). Nor can top-lock be left in CLS_CACHED
 352                  * state, because lov maintains an invariant that all
 353                  * sub-locks exist in CLS_CACHED (this allows cached top-lock
 354                  * to be reused immediately). Nor can we wait for top-lock
 355                  * state to change, because this can be synchronous to the
 356                  * current thread.
 357                          *
 358                  * We know for sure that lov_lock_unuse() will be called at
 359                  * least one more time to finish un-using, so leave a mark on
 360                  * the top-lock, that will be seen by the next call to
 361                  * lov_lock_unuse().
 362                  */
 363                 lov->lls_unuse_race = 1;
 364                 break;
 365         case CLS_CACHED:
 366                 /*
 367                  * if a sub-lock is canceled move its top-lock into CLS_NEW
 368                  * state to preserve an invariant that a top-lock in
 369                  * CLS_CACHED is immediately ready for re-use (i.e., has all
 370                  * sub-locks), and so that next attempt to re-use the top-lock
 371                  * enqueues missing sub-lock.
 372                  */
 373                 cl_lock_state_set(env, parent, CLS_NEW);
 374                 /*
 375                  * if last sub-lock is canceled, destroy the top-lock (which
 376                  * is now `empty') proactively.
 377                  */
 378                 if (lov->lls_nr_filled == 0) {
 379                         /* ... but unfortunately, this cannot be done easily,
 380                          * as cancellation of a top-lock might acquire mutices
 381                          * of its other sub-locks, violating lock ordering,
 382                          * see cl_lock_{cancel,delete}() preconditions.
 383                          *
 384                          * To work around this, the mutex of this sub-lock is
 385                          * released, top-lock is destroyed, and sub-lock mutex
 386                          * acquired again. The list of parents has to be
 387                          * re-scanned from the beginning after this.
 388                          *
 389                          * Only do this if no mutices other than on @child and
 390                          * @parent are held by the current thread.
 391                          *
 392                          * TODO: The lock modal here is too complex, because
 393                          * the lock may be canceled and deleted by voluntarily:
 394                          *    cl_lock_request
 395                          *      -> osc_lock_enqueue_wait
 396                          *        -> osc_lock_cancel_wait
 397                          *          -> cl_lock_delete
 398                          *            -> lovsub_lock_delete
 399                          *              -> cl_lock_cancel/delete
 400                          *                -> ...
 401                          *
 402                          * The better choice is to spawn a kernel thread for
 403                          * this purpose. -jay
 404                          */
 405                         if (cl_lock_nr_mutexed(env) == 2) {
 406                                 cl_lock_mutex_put(env, child);
 407                                 cl_lock_cancel(env, parent);
 408                                 cl_lock_delete(env, parent);
 409                                 result = 1;
 410                         }
 411                 }
 412                 break;
 413         case CLS_HELD:
 414         default:
 415                 CERROR("Impossible state: %i\n", parent->cll_state);
 416                 LBUG();
 417         }
 418
 419         RETURN(result);
 420 }
 421
 422 /**
 423  * An implementation of cl_lock_operations::clo_delete() method. This is
 424  * invoked in "bottom-to-top" delete, when lock destruction starts from the
 425  * sub-lock (e.g, as a result of ldlm lock LRU policy).
 426  */
 427 static void lovsub_lock_delete(const struct lu_env *env,
 428                                const struct cl_lock_slice *slice)
 429 {
 430         struct cl_lock     *child = slice->cls_lock;
 431         struct lovsub_lock *sub   = cl2lovsub_lock(slice);
 432         int restart;
 433
 434         LASSERT(cl_lock_is_mutexed(child));
 435
 436         ENTRY;
 437         /*
 438          * Destruction of a sub-lock might take multiple iterations, because
 439          * when the last sub-lock of a given top-lock is deleted, top-lock is
 440          * canceled proactively, and this requires to release sub-lock
 441          * mutex. Once sub-lock mutex has been released, list of its parents
 442          * has to be re-scanned from the beginning.
 443          */
 444         do {
 445                 struct lov_lock      *lov;
 446                 struct lov_lock_link *scan;
 447                 struct lov_lock_link *temp;
 448                 struct lov_lock_sub  *subdata;
 449
 450                 restart = 0;
 451                 list_for_each_entry_safe(scan, temp,
 452                                          &sub->lss_parents, lll_list) {
 453                         lov     = scan->lll_super;
 454                         subdata = &lov->lls_sub[scan->lll_idx];
 455                         lovsub_parent_lock(env, lov);
 456                         subdata->sub_got = subdata->sub_descr;
 457                         lov_lock_unlink(env, scan, sub);
 458                         restart = lovsub_lock_delete_one(env, child, lov);
 459                         lovsub_parent_unlock(env, lov);
 460
 461                         if (restart) {
 462                                 cl_lock_mutex_get(env, child);
 463                                 break;
 464                         }
 465                }
 466         } while (restart);
 467         EXIT;
 468 }
 469
 470 static int lovsub_lock_print(const struct lu_env *env, void *cookie,
 471                              lu_printer_t p, const struct cl_lock_slice *slice)
 472 {
 473         struct lovsub_lock   *sub = cl2lovsub_lock(slice);
 474         struct lov_lock      *lov;
 475         struct lov_lock_link *scan;
 476
 477         list_for_each_entry(scan, &sub->lss_parents, lll_list) {
 478                 lov = scan->lll_super;
 479                 (*p)(env, cookie, "[%d %p ", scan->lll_idx, lov);
 480                 if (lov != NULL)
 481                         cl_lock_descr_print(env, cookie, p,
 482                                             &lov->lls_cl.cls_lock->cll_descr);
 483                 (*p)(env, cookie, "] ");
 484         }
 485         return 0;
 486 }
 487
 488 static const struct cl_lock_operations lovsub_lock_ops = {
 489         .clo_fini    = lovsub_lock_fini,
 490         .clo_state   = lovsub_lock_state,
 491         .clo_delete  = lovsub_lock_delete,
 492         .clo_modify  = lovsub_lock_modify,
 493         .clo_closure = lovsub_lock_closure,
 494         .clo_weigh   = lovsub_lock_weigh,
 495         .clo_print   = lovsub_lock_print
 496 };
 497
 498 int lovsub_lock_init(const struct lu_env *env, struct cl_object *obj,
 499                      struct cl_lock *lock, const struct cl_io *io)
 500 {
 501         struct lovsub_lock *lsk;
 502         int result;
 503
 504         ENTRY;
 505         OBD_SLAB_ALLOC_PTR_GFP(lsk, lovsub_lock_kmem, CFS_ALLOC_IO);
 506         if (lsk != NULL) {
 507                 CFS_INIT_LIST_HEAD(&lsk->lss_parents);
 508                 cl_lock_slice_add(lock, &lsk->lss_cl, obj, &lovsub_lock_ops);
 509                 result = 0;
 510         } else
 511                 result = -ENOMEM;
 512         RETURN(result);
 513 }
 514
 515 /** @} lov */