lustre/lov/lov_lock.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) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2016, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * Implementation of cl_lock for LOV layer.
  33  *
  34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_LOV
  38
  39 #include "lov_cl_internal.h"
  40
  41 /** \addtogroup lov
  42  *  @{
  43  */
  44
  45 /*****************************************************************************
  46  *
  47  * Lov lock operations.
  48  *
  49  */
  50
  51 static struct lov_sublock_env *lov_sublock_env_get(const struct lu_env *env,
  52                                                    const struct cl_lock *parent,
  53                                                    struct lov_lock_sub *lls)
  54 {
  55         struct lov_sublock_env *subenv;
  56         struct lov_io          *lio    = lov_env_io(env);
  57         struct cl_io           *io     = lio->lis_cl.cis_io;
  58         struct lov_io_sub      *sub;
  59
  60         subenv = &lov_env_session(env)->ls_subenv;
  61
  62         /*
  63          * FIXME: We tend to use the subio's env & io to call the sublock
  64          * lock operations because osc lock sometimes stores some control
  65          * variables in thread's IO infomation(Now only lockless information).
  66          * However, if the lock's host(object) is different from the object
  67          * for current IO, we have no way to get the subenv and subio because
  68          * they are not initialized at all. As a temp fix, in this case,
  69          * we still borrow the parent's env to call sublock operations.
  70          */
  71         if (!io || !cl_object_same(io->ci_obj, parent->cll_descr.cld_obj)) {
  72                 subenv->lse_env = env;
  73                 subenv->lse_io = io;
  74         } else {
  75                 sub = lov_sub_get(env, lio, lls->sub_index);
  76                 if (!IS_ERR(sub)) {
  77                         subenv->lse_env = sub->sub_env;
  78                         subenv->lse_io  = &sub->sub_io;
  79                 } else {
  80                         subenv = (void *)sub;
  81                 }
  82         }
  83         return subenv;
  84 }
  85
  86 static int lov_sublock_init(const struct lu_env *env,
  87                             const struct cl_lock *parent,
  88                             struct lov_lock_sub *lls)
  89 {
  90         struct lov_sublock_env *subenv;
  91         int result;
  92         ENTRY;
  93
  94         subenv = lov_sublock_env_get(env, parent, lls);
  95         if (!IS_ERR(subenv)) {
  96                 result = cl_lock_init(subenv->lse_env, &lls->sub_lock,
  97                                       subenv->lse_io);
  98         } else {
  99                 /* error occurs. */
 100                 result = PTR_ERR(subenv);
 101         }
 102         RETURN(result);
 103 }
 104
 105 /**
 106  * Creates sub-locks for a given lov_lock for the first time.
 107  *
 108  * Goes through all sub-objects of top-object, and creates sub-locks on every
 109  * sub-object intersecting with top-lock extent. This is complicated by the
 110  * fact that top-lock (that is being created) can be accessed concurrently
 111  * through already created sub-locks (possibly shared with other top-locks).
 112  */
 113 static struct lov_lock *lov_lock_sub_init(const struct lu_env *env,
 114                                           const struct cl_object *obj,
 115                                           struct cl_lock *lock)
 116 {
 117         struct lov_object *lov = cl2lov(obj);
 118         struct lov_lock *lovlck;
 119         struct lu_extent ext;
 120         loff_t start;
 121         loff_t end;
 122         int result = 0;
 123         int i;
 124         int index;
 125         int nr;
 126
 127         ENTRY;
 128
 129         ext.e_start = cl_offset(obj, lock->cll_descr.cld_start);
 130         if (lock->cll_descr.cld_end == CL_PAGE_EOF)
 131                 ext.e_end = OBD_OBJECT_EOF;
 132         else
 133                 ext.e_end  = cl_offset(obj, lock->cll_descr.cld_end + 1);
 134
 135         nr = 0;
 136         for (index = lov_lsm_entry(lov->lo_lsm, ext.e_start);
 137              index >= 0 && index < lov->lo_lsm->lsm_entry_count; index++) {
 138                 struct lov_layout_raid0 *r0 = lov_r0(lov, index);
 139
 140                 /* assume lsm entries are sorted. */
 141                 if (!lu_extent_is_overlapped(&ext,
 142                                              &lov_lse(lov, index)->lsme_extent))
 143                         break;
 144
 145                 for (i = 0; i < r0->lo_nr; i++) {
 146                         if (likely(r0->lo_sub[i] != NULL) && /* spare layout */
 147                             lov_stripe_intersects(lov->lo_lsm, index, i,
 148                                                   &ext, &start, &end))
 149                                 nr++;
 150                 }
 151         }
 152         /**
 153          * Aggressive lock request (from cl_setattr_ost) which asks for
 154          * [eof, -1) lock, could come across uninstantiated layout extent,
 155          * hence a 0 nr is possible.
 156          */
 157
 158         OBD_ALLOC_LARGE(lovlck, offsetof(struct lov_lock, lls_sub[nr]));
 159         if (lovlck == NULL)
 160                 RETURN(ERR_PTR(-ENOMEM));
 161
 162         lovlck->lls_nr = nr;
 163         nr = 0;
 164         for (index = lov_lsm_entry(lov->lo_lsm, ext.e_start);
 165              index >= 0 && index < lov->lo_lsm->lsm_entry_count; index++) {
 166                 struct lov_layout_raid0 *r0 = lov_r0(lov, index);
 167
 168                 /* assume lsm entries are sorted. */
 169                 if (!lu_extent_is_overlapped(&ext,
 170                                              &lov_lse(lov, index)->lsme_extent))
 171                         break;
 172                 for (i = 0; i < r0->lo_nr; ++i) {
 173                         struct lov_lock_sub *lls = &lovlck->lls_sub[nr];
 174                         struct cl_lock_descr *descr = &lls->sub_lock.cll_descr;
 175
 176                         if (unlikely(r0->lo_sub[i] == NULL) ||
 177                             !lov_stripe_intersects(lov->lo_lsm, index, i,
 178                                                    &ext, &start, &end))
 179                                 continue;
 180
 181                         LASSERT(descr->cld_obj == NULL);
 182                         descr->cld_obj   = lovsub2cl(r0->lo_sub[i]);
 183                         descr->cld_start = cl_index(descr->cld_obj, start);
 184                         descr->cld_end   = cl_index(descr->cld_obj, end);
 185                         descr->cld_mode  = lock->cll_descr.cld_mode;
 186                         descr->cld_gid   = lock->cll_descr.cld_gid;
 187                         descr->cld_enq_flags = lock->cll_descr.cld_enq_flags;
 188
 189                         lls->sub_index = lov_comp_index(index, i);
 190
 191                         /* initialize sub lock */
 192                         result = lov_sublock_init(env, lock, lls);
 193                         if (result < 0)
 194                                 break;
 195
 196                         lls->sub_initialized = 1;
 197                         nr++;
 198                 }
 199         }
 200         LASSERT(ergo(result == 0, nr == lovlck->lls_nr));
 201
 202         if (result != 0) {
 203                 for (i = 0; i < nr; ++i) {
 204                         if (!lovlck->lls_sub[i].sub_initialized)
 205                                 break;
 206
 207                         cl_lock_fini(env, &lovlck->lls_sub[i].sub_lock);
 208                 }
 209
 210                 OBD_FREE_LARGE(lovlck,
 211                                 offsetof(struct lov_lock, lls_sub[nr]));
 212                 lovlck = ERR_PTR(result);
 213         }
 214
 215         RETURN(lovlck);
 216 }
 217
 218 static void lov_lock_fini(const struct lu_env *env,
 219                           struct cl_lock_slice *slice)
 220 {
 221         struct lov_lock *lovlck;
 222         int i;
 223
 224         ENTRY;
 225         lovlck = cl2lov_lock(slice);
 226         for (i = 0; i < lovlck->lls_nr; ++i) {
 227                 LASSERT(!lovlck->lls_sub[i].sub_is_enqueued);
 228                 if (lovlck->lls_sub[i].sub_initialized)
 229                         cl_lock_fini(env, &lovlck->lls_sub[i].sub_lock);
 230         }
 231         OBD_FREE_LARGE(lovlck,
 232                        offsetof(struct lov_lock, lls_sub[lovlck->lls_nr]));
 233         EXIT;
 234 }
 235
 236 /**
 237  * Implementation of cl_lock_operations::clo_enqueue() for lov layer. This
 238  * function is rather subtle, as it enqueues top-lock (i.e., advances top-lock
 239  * state machine from CLS_QUEUING to CLS_ENQUEUED states) by juggling sub-lock
 240  * state machines in the face of sub-locks sharing (by multiple top-locks),
 241  * and concurrent sub-lock cancellations.
 242  */
 243 static int lov_lock_enqueue(const struct lu_env *env,
 244                             const struct cl_lock_slice *slice,
 245                             struct cl_io *io, struct cl_sync_io *anchor)
 246 {
 247         struct cl_lock          *lock   = slice->cls_lock;
 248         struct lov_lock         *lovlck = cl2lov_lock(slice);
 249         int                     i;
 250         int                     rc      = 0;
 251
 252         ENTRY;
 253
 254         for (i = 0; i < lovlck->lls_nr; ++i) {
 255                 struct lov_lock_sub     *lls = &lovlck->lls_sub[i];
 256                 struct lov_sublock_env  *subenv;
 257
 258                 subenv = lov_sublock_env_get(env, lock, lls);
 259                 if (IS_ERR(subenv)) {
 260                         rc = PTR_ERR(subenv);
 261                         break;
 262                 }
 263
 264                 rc = cl_lock_enqueue(subenv->lse_env, subenv->lse_io,
 265                                      &lls->sub_lock, anchor);
 266                 if (rc != 0)
 267                         break;
 268
 269                 lls->sub_is_enqueued = 1;
 270         }
 271         RETURN(rc);
 272 }
 273
 274 static void lov_lock_cancel(const struct lu_env *env,
 275                             const struct cl_lock_slice *slice)
 276 {
 277         struct cl_lock  *lock   = slice->cls_lock;
 278         struct lov_lock *lovlck = cl2lov_lock(slice);
 279         int i;
 280
 281         ENTRY;
 282
 283         for (i = 0; i < lovlck->lls_nr; ++i) {
 284                 struct lov_lock_sub     *lls = &lovlck->lls_sub[i];
 285                 struct cl_lock          *sublock = &lls->sub_lock;
 286                 struct lov_sublock_env  *subenv;
 287
 288                 if (!lls->sub_is_enqueued)
 289                         continue;
 290
 291                 lls->sub_is_enqueued = 0;
 292                 subenv = lov_sublock_env_get(env, lock, lls);
 293                 if (!IS_ERR(subenv)) {
 294                         cl_lock_cancel(subenv->lse_env, sublock);
 295                 } else {
 296                         CL_LOCK_DEBUG(D_ERROR, env, slice->cls_lock,
 297                                       "lov_lock_cancel fails with %ld.\n",
 298                                       PTR_ERR(subenv));
 299                 }
 300         }
 301 }
 302
 303 static int lov_lock_print(const struct lu_env *env, void *cookie,
 304                           lu_printer_t p, const struct cl_lock_slice *slice)
 305 {
 306         struct lov_lock *lck = cl2lov_lock(slice);
 307         int              i;
 308
 309         (*p)(env, cookie, "%d\n", lck->lls_nr);
 310         for (i = 0; i < lck->lls_nr; ++i) {
 311                 struct lov_lock_sub *sub;
 312
 313                 sub = &lck->lls_sub[i];
 314                 (*p)(env, cookie, "    %d %x: ", i, sub->sub_is_enqueued);
 315                 cl_lock_print(env, cookie, p, &sub->sub_lock);
 316         }
 317         return 0;
 318 }
 319
 320 static const struct cl_lock_operations lov_lock_ops = {
 321         .clo_fini      = lov_lock_fini,
 322         .clo_enqueue   = lov_lock_enqueue,
 323         .clo_cancel    = lov_lock_cancel,
 324         .clo_print     = lov_lock_print
 325 };
 326
 327 int lov_lock_init_composite(const struct lu_env *env, struct cl_object *obj,
 328                             struct cl_lock *lock, const struct cl_io *io)
 329 {
 330         struct lov_lock *lck;
 331         int result = 0;
 332
 333         ENTRY;
 334         lck = lov_lock_sub_init(env, obj, lock);
 335         if (!IS_ERR(lck))
 336                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_lock_ops);
 337         else
 338                 result = PTR_ERR(lck);
 339         RETURN(result);
 340 }
 341
 342 static void lov_empty_lock_fini(const struct lu_env *env,
 343                                 struct cl_lock_slice *slice)
 344 {
 345         struct lov_lock *lck = cl2lov_lock(slice);
 346         OBD_SLAB_FREE_PTR(lck, lov_lock_kmem);
 347 }
 348
 349 static int lov_empty_lock_print(const struct lu_env *env, void *cookie,
 350                         lu_printer_t p, const struct cl_lock_slice *slice)
 351 {
 352         (*p)(env, cookie, "empty\n");
 353         return 0;
 354 }
 355
 356 /* XXX: more methods will be added later. */
 357 static const struct cl_lock_operations lov_empty_lock_ops = {
 358         .clo_fini  = lov_empty_lock_fini,
 359         .clo_print = lov_empty_lock_print
 360 };
 361
 362 int lov_lock_init_empty(const struct lu_env *env, struct cl_object *obj,
 363                         struct cl_lock *lock, const struct cl_io *io)
 364 {
 365         struct lov_lock *lck;
 366         int result = -ENOMEM;
 367
 368         ENTRY;
 369         OBD_SLAB_ALLOC_PTR_GFP(lck, lov_lock_kmem, GFP_NOFS);
 370         if (lck != NULL) {
 371                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_empty_lock_ops);
 372                 result = 0;
 373         }
 374         RETURN(result);
 375 }
 376
 377 /** @} lov */