lustre/mdt/mdt_lvb.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, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012, 2017, Intel Corporation.
  25  * Use is subject to license terms.
  26  *
  27  * lustre/mdt/mdt_lvb.c
  28  *
  29  * Author: Jinshan Xiong <jinshan.xiong@intel.com>
  30  */
  31
  32 #define DEBUG_SUBSYSTEM S_MDS
  33 #include <lustre_swab.h>
  34 #include "mdt_internal.h"
  35
  36 /* Called with res->lr_lvb_sem held */
  37 static int mdt_lvbo_init(struct ldlm_resource *res)
  38 {
  39         if (IS_LQUOTA_RES(res)) {
  40                 struct mdt_device       *mdt;
  41
  42                 mdt = ldlm_res_to_ns(res)->ns_lvbp;
  43                 if (mdt->mdt_qmt_dev == NULL)
  44                         return 0;
  45
  46                 /* call lvbo init function of quota master */
  47                 return qmt_hdls.qmth_lvbo_init(mdt->mdt_qmt_dev, res);
  48         }
  49         return 0;
  50 }
  51
  52 int mdt_dom_lvb_alloc(struct ldlm_resource *res)
  53 {
  54         struct ost_lvb *lvb;
  55
  56         mutex_lock(&res->lr_lvb_mutex);
  57         if (res->lr_lvb_data == NULL) {
  58                 OBD_ALLOC_PTR(lvb);
  59                 if (lvb == NULL) {
  60                         mutex_unlock(&res->lr_lvb_mutex);
  61                         return -ENOMEM;
  62                 }
  63
  64                 res->lr_lvb_data = lvb;
  65                 res->lr_lvb_len = sizeof(*lvb);
  66
  67                 /* Store error in LVB to inidicate it has no data yet.
  68                  */
  69                 OST_LVB_SET_ERR(lvb->lvb_blocks, -ENODATA);
  70         }
  71         mutex_unlock(&res->lr_lvb_mutex);
  72         return 0;
  73 }
  74
  75 int mdt_dom_lvb_is_valid(struct ldlm_resource *res)
  76 {
  77         struct ost_lvb *res_lvb = res->lr_lvb_data;
  78
  79         return !(res_lvb == NULL || OST_LVB_IS_ERR(res_lvb->lvb_blocks));
  80 }
  81
  82 int mdt_dom_disk_lvbo_update(const struct lu_env *env, struct mdt_object *mo,
  83                              struct ldlm_resource *res, bool increase_only)
  84 {
  85         struct mdt_thread_info *info = mdt_th_info(env);
  86         const struct lu_fid *fid = mdt_object_fid(mo);
  87         struct ost_lvb *lvb;
  88         struct md_attr *ma;
  89         int rc = 0;
  90
  91         ENTRY;
  92
  93         lvb = res->lr_lvb_data;
  94         LASSERT(lvb);
  95
  96         if (!mdt_object_exists(mo) || mdt_object_remote(mo))
  97                 RETURN(-ENOENT);
  98
  99         ma = &info->mti_attr;
 100         ma->ma_valid = 0;
 101         ma->ma_need = MA_INODE;
 102         rc = mo_attr_get(env, mdt_object_child(mo), ma);
 103         if (rc)
 104                 RETURN(rc);
 105
 106         lock_res(res);
 107         if (ma->ma_attr.la_size > lvb->lvb_size || !increase_only) {
 108                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb size from disk: "
 109                        "%llu -> %llu\n", PFID(fid),
 110                        lvb->lvb_size, ma->ma_attr.la_size);
 111                 lvb->lvb_size = ma->ma_attr.la_size;
 112         }
 113
 114         if (ma->ma_attr.la_mtime > lvb->lvb_mtime || !increase_only) {
 115                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb mtime from disk: "
 116                        "%llu -> %llu\n", PFID(fid),
 117                        lvb->lvb_mtime, ma->ma_attr.la_mtime);
 118                 lvb->lvb_mtime = ma->ma_attr.la_mtime;
 119         }
 120         if (ma->ma_attr.la_atime > lvb->lvb_atime || !increase_only) {
 121                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb atime from disk: "
 122                        "%llu -> %llu\n", PFID(fid),
 123                        lvb->lvb_atime, ma->ma_attr.la_atime);
 124                 lvb->lvb_atime = ma->ma_attr.la_atime;
 125         }
 126         if (ma->ma_attr.la_ctime > lvb->lvb_ctime || !increase_only) {
 127                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb ctime from disk: "
 128                        "%llu -> %llu\n", PFID(fid),
 129                        lvb->lvb_ctime, ma->ma_attr.la_ctime);
 130                 lvb->lvb_ctime = ma->ma_attr.la_ctime;
 131         }
 132         if (ma->ma_attr.la_blocks > lvb->lvb_blocks || !increase_only) {
 133                 CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb blocks from disk: "
 134                        "%llu -> %llu\n", PFID(fid), lvb->lvb_blocks,
 135                        (unsigned long long)ma->ma_attr.la_blocks);
 136                 lvb->lvb_blocks = ma->ma_attr.la_blocks;
 137         }
 138         unlock_res(res);
 139
 140         RETURN(rc);
 141 }
 142
 143 int mdt_dom_lvbo_update(struct ldlm_resource *res, struct ldlm_lock *lock,
 144                         struct ptlrpc_request *req, bool increase_only)
 145 {
 146         struct obd_export *exp = lock ? lock->l_export : NULL;
 147         struct mdt_device *mdt;
 148         struct mdt_object *mo;
 149         struct mdt_thread_info *info;
 150         struct ost_lvb *lvb;
 151         struct lu_env env;
 152         struct lu_fid *fid;
 153         int rc = 0;
 154
 155         ENTRY;
 156
 157         /* Before going further let's check that OBD and export are healthy.
 158          */
 159         if (exp != NULL &&
 160             (exp->exp_disconnected || exp->exp_failed ||
 161              exp->exp_obd->obd_stopping)) {
 162                 CDEBUG(D_INFO, "Skip LVB update, export is %s, obd is %s\n",
 163                        exp->exp_failed ? "failed" : "disconnected",
 164                        exp->exp_obd->obd_stopping ? "stopping" : "OK");
 165                 RETURN(0);
 166         }
 167
 168         rc = mdt_dom_lvb_alloc(res);
 169         if (rc < 0)
 170                 RETURN(rc);
 171
 172         mdt = ldlm_res_to_ns(res)->ns_lvbp;
 173         if (mdt == NULL)
 174                 RETURN(-ENOENT);
 175
 176         rc = lu_env_init(&env, LCT_MD_THREAD);
 177         if (rc)
 178                 RETURN(rc);
 179
 180         info = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
 181         if (info == NULL)
 182                 GOTO(out_env, rc = -ENOMEM);
 183
 184         memset(info, 0, sizeof *info);
 185         info->mti_env = &env;
 186         info->mti_exp = req ? req->rq_export : NULL;
 187         info->mti_mdt = mdt;
 188
 189         fid = &info->mti_tmp_fid2;
 190         fid_extract_from_res_name(fid, &res->lr_name);
 191
 192         lvb = res->lr_lvb_data;
 193         LASSERT(lvb);
 194
 195         /* Update the LVB from the network message */
 196         if (req != NULL) {
 197                 struct ost_lvb *rpc_lvb;
 198
 199                 rpc_lvb = req_capsule_server_swab_get(&req->rq_pill,
 200                                                       &RMF_DLM_LVB,
 201                                                       lustre_swab_ost_lvb);
 202                 if (rpc_lvb == NULL)
 203                         goto disk_update;
 204
 205                 lock_res(res);
 206                 if (rpc_lvb->lvb_size > lvb->lvb_size || !increase_only) {
 207                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb size: "
 208                                "%llu -> %llu\n", PFID(fid),
 209                                lvb->lvb_size, rpc_lvb->lvb_size);
 210                         lvb->lvb_size = rpc_lvb->lvb_size;
 211                 }
 212                 if (rpc_lvb->lvb_mtime > lvb->lvb_mtime || !increase_only) {
 213                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb mtime: "
 214                                "%llu -> %llu\n", PFID(fid),
 215                                lvb->lvb_mtime, rpc_lvb->lvb_mtime);
 216                         lvb->lvb_mtime = rpc_lvb->lvb_mtime;
 217                 }
 218                 if (rpc_lvb->lvb_atime > lvb->lvb_atime || !increase_only) {
 219                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb atime: "
 220                                "%llu -> %llu\n", PFID(fid),
 221                                lvb->lvb_atime, rpc_lvb->lvb_atime);
 222                         lvb->lvb_atime = rpc_lvb->lvb_atime;
 223                 }
 224                 if (rpc_lvb->lvb_ctime > lvb->lvb_ctime || !increase_only) {
 225                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb ctime: "
 226                                "%llu -> %llu\n", PFID(fid),
 227                                lvb->lvb_ctime, rpc_lvb->lvb_ctime);
 228                         lvb->lvb_ctime = rpc_lvb->lvb_ctime;
 229                 }
 230                 if (rpc_lvb->lvb_blocks > lvb->lvb_blocks || !increase_only) {
 231                         CDEBUG(D_DLMTRACE, "res: "DFID" updating lvb blocks: "
 232                                "%llu -> %llu\n", PFID(fid),
 233                                lvb->lvb_blocks, rpc_lvb->lvb_blocks);
 234                         lvb->lvb_blocks = rpc_lvb->lvb_blocks;
 235                 }
 236                 unlock_res(res);
 237         }
 238
 239 disk_update:
 240         /* Update the LVB from the disk inode */
 241         mo = mdt_object_find(&env, mdt, fid);
 242         if (IS_ERR(mo))
 243                 GOTO(out_env, rc = PTR_ERR(mo));
 244
 245         rc = mdt_dom_disk_lvbo_update(&env, mo, res, !!increase_only);
 246         mdt_object_put(&env, mo);
 247 out_env:
 248         lu_env_fini(&env);
 249         return rc;
 250 }
 251
 252 static int mdt_lvbo_update(struct ldlm_resource *res, struct ldlm_lock *lock,
 253                            struct ptlrpc_request *req, int increase_only)
 254 {
 255         ENTRY;
 256
 257         if (IS_LQUOTA_RES(res)) {
 258                 struct mdt_device *mdt;
 259
 260                 mdt = ldlm_res_to_ns(res)->ns_lvbp;
 261                 if (mdt->mdt_qmt_dev == NULL)
 262                         return 0;
 263
 264                 /* call lvbo update function of quota master */
 265                 return qmt_hdls.qmth_lvbo_update(mdt->mdt_qmt_dev, res, req,
 266                                                  increase_only);
 267         }
 268
 269         /* Data-on-MDT lvbo update.
 270          * Like a ldlm_lock_init() the lock can be skipped and that means
 271          * it is DOM resource because lvbo_update() without lock is called
 272          * by MDT for DOM objects only.
 273          */
 274         if (lock == NULL || ldlm_has_dom(lock))
 275                 return mdt_dom_lvbo_update(res, lock, req, !!increase_only);
 276         return 0;
 277 }
 278
 279
 280 static int mdt_lvbo_size(struct ldlm_lock *lock)
 281 {
 282         struct mdt_device *mdt;
 283
 284         /* resource on server side never changes. */
 285         mdt = ldlm_res_to_ns(lock->l_resource)->ns_lvbp;
 286         LASSERT(mdt != NULL);
 287
 288         if (IS_LQUOTA_RES(lock->l_resource)) {
 289                 if (mdt->mdt_qmt_dev == NULL)
 290                         return 0;
 291
 292                 /* call lvbo size function of quota master */
 293                 return qmt_hdls.qmth_lvbo_size(mdt->mdt_qmt_dev, lock);
 294         }
 295
 296         if (ldlm_has_dom(lock))
 297                 return sizeof(struct ost_lvb);
 298
 299         if (ldlm_has_layout(lock))
 300                 return mdt->mdt_max_mdsize;
 301
 302         return 0;
 303 }
 304
 305 static int mdt_lvbo_fill(struct ldlm_lock *lock, void *lvb, int lvblen)
 306 {
 307         struct lu_env env;
 308         struct mdt_thread_info *info;
 309         struct mdt_device *mdt;
 310         struct lu_fid *fid;
 311         struct mdt_object *obj = NULL;
 312         struct md_object *child = NULL;
 313         int rc;
 314         ENTRY;
 315
 316         mdt = ldlm_lock_to_ns(lock)->ns_lvbp;
 317         if (IS_LQUOTA_RES(lock->l_resource)) {
 318                 if (mdt->mdt_qmt_dev == NULL)
 319                         RETURN(0);
 320
 321                 /* call lvbo fill function of quota master */
 322                 rc = qmt_hdls.qmth_lvbo_fill(mdt->mdt_qmt_dev, lock, lvb,
 323                                              lvblen);
 324                 RETURN(rc);
 325         }
 326
 327         /* LVB for DoM lock is needed only for glimpse,
 328          * don't fill DoM data if there is layout lock */
 329         if (ldlm_has_dom(lock)) {
 330                 struct ldlm_resource *res = lock->l_resource;
 331                 int lvb_len = sizeof(struct ost_lvb);
 332
 333                 if (!mdt_dom_lvb_is_valid(res))
 334                         mdt_dom_lvbo_update(lock->l_resource, lock, NULL, 0);
 335
 336                 if (lvb_len > lvblen)
 337                         lvb_len = lvblen;
 338
 339                 lock_res(res);
 340                 memcpy(lvb, res->lr_lvb_data, lvb_len);
 341                 unlock_res(res);
 342
 343                 RETURN(lvb_len);
 344         }
 345
 346         /* Only fill layout if layout lock is granted */
 347         if (!ldlm_has_layout(lock) || lock->l_granted_mode != lock->l_req_mode)
 348                 RETURN(0);
 349
 350         /* XXX create an env to talk to mdt stack. We should get this env from
 351          * ptlrpc_thread->t_env. */
 352         rc = lu_env_init(&env, LCT_MD_THREAD);
 353         /* Likely ENOMEM */
 354         if (rc)
 355                 RETURN(rc);
 356
 357         info = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
 358         /* Likely ENOMEM */
 359         if (info == NULL)
 360                 GOTO(out, rc = -ENOMEM);
 361
 362         memset(info, 0, sizeof *info);
 363         info->mti_env = &env;
 364         info->mti_exp = lock->l_export;
 365         info->mti_mdt = mdt;
 366
 367         /* XXX get fid by resource id. why don't include fid in ldlm_resource */
 368         fid = &info->mti_tmp_fid2;
 369         fid_extract_from_res_name(fid, &lock->l_resource->lr_name);
 370
 371         obj = mdt_object_find(&env, info->mti_mdt, fid);
 372         if (IS_ERR(obj))
 373                 GOTO(out, rc = PTR_ERR(obj));
 374
 375         if (!mdt_object_exists(obj) || mdt_object_remote(obj))
 376                 GOTO(out, rc = -ENOENT);
 377
 378         child = mdt_object_child(obj);
 379
 380         /* get the length of lsm */
 381         rc = mo_xattr_get(&env, child, &LU_BUF_NULL, XATTR_NAME_LOV);
 382         if (rc < 0)
 383                 GOTO(out, rc);
 384         if (rc > 0) {
 385                 struct lu_buf *lmm = NULL;
 386                 if (lvblen < rc) {
 387                         int level;
 388
 389                         /* The layout EA may be larger than mdt_max_mdsize
 390                          * and in that case mdt_max_mdsize is just updated
 391                          * but if EA size is less than mdt_max_mdsize then
 392                          * it is an error in lvblen value provided. */
 393                         if (rc > info->mti_mdt->mdt_max_mdsize) {
 394                                 info->mti_mdt->mdt_max_mdsize = rc;
 395                                 level = D_INFO;
 396                         } else {
 397                                 level = D_ERROR;
 398                         }
 399                         CDEBUG_LIMIT(level, "%s: small buffer size %d for EA "
 400                                      "%d (max_mdsize %d): rc = %d\n",
 401                                      mdt_obd_name(mdt), lvblen, rc,
 402                                      info->mti_mdt->mdt_max_mdsize, -ERANGE);
 403                         GOTO(out, rc = -ERANGE);
 404                 }
 405                 lmm = &info->mti_buf;
 406                 lmm->lb_buf = lvb;
 407                 lmm->lb_len = rc;
 408                 rc = mo_xattr_get(&env, child, lmm, XATTR_NAME_LOV);
 409                 if (rc < 0)
 410                         GOTO(out, rc);
 411         }
 412
 413 out:
 414         if (obj != NULL && !IS_ERR(obj))
 415                 mdt_object_put(&env, obj);
 416         lu_env_fini(&env);
 417         RETURN(rc < 0 ? 0 : rc);
 418 }
 419
 420 static int mdt_lvbo_free(struct ldlm_resource *res)
 421 {
 422         if (IS_LQUOTA_RES(res)) {
 423                 struct mdt_device       *mdt;
 424
 425                 mdt = ldlm_res_to_ns(res)->ns_lvbp;
 426                 if (mdt->mdt_qmt_dev == NULL)
 427                         return 0;
 428
 429                 /* call lvbo free function of quota master */
 430                 return qmt_hdls.qmth_lvbo_free(mdt->mdt_qmt_dev, res);
 431         }
 432
 433         /* Data-on-MDT lvbo free */
 434         if (res->lr_lvb_data != NULL)
 435                 OBD_FREE(res->lr_lvb_data, res->lr_lvb_len);
 436         return 0;
 437 }
 438
 439 struct ldlm_valblock_ops mdt_lvbo = {
 440         .lvbo_init      = mdt_lvbo_init,
 441         .lvbo_update    = mdt_lvbo_update,
 442         .lvbo_size      = mdt_lvbo_size,
 443         .lvbo_fill      = mdt_lvbo_fill,
 444         .lvbo_free      = mdt_lvbo_free
 445 };