lustre/mdc/mdc_dev.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) 2017, Intel Corporation.
  24  */
  25 /*
  26  * This file is part of Lustre, http://www.lustre.org/
  27  *
  28  * Implementation of cl_device, cl_req for MDC layer.
  29  *
  30  * Author: Mikhail Pershin <mike.pershin@intel.com>
  31  */
  32
  33 #define DEBUG_SUBSYSTEM S_MDC
  34
  35 #include <obd_class.h>
  36 #include <lustre_osc.h>
  37 #include <linux/falloc.h>
  38 #include <uapi/linux/lustre/lustre_param.h>
  39
  40 #include "mdc_internal.h"
  41
  42 static void mdc_lock_build_policy(const struct lu_env *env,
  43                                   const struct cl_lock *lock,
  44                                   union ldlm_policy_data *policy)
  45 {
  46         memset(policy, 0, sizeof *policy);
  47         policy->l_inodebits.bits = MDS_INODELOCK_DOM;
  48         if (lock) {
  49                 policy->l_inodebits.li_gid = lock->cll_descr.cld_gid;
  50         }
  51 }
  52
  53 int mdc_ldlm_glimpse_ast(struct ldlm_lock *dlmlock, void *data)
  54 {
  55         return osc_ldlm_glimpse_ast(dlmlock, data);
  56 }
  57
  58 static void mdc_lock_build_einfo(const struct lu_env *env,
  59                                  const struct cl_lock *lock,
  60                                  struct osc_object *osc,
  61                                  struct ldlm_enqueue_info *einfo)
  62 {
  63         einfo->ei_type = LDLM_IBITS;
  64         einfo->ei_mode = osc_cl_lock2ldlm(lock->cll_descr.cld_mode);
  65         einfo->ei_cb_bl = mdc_ldlm_blocking_ast;
  66         einfo->ei_cb_cp = ldlm_completion_ast;
  67         einfo->ei_cb_gl = mdc_ldlm_glimpse_ast;
  68         einfo->ei_cbdata = osc; /* value to be put into ->l_ast_data */
  69         einfo->ei_req_slot = 1;
  70 }
  71
  72 static void mdc_lock_lvb_update(const struct lu_env *env,
  73                                 struct osc_object *osc,
  74                                 struct ldlm_lock *dlmlock,
  75                                 struct ost_lvb *lvb);
  76
  77 static int mdc_set_dom_lock_data(struct ldlm_lock *lock, void *data)
  78 {
  79         int set = 0;
  80
  81         LASSERT(lock != NULL);
  82         LASSERT(lock->l_glimpse_ast == mdc_ldlm_glimpse_ast);
  83
  84         lock_res_and_lock(lock);
  85
  86         if (lock->l_ast_data == NULL)
  87                 lock->l_ast_data = data;
  88         if (lock->l_ast_data == data)
  89                 set = 1;
  90
  91         unlock_res_and_lock(lock);
  92
  93         return set;
  94 }
  95
  96 static int mdc_dom_lock_match(const struct lu_env *env, struct obd_export *exp,
  97                               struct ldlm_res_id *res_id, enum ldlm_type type,
  98                               union ldlm_policy_data *policy,
  99                               enum ldlm_mode mode, __u64 *flags,
 100                               struct osc_object *obj,
 101                               struct lustre_handle *lockh,
 102                               enum ldlm_match_flags match_flags)
 103 {
 104         struct obd_device *obd = exp->exp_obd;
 105         __u64 lflags = *flags;
 106         enum ldlm_mode rc;
 107
 108         ENTRY;
 109
 110         rc = ldlm_lock_match_with_skip(obd->obd_namespace, lflags, 0,
 111                              res_id, type, policy, mode, lockh, match_flags);
 112         if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
 113                 RETURN(rc);
 114
 115         if (obj != NULL) {
 116                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
 117
 118                 LASSERT(lock != NULL);
 119                 if (mdc_set_dom_lock_data(lock, obj)) {
 120                         lock_res_and_lock(lock);
 121                         if (!ldlm_is_lvb_cached(lock)) {
 122                                 LASSERT(lock->l_ast_data == obj);
 123                                 mdc_lock_lvb_update(env, obj, lock, NULL);
 124                                 ldlm_set_lvb_cached(lock);
 125                         }
 126                         unlock_res_and_lock(lock);
 127                 } else {
 128                         ldlm_lock_decref(lockh, rc);
 129                         rc = 0;
 130                 }
 131                 LDLM_LOCK_PUT(lock);
 132         }
 133         RETURN(rc);
 134 }
 135
 136 /**
 137  * Finds an existing lock covering a page with given index.
 138  * Copy of osc_obj_dlmlock_at_pgoff() but for DoM IBITS lock.
 139  */
 140 static struct ldlm_lock *mdc_dlmlock_at_pgoff(const struct lu_env *env,
 141                                               struct osc_object *obj,
 142                                               pgoff_t index,
 143                                               enum osc_dap_flags dap_flags)
 144 {
 145         struct osc_thread_info *info = osc_env_info(env);
 146         struct ldlm_res_id *resname = &info->oti_resname;
 147         union ldlm_policy_data *policy = &info->oti_policy;
 148         struct lustre_handle lockh;
 149         struct ldlm_lock *lock = NULL;
 150         enum ldlm_mode mode;
 151         __u64 flags;
 152         enum ldlm_match_flags match_flags = 0;
 153
 154         ENTRY;
 155
 156         fid_build_reg_res_name(lu_object_fid(osc2lu(obj)), resname);
 157         mdc_lock_build_policy(env, NULL, policy);
 158         policy->l_inodebits.li_gid = LDLM_GID_ANY;
 159
 160         flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_CBPENDING;
 161         if (dap_flags & OSC_DAP_FL_TEST_LOCK)
 162                 flags |= LDLM_FL_TEST_LOCK;
 163
 164         if (dap_flags & OSC_DAP_FL_AST)
 165                 match_flags |= LDLM_MATCH_AST;
 166
 167         if (dap_flags & OSC_DAP_FL_CANCELING)
 168                 match_flags |= LDLM_MATCH_UNREF;
 169
 170 again:
 171         /* Next, search for already existing extent locks that will cover us */
 172         /* If we're trying to read, we also search for an existing PW lock.  The
 173          * VFS and page cache already protect us locally, so lots of readers/
 174          * writers can share a single PW lock. */
 175         mode = mdc_dom_lock_match(env, osc_export(obj), resname, LDLM_IBITS,
 176                                   policy, LCK_PR | LCK_PW | LCK_GROUP, &flags,
 177                                   obj, &lockh, match_flags);
 178         if (mode != 0) {
 179                 lock = ldlm_handle2lock(&lockh);
 180                 /* RACE: the lock is cancelled so let's try again */
 181                 if (unlikely(lock == NULL))
 182                         goto again;
 183         }
 184
 185         RETURN(lock);
 186 }
 187
 188 /**
 189  * Check if page @page is covered by an extra lock or discard it.
 190  */
 191 static bool mdc_check_and_discard_cb(const struct lu_env *env, struct cl_io *io,
 192                                      void **pvec, int count, void *cbdata)
 193 {
 194         struct osc_thread_info *info = osc_env_info(env);
 195         struct osc_object *osc = cbdata;
 196         pgoff_t index;
 197         int i;
 198
 199         for (i = 0; i < count; i++) {
 200                 struct osc_page *ops = pvec[i];
 201
 202                 index = osc_index(ops);
 203                 if (index >= info->oti_fn_index) {
 204                         struct ldlm_lock *tmp;
 205                         struct cl_page *page = ops->ops_cl.cpl_page;
 206
 207                         /* refresh non-overlapped index */
 208                         tmp = mdc_dlmlock_at_pgoff(env, osc, index,
 209                                         OSC_DAP_FL_TEST_LOCK | OSC_DAP_FL_AST);
 210                         if (tmp != NULL) {
 211                                 info->oti_fn_index = CL_PAGE_EOF;
 212                                 LDLM_LOCK_PUT(tmp);
 213                         } else if (cl_page_own(env, io, page) == 0) {
 214                                 /* discard the page */
 215                                 cl_page_discard(env, io, page);
 216                                 cl_page_disown(env, io, page);
 217                         } else {
 218                                 if (page->cp_type != CPT_TRANSIENT)
 219                                         LASSERT(page->cp_state == CPS_FREEING);
 220                         }
 221                 }
 222
 223                 info->oti_next_index = index + 1;
 224         }
 225         return true;
 226 }
 227
 228 /**
 229  * Discard pages protected by the given lock. This function traverses radix
 230  * tree to find all covering pages and discard them. If a page is being covered
 231  * by other locks, it should remain in cache.
 232  *
 233  * If error happens on any step, the process continues anyway (the reasoning
 234  * behind this being that lock cancellation cannot be delayed indefinitely).
 235  */
 236 static int mdc_lock_discard_pages(const struct lu_env *env,
 237                                   struct osc_object *osc,
 238                                   pgoff_t start, pgoff_t end,
 239                                   bool discard)
 240 {
 241         struct osc_thread_info *info = osc_env_info(env);
 242         struct cl_io *io = &info->oti_io;
 243         osc_page_gang_cbt cb;
 244         int result;
 245
 246         ENTRY;
 247
 248         io->ci_obj = cl_object_top(osc2cl(osc));
 249         io->ci_ignore_layout = 1;
 250         result = cl_io_init(env, io, CIT_MISC, io->ci_obj);
 251         if (result != 0)
 252                 GOTO(out, result);
 253
 254         cb = discard ? osc_discard_cb : mdc_check_and_discard_cb;
 255         info->oti_fn_index = info->oti_next_index = start;
 256
 257         osc_page_gang_lookup(env, io, osc, info->oti_next_index,
 258                              end, cb, (void *)osc);
 259 out:
 260         cl_io_fini(env, io);
 261         RETURN(result);
 262 }
 263
 264 static int mdc_lock_flush(const struct lu_env *env, struct osc_object *obj,
 265                           pgoff_t start, pgoff_t end, enum cl_lock_mode mode,
 266                           bool discard)
 267 {
 268         int result = 0;
 269         int rc;
 270
 271         ENTRY;
 272
 273         if (mode == CLM_WRITE) {
 274                 result = osc_cache_writeback_range(env, obj, start, end, 1,
 275                                                    discard);
 276                 CDEBUG(D_CACHE, "object %p: [%lu -> %lu] %d pages were %s.\n",
 277                        obj, start, end, result,
 278                        discard ? "discarded" : "written back");
 279                 if (result > 0)
 280                         result = 0;
 281         }
 282
 283         /* Avoid lock matching with CLM_WRITE, there can be no other locks */
 284         rc = mdc_lock_discard_pages(env, obj, start, end,
 285                                     mode == CLM_WRITE || discard);
 286         if (result == 0 && rc < 0)
 287                 result = rc;
 288
 289         RETURN(result);
 290 }
 291
 292 static void mdc_lock_lockless_cancel(const struct lu_env *env,
 293                                      const struct cl_lock_slice *slice)
 294 {
 295         struct osc_lock *ols = cl2osc_lock(slice);
 296         struct osc_object *osc = cl2osc(slice->cls_obj);
 297         struct cl_lock_descr *descr = &slice->cls_lock->cll_descr;
 298         int rc;
 299
 300         LASSERT(ols->ols_dlmlock == NULL);
 301         rc = mdc_lock_flush(env, osc, descr->cld_start, descr->cld_end,
 302                             descr->cld_mode, 0);
 303         if (rc != 0)
 304                 CERROR("Pages for lockless lock %p were not purged(%d)\n",
 305                        ols, rc);
 306
 307         osc_lock_wake_waiters(env, osc, ols);
 308 }
 309
 310 /**
 311  * Helper for osc_dlm_blocking_ast() handling discrepancies between cl_lock
 312  * and ldlm_lock caches.
 313  */
 314 static int mdc_dlm_canceling(const struct lu_env *env,
 315                              struct ldlm_lock *dlmlock)
 316 {
 317         struct cl_object *obj = NULL;
 318         int result = 0;
 319         bool discard;
 320         enum cl_lock_mode mode = CLM_READ;
 321
 322         ENTRY;
 323
 324         lock_res_and_lock(dlmlock);
 325         if (!ldlm_is_granted(dlmlock)) {
 326                 dlmlock->l_ast_data = NULL;
 327                 unlock_res_and_lock(dlmlock);
 328                 RETURN(0);
 329         }
 330
 331         discard = ldlm_is_discard_data(dlmlock);
 332         if (dlmlock->l_granted_mode & (LCK_PW | LCK_GROUP))
 333                 mode = CLM_WRITE;
 334
 335         if (dlmlock->l_ast_data != NULL) {
 336                 obj = osc2cl(dlmlock->l_ast_data);
 337                 cl_object_get(obj);
 338         }
 339         unlock_res_and_lock(dlmlock);
 340
 341         /* if l_ast_data is NULL, the dlmlock was enqueued by AGL or
 342          * the object has been destroyed. */
 343         if (obj != NULL) {
 344                 struct cl_attr *attr = &osc_env_info(env)->oti_attr;
 345
 346                 /* Destroy pages covered by the extent of the DLM lock */
 347                 result = mdc_lock_flush(env, cl2osc(obj), 0,
 348                                         CL_PAGE_EOF, mode, discard);
 349                 /* Losing a lock, set KMS to 0.
 350                  * NB: assumed that DOM lock covers whole data on MDT.
 351                  */
 352                 /* losing a lock, update kms */
 353                 lock_res_and_lock(dlmlock);
 354                 dlmlock->l_ast_data = NULL;
 355                 cl_object_attr_lock(obj);
 356                 attr->cat_kms = 0;
 357                 cl_object_attr_update(env, obj, attr, CAT_KMS);
 358                 cl_object_attr_unlock(obj);
 359                 unlock_res_and_lock(dlmlock);
 360                 cl_object_put(env, obj);
 361         }
 362         RETURN(result);
 363 }
 364
 365 int mdc_ldlm_blocking_ast(struct ldlm_lock *dlmlock,
 366                           struct ldlm_lock_desc *new, void *data, int reason)
 367 {
 368         int rc = 0;
 369
 370         ENTRY;
 371
 372         switch (reason) {
 373         case LDLM_CB_BLOCKING: {
 374                 struct lustre_handle lockh;
 375
 376                 ldlm_lock2handle(dlmlock, &lockh);
 377                 rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
 378                 if (rc == -ENODATA)
 379                         rc = 0;
 380                 break;
 381         }
 382         case LDLM_CB_CANCELING: {
 383                 struct lu_env *env;
 384                 __u16 refcheck;
 385
 386                 /*
 387                  * This can be called in the context of outer IO, e.g.,
 388                  *
 389                  *    osc_enqueue_base()->...
 390                  *      ->ldlm_prep_elc_req()->...
 391                  *        ->ldlm_cancel_callback()->...
 392                  *          ->osc_ldlm_blocking_ast()
 393                  *
 394                  * new environment has to be created to not corrupt outer
 395                  * context.
 396                  */
 397                 env = cl_env_get(&refcheck);
 398                 if (IS_ERR(env)) {
 399                         rc = PTR_ERR(env);
 400                         break;
 401                 }
 402
 403                 rc = mdc_dlm_canceling(env, dlmlock);
 404                 cl_env_put(env, &refcheck);
 405                 break;
 406         }
 407         default:
 408                 LBUG();
 409         }
 410         RETURN(rc);
 411 }
 412
 413 /**
 414  * Updates object attributes from a lock value block (lvb) received together
 415  * with the DLM lock reply from the server.
 416  * This can be optimized to not update attributes when lock is a result of a
 417  * local match.
 418  *
 419  * Called under lock and resource spin-locks.
 420  */
 421 void mdc_lock_lvb_update(const struct lu_env *env, struct osc_object *osc,
 422                          struct ldlm_lock *dlmlock, struct ost_lvb *lvb)
 423 {
 424         struct cl_object *obj = osc2cl(osc);
 425         struct lov_oinfo *oinfo = osc->oo_oinfo;
 426         struct cl_attr *attr = &osc_env_info(env)->oti_attr;
 427         unsigned valid = CAT_BLOCKS | CAT_ATIME | CAT_CTIME | CAT_MTIME |
 428                          CAT_SIZE;
 429         unsigned int setkms = 0;
 430
 431         ENTRY;
 432
 433         if (lvb == NULL) {
 434                 LASSERT(dlmlock != NULL);
 435                 lvb = &dlmlock->l_ost_lvb;
 436         }
 437         cl_lvb2attr(attr, lvb);
 438
 439         cl_object_attr_lock(obj);
 440         if (dlmlock != NULL) {
 441                 __u64 size;
 442
 443                 check_res_locked(dlmlock->l_resource);
 444                 size = lvb->lvb_size;
 445
 446                 if (size >= oinfo->loi_kms) {
 447                         valid |= CAT_KMS;
 448                         attr->cat_kms = size;
 449                         setkms = 1;
 450                 }
 451                 ldlm_lock_allow_match_locked(dlmlock);
 452         }
 453
 454         /* The size should not be less than the kms */
 455         if (attr->cat_size < oinfo->loi_kms)
 456                 attr->cat_size = oinfo->loi_kms;
 457
 458         LDLM_DEBUG(dlmlock, "acquired size %llu, setting rss=%llu;%s "
 459                    "kms=%llu, end=%llu", lvb->lvb_size, attr->cat_size,
 460                    setkms ? "" : " leaving",
 461                    setkms ? attr->cat_kms : oinfo->loi_kms,
 462                    dlmlock ? dlmlock->l_policy_data.l_extent.end : -1ull);
 463
 464         cl_object_attr_update(env, obj, attr, valid);
 465         cl_object_attr_unlock(obj);
 466         EXIT;
 467 }
 468
 469 static void mdc_lock_granted(const struct lu_env *env, struct osc_lock *oscl,
 470                              struct lustre_handle *lockh)
 471 {
 472         struct osc_object *osc = cl2osc(oscl->ols_cl.cls_obj);
 473         struct ldlm_lock *dlmlock;
 474
 475         ENTRY;
 476
 477         dlmlock = ldlm_handle2lock_long(lockh, 0);
 478         LASSERT(dlmlock != NULL);
 479
 480         /* lock reference taken by ldlm_handle2lock_long() is
 481          * owned by osc_lock and released in osc_lock_detach()
 482          */
 483         lu_ref_add_atomic(&dlmlock->l_reference, "osc_lock", oscl);
 484         oscl->ols_has_ref = 1;
 485
 486         LASSERT(oscl->ols_dlmlock == NULL);
 487         oscl->ols_dlmlock = dlmlock;
 488
 489         /* This may be a matched lock for glimpse request, do not hold
 490          * lock reference in that case. */
 491         if (!oscl->ols_glimpse) {
 492                 /* hold a refc for non glimpse lock which will
 493                  * be released in osc_lock_cancel() */
 494                 lustre_handle_copy(&oscl->ols_handle, lockh);
 495                 ldlm_lock_addref(lockh, oscl->ols_einfo.ei_mode);
 496                 oscl->ols_hold = 1;
 497         }
 498
 499         /* Lock must have been granted. */
 500         lock_res_and_lock(dlmlock);
 501         if (ldlm_is_granted(dlmlock)) {
 502                 struct cl_lock_descr *descr = &oscl->ols_cl.cls_lock->cll_descr;
 503
 504                 /* extend the lock extent, otherwise it will have problem when
 505                  * we decide whether to grant a lockless lock. */
 506                 descr->cld_mode = osc_ldlm2cl_lock(dlmlock->l_granted_mode);
 507                 descr->cld_start = 0;
 508                 descr->cld_end = CL_PAGE_EOF;
 509
 510                 /* no lvb update for matched lock */
 511                 if (!ldlm_is_lvb_cached(dlmlock)) {
 512                         LASSERT(oscl->ols_flags & LDLM_FL_LVB_READY);
 513                         LASSERT(osc == dlmlock->l_ast_data);
 514                         mdc_lock_lvb_update(env, osc, dlmlock, NULL);
 515                         ldlm_set_lvb_cached(dlmlock);
 516                 }
 517         }
 518         unlock_res_and_lock(dlmlock);
 519
 520         LASSERT(oscl->ols_state != OLS_GRANTED);
 521         oscl->ols_state = OLS_GRANTED;
 522         EXIT;
 523 }
 524
 525 /**
 526  * Lock upcall function that is executed either when a reply to ENQUEUE rpc is
 527  * received from a server, or after mdc_enqueue_send() matched a local DLM
 528  * lock.
 529  */
 530 static int mdc_lock_upcall(void *cookie, struct lustre_handle *lockh,
 531                            int errcode)
 532 {
 533         struct osc_lock *oscl = cookie;
 534         struct cl_lock_slice *slice = &oscl->ols_cl;
 535         struct lu_env *env;
 536         int rc;
 537
 538         ENTRY;
 539
 540         env = cl_env_percpu_get();
 541         /* should never happen, similar to osc_ldlm_blocking_ast(). */
 542         LASSERT(!IS_ERR(env));
 543
 544         rc = ldlm_error2errno(errcode);
 545         if (oscl->ols_state == OLS_ENQUEUED) {
 546                 oscl->ols_state = OLS_UPCALL_RECEIVED;
 547         } else if (oscl->ols_state == OLS_CANCELLED) {
 548                 rc = -EIO;
 549         } else {
 550                 CERROR("Impossible state: %d\n", oscl->ols_state);
 551                 LBUG();
 552         }
 553
 554         CDEBUG(D_INODE, "rc %d, err %d\n", rc, errcode);
 555         if (rc == 0)
 556                 mdc_lock_granted(env, oscl, lockh);
 557
 558         /* Error handling, some errors are tolerable. */
 559         if (oscl->ols_glimpse && rc == -ENAVAIL) {
 560                 LASSERT(oscl->ols_flags & LDLM_FL_LVB_READY);
 561                 mdc_lock_lvb_update(env, cl2osc(slice->cls_obj),
 562                                     NULL, &oscl->ols_lvb);
 563                 /* Hide the error. */
 564                 rc = 0;
 565         }
 566
 567         if (oscl->ols_owner != NULL)
 568                 cl_sync_io_note(env, oscl->ols_owner, rc);
 569         cl_env_percpu_put(env);
 570
 571         RETURN(rc);
 572 }
 573
 574 /* This is needed only for old servers (before 2.14) support */
 575 int mdc_fill_lvb(struct req_capsule *pill, struct ost_lvb *lvb)
 576 {
 577         struct mdt_body *body;
 578
 579         /* get LVB data from mdt_body otherwise */
 580         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 581         if (!body)
 582                 RETURN(-EPROTO);
 583
 584         if (!(body->mbo_valid & OBD_MD_DOM_SIZE))
 585                 RETURN(-EPROTO);
 586
 587         mdc_body2lvb(body, lvb);
 588         RETURN(0);
 589 }
 590
 591 static int mdc_enqueue_fini(struct obd_export *exp, struct ptlrpc_request *req,
 592                             osc_enqueue_upcall_f upcall, void *cookie,
 593                             struct lustre_handle *lockh, enum ldlm_mode mode,
 594                             __u64 *flags, int errcode)
 595 {
 596         struct osc_lock *ols = cookie;
 597         bool glimpse = *flags & LDLM_FL_HAS_INTENT;
 598         int rc = 0;
 599
 600         ENTRY;
 601
 602         /* needed only for glimpse from an old server (< 2.14) */
 603         if (glimpse && !exp_connect_dom_lvb(exp) && errcode >= 0)
 604                 rc = mdc_fill_lvb(&req->rq_pill, &ols->ols_lvb);
 605
 606         if (glimpse && errcode == ELDLM_LOCK_ABORTED) {
 607                 struct ldlm_reply *rep;
 608
 609                 rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
 610                 if (likely(rep)) {
 611                         rep->lock_policy_res2 =
 612                                 ptlrpc_status_ntoh(rep->lock_policy_res2);
 613                         if (rep->lock_policy_res2)
 614                                 errcode = rep->lock_policy_res2;
 615                 } else {
 616                         rc = -EPROTO;
 617                 }
 618                 *flags |= LDLM_FL_LVB_READY;
 619         } else if (errcode == ELDLM_OK) {
 620                 struct ldlm_lock *lock;
 621
 622                 /* Callers have references, should be valid always */
 623                 lock = ldlm_handle2lock(lockh);
 624
 625                 /* At this point ols_lvb must be filled with correct LVB either
 626                  * by mdc_fill_lvb() above or by ldlm_cli_enqueue_fini().
 627                  * DoM uses l_ost_lvb to store LVB data, so copy it here from
 628                  * just updated ols_lvb.
 629                  */
 630                 lock_res_and_lock(lock);
 631                 memcpy(&lock->l_ost_lvb, &ols->ols_lvb,
 632                        sizeof(lock->l_ost_lvb));
 633                 unlock_res_and_lock(lock);
 634                 LDLM_LOCK_PUT(lock);
 635                 *flags |= LDLM_FL_LVB_READY;
 636         }
 637
 638         /* Call the update callback. */
 639         rc = (*upcall)(cookie, lockh, rc < 0 ? rc : errcode);
 640
 641         /* release the reference taken in ldlm_cli_enqueue() */
 642         if (errcode == ELDLM_LOCK_MATCHED)
 643                 errcode = ELDLM_OK;
 644         if (errcode == ELDLM_OK && lustre_handle_is_used(lockh))
 645                 ldlm_lock_decref(lockh, mode);
 646
 647         RETURN(rc);
 648 }
 649
 650 static int mdc_enqueue_interpret(const struct lu_env *env,
 651                                  struct ptlrpc_request *req,
 652                                  void *args, int rc)
 653 {
 654         struct osc_enqueue_args *aa = args;
 655         struct ldlm_lock *lock;
 656         struct lustre_handle *lockh = &aa->oa_lockh;
 657         enum ldlm_mode mode = aa->oa_mode;
 658         struct ldlm_enqueue_info einfo = {
 659                 .ei_type = aa->oa_type,
 660                 .ei_mode = mode,
 661         };
 662
 663         ENTRY;
 664
 665         LASSERT(!aa->oa_speculative);
 666
 667         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
 668          * be valid. */
 669         lock = ldlm_handle2lock(lockh);
 670         LASSERTF(lock != NULL,
 671                  "lockh %#llx, req %px, aa %px - client evicted?\n",
 672                  lockh->cookie, req, aa);
 673
 674         /* Take an additional reference so that a blocking AST that
 675          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
 676          * to arrive after an upcall has been executed by
 677          * mdc_enqueue_fini().
 678          */
 679         ldlm_lock_addref(lockh, mode);
 680
 681         /* Let cl_lock_state_wait fail with -ERESTARTSYS to unuse sublocks. */
 682         CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_HANG, 2);
 683
 684         /* Let CP AST to grant the lock first. */
 685         CFS_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
 686
 687         /* Complete obtaining the lock procedure. */
 688         rc = ldlm_cli_enqueue_fini(aa->oa_exp, &req->rq_pill, &einfo, 1,
 689                                    aa->oa_flags, aa->oa_lvb, aa->oa_lvb ?
 690                                    sizeof(*aa->oa_lvb) : 0, lockh, rc, true);
 691         /* Complete mdc stuff. */
 692         rc = mdc_enqueue_fini(aa->oa_exp, req, aa->oa_upcall, aa->oa_cookie,
 693                               lockh, mode, aa->oa_flags, rc);
 694
 695         CFS_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
 696
 697         ldlm_lock_decref(lockh, mode);
 698         LDLM_LOCK_PUT(lock);
 699         RETURN(rc);
 700 }
 701
 702 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
 703  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
 704  * other synchronous requests, however keeping some locks and trying to obtain
 705  * others may take a considerable amount of time in a case of ost failure; and
 706  * when other sync requests do not get released lock from a client, the client
 707  * is excluded from the cluster -- such scenarious make the life difficult, so
 708  * release locks just after they are obtained. */
 709 static int mdc_enqueue_send(const struct lu_env *env, struct obd_export *exp,
 710                             struct ldlm_res_id *res_id, __u64 *flags,
 711                             union ldlm_policy_data *policy, struct ost_lvb *lvb,
 712                             osc_enqueue_upcall_f upcall, void *cookie,
 713                             struct ldlm_enqueue_info *einfo, int async)
 714 {
 715         struct obd_device *obd = exp->exp_obd;
 716         struct lustre_handle lockh = { 0 };
 717         struct ptlrpc_request *req = NULL;
 718         struct ldlm_intent *lit;
 719         enum ldlm_mode mode;
 720         bool glimpse = *flags & LDLM_FL_HAS_INTENT;
 721         __u64 search_flags = *flags;
 722         __u64 match_flags = 0;
 723         LIST_HEAD(cancels);
 724         int rc, count;
 725         int lvb_size;
 726         bool compat_glimpse = glimpse && !exp_connect_dom_lvb(exp);
 727
 728         ENTRY;
 729
 730         mode = einfo->ei_mode;
 731         if (einfo->ei_mode == LCK_PR)
 732                 mode |= LCK_PW;
 733
 734         search_flags |= LDLM_FL_LVB_READY;
 735         if (glimpse)
 736                 search_flags |= LDLM_FL_BLOCK_GRANTED;
 737         if (mode == LCK_GROUP)
 738                 match_flags = LDLM_MATCH_GROUP;
 739         mode = ldlm_lock_match_with_skip(obd->obd_namespace, search_flags, 0,
 740                                          res_id, einfo->ei_type, policy, mode,
 741                                          &lockh, match_flags);
 742         if (mode) {
 743                 struct ldlm_lock *matched;
 744
 745                 if (*flags & LDLM_FL_TEST_LOCK)
 746                         RETURN(ELDLM_OK);
 747
 748                 matched = ldlm_handle2lock(&lockh);
 749
 750                 if (CFS_FAIL_CHECK(OBD_FAIL_MDC_GLIMPSE_DDOS))
 751                         ldlm_set_kms_ignore(matched);
 752
 753                 if (mdc_set_dom_lock_data(matched, einfo->ei_cbdata)) {
 754                         *flags |= LDLM_FL_LVB_READY;
 755
 756                         /* We already have a lock, and it's referenced. */
 757                         (*upcall)(cookie, &lockh, ELDLM_LOCK_MATCHED);
 758
 759                         ldlm_lock_decref(&lockh, mode);
 760                         LDLM_LOCK_PUT(matched);
 761                         RETURN(ELDLM_OK);
 762                 }
 763                 ldlm_lock_decref(&lockh, mode);
 764                 LDLM_LOCK_PUT(matched);
 765         }
 766
 767         if (*flags & (LDLM_FL_TEST_LOCK | LDLM_FL_MATCH_LOCK))
 768                 RETURN(-ENOLCK);
 769
 770         /* Glimpse is intent on old server */
 771         req = ptlrpc_request_alloc(class_exp2cliimp(exp), compat_glimpse ?
 772                                    &RQF_LDLM_INTENT : &RQF_LDLM_ENQUEUE);
 773         if (req == NULL)
 774                 RETURN(-ENOMEM);
 775
 776         /* For WRITE lock cancel other locks on resource early if any */
 777         if (einfo->ei_mode & LCK_PW)
 778                 count = mdc_resource_get_unused_res(exp, res_id, &cancels,
 779                                                     einfo->ei_mode,
 780                                                     MDS_INODELOCK_DOM);
 781         else
 782                 count = 0;
 783
 784         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
 785         if (rc < 0) {
 786                 ptlrpc_request_free(req);
 787                 RETURN(rc);
 788         }
 789
 790         if (compat_glimpse) {
 791                 /* pack the glimpse intent */
 792                 lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
 793                 lit->opc = IT_GLIMPSE;
 794         }
 795
 796         /* users of mdc_enqueue() can pass this flag for ldlm_lock_match() */
 797         *flags &= ~LDLM_FL_BLOCK_GRANTED;
 798
 799         if (compat_glimpse) {
 800                 req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, 0);
 801                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, 0);
 802                 lvb_size = 0;
 803         } else {
 804                 lvb_size = sizeof(*lvb);
 805                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
 806                                      lvb_size);
 807         }
 808         ptlrpc_request_set_replen(req);
 809
 810         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
 811                               lvb_size, LVB_T_OST, &lockh, async);
 812         if (async) {
 813                 if (!rc) {
 814                         struct osc_enqueue_args *aa;
 815
 816                         aa = ptlrpc_req_async_args(aa, req);
 817                         aa->oa_exp = exp;
 818                         aa->oa_mode = einfo->ei_mode;
 819                         aa->oa_type = einfo->ei_type;
 820                         lustre_handle_copy(&aa->oa_lockh, &lockh);
 821                         aa->oa_upcall = upcall;
 822                         aa->oa_cookie = cookie;
 823                         aa->oa_speculative = false;
 824                         aa->oa_flags = flags;
 825                         aa->oa_lvb = compat_glimpse ? NULL : lvb;
 826
 827                         req->rq_interpret_reply = mdc_enqueue_interpret;
 828                         ptlrpcd_add_req(req);
 829                 } else {
 830                         ptlrpc_req_finished(req);
 831                 }
 832                 RETURN(rc);
 833         }
 834
 835         rc = mdc_enqueue_fini(exp, req, upcall, cookie, &lockh, einfo->ei_mode,
 836                               flags, rc);
 837         ptlrpc_req_finished(req);
 838         RETURN(rc);
 839 }
 840
 841 /**
 842  * Implementation of cl_lock_operations::clo_enqueue() method for osc
 843  * layer. This initiates ldlm enqueue:
 844  *
 845  *     - cancels conflicting locks early (osc_lock_enqueue_wait());
 846  *
 847  *     - calls osc_enqueue_base() to do actual enqueue.
 848  *
 849  * osc_enqueue_base() is supplied with an upcall function that is executed
 850  * when lock is received either after a local cached ldlm lock is matched, or
 851  * when a reply from the server is received.
 852  *
 853  * This function does not wait for the network communication to complete.
 854  */
 855 static int mdc_lock_enqueue(const struct lu_env *env,
 856                             const struct cl_lock_slice *slice,
 857                             struct cl_io *unused, struct cl_sync_io *anchor)
 858 {
 859         struct osc_thread_info *info = osc_env_info(env);
 860         struct osc_io *oio = osc_env_io(env);
 861         struct osc_object *osc = cl2osc(slice->cls_obj);
 862         struct osc_lock *oscl = cl2osc_lock(slice);
 863         struct cl_lock *lock = slice->cls_lock;
 864         struct ldlm_res_id *resname = &info->oti_resname;
 865         union ldlm_policy_data *policy = &info->oti_policy;
 866         osc_enqueue_upcall_f upcall = mdc_lock_upcall;
 867         void *cookie = (void *)oscl;
 868         bool async = false;
 869         int result;
 870
 871         ENTRY;
 872
 873         LASSERTF(ergo(oscl->ols_glimpse, lock->cll_descr.cld_mode <= CLM_READ),
 874                 "lock = %px, ols = %px\n", lock, oscl);
 875
 876         if (oscl->ols_state == OLS_GRANTED)
 877                 RETURN(0);
 878
 879         /* Lockahead is not supported on MDT yet */
 880         if (oscl->ols_flags & LDLM_FL_NO_EXPANSION) {
 881                 result = -EOPNOTSUPP;
 882                 RETURN(result);
 883         }
 884
 885         if (oscl->ols_flags & LDLM_FL_TEST_LOCK)
 886                 GOTO(enqueue_base, 0);
 887
 888         if (oscl->ols_glimpse) {
 889                 LASSERT(equi(oscl->ols_speculative, anchor == NULL));
 890                 async = true;
 891                 GOTO(enqueue_base, 0);
 892         }
 893
 894         result = osc_lock_enqueue_wait(env, osc, oscl);
 895         if (result < 0)
 896                 GOTO(out, result);
 897
 898         /* we can grant lockless lock right after all conflicting locks
 899          * are canceled. */
 900         if (osc_lock_is_lockless(oscl)) {
 901                 oscl->ols_state = OLS_GRANTED;
 902                 oio->oi_lockless = 1;
 903                 RETURN(0);
 904         }
 905
 906 enqueue_base:
 907         oscl->ols_state = OLS_ENQUEUED;
 908         if (anchor != NULL) {
 909                 atomic_inc(&anchor->csi_sync_nr);
 910                 oscl->ols_owner = anchor;
 911         }
 912
 913         /**
 914          * DLM lock's ast data must be osc_object;
 915          * DLM's enqueue callback set to osc_lock_upcall() with cookie as
 916          * osc_lock.
 917          */
 918         fid_build_reg_res_name(lu_object_fid(osc2lu(osc)), resname);
 919         mdc_lock_build_policy(env, lock, policy);
 920         LASSERT(!oscl->ols_speculative);
 921         result = mdc_enqueue_send(env, osc_export(osc), resname,
 922                                   &oscl->ols_flags, policy, &oscl->ols_lvb,
 923                                   upcall, cookie, &oscl->ols_einfo, async);
 924         if (result == 0) {
 925                 if (osc_lock_is_lockless(oscl)) {
 926                         oio->oi_lockless = 1;
 927                 } else if (!async) {
 928                         LASSERT(oscl->ols_state == OLS_GRANTED);
 929                         LASSERT(oscl->ols_hold);
 930                         LASSERT(oscl->ols_dlmlock != NULL);
 931                 }
 932         }
 933 out:
 934         if (result < 0) {
 935                 oscl->ols_state = OLS_CANCELLED;
 936                 osc_lock_wake_waiters(env, osc, oscl);
 937
 938                 if (anchor != NULL)
 939                         cl_sync_io_note(env, anchor, result);
 940         }
 941         RETURN(result);
 942 }
 943
 944 static const struct cl_lock_operations mdc_lock_lockless_ops = {
 945         .clo_fini = osc_lock_fini,
 946         .clo_enqueue = mdc_lock_enqueue,
 947         .clo_cancel = mdc_lock_lockless_cancel,
 948         .clo_print = osc_lock_print
 949 };
 950
 951 static const struct cl_lock_operations mdc_lock_ops = {
 952         .clo_fini       = osc_lock_fini,
 953         .clo_enqueue    = mdc_lock_enqueue,
 954         .clo_cancel     = osc_lock_cancel,
 955         .clo_print      = osc_lock_print,
 956 };
 957
 958 static int mdc_lock_init(const struct lu_env *env, struct cl_object *obj,
 959                          struct cl_lock *lock, const struct cl_io *io)
 960 {
 961         struct osc_lock *ols;
 962         __u32 enqflags = lock->cll_descr.cld_enq_flags;
 963         __u64 flags = osc_enq2ldlm_flags(enqflags);
 964
 965         ENTRY;
 966
 967         /* Ignore AGL for Data-on-MDT, stat returns size data */
 968         if ((enqflags & CEF_SPECULATIVE) != 0)
 969                 RETURN(0);
 970
 971         OBD_SLAB_ALLOC_PTR_GFP(ols, osc_lock_kmem, GFP_NOFS);
 972         if (unlikely(ols == NULL))
 973                 RETURN(-ENOMEM);
 974
 975         ols->ols_state = OLS_NEW;
 976         spin_lock_init(&ols->ols_lock);
 977         INIT_LIST_HEAD(&ols->ols_waiting_list);
 978         INIT_LIST_HEAD(&ols->ols_wait_entry);
 979         INIT_LIST_HEAD(&ols->ols_nextlock_oscobj);
 980         ols->ols_lockless_ops = &mdc_lock_lockless_ops;
 981
 982         ols->ols_flags = flags;
 983         ols->ols_speculative = !!(enqflags & CEF_SPECULATIVE);
 984
 985         if (ols->ols_flags & LDLM_FL_HAS_INTENT) {
 986                 ols->ols_flags |= LDLM_FL_BLOCK_GRANTED;
 987                 ols->ols_glimpse = 1;
 988         }
 989         mdc_lock_build_einfo(env, lock, cl2osc(obj), &ols->ols_einfo);
 990
 991         cl_lock_slice_add(lock, &ols->ols_cl, obj, &mdc_lock_ops);
 992
 993         if (!(enqflags & CEF_MUST))
 994                 osc_lock_to_lockless(env, ols, (enqflags & CEF_NEVER));
 995
 996         if (io->ci_type == CIT_WRITE || cl_io_is_mkwrite(io))
 997                 osc_lock_set_writer(env, io, obj, ols);
 998         else if (io->ci_type == CIT_READ ||
 999                  (io->ci_type == CIT_FAULT && !io->u.ci_fault.ft_mkwrite))
1000                 osc_lock_set_reader(env, io, obj, ols);
1001
1002         LDLM_DEBUG_NOLOCK("lock %p, mdc lock %p, flags %llx",
1003                           lock, ols, ols->ols_flags);
1004         RETURN(0);
1005 }
1006
1007 /**
1008  * IO operations.
1009  *
1010  * An implementation of cl_io_operations specific methods for MDC layer.
1011  *
1012  */
1013 static int mdc_async_upcall(void *a, int rc)
1014 {
1015         struct osc_async_cbargs *args = a;
1016
1017         args->opc_rc = rc;
1018         complete(&args->opc_sync);
1019         return 0;
1020 }
1021
1022 static int mdc_get_lock_handle(const struct lu_env *env, struct osc_object *osc,
1023                                pgoff_t index, struct lustre_handle *lh)
1024 {
1025         struct ldlm_lock *lock;
1026
1027         /* find DOM lock protecting object */
1028         lock = mdc_dlmlock_at_pgoff(env, osc, index,
1029                                     OSC_DAP_FL_TEST_LOCK |
1030                                     OSC_DAP_FL_CANCELING);
1031         if (lock == NULL) {
1032                 struct ldlm_resource *res;
1033                 struct ldlm_res_id *resname;
1034
1035                 resname = &osc_env_info(env)->oti_resname;
1036                 fid_build_reg_res_name(lu_object_fid(osc2lu(osc)), resname);
1037                 res = ldlm_resource_get(osc_export(osc)->
1038                                                         exp_obd->obd_namespace,
1039                                         resname, LDLM_IBITS, 0);
1040                 if (IS_ERR(res))
1041                         CERROR("No lock resource for "DFID"\n",
1042                                 PFID(lu_object_fid(osc2lu(osc))));
1043                 else
1044                         ldlm_resource_dump(D_ERROR, res);
1045                 dump_stack();
1046                 return -ENOENT;
1047         } else {
1048                 *lh = lock->l_remote_handle;
1049                 LDLM_LOCK_PUT(lock);
1050         }
1051         return 0;
1052 }
1053
1054 static int mdc_io_setattr_start(const struct lu_env *env,
1055                                 const struct cl_io_slice *slice)
1056 {
1057         struct cl_io *io = slice->cis_io;
1058         struct osc_io *oio = cl2osc_io(env, slice);
1059         struct cl_object *obj = slice->cis_obj;
1060         struct lov_oinfo *loi = cl2osc(obj)->oo_oinfo;
1061         struct cl_attr *attr = &osc_env_info(env)->oti_attr;
1062         struct obdo *oa = &oio->oi_oa;
1063         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
1064         __u64 size = io->u.ci_setattr.sa_attr.lvb_size;
1065         unsigned int ia_avalid = io->u.ci_setattr.sa_avalid;
1066         enum op_xvalid ia_xvalid = io->u.ci_setattr.sa_xvalid;
1067         int rc;
1068
1069         /* silently ignore non-truncate setattr for Data-on-MDT object */
1070         if (cl_io_is_trunc(io)) {
1071                 /* truncate cache dirty pages first */
1072                 rc = osc_cache_truncate_start(env, cl2osc(obj), size,
1073                                               &oio->oi_trunc);
1074                 if (rc < 0)
1075                         return rc;
1076         } else if (cl_io_is_fallocate(io) &&
1077                    io->u.ci_setattr.sa_falloc_mode & FALLOC_FL_PUNCH_HOLE) {
1078                 rc = osc_punch_start(env, io, obj);
1079                 if (rc < 0)
1080                         return rc;
1081         }
1082
1083         if (oio->oi_lockless == 0) {
1084                 cl_object_attr_lock(obj);
1085                 rc = cl_object_attr_get(env, obj, attr);
1086                 if (rc == 0) {
1087                         struct ost_lvb *lvb = &io->u.ci_setattr.sa_attr;
1088                         unsigned int cl_valid = 0;
1089
1090                         if (ia_avalid & ATTR_SIZE) {
1091                                 attr->cat_size = size;
1092                                 attr->cat_kms = size;
1093                                 cl_valid = (CAT_SIZE | CAT_KMS);
1094                         }
1095                         if (ia_avalid & ATTR_MTIME_SET) {
1096                                 attr->cat_mtime = lvb->lvb_mtime;
1097                                 cl_valid |= CAT_MTIME;
1098                         }
1099                         if (ia_avalid & ATTR_ATIME_SET) {
1100                                 attr->cat_atime = lvb->lvb_atime;
1101                                 cl_valid |= CAT_ATIME;
1102                         }
1103                         if (ia_xvalid & OP_XVALID_CTIME_SET) {
1104                                 attr->cat_ctime = lvb->lvb_ctime;
1105                                 cl_valid |= CAT_CTIME;
1106                         }
1107                         rc = cl_object_attr_update(env, obj, attr, cl_valid);
1108                 }
1109                 cl_object_attr_unlock(obj);
1110                 if (rc < 0)
1111                         return rc;
1112         }
1113
1114         if (!(ia_avalid & ATTR_SIZE) && !cl_io_is_fallocate(io))
1115                 return 0;
1116
1117         memset(oa, 0, sizeof(*oa));
1118         oa->o_oi = loi->loi_oi;
1119         oa->o_mtime = attr->cat_mtime;
1120         oa->o_atime = attr->cat_atime;
1121         oa->o_ctime = attr->cat_ctime;
1122         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLATIME |
1123                       OBD_MD_FLCTIME | OBD_MD_FLMTIME | OBD_MD_FLSIZE |
1124                       OBD_MD_FLBLOCKS;
1125
1126         if (oio->oi_lockless) {
1127                 oa->o_flags = OBD_FL_SRVLOCK;
1128                 oa->o_valid |= OBD_MD_FLFLAGS;
1129         } else {
1130                 rc = mdc_get_lock_handle(env, cl2osc(obj), CL_PAGE_EOF,
1131                                          &oa->o_handle);
1132                 if (!rc)
1133                         oa->o_valid |= OBD_MD_FLHANDLE;
1134         }
1135
1136         init_completion(&cbargs->opc_sync);
1137         if (cl_io_is_fallocate(io)) {
1138                 int falloc_mode = io->u.ci_setattr.sa_falloc_mode;
1139
1140                 oa->o_size = io->u.ci_setattr.sa_falloc_offset;
1141                 oa->o_blocks = io->u.ci_setattr.sa_falloc_end;
1142                 rc = osc_fallocate_base(osc_export(cl2osc(obj)), oa,
1143                                         mdc_async_upcall, cbargs, falloc_mode);
1144         } else {
1145                 oa->o_size = size;
1146                 oa->o_blocks = OBD_OBJECT_EOF;
1147                 rc = osc_punch_send(osc_export(cl2osc(obj)), oa,
1148                                     mdc_async_upcall, cbargs);
1149         }
1150         cbargs->opc_rpc_sent = rc == 0;
1151         return rc;
1152 }
1153
1154 static int mdc_io_read_ahead(const struct lu_env *env,
1155                              const struct cl_io_slice *ios,
1156                              pgoff_t start, struct cl_read_ahead *ra)
1157 {
1158         struct osc_object *osc = cl2osc(ios->cis_obj);
1159         struct osc_io *oio = cl2osc_io(env, ios);
1160         struct ldlm_lock *dlmlock;
1161
1162         ENTRY;
1163
1164         dlmlock = mdc_dlmlock_at_pgoff(env, osc, start, 0);
1165         if (dlmlock == NULL)
1166                 RETURN(-ENODATA);
1167
1168         oio->oi_is_readahead = 1;
1169         if (dlmlock->l_req_mode != LCK_PR) {
1170                 struct lustre_handle lockh;
1171
1172                 ldlm_lock2handle(dlmlock, &lockh);
1173                 ldlm_lock_addref(&lockh, LCK_PR);
1174                 ldlm_lock_decref(&lockh, dlmlock->l_req_mode);
1175         }
1176
1177         ra->cra_rpc_pages = osc_cli(osc)->cl_max_pages_per_rpc;
1178         ra->cra_end_idx = CL_PAGE_EOF;
1179         ra->cra_release = osc_read_ahead_release;
1180         ra->cra_dlmlock = dlmlock;
1181         ra->cra_oio = oio;
1182
1183         RETURN(0);
1184 }
1185
1186 static int mdc_io_fsync_start(const struct lu_env *env,
1187                               const struct cl_io_slice *slice)
1188 {
1189         struct cl_io *io = slice->cis_io;
1190         struct cl_fsync_io *fio = &io->u.ci_fsync;
1191         struct cl_object *obj = slice->cis_obj;
1192         struct osc_object *osc = cl2osc(obj);
1193         int result = 0;
1194
1195         ENTRY;
1196
1197         if (fio->fi_mode == CL_FSYNC_RECLAIM) {
1198                 struct client_obd *cli = osc_cli(osc);
1199
1200                 if (!atomic_long_read(&cli->cl_unstable_count)) {
1201                         /* Stop flush when there are no unstable pages? */
1202                         CDEBUG(D_CACHE, "unstable count is zero\n");
1203                         RETURN(0);
1204                 }
1205         }
1206
1207         /* a MDC lock always covers whole object, do sync for whole
1208          * possible range despite of supplied start/end values.
1209          */
1210         result = osc_cache_writeback_range(env, osc, 0, CL_PAGE_EOF, 0,
1211                                            fio->fi_mode == CL_FSYNC_DISCARD);
1212         if (result > 0) {
1213                 fio->fi_nr_written += result;
1214                 result = 0;
1215         }
1216         if (fio->fi_mode == CL_FSYNC_ALL || fio->fi_mode == CL_FSYNC_RECLAIM) {
1217                 struct osc_io *oio = cl2osc_io(env, slice);
1218                 struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
1219                 int rc;
1220
1221                 if (fio->fi_mode == CL_FSYNC_ALL) {
1222                         rc = osc_cache_wait_range(env, osc, 0, CL_PAGE_EOF);
1223                         if (result == 0)
1224                                 result = rc;
1225                 }
1226                 /* Use OSC sync code because it is asynchronous.
1227                  * It is to be added into MDC and avoid the using of
1228                  * OST_SYNC at both MDC and MDT.
1229                  */
1230                 rc = osc_fsync_ost(env, osc, fio);
1231                 if (result == 0) {
1232                         cbargs->opc_rpc_sent = 1;
1233                         result = rc;
1234                 }
1235         }
1236
1237         RETURN(result);
1238 }
1239
1240 struct mdc_data_version_args {
1241         struct osc_io *dva_oio;
1242 };
1243
1244 static int
1245 mdc_data_version_interpret(const struct lu_env *env, struct ptlrpc_request *req,
1246                            void *args, int rc)
1247 {
1248         struct mdc_data_version_args *dva = args;
1249         struct osc_io *oio = dva->dva_oio;
1250         const struct mdt_body *body;
1251
1252         ENTRY;
1253         if (rc < 0)
1254                 GOTO(out, rc);
1255
1256         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
1257         if (body == NULL)
1258                 GOTO(out, rc = -EPROTO);
1259
1260         /* Prepare OBDO from mdt_body for CLIO */
1261         oio->oi_oa.o_valid = body->mbo_valid;
1262         oio->oi_oa.o_flags = body->mbo_flags;
1263         oio->oi_oa.o_data_version = body->mbo_version;
1264         oio->oi_oa.o_layout_version = body->mbo_layout_gen;
1265         EXIT;
1266 out:
1267         oio->oi_cbarg.opc_rc = rc;
1268         complete(&oio->oi_cbarg.opc_sync);
1269         return 0;
1270 }
1271
1272 static int mdc_io_data_version_start(const struct lu_env *env,
1273                                      const struct cl_io_slice *slice)
1274 {
1275         struct cl_data_version_io *dv = &slice->cis_io->u.ci_data_version;
1276         struct osc_io *oio = cl2osc_io(env, slice);
1277         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
1278         struct osc_object *obj = cl2osc(slice->cis_obj);
1279         struct obd_export *exp = osc_export(obj);
1280         struct ptlrpc_request *req;
1281         struct mdt_body *body;
1282         struct mdc_data_version_args *dva;
1283         int rc;
1284
1285         ENTRY;
1286
1287         memset(&oio->oi_oa, 0, sizeof(oio->oi_oa));
1288         oio->oi_oa.o_oi.oi_fid = *lu_object_fid(osc2lu(obj));
1289         oio->oi_oa.o_valid = OBD_MD_FLID;
1290
1291         init_completion(&cbargs->opc_sync);
1292
1293         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_GETATTR);
1294         if (req == NULL)
1295                 RETURN(-ENOMEM);
1296
1297         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
1298         if (rc < 0) {
1299                 ptlrpc_request_free(req);
1300                 RETURN(rc);
1301         }
1302
1303         body = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
1304         body->mbo_fid1 = *lu_object_fid(osc2lu(obj));
1305         body->mbo_valid = OBD_MD_FLID;
1306         /* Indicate that data version is needed */
1307         body->mbo_valid |= OBD_MD_FLDATAVERSION;
1308         body->mbo_flags = 0;
1309
1310         if (dv->dv_flags & (LL_DV_RD_FLUSH | LL_DV_WR_FLUSH)) {
1311                 body->mbo_valid |= OBD_MD_FLFLAGS;
1312                 body->mbo_flags |= OBD_FL_SRVLOCK;
1313                 if (dv->dv_flags & LL_DV_WR_FLUSH)
1314                         body->mbo_flags |= OBD_FL_FLUSH;
1315         }
1316
1317         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, 0);
1318         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, 0);
1319         req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX, RCL_SERVER, 0);
1320         ptlrpc_request_set_replen(req);
1321
1322         req->rq_interpret_reply = mdc_data_version_interpret;
1323         dva = ptlrpc_req_async_args(dva, req);
1324         dva->dva_oio = oio;
1325
1326         ptlrpcd_add_req(req);
1327
1328         RETURN(0);
1329 }
1330
1331 static void mdc_io_data_version_end(const struct lu_env *env,
1332                                     const struct cl_io_slice *slice)
1333 {
1334         struct cl_data_version_io *dv = &slice->cis_io->u.ci_data_version;
1335         struct osc_io *oio = cl2osc_io(env, slice);
1336         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
1337
1338         ENTRY;
1339         wait_for_completion(&cbargs->opc_sync);
1340
1341         if (cbargs->opc_rc != 0) {
1342                 slice->cis_io->ci_result = cbargs->opc_rc;
1343         } else {
1344                 slice->cis_io->ci_result = 0;
1345                 if (!(oio->oi_oa.o_valid &
1346                       (OBD_MD_LAYOUT_VERSION | OBD_MD_FLDATAVERSION)))
1347                         slice->cis_io->ci_result = -EOPNOTSUPP;
1348
1349                 if (oio->oi_oa.o_valid & OBD_MD_LAYOUT_VERSION)
1350                         dv->dv_layout_version = oio->oi_oa.o_layout_version;
1351                 if (oio->oi_oa.o_valid & OBD_MD_FLDATAVERSION)
1352                         dv->dv_data_version = oio->oi_oa.o_data_version;
1353         }
1354
1355         EXIT;
1356 }
1357
1358 static const struct cl_io_operations mdc_io_ops = {
1359         .op = {
1360                 [CIT_READ] = {
1361                         .cio_iter_init = osc_io_iter_init,
1362                         .cio_iter_fini = osc_io_rw_iter_fini,
1363                         .cio_start     = osc_io_read_start,
1364                 },
1365                 [CIT_WRITE] = {
1366                         .cio_iter_init = osc_io_iter_init,
1367                         .cio_iter_fini = osc_io_rw_iter_fini,
1368                         .cio_start     = osc_io_write_start,
1369                         .cio_end       = osc_io_end,
1370                 },
1371                 [CIT_SETATTR] = {
1372                         .cio_iter_init = osc_io_iter_init,
1373                         .cio_iter_fini = osc_io_iter_fini,
1374                         .cio_start     = mdc_io_setattr_start,
1375                         .cio_end       = osc_io_setattr_end,
1376                 },
1377                 [CIT_DATA_VERSION] = {
1378                         .cio_start = mdc_io_data_version_start,
1379                         .cio_end   = mdc_io_data_version_end,
1380                 },
1381                 [CIT_FAULT] = {
1382                         .cio_iter_init = osc_io_iter_init,
1383                         .cio_iter_fini = osc_io_iter_fini,
1384                         .cio_start     = osc_io_fault_start,
1385                         .cio_end       = osc_io_end,
1386                 },
1387                 [CIT_FSYNC] = {
1388                         .cio_start = mdc_io_fsync_start,
1389                         .cio_end   = osc_io_fsync_end,
1390                 },
1391                 [CIT_LSEEK] = {
1392                         .cio_start  = osc_io_lseek_start,
1393                         .cio_end    = osc_io_lseek_end,
1394                 },
1395         },
1396         .cio_read_ahead   = mdc_io_read_ahead,
1397         .cio_lru_reserve  = osc_io_lru_reserve,
1398         .cio_submit       = osc_io_submit,
1399         .cio_commit_async = osc_io_commit_async,
1400         .cio_extent_release = osc_io_extent_release,
1401 };
1402
1403 static int mdc_io_init(const struct lu_env *env, struct cl_object *obj,
1404                        struct cl_io *io)
1405 {
1406         struct osc_io *oio = osc_env_io(env);
1407
1408         CL_IO_SLICE_CLEAN(oio, oi_cl);
1409         cl_io_slice_add(io, &oio->oi_cl, obj, &mdc_io_ops);
1410         return 0;
1411 }
1412
1413 static void mdc_build_res_name(struct osc_object *osc,
1414                                    struct ldlm_res_id *resname)
1415 {
1416         fid_build_reg_res_name(lu_object_fid(osc2lu(osc)), resname);
1417 }
1418
1419 /**
1420  * Implementation of struct cl_req_operations::cro_attr_set() for MDC
1421  * layer. MDC is responsible for struct obdo::o_id and struct obdo::o_seq
1422  * fields.
1423  */
1424 static void mdc_req_attr_set(const struct lu_env *env, struct cl_object *obj,
1425                              struct cl_req_attr *attr)
1426 {
1427         u64 flags = attr->cra_flags;
1428
1429         /* Copy object FID to cl_attr */
1430         attr->cra_oa->o_oi.oi_fid = *lu_object_fid(&obj->co_lu);
1431
1432         if (flags & OBD_MD_FLGROUP)
1433                 attr->cra_oa->o_valid |= OBD_MD_FLGROUP;
1434
1435         if (flags & OBD_MD_FLID)
1436                 attr->cra_oa->o_valid |= OBD_MD_FLID;
1437
1438         if (flags & OBD_MD_FLHANDLE) {
1439                 struct osc_page *opg;
1440
1441                 opg = osc_cl_page_osc(attr->cra_page, cl2osc(obj));
1442                 if (!opg->ops_srvlock) {
1443                         int rc;
1444
1445                         rc = mdc_get_lock_handle(env, cl2osc(obj),
1446                                                  osc_index(opg),
1447                                                  &attr->cra_oa->o_handle);
1448                         if (rc) {
1449                                 CL_PAGE_DEBUG(D_ERROR, env, attr->cra_page,
1450                                               "uncovered page!\n");
1451                                 LBUG();
1452                         } else {
1453                                 attr->cra_oa->o_valid |= OBD_MD_FLHANDLE;
1454                         }
1455                 }
1456         }
1457 }
1458
1459 static int mdc_attr_get(const struct lu_env *env, struct cl_object *obj,
1460                         struct cl_attr *attr)
1461 {
1462         struct lov_oinfo *oinfo = cl2osc(obj)->oo_oinfo;
1463
1464         if (OST_LVB_IS_ERR(oinfo->loi_lvb.lvb_blocks))
1465                 return OST_LVB_GET_ERR(oinfo->loi_lvb.lvb_blocks);
1466
1467         return osc_attr_get(env, obj, attr);
1468 }
1469
1470 static int mdc_object_ast_clear(struct ldlm_lock *lock, void *data)
1471 {
1472         struct osc_object *osc = (struct osc_object *)data;
1473         struct ost_lvb *lvb = &lock->l_ost_lvb;
1474         struct lov_oinfo *oinfo;
1475         ENTRY;
1476
1477         if (lock->l_ast_data == data) {
1478                 lock->l_ast_data = NULL;
1479
1480                 LASSERT(osc != NULL);
1481                 LASSERT(osc->oo_oinfo != NULL);
1482                 LASSERT(lvb != NULL);
1483
1484                 /* Updates lvb in lock by the cached oinfo */
1485                 oinfo = osc->oo_oinfo;
1486
1487                 LDLM_DEBUG(lock, "update lock size %llu blocks %llu [cma]time: "
1488                            "%llu %llu %llu by oinfo size %llu blocks %llu "
1489                            "[cma]time %llu %llu %llu", lvb->lvb_size,
1490                            lvb->lvb_blocks, lvb->lvb_ctime, lvb->lvb_mtime,
1491                            lvb->lvb_atime, oinfo->loi_lvb.lvb_size,
1492                            oinfo->loi_lvb.lvb_blocks, oinfo->loi_lvb.lvb_ctime,
1493                            oinfo->loi_lvb.lvb_mtime, oinfo->loi_lvb.lvb_atime);
1494                 LASSERT(oinfo->loi_lvb.lvb_size >= oinfo->loi_kms);
1495
1496                 cl_object_attr_lock(&osc->oo_cl);
1497                 memcpy(lvb, &oinfo->loi_lvb, sizeof(oinfo->loi_lvb));
1498                 cl_object_attr_unlock(&osc->oo_cl);
1499                 ldlm_clear_lvb_cached(lock);
1500         }
1501         RETURN(LDLM_ITER_CONTINUE);
1502 }
1503
1504 static int mdc_object_prune(const struct lu_env *env, struct cl_object *obj)
1505 {
1506         struct osc_object *osc = cl2osc(obj);
1507         struct ldlm_res_id *resname = &osc_env_info(env)->oti_resname;
1508
1509         /* DLM locks don't hold a reference of osc_object so we have to
1510          * clear it before the object is being destroyed. */
1511         osc_build_res_name(osc, resname);
1512         ldlm_resource_iterate(osc_export(osc)->exp_obd->obd_namespace, resname,
1513                               mdc_object_ast_clear, osc);
1514         return 0;
1515 }
1516
1517 static int mdc_object_flush(const struct lu_env *env, struct cl_object *obj,
1518                             struct ldlm_lock *lock)
1519 {
1520         /* if lock cancel is initiated from llite then it is combined
1521          * lock with DOM bit and it may have no l_ast_data initialized yet,
1522          * so init it here with given osc_object.
1523          */
1524         mdc_set_dom_lock_data(lock, cl2osc(obj));
1525         RETURN(mdc_dlm_canceling(env, lock));
1526 }
1527
1528 static const struct cl_object_operations mdc_ops = {
1529         .coo_page_init = osc_page_init,
1530         .coo_lock_init = mdc_lock_init,
1531         .coo_io_init = mdc_io_init,
1532         .coo_attr_get = mdc_attr_get,
1533         .coo_attr_update = osc_attr_update,
1534         .coo_glimpse = osc_object_glimpse,
1535         .coo_req_attr_set = mdc_req_attr_set,
1536         .coo_prune = mdc_object_prune,
1537         .coo_object_flush = mdc_object_flush
1538 };
1539
1540 static const struct osc_object_operations mdc_object_ops = {
1541         .oto_build_res_name = mdc_build_res_name,
1542         .oto_dlmlock_at_pgoff = mdc_dlmlock_at_pgoff,
1543 };
1544
1545 static int mdc_object_init(const struct lu_env *env, struct lu_object *obj,
1546                            const struct lu_object_conf *conf)
1547 {
1548         struct osc_object *osc = lu2osc(obj);
1549
1550         if (osc->oo_initialized)
1551                 return 0;
1552
1553         osc->oo_initialized = true;
1554
1555         return osc_object_init(env, obj, conf);
1556 }
1557
1558 static void mdc_object_free(const struct lu_env *env, struct lu_object *obj)
1559 {
1560         osc_object_free(env, obj);
1561 }
1562
1563 static const struct lu_object_operations mdc_lu_obj_ops = {
1564         .loo_object_init = mdc_object_init,
1565         .loo_object_delete = NULL,
1566         .loo_object_release = NULL,
1567         .loo_object_free = mdc_object_free,
1568         .loo_object_print = osc_object_print,
1569         .loo_object_invariant = NULL
1570 };
1571
1572 static struct lu_object *mdc_object_alloc(const struct lu_env *env,
1573                                           const struct lu_object_header *unused,
1574                                           struct lu_device *dev)
1575 {
1576         struct osc_object *osc;
1577         struct lu_object  *obj;
1578
1579         OBD_SLAB_ALLOC_PTR_GFP(osc, osc_object_kmem, GFP_NOFS);
1580         if (osc != NULL) {
1581                 obj = osc2lu(osc);
1582                 lu_object_init(obj, NULL, dev);
1583                 osc->oo_cl.co_ops = &mdc_ops;
1584                 obj->lo_ops = &mdc_lu_obj_ops;
1585                 osc->oo_obj_ops = &mdc_object_ops;
1586                 osc->oo_initialized = false;
1587         } else {
1588                 obj = NULL;
1589         }
1590         return obj;
1591 }
1592
1593 static int mdc_process_config(const struct lu_env *env, struct lu_device *d,
1594                               struct lustre_cfg *cfg)
1595 {
1596         size_t count  = class_modify_config(cfg, PARAM_MDC,
1597                                             &d->ld_obd->obd_kset.kobj);
1598         return count > 0 ? 0 : count;
1599 }
1600
1601 const struct lu_device_operations mdc_lu_ops = {
1602         .ldo_object_alloc = mdc_object_alloc,
1603         .ldo_process_config = mdc_process_config,
1604         .ldo_recovery_complete = NULL,
1605 };
1606
1607 static struct lu_device *mdc_device_alloc(const struct lu_env *env,
1608                                           struct lu_device_type *t,
1609                                           struct lustre_cfg *cfg)
1610 {
1611         struct lu_device *d;
1612         struct osc_device *oc;
1613         struct obd_device *obd;
1614         int rc;
1615
1616         OBD_ALLOC_PTR(oc);
1617         if (oc == NULL)
1618                 RETURN(ERR_PTR(-ENOMEM));
1619
1620         cl_device_init(&oc->osc_cl, t);
1621         d = osc2lu_dev(oc);
1622         d->ld_ops = &mdc_lu_ops;
1623
1624         /* Setup MDC OBD */
1625         obd = class_name2obd(lustre_cfg_string(cfg, 0));
1626         if (obd == NULL)
1627                 RETURN(ERR_PTR(-ENODEV));
1628
1629         rc = mdc_setup(obd, cfg);
1630         if (rc < 0) {
1631                 osc_device_free(env, d);
1632                 RETURN(ERR_PTR(rc));
1633         }
1634         oc->osc_exp = obd->obd_self_export;
1635         oc->osc_stats.os_init = ktime_get_real();
1636         RETURN(d);
1637 }
1638
1639 static const struct lu_device_type_operations mdc_device_type_ops = {
1640         .ldto_device_alloc = mdc_device_alloc,
1641         .ldto_device_free = osc_device_free,
1642         .ldto_device_init = osc_device_init,
1643         .ldto_device_fini = osc_device_fini
1644 };
1645
1646 struct lu_device_type mdc_device_type = {
1647         .ldt_tags = LU_DEVICE_CL,
1648         .ldt_name = LUSTRE_MDC_NAME,
1649         .ldt_ops = &mdc_device_type_ops,
1650         .ldt_ctx_tags = LCT_CL_THREAD
1651 };
1652
1653 /** @} osc */