lustre/mdc/mdc_request.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) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, 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
  33 #define DEBUG_SUBSYSTEM S_MDC
  34
  35 #include <linux/init.h>
  36 #include <linux/kthread.h>
  37 #include <linux/module.h>
  38 #include <linux/pagemap.h>
  39 #include <linux/user_namespace.h>
  40 #include <linux/utsname.h>
  41 #ifdef HAVE_UIDGID_HEADER
  42 # include <linux/uidgid.h>
  43 #endif
  44
  45 #include <lustre_errno.h>
  46
  47 #include <cl_object.h>
  48 #include <llog_swab.h>
  49 #include <lprocfs_status.h>
  50 #include <lustre_acl.h>
  51 #include <lustre_fid.h>
  52 #include <uapi/linux/lustre/lustre_ioctl.h>
  53 #include <lustre_kernelcomm.h>
  54 #include <lustre_lmv.h>
  55 #include <lustre_log.h>
  56 #include <lustre_swab.h>
  57 #include <obd_class.h>
  58 #include <lustre_osc.h>
  59
  60 #include "mdc_internal.h"
  61
  62 #define REQUEST_MINOR 244
  63
  64 static int mdc_cleanup(struct obd_device *obd);
  65
  66 static inline int mdc_queue_wait(struct ptlrpc_request *req)
  67 {
  68         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
  69         int rc;
  70
  71         /* obd_get_request_slot() ensures that this client has no more
  72          * than cl_max_rpcs_in_flight RPCs simultaneously inf light
  73          * against an MDT. */
  74         rc = obd_get_request_slot(cli);
  75         if (rc != 0)
  76                 return rc;
  77
  78         rc = ptlrpc_queue_wait(req);
  79         obd_put_request_slot(cli);
  80
  81         return rc;
  82 }
  83
  84 /*
  85  * Send MDS_GET_ROOT RPC to fetch root FID.
  86  *
  87  * If \a fileset is not NULL it should contain a subdirectory off
  88  * the ROOT/ directory to be mounted on the client. Return the FID
  89  * of the subdirectory to the client to mount onto its mountpoint.
  90  *
  91  * \param[in]   imp     MDC import
  92  * \param[in]   fileset fileset name, which could be NULL
  93  * \param[out]  rootfid root FID of this mountpoint
  94  * \param[out]  pc      root capa will be unpacked and saved in this pointer
  95  *
  96  * \retval      0 on success, negative errno on failure
  97  */
  98 static int mdc_get_root(struct obd_export *exp, const char *fileset,
  99                          struct lu_fid *rootfid)
 100 {
 101         struct ptlrpc_request   *req;
 102         struct mdt_body         *body;
 103         int                      rc;
 104
 105         ENTRY;
 106
 107         if (fileset && !(exp_connect_flags(exp) & OBD_CONNECT_SUBTREE))
 108                 RETURN(-ENOTSUPP);
 109
 110         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 111                                 &RQF_MDS_GET_ROOT);
 112         if (req == NULL)
 113                 RETURN(-ENOMEM);
 114
 115         if (fileset != NULL)
 116                 req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 117                                      strlen(fileset) + 1);
 118         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_ROOT);
 119         if (rc) {
 120                 ptlrpc_request_free(req);
 121                 RETURN(rc);
 122         }
 123         mdc_pack_body(req, NULL, 0, 0, -1, 0);
 124         if (fileset != NULL) {
 125                 char *name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 126
 127                 memcpy(name, fileset, strlen(fileset));
 128         }
 129         lustre_msg_add_flags(req->rq_reqmsg, LUSTRE_IMP_FULL);
 130         req->rq_send_state = LUSTRE_IMP_FULL;
 131
 132         ptlrpc_request_set_replen(req);
 133
 134         rc = ptlrpc_queue_wait(req);
 135         if (rc)
 136                 GOTO(out, rc);
 137
 138         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 139         if (body == NULL)
 140                 GOTO(out, rc = -EPROTO);
 141
 142         *rootfid = body->mbo_fid1;
 143         CDEBUG(D_NET, "root fid="DFID", last_committed=%llu\n",
 144                PFID(rootfid), lustre_msg_get_last_committed(req->rq_repmsg));
 145         EXIT;
 146 out:
 147         ptlrpc_req_finished(req);
 148
 149         return rc;
 150 }
 151
 152 /*
 153  * This function now is known to always saying that it will receive 4 buffers
 154  * from server. Even for cases when acl_size and md_size is zero, RPC header
 155  * will contain 4 fields and RPC itself will contain zero size fields. This is
 156  * because mdt_getattr*() _always_ returns 4 fields, but if acl is not needed
 157  * and thus zero, it shrinks it, making zero size. The same story about
 158  * md_size. And this is course of problem when client waits for smaller number
 159  * of fields. This issue will be fixed later when client gets aware of RPC
 160  * layouts.  --umka
 161  */
 162 static int mdc_getattr_common(struct obd_export *exp,
 163                               struct ptlrpc_request *req)
 164 {
 165         struct req_capsule *pill = &req->rq_pill;
 166         struct mdt_body    *body;
 167         void               *eadata;
 168         int                 rc;
 169         ENTRY;
 170
 171         /* Request message already built. */
 172         rc = ptlrpc_queue_wait(req);
 173         if (rc != 0)
 174                 RETURN(rc);
 175
 176         /* sanity check for the reply */
 177         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 178         if (body == NULL)
 179                 RETURN(-EPROTO);
 180
 181         CDEBUG(D_NET, "mode: %o\n", body->mbo_mode);
 182
 183         mdc_update_max_ea_from_body(exp, body);
 184         if (body->mbo_eadatasize != 0) {
 185                 eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 186                                                       body->mbo_eadatasize);
 187                 if (eadata == NULL)
 188                         RETURN(-EPROTO);
 189         }
 190
 191         RETURN(0);
 192 }
 193
 194 static void mdc_reset_acl_req(struct ptlrpc_request *req)
 195 {
 196         spin_lock(&req->rq_early_free_lock);
 197         sptlrpc_cli_free_repbuf(req);
 198         req->rq_repbuf = NULL;
 199         req->rq_repbuf_len = 0;
 200         req->rq_repdata = NULL;
 201         req->rq_reqdata_len = 0;
 202         spin_unlock(&req->rq_early_free_lock);
 203 }
 204
 205 static int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data,
 206                        struct ptlrpc_request **request)
 207 {
 208         struct ptlrpc_request *req;
 209         struct obd_import *imp = class_exp2cliimp(exp);
 210         __u32 acl_bufsize = LUSTRE_POSIX_ACL_MAX_SIZE_OLD;
 211         int rc;
 212         ENTRY;
 213
 214         /* Single MDS without an LMV case */
 215         if (op_data->op_flags & MF_GET_MDT_IDX) {
 216                 op_data->op_mds = 0;
 217                 RETURN(0);
 218         }
 219
 220         *request = NULL;
 221         req = ptlrpc_request_alloc(imp, &RQF_MDS_GETATTR);
 222         if (req == NULL)
 223                 RETURN(-ENOMEM);
 224
 225         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
 226         if (rc) {
 227                 ptlrpc_request_free(req);
 228                 RETURN(rc);
 229         }
 230
 231 again:
 232         mdc_pack_body(req, &op_data->op_fid1, op_data->op_valid,
 233                       op_data->op_mode, -1, 0);
 234         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
 235         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 236                              op_data->op_mode);
 237         ptlrpc_request_set_replen(req);
 238
 239         rc = mdc_getattr_common(exp, req);
 240         if (rc) {
 241                 if (rc == -ERANGE) {
 242                         acl_bufsize = MIN(imp->imp_connect_data.ocd_max_easize,
 243                                           XATTR_SIZE_MAX);
 244                         mdc_reset_acl_req(req);
 245                         goto again;
 246                 }
 247
 248                 ptlrpc_req_finished(req);
 249         } else {
 250                 *request = req;
 251         }
 252
 253         RETURN(rc);
 254 }
 255
 256 static int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data,
 257                             struct ptlrpc_request **request)
 258 {
 259         struct ptlrpc_request *req;
 260         struct obd_import *imp = class_exp2cliimp(exp);
 261         __u32 acl_bufsize = LUSTRE_POSIX_ACL_MAX_SIZE_OLD;
 262         int rc;
 263         ENTRY;
 264
 265         *request = NULL;
 266         req = ptlrpc_request_alloc(imp, &RQF_MDS_GETATTR_NAME);
 267         if (req == NULL)
 268                 RETURN(-ENOMEM);
 269
 270         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 271                              op_data->op_namelen + 1);
 272
 273         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR_NAME);
 274         if (rc) {
 275                 ptlrpc_request_free(req);
 276                 RETURN(rc);
 277         }
 278
 279         if (op_data->op_name) {
 280                 char *name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 281                 LASSERT(strnlen(op_data->op_name, op_data->op_namelen) ==
 282                                 op_data->op_namelen);
 283                 memcpy(name, op_data->op_name, op_data->op_namelen);
 284         }
 285
 286 again:
 287         mdc_pack_body(req, &op_data->op_fid1, op_data->op_valid,
 288                       op_data->op_mode, op_data->op_suppgids[0], 0);
 289         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 290                              op_data->op_mode);
 291         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
 292         ptlrpc_request_set_replen(req);
 293
 294         rc = mdc_getattr_common(exp, req);
 295         if (rc) {
 296                 if (rc == -ERANGE) {
 297                         acl_bufsize = MIN(imp->imp_connect_data.ocd_max_easize,
 298                                           XATTR_SIZE_MAX);
 299                         mdc_reset_acl_req(req);
 300                         goto again;
 301                 }
 302
 303                 ptlrpc_req_finished(req);
 304         } else {
 305                 *request = req;
 306         }
 307
 308         RETURN(rc);
 309 }
 310
 311 static int mdc_xattr_common(struct obd_export *exp,const struct req_format *fmt,
 312                             const struct lu_fid *fid, int opcode, u64 valid,
 313                             const char *xattr_name, const char *input,
 314                             int input_size, int output_size, int flags,
 315                             __u32 suppgid, struct ptlrpc_request **request)
 316 {
 317         struct ptlrpc_request *req;
 318         int   xattr_namelen = 0;
 319         char *tmp;
 320         int   rc;
 321         ENTRY;
 322
 323         *request = NULL;
 324         req = ptlrpc_request_alloc(class_exp2cliimp(exp), fmt);
 325         if (req == NULL)
 326                 RETURN(-ENOMEM);
 327
 328         if (xattr_name) {
 329                 xattr_namelen = strlen(xattr_name) + 1;
 330                 req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 331                                      xattr_namelen);
 332         }
 333         if (input_size)
 334                 LASSERT(input);
 335         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 336                              input_size);
 337
 338         /* get SELinux policy info if any */
 339         rc = sptlrpc_get_sepol(req);
 340         if (rc < 0) {
 341                 ptlrpc_request_free(req);
 342                 RETURN(rc);
 343         }
 344         req_capsule_set_size(&req->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
 345                              strlen(req->rq_sepol) ?
 346                              strlen(req->rq_sepol) + 1 : 0);
 347
 348         /* Flush local XATTR locks to get rid of a possible cancel RPC */
 349         if (opcode == MDS_REINT && fid_is_sane(fid) &&
 350             exp->exp_connect_data.ocd_ibits_known & MDS_INODELOCK_XATTR) {
 351                 struct list_head cancels = LIST_HEAD_INIT(cancels);
 352                 int count;
 353
 354                 /* Without that packing would fail */
 355                 if (input_size == 0)
 356                         req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
 357                                              RCL_CLIENT, 0);
 358
 359                 count = mdc_resource_get_unused(exp, fid,
 360                                                 &cancels, LCK_EX,
 361                                                 MDS_INODELOCK_XATTR);
 362
 363                 rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 364                 if (rc) {
 365                         ptlrpc_request_free(req);
 366                         RETURN(rc);
 367                 }
 368         } else {
 369                 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, opcode);
 370                 if (rc) {
 371                         ptlrpc_request_free(req);
 372                         RETURN(rc);
 373                 }
 374         }
 375
 376         if (opcode == MDS_REINT) {
 377                 struct mdt_rec_setxattr *rec;
 378
 379                 CLASSERT(sizeof(struct mdt_rec_setxattr) ==
 380                          sizeof(struct mdt_rec_reint));
 381                 rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
 382                 rec->sx_opcode = REINT_SETXATTR;
 383                 rec->sx_fsuid  = from_kuid(&init_user_ns, current_fsuid());
 384                 rec->sx_fsgid  = from_kgid(&init_user_ns, current_fsgid());
 385                 rec->sx_cap    = cfs_curproc_cap_pack();
 386                 rec->sx_suppgid1 = suppgid;
 387                 rec->sx_suppgid2 = -1;
 388                 rec->sx_fid    = *fid;
 389                 rec->sx_valid  = valid | OBD_MD_FLCTIME;
 390                 rec->sx_time   = ktime_get_real_seconds();
 391                 rec->sx_size   = output_size;
 392                 rec->sx_flags  = flags;
 393         } else {
 394                 mdc_pack_body(req, fid, valid, output_size, suppgid, flags);
 395         }
 396
 397         if (xattr_name) {
 398                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 399                 memcpy(tmp, xattr_name, xattr_namelen);
 400         }
 401         if (input_size) {
 402                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_EADATA);
 403                 memcpy(tmp, input, input_size);
 404         }
 405
 406         mdc_file_sepol_pack(req);
 407
 408         if (req_capsule_has_field(&req->rq_pill, &RMF_EADATA, RCL_SERVER))
 409                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
 410                                      RCL_SERVER, output_size);
 411         ptlrpc_request_set_replen(req);
 412
 413         /* make rpc */
 414         if (opcode == MDS_REINT)
 415                 mdc_get_mod_rpc_slot(req, NULL);
 416
 417         rc = ptlrpc_queue_wait(req);
 418
 419         if (opcode == MDS_REINT)
 420                 mdc_put_mod_rpc_slot(req, NULL);
 421
 422         if (rc)
 423                 ptlrpc_req_finished(req);
 424         else
 425                 *request = req;
 426         RETURN(rc);
 427 }
 428
 429 static int mdc_setxattr(struct obd_export *exp, const struct lu_fid *fid,
 430                         u64 obd_md_valid, const char *name,
 431                         const void *value, size_t value_size,
 432                         unsigned int xattr_flags, u32 suppgid,
 433                         struct ptlrpc_request **req)
 434 {
 435         LASSERT(obd_md_valid == OBD_MD_FLXATTR ||
 436                 obd_md_valid == OBD_MD_FLXATTRRM);
 437
 438         return mdc_xattr_common(exp, &RQF_MDS_REINT_SETXATTR,
 439                                 fid, MDS_REINT, obd_md_valid, name,
 440                                 value, value_size, 0, xattr_flags, suppgid,
 441                                 req);
 442 }
 443
 444 static int mdc_getxattr(struct obd_export *exp, const struct lu_fid *fid,
 445                         u64 obd_md_valid, const char *name, size_t buf_size,
 446                         struct ptlrpc_request **req)
 447 {
 448         struct mdt_body *body;
 449         int rc;
 450
 451         LASSERT(obd_md_valid == OBD_MD_FLXATTR ||
 452                 obd_md_valid == OBD_MD_FLXATTRLS);
 453
 454         /* The below message is checked in sanity-selinux.sh test_20d */
 455         CDEBUG(D_INFO, "%s: get xattr '%s' for "DFID"\n",
 456                exp->exp_obd->obd_name, name, PFID(fid));
 457         rc = mdc_xattr_common(exp, &RQF_MDS_GETXATTR, fid, MDS_GETXATTR,
 458                               obd_md_valid, name, NULL, 0, buf_size, 0, -1,
 459                               req);
 460         if (rc < 0)
 461                 GOTO(out, rc);
 462
 463         body = req_capsule_server_get(&(*req)->rq_pill, &RMF_MDT_BODY);
 464         if (body == NULL)
 465                 GOTO(out, rc = -EPROTO);
 466
 467         /* only detect the xattr size */
 468         if (buf_size == 0) {
 469                 /* LU-11109: Older MDTs do not distinguish
 470                  * between nonexistent xattrs and zero length
 471                  * values in this case. Newer MDTs will return
 472                  * -ENODATA or set OBD_MD_FLXATTR. */
 473                 GOTO(out, rc = body->mbo_eadatasize);
 474         }
 475
 476         if (body->mbo_eadatasize == 0) {
 477                 /* LU-11109: Newer MDTs set OBD_MD_FLXATTR on
 478                  * success so that we can distinguish between
 479                  * zero length value and nonexistent xattr.
 480                  *
 481                  * If OBD_MD_FLXATTR is not set then we keep
 482                  * the old behavior and return -ENODATA for
 483                  * getxattr() when mbo_eadatasize is 0. But
 484                  * -ENODATA only makes sense for getxattr()
 485                  * and not for listxattr(). */
 486                 if (body->mbo_valid & OBD_MD_FLXATTR)
 487                         GOTO(out, rc = 0);
 488                 else if (obd_md_valid == OBD_MD_FLXATTR)
 489                         GOTO(out, rc = -ENODATA);
 490                 else
 491                         GOTO(out, rc = 0);
 492         }
 493
 494         GOTO(out, rc = body->mbo_eadatasize);
 495 out:
 496         if (rc < 0) {
 497                 ptlrpc_req_finished(*req);
 498                 *req = NULL;
 499         }
 500
 501         return rc;
 502 }
 503
 504 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
 505 static int mdc_unpack_acl(struct ptlrpc_request *req, struct lustre_md *md)
 506 {
 507         struct req_capsule     *pill = &req->rq_pill;
 508         struct mdt_body        *body = md->body;
 509         struct posix_acl       *acl;
 510         void                   *buf;
 511         int                     rc;
 512         ENTRY;
 513
 514         if (!body->mbo_aclsize)
 515                 RETURN(0);
 516
 517         buf = req_capsule_server_sized_get(pill, &RMF_ACL, body->mbo_aclsize);
 518
 519         if (!buf)
 520                 RETURN(-EPROTO);
 521
 522         acl = posix_acl_from_xattr(&init_user_ns, buf, body->mbo_aclsize);
 523         if (acl == NULL)
 524                 RETURN(0);
 525         if (IS_ERR(acl)) {
 526                 rc = PTR_ERR(acl);
 527                 CERROR("convert xattr to acl: %d\n", rc);
 528                 RETURN(rc);
 529         }
 530
 531         rc = posix_acl_valid(&init_user_ns, acl);
 532         if (rc) {
 533                 CERROR("validate acl: %d\n", rc);
 534                 posix_acl_release(acl);
 535                 RETURN(rc);
 536         }
 537
 538         md->posix_acl = acl;
 539         RETURN(0);
 540 }
 541 #else
 542 #define mdc_unpack_acl(req, md) 0
 543 #endif
 544
 545 int mdc_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,
 546                       struct obd_export *dt_exp, struct obd_export *md_exp,
 547                       struct lustre_md *md)
 548 {
 549         struct req_capsule *pill = &req->rq_pill;
 550         int rc;
 551         ENTRY;
 552
 553         LASSERT(md);
 554         memset(md, 0, sizeof(*md));
 555
 556         md->body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 557         LASSERT(md->body != NULL);
 558
 559         if (md->body->mbo_valid & OBD_MD_FLEASIZE) {
 560                 if (!S_ISREG(md->body->mbo_mode)) {
 561                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, should be a "
 562                                "regular file, but is not\n");
 563                         GOTO(out, rc = -EPROTO);
 564                 }
 565
 566                 if (md->body->mbo_eadatasize == 0) {
 567                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, "
 568                                "but eadatasize 0\n");
 569                         GOTO(out, rc = -EPROTO);
 570                 }
 571
 572                 md->layout.lb_len = md->body->mbo_eadatasize;
 573                 md->layout.lb_buf = req_capsule_server_sized_get(pill,
 574                                                         &RMF_MDT_MD,
 575                                                         md->layout.lb_len);
 576                 if (md->layout.lb_buf == NULL)
 577                         GOTO(out, rc = -EPROTO);
 578         } else if (md->body->mbo_valid & OBD_MD_FLDIREA) {
 579                 const union lmv_mds_md *lmv;
 580                 size_t lmv_size;
 581
 582                 if (!S_ISDIR(md->body->mbo_mode)) {
 583                         CDEBUG(D_INFO, "OBD_MD_FLDIREA set, should be a "
 584                                "directory, but is not\n");
 585                         GOTO(out, rc = -EPROTO);
 586                 }
 587
 588                 if (md->body->mbo_valid & OBD_MD_MEA) {
 589                         lmv_size = md->body->mbo_eadatasize;
 590                         if (lmv_size == 0) {
 591                                 CDEBUG(D_INFO, "OBD_MD_FLDIREA is set, "
 592                                        "but eadatasize 0\n");
 593                                 RETURN(-EPROTO);
 594                         }
 595
 596                         lmv = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 597                                                            lmv_size);
 598                         if (lmv == NULL)
 599                                 GOTO(out, rc = -EPROTO);
 600
 601                         rc = md_unpackmd(md_exp, &md->lmv, lmv, lmv_size);
 602                         if (rc < 0)
 603                                 GOTO(out, rc);
 604
 605                         if (rc < (int)sizeof(*md->lmv)) {
 606                                 struct lmv_foreign_md *lfm = md->lfm;
 607
 608                                 /* short (< sizeof(struct lmv_stripe_md))
 609                                  * foreign LMV case
 610                                  */
 611                                 if (lfm->lfm_magic != LMV_MAGIC_FOREIGN) {
 612                                         CDEBUG(D_INFO,
 613                                                "lmv size too small: %d < %d\n",
 614                                                rc, (int)sizeof(*md->lmv));
 615                                         GOTO(out, rc = -EPROTO);
 616                                 }
 617                         }
 618                 }
 619
 620                 /* since 2.12.58 intent_getattr fetches default LMV */
 621                 if (md->body->mbo_valid & OBD_MD_DEFAULT_MEA) {
 622                         lmv_size = sizeof(struct lmv_user_md);
 623                         lmv = req_capsule_server_sized_get(pill,
 624                                                            &RMF_DEFAULT_MDT_MD,
 625                                                            lmv_size);
 626                         if (!lmv)
 627                                 GOTO(out, rc = -EPROTO);
 628
 629                         rc = md_unpackmd(md_exp, &md->default_lmv, lmv,
 630                                          lmv_size);
 631                         if (rc < 0)
 632                                 GOTO(out, rc);
 633
 634                         if (rc < (int)sizeof(*md->default_lmv)) {
 635                                 CDEBUG(D_INFO,
 636                                        "default lmv size too small: %d < %d\n",
 637                                         rc, (int)sizeof(*md->default_lmv));
 638                                 GOTO(out, rc = -EPROTO);
 639                         }
 640                 }
 641         }
 642         rc = 0;
 643
 644         if (md->body->mbo_valid & OBD_MD_FLACL) {
 645                 /* for ACL, it's possible that FLACL is set but aclsize is zero.
 646                  * only when aclsize != 0 there's an actual segment for ACL
 647                  * in reply buffer.
 648                  */
 649                 if (md->body->mbo_aclsize) {
 650                         rc = mdc_unpack_acl(req, md);
 651                         if (rc)
 652                                 GOTO(out, rc);
 653 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
 654                 } else {
 655                         md->posix_acl = NULL;
 656 #endif
 657                 }
 658         }
 659
 660         EXIT;
 661 out:
 662         if (rc) {
 663 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
 664                 posix_acl_release(md->posix_acl);
 665 #endif
 666         }
 667         return rc;
 668 }
 669
 670 int mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
 671 {
 672         ENTRY;
 673         RETURN(0);
 674 }
 675
 676 void mdc_replay_open(struct ptlrpc_request *req)
 677 {
 678         struct md_open_data *mod = req->rq_cb_data;
 679         struct ptlrpc_request *close_req;
 680         struct obd_client_handle *och;
 681         struct lustre_handle old_open_handle = { };
 682         struct mdt_body *body;
 683         ENTRY;
 684
 685         if (mod == NULL) {
 686                 DEBUG_REQ(D_ERROR, req,
 687                           "cannot properly replay without open data");
 688                 EXIT;
 689                 return;
 690         }
 691
 692         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 693         LASSERT(body != NULL);
 694
 695         spin_lock(&req->rq_lock);
 696         och = mod->mod_och;
 697         if (och && och->och_open_handle.cookie)
 698                 req->rq_early_free_repbuf = 1;
 699         else
 700                 req->rq_early_free_repbuf = 0;
 701         spin_unlock(&req->rq_lock);
 702
 703         if (req->rq_early_free_repbuf) {
 704                 struct lustre_handle *file_open_handle;
 705
 706                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 707
 708                 file_open_handle = &och->och_open_handle;
 709                 CDEBUG(D_HA, "updating handle from %#llx to %#llx\n",
 710                        file_open_handle->cookie, body->mbo_open_handle.cookie);
 711                 old_open_handle = *file_open_handle;
 712                 *file_open_handle = body->mbo_open_handle;
 713         }
 714
 715         close_req = mod->mod_close_req;
 716         if (close_req) {
 717                 __u32 opc = lustre_msg_get_opc(close_req->rq_reqmsg);
 718                 struct mdt_ioepoch *epoch;
 719
 720                 LASSERT(opc == MDS_CLOSE);
 721                 epoch = req_capsule_client_get(&close_req->rq_pill,
 722                                                &RMF_MDT_EPOCH);
 723                 LASSERT(epoch);
 724
 725                 if (req->rq_early_free_repbuf)
 726                         LASSERT(old_open_handle.cookie ==
 727                                 epoch->mio_open_handle.cookie);
 728
 729                 DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
 730                 epoch->mio_open_handle = body->mbo_open_handle;
 731         }
 732         EXIT;
 733 }
 734
 735 void mdc_commit_open(struct ptlrpc_request *req)
 736 {
 737         struct md_open_data *mod = req->rq_cb_data;
 738         if (mod == NULL)
 739                 return;
 740
 741         /**
 742          * No need to touch md_open_data::mod_och, it holds a reference on
 743          * \var mod and will zero references to each other, \var mod will be
 744          * freed after that when md_open_data::mod_och will put the reference.
 745          */
 746
 747         /**
 748          * Do not let open request to disappear as it still may be needed
 749          * for close rpc to happen (it may happen on evict only, otherwise
 750          * ptlrpc_request::rq_replay does not let mdc_commit_open() to be
 751          * called), just mark this rpc as committed to distinguish these 2
 752          * cases, see mdc_close() for details. The open request reference will
 753          * be put along with freeing \var mod.
 754          */
 755         ptlrpc_request_addref(req);
 756         spin_lock(&req->rq_lock);
 757         req->rq_committed = 1;
 758         spin_unlock(&req->rq_lock);
 759         req->rq_cb_data = NULL;
 760         obd_mod_put(mod);
 761 }
 762
 763 int mdc_set_open_replay_data(struct obd_export *exp,
 764                              struct obd_client_handle *och,
 765                              struct lookup_intent *it)
 766 {
 767         struct md_open_data     *mod;
 768         struct mdt_rec_create   *rec;
 769         struct mdt_body         *body;
 770         struct ptlrpc_request   *open_req = it->it_request;
 771         struct obd_import       *imp = open_req->rq_import;
 772         ENTRY;
 773
 774         if (!open_req->rq_replay)
 775                 RETURN(0);
 776
 777         rec = req_capsule_client_get(&open_req->rq_pill, &RMF_REC_REINT);
 778         body = req_capsule_server_get(&open_req->rq_pill, &RMF_MDT_BODY);
 779         LASSERT(rec != NULL);
 780         /* Incoming message in my byte order (it's been swabbed). */
 781         /* Outgoing messages always in my byte order. */
 782         LASSERT(body != NULL);
 783
 784         /* Only if the import is replayable, we set replay_open data */
 785         if (och && imp->imp_replayable) {
 786                 mod = obd_mod_alloc();
 787                 if (mod == NULL) {
 788                         DEBUG_REQ(D_ERROR, open_req,
 789                                   "cannot allocate md_open_data");
 790                         RETURN(0);
 791                 }
 792
 793                 /**
 794                  * Take a reference on \var mod, to be freed on mdc_close().
 795                  * It protects \var mod from being freed on eviction (commit
 796                  * callback is called despite rq_replay flag).
 797                  * Another reference for \var och.
 798                  */
 799                 obd_mod_get(mod);
 800                 obd_mod_get(mod);
 801
 802                 spin_lock(&open_req->rq_lock);
 803                 och->och_mod = mod;
 804                 mod->mod_och = och;
 805                 mod->mod_is_create = it_disposition(it, DISP_OPEN_CREATE) ||
 806                                      it_disposition(it, DISP_OPEN_STRIPE);
 807                 mod->mod_open_req = open_req;
 808                 open_req->rq_cb_data = mod;
 809                 open_req->rq_commit_cb = mdc_commit_open;
 810                 open_req->rq_early_free_repbuf = 1;
 811                 spin_unlock(&open_req->rq_lock);
 812         }
 813
 814         rec->cr_fid2 = body->mbo_fid1;
 815         rec->cr_open_handle_old = body->mbo_open_handle;
 816         open_req->rq_replay_cb = mdc_replay_open;
 817         if (!fid_is_sane(&body->mbo_fid1)) {
 818                 DEBUG_REQ(D_ERROR, open_req,
 819                           "saving replay request with insane FID " DFID,
 820                           PFID(&body->mbo_fid1));
 821                 LBUG();
 822         }
 823
 824         DEBUG_REQ(D_RPCTRACE, open_req, "Set up open replay data");
 825         RETURN(0);
 826 }
 827
 828 static void mdc_free_open(struct md_open_data *mod)
 829 {
 830         int committed = 0;
 831
 832         if (mod->mod_is_create == 0 &&
 833             imp_connect_disp_stripe(mod->mod_open_req->rq_import))
 834                 committed = 1;
 835
 836         /**
 837          * No reason to asssert here if the open request has
 838          * rq_replay == 1. It means that mdc_close failed, and
 839          * close request wasn`t sent. It is not fatal to client.
 840          * The worst thing is eviction if the client gets open lock
 841          **/
 842
 843         DEBUG_REQ(D_RPCTRACE, mod->mod_open_req,
 844                   "free open request, rq_replay=%d",
 845                   mod->mod_open_req->rq_replay);
 846
 847         ptlrpc_request_committed(mod->mod_open_req, committed);
 848         if (mod->mod_close_req)
 849                 ptlrpc_request_committed(mod->mod_close_req, committed);
 850 }
 851
 852 int mdc_clear_open_replay_data(struct obd_export *exp,
 853                                struct obd_client_handle *och)
 854 {
 855         struct md_open_data *mod = och->och_mod;
 856         ENTRY;
 857
 858         /**
 859          * It is possible to not have \var mod in a case of eviction between
 860          * lookup and ll_file_open().
 861          **/
 862         if (mod == NULL)
 863                 RETURN(0);
 864
 865         LASSERT(mod != LP_POISON);
 866         LASSERT(mod->mod_open_req != NULL);
 867
 868         spin_lock(&mod->mod_open_req->rq_lock);
 869         if (mod->mod_och)
 870                 mod->mod_och->och_open_handle.cookie = 0;
 871         mod->mod_open_req->rq_early_free_repbuf = 0;
 872         spin_unlock(&mod->mod_open_req->rq_lock);
 873         mdc_free_open(mod);
 874
 875         mod->mod_och = NULL;
 876         och->och_mod = NULL;
 877         obd_mod_put(mod);
 878
 879         RETURN(0);
 880 }
 881
 882 static int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
 883                      struct md_open_data *mod, struct ptlrpc_request **request)
 884 {
 885         struct obd_device     *obd = class_exp2obd(exp);
 886         struct ptlrpc_request *req;
 887         struct req_format     *req_fmt;
 888         size_t                 u32_count = 0;
 889         int                    rc;
 890         int                    saved_rc = 0;
 891         ENTRY;
 892
 893         CDEBUG(D_INODE, "%s: "DFID" file closed with intent: %x\n",
 894                exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
 895                op_data->op_bias);
 896
 897         if (op_data->op_bias & MDS_CLOSE_INTENT) {
 898                 req_fmt = &RQF_MDS_CLOSE_INTENT;
 899                 if (op_data->op_bias & MDS_HSM_RELEASE) {
 900                         /* allocate a FID for volatile file */
 901                         rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2,
 902                                            op_data);
 903                         if (rc < 0) {
 904                                 CERROR("%s: "DFID" allocating FID: rc = %d\n",
 905                                        obd->obd_name, PFID(&op_data->op_fid1),
 906                                        rc);
 907                                 /* save the errcode and proceed to close */
 908                                 saved_rc = rc;
 909                         }
 910                 }
 911                 if (op_data->op_bias & MDS_CLOSE_RESYNC_DONE) {
 912                         size_t count = op_data->op_data_size / sizeof(__u32);
 913
 914                         if (count > INLINE_RESYNC_ARRAY_SIZE)
 915                                 u32_count = count;
 916                 }
 917         } else {
 918                 req_fmt = &RQF_MDS_CLOSE;
 919         }
 920
 921         *request = NULL;
 922         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_CLOSE))
 923                 req = NULL;
 924         else
 925                 req = ptlrpc_request_alloc(class_exp2cliimp(exp), req_fmt);
 926
 927         /* Ensure that this close's handle is fixed up during replay. */
 928         if (likely(mod != NULL)) {
 929                 LASSERTF(mod->mod_open_req != NULL &&
 930                          mod->mod_open_req->rq_type != LI_POISON,
 931                          "POISONED open %p!\n", mod->mod_open_req);
 932
 933                 mod->mod_close_req = req;
 934
 935                 DEBUG_REQ(D_RPCTRACE, mod->mod_open_req, "matched open");
 936                 /* We no longer want to preserve this open for replay even
 937                  * though the open was committed. b=3632, b=3633 */
 938                 spin_lock(&mod->mod_open_req->rq_lock);
 939                 mod->mod_open_req->rq_replay = 0;
 940                 spin_unlock(&mod->mod_open_req->rq_lock);
 941         } else {
 942                 CDEBUG(D_HA, "couldn't find open req; expecting close error\n");
 943         }
 944         if (req == NULL) {
 945                 /**
 946                  * TODO: repeat close after errors
 947                  */
 948                 CWARN("%s: close of FID "DFID" failed, file reference will be "
 949                       "dropped when this client unmounts or is evicted\n",
 950                       obd->obd_name, PFID(&op_data->op_fid1));
 951                 GOTO(out, rc = -ENOMEM);
 952         }
 953
 954         if (u32_count > 0)
 955                 req_capsule_set_size(&req->rq_pill, &RMF_U32, RCL_CLIENT,
 956                                      u32_count * sizeof(__u32));
 957
 958         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_CLOSE);
 959         if (rc) {
 960                 ptlrpc_request_free(req);
 961                 req = NULL;
 962                 GOTO(out, rc);
 963         }
 964
 965         /* To avoid a livelock (bug 7034), we need to send CLOSE RPCs to a
 966          * portal whose threads are not taking any DLM locks and are therefore
 967          * always progressing */
 968         req->rq_request_portal = MDS_READPAGE_PORTAL;
 969         ptlrpc_at_set_req_timeout(req);
 970
 971         if (!(exp_connect_flags2(exp) & OBD_CONNECT2_LSOM))
 972                 op_data->op_xvalid &= ~(OP_XVALID_LAZYSIZE |
 973                                         OP_XVALID_LAZYBLOCKS);
 974
 975         mdc_close_pack(req, op_data);
 976
 977         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 978                              obd->u.cli.cl_default_mds_easize);
 979
 980         ptlrpc_request_set_replen(req);
 981
 982         mdc_get_mod_rpc_slot(req, NULL);
 983         rc = ptlrpc_queue_wait(req);
 984         mdc_put_mod_rpc_slot(req, NULL);
 985
 986         if (req->rq_repmsg == NULL) {
 987                 CDEBUG(D_RPCTRACE, "request %p failed to send: rc = %d\n", req,
 988                        req->rq_status);
 989                 if (rc == 0)
 990                         rc = req->rq_status ?: -EIO;
 991         } else if (rc == 0 || rc == -EAGAIN) {
 992                 struct mdt_body *body;
 993
 994                 rc = lustre_msg_get_status(req->rq_repmsg);
 995                 if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 996                         DEBUG_REQ(D_ERROR, req,
 997                                   "type = PTL_RPC_MSG_ERR: rc = %d", rc);
 998                         if (rc > 0)
 999                                 rc = -rc;
1000                 }
1001                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
1002                 if (body == NULL)
1003                         rc = -EPROTO;
1004         } else if (rc == -ESTALE) {
1005                 /**
1006                  * it can be allowed error after 3633 if open was committed and
1007                  * server failed before close was sent. Let's check if mod
1008                  * exists and return no error in that case
1009                  */
1010                 if (mod) {
1011                         DEBUG_REQ(D_HA, req, "Reset ESTALE = %d", rc);
1012                         LASSERT(mod->mod_open_req != NULL);
1013                         if (mod->mod_open_req->rq_committed)
1014                                 rc = 0;
1015                 }
1016         }
1017
1018 out:
1019         if (mod) {
1020                 if (rc != 0)
1021                         mod->mod_close_req = NULL;
1022                 /* Since now, mod is accessed through open_req only,
1023                  * thus close req does not keep a reference on mod anymore. */
1024                 obd_mod_put(mod);
1025         }
1026         *request = req;
1027
1028         RETURN(rc < 0 ? rc : saved_rc);
1029 }
1030
1031 static int mdc_getpage(struct obd_export *exp, const struct lu_fid *fid,
1032                        u64 offset, struct page **pages, int npages,
1033                        struct ptlrpc_request **request)
1034 {
1035         struct ptlrpc_request   *req;
1036         struct ptlrpc_bulk_desc *desc;
1037         int                      i;
1038         wait_queue_head_t        waitq;
1039         int                      resends = 0;
1040         struct l_wait_info       lwi;
1041         int                      rc;
1042         ENTRY;
1043
1044         *request = NULL;
1045         init_waitqueue_head(&waitq);
1046
1047 restart_bulk:
1048         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_READPAGE);
1049         if (req == NULL)
1050                 RETURN(-ENOMEM);
1051
1052         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_READPAGE);
1053         if (rc) {
1054                 ptlrpc_request_free(req);
1055                 RETURN(rc);
1056         }
1057
1058         req->rq_request_portal = MDS_READPAGE_PORTAL;
1059         ptlrpc_at_set_req_timeout(req);
1060
1061         desc = ptlrpc_prep_bulk_imp(req, npages, 1,
1062                                     PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
1063                                     MDS_BULK_PORTAL,
1064                                     &ptlrpc_bulk_kiov_pin_ops);
1065         if (desc == NULL) {
1066                 ptlrpc_req_finished(req);
1067                 RETURN(-ENOMEM);
1068         }
1069
1070         /* NB req now owns desc and will free it when it gets freed */
1071         for (i = 0; i < npages; i++)
1072                 desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
1073                                                  PAGE_SIZE);
1074
1075         mdc_readdir_pack(req, offset, PAGE_SIZE * npages, fid);
1076
1077         ptlrpc_request_set_replen(req);
1078         rc = ptlrpc_queue_wait(req);
1079         if (rc) {
1080                 ptlrpc_req_finished(req);
1081                 if (rc != -ETIMEDOUT)
1082                         RETURN(rc);
1083
1084                 resends++;
1085                 if (!client_should_resend(resends, &exp->exp_obd->u.cli)) {
1086                         CERROR("%s: too many resend retries: rc = %d\n",
1087                                exp->exp_obd->obd_name, -EIO);
1088                         RETURN(-EIO);
1089                 }
1090                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL,
1091                                        NULL);
1092                 l_wait_event(waitq, 0, &lwi);
1093
1094                 goto restart_bulk;
1095         }
1096
1097         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1098                                           req->rq_bulk->bd_nob_transferred);
1099         if (rc < 0) {
1100                 ptlrpc_req_finished(req);
1101                 RETURN(rc);
1102         }
1103
1104         if (req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK) {
1105                 CERROR("%s: unexpected bytes transferred: %d (%ld expected)\n",
1106                        exp->exp_obd->obd_name, req->rq_bulk->bd_nob_transferred,
1107                        PAGE_SIZE * npages);
1108                 ptlrpc_req_finished(req);
1109                 RETURN(-EPROTO);
1110         }
1111
1112         *request = req;
1113         RETURN(0);
1114 }
1115
1116 static void mdc_release_page(struct page *page, int remove)
1117 {
1118         if (remove) {
1119                 lock_page(page);
1120                 if (likely(page->mapping != NULL))
1121                         delete_from_page_cache(page);
1122                 unlock_page(page);
1123         }
1124         put_page(page);
1125 }
1126
1127 static struct page *mdc_page_locate(struct address_space *mapping, __u64 *hash,
1128                                     __u64 *start, __u64 *end, int hash64)
1129 {
1130         /*
1131          * Complement of hash is used as an index so that
1132          * radix_tree_gang_lookup() can be used to find a page with starting
1133          * hash _smaller_ than one we are looking for.
1134          */
1135         unsigned long offset = hash_x_index(*hash, hash64);
1136         struct page *page;
1137         unsigned long flags;
1138         int found;
1139
1140         xa_lock_irqsave(&mapping->i_pages, flags);
1141         found = radix_tree_gang_lookup(&mapping->page_tree,
1142                                        (void **)&page, offset, 1);
1143         if (found > 0 && !radix_tree_exceptional_entry(page)) {
1144                 struct lu_dirpage *dp;
1145
1146                 get_page(page);
1147                 xa_unlock_irqrestore(&mapping->i_pages, flags);
1148                 /*
1149                  * In contrast to find_lock_page() we are sure that directory
1150                  * page cannot be truncated (while DLM lock is held) and,
1151                  * hence, can avoid restart.
1152                  *
1153                  * In fact, page cannot be locked here at all, because
1154                  * mdc_read_page_remote does synchronous io.
1155                  */
1156                 wait_on_page_locked(page);
1157                 if (PageUptodate(page)) {
1158                         dp = kmap(page);
1159                         if (BITS_PER_LONG == 32 && hash64) {
1160                                 *start = le64_to_cpu(dp->ldp_hash_start) >> 32;
1161                                 *end   = le64_to_cpu(dp->ldp_hash_end) >> 32;
1162                                 *hash  = *hash >> 32;
1163                         } else {
1164                                 *start = le64_to_cpu(dp->ldp_hash_start);
1165                                 *end   = le64_to_cpu(dp->ldp_hash_end);
1166                         }
1167                         if (unlikely(*start == 1 && *hash == 0))
1168                                 *hash = *start;
1169                         else
1170                                 LASSERTF(*start <= *hash, "start = %#llx"
1171                                          ",end = %#llx,hash = %#llx\n",
1172                                          *start, *end, *hash);
1173                         CDEBUG(D_VFSTRACE, "offset %lx [%#llx %#llx],"
1174                               " hash %#llx\n", offset, *start, *end, *hash);
1175                         if (*hash > *end) {
1176                                 kunmap(page);
1177                                 mdc_release_page(page, 0);
1178                                 page = NULL;
1179                         } else if (*end != *start && *hash == *end) {
1180                                 /*
1181                                  * upon hash collision, remove this page,
1182                                  * otherwise put page reference, and
1183                                  * mdc_read_page_remote() will issue RPC to
1184                                  * fetch the page we want.
1185                                  */
1186                                 kunmap(page);
1187                                 mdc_release_page(page,
1188                                     le32_to_cpu(dp->ldp_flags) & LDF_COLLIDE);
1189                                 page = NULL;
1190                         }
1191                 } else {
1192                         put_page(page);
1193                         page = ERR_PTR(-EIO);
1194                 }
1195         } else {
1196                 xa_unlock_irqrestore(&mapping->i_pages, flags);
1197                 page = NULL;
1198         }
1199         return page;
1200 }
1201
1202 /*
1203  * Adjust a set of pages, each page containing an array of lu_dirpages,
1204  * so that each page can be used as a single logical lu_dirpage.
1205  *
1206  * A lu_dirpage is laid out as follows, where s = ldp_hash_start,
1207  * e = ldp_hash_end, f = ldp_flags, p = padding, and each "ent" is a
1208  * struct lu_dirent.  It has size up to LU_PAGE_SIZE. The ldp_hash_end
1209  * value is used as a cookie to request the next lu_dirpage in a
1210  * directory listing that spans multiple pages (two in this example):
1211  *   ________
1212  *  |        |
1213  * .|--------v-------   -----.
1214  * |s|e|f|p|ent|ent| ... |ent|
1215  * '--|--------------   -----'   Each PAGE contains a single
1216  *    '------.                   lu_dirpage.
1217  * .---------v-------   -----.
1218  * |s|e|f|p|ent| 0 | ... | 0 |
1219  * '-----------------   -----'
1220  *
1221  * However, on hosts where the native VM page size (PAGE_SIZE) is
1222  * larger than LU_PAGE_SIZE, a single host page may contain multiple
1223  * lu_dirpages. After reading the lu_dirpages from the MDS, the
1224  * ldp_hash_end of the first lu_dirpage refers to the one immediately
1225  * after it in the same PAGE (arrows simplified for brevity, but
1226  * in general e0==s1, e1==s2, etc.):
1227  *
1228  * .--------------------   -----.
1229  * |s0|e0|f0|p|ent|ent| ... |ent|
1230  * |---v----------------   -----|
1231  * |s1|e1|f1|p|ent|ent| ... |ent|
1232  * |---v----------------   -----|  Here, each PAGE contains
1233  *             ...                 multiple lu_dirpages.
1234  * |---v----------------   -----|
1235  * |s'|e'|f'|p|ent|ent| ... |ent|
1236  * '---|----------------   -----'
1237  *     v
1238  * .----------------------------.
1239  * |        next PAGE           |
1240  *
1241  * This structure is transformed into a single logical lu_dirpage as follows:
1242  *
1243  * - Replace e0 with e' so the request for the next lu_dirpage gets the page
1244  *   labeled 'next PAGE'.
1245  *
1246  * - Copy the LDF_COLLIDE flag from f' to f0 to correctly reflect whether
1247  *   a hash collision with the next page exists.
1248  *
1249  * - Adjust the lde_reclen of the ending entry of each lu_dirpage to span
1250  *   to the first entry of the next lu_dirpage.
1251  */
1252 #if PAGE_SIZE > LU_PAGE_SIZE
1253 static void mdc_adjust_dirpages(struct page **pages, int cfs_pgs, int lu_pgs)
1254 {
1255         int i;
1256
1257         for (i = 0; i < cfs_pgs; i++) {
1258                 struct lu_dirpage *dp = kmap(pages[i]);
1259                 struct lu_dirpage *first = dp;
1260                 struct lu_dirent *end_dirent = NULL;
1261                 struct lu_dirent *ent;
1262                 __u64 hash_end = dp->ldp_hash_end;
1263                 __u32 flags = dp->ldp_flags;
1264
1265                 while (--lu_pgs > 0) {
1266                         ent = lu_dirent_start(dp);
1267                         for (end_dirent = ent; ent != NULL;
1268                              end_dirent = ent, ent = lu_dirent_next(ent));
1269
1270                         /* Advance dp to next lu_dirpage. */
1271                         dp = (struct lu_dirpage *)((char *)dp + LU_PAGE_SIZE);
1272
1273                         /* Check if we've reached the end of the PAGE. */
1274                         if (!((unsigned long)dp & ~PAGE_MASK))
1275                                 break;
1276
1277                         /* Save the hash and flags of this lu_dirpage. */
1278                         hash_end = dp->ldp_hash_end;
1279                         flags = dp->ldp_flags;
1280
1281                         /* Check if lu_dirpage contains no entries. */
1282                         if (end_dirent == NULL)
1283                                 break;
1284
1285                         /* Enlarge the end entry lde_reclen from 0 to
1286                          * first entry of next lu_dirpage. */
1287                         LASSERT(le16_to_cpu(end_dirent->lde_reclen) == 0);
1288                         end_dirent->lde_reclen =
1289                                 cpu_to_le16((char *)(dp->ldp_entries) -
1290                                             (char *)end_dirent);
1291                 }
1292
1293                 first->ldp_hash_end = hash_end;
1294                 first->ldp_flags &= ~cpu_to_le32(LDF_COLLIDE);
1295                 first->ldp_flags |= flags & cpu_to_le32(LDF_COLLIDE);
1296
1297                 kunmap(pages[i]);
1298         }
1299         LASSERTF(lu_pgs == 0, "left = %d\n", lu_pgs);
1300 }
1301 #else
1302 #define mdc_adjust_dirpages(pages, cfs_pgs, lu_pgs) do {} while (0)
1303 #endif  /* PAGE_SIZE > LU_PAGE_SIZE */
1304
1305 /* parameters for readdir page */
1306 struct readpage_param {
1307         struct md_op_data       *rp_mod;
1308         __u64                   rp_off;
1309         int                     rp_hash64;
1310         struct obd_export       *rp_exp;
1311         struct md_callback      *rp_cb;
1312 };
1313
1314 /**
1315  * Read pages from server.
1316  *
1317  * Page in MDS_READPAGE RPC is packed in LU_PAGE_SIZE, and each page contains
1318  * a header lu_dirpage which describes the start/end hash, and whether this
1319  * page is empty (contains no dir entry) or hash collide with next page.
1320  * After client receives reply, several pages will be integrated into dir page
1321  * in PAGE_SIZE (if PAGE_SIZE greater than LU_PAGE_SIZE), and the
1322  * lu_dirpage for this integrated page will be adjusted.
1323  **/
1324 static int mdc_read_page_remote(void *data, struct page *page0)
1325 {
1326         struct readpage_param *rp = data;
1327         struct page **page_pool;
1328         struct page *page;
1329         struct lu_dirpage *dp;
1330         struct md_op_data *op_data = rp->rp_mod;
1331         struct ptlrpc_request *req;
1332         int max_pages;
1333         struct inode *inode;
1334         struct lu_fid *fid;
1335         int rd_pgs = 0; /* number of pages actually read */
1336         int npages;
1337         int i;
1338         int rc;
1339         ENTRY;
1340
1341         max_pages = rp->rp_exp->exp_obd->u.cli.cl_max_pages_per_rpc;
1342         inode = op_data->op_data;
1343         fid = &op_data->op_fid1;
1344         LASSERT(inode != NULL);
1345
1346         OBD_ALLOC(page_pool, sizeof(page_pool[0]) * max_pages);
1347         if (page_pool != NULL) {
1348                 page_pool[0] = page0;
1349         } else {
1350                 page_pool = &page0;
1351                 max_pages = 1;
1352         }
1353
1354         for (npages = 1; npages < max_pages; npages++) {
1355                 page = __page_cache_alloc(mapping_gfp_mask(inode->i_mapping)
1356                                           | __GFP_COLD);
1357                 if (page == NULL)
1358                         break;
1359                 page_pool[npages] = page;
1360         }
1361
1362         rc = mdc_getpage(rp->rp_exp, fid, rp->rp_off, page_pool, npages, &req);
1363         if (rc < 0) {
1364                 /* page0 is special, which was added into page cache early */
1365                 delete_from_page_cache(page0);
1366         } else {
1367                 int lu_pgs;
1368
1369                 rd_pgs = (req->rq_bulk->bd_nob_transferred + PAGE_SIZE - 1) >>
1370                         PAGE_SHIFT;
1371                 lu_pgs = req->rq_bulk->bd_nob_transferred >> LU_PAGE_SHIFT;
1372                 LASSERT(!(req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK));
1373
1374                 CDEBUG(D_INODE, "read %d(%d) pages\n", rd_pgs, lu_pgs);
1375
1376                 mdc_adjust_dirpages(page_pool, rd_pgs, lu_pgs);
1377
1378                 SetPageUptodate(page0);
1379         }
1380         unlock_page(page0);
1381
1382         ptlrpc_req_finished(req);
1383         CDEBUG(D_CACHE, "read %d/%d pages\n", rd_pgs, npages);
1384         for (i = 1; i < npages; i++) {
1385                 unsigned long   offset;
1386                 __u64           hash;
1387                 int ret;
1388
1389                 page = page_pool[i];
1390
1391                 if (rc < 0 || i >= rd_pgs) {
1392                         put_page(page);
1393                         continue;
1394                 }
1395
1396                 SetPageUptodate(page);
1397
1398                 dp = kmap(page);
1399                 hash = le64_to_cpu(dp->ldp_hash_start);
1400                 kunmap(page);
1401
1402                 offset = hash_x_index(hash, rp->rp_hash64);
1403
1404                 prefetchw(&page->flags);
1405                 ret = add_to_page_cache_lru(page, inode->i_mapping, offset,
1406                                             GFP_KERNEL);
1407                 if (ret == 0)
1408                         unlock_page(page);
1409                 else
1410                         CDEBUG(D_VFSTRACE, "page %lu add to page cache failed:"
1411                                " rc = %d\n", offset, ret);
1412                 put_page(page);
1413         }
1414
1415         if (page_pool != &page0)
1416                 OBD_FREE(page_pool, sizeof(page_pool[0]) * max_pages);
1417
1418         RETURN(rc);
1419 }
1420
1421 /**
1422  * Read dir page from cache first, if it can not find it, read it from
1423  * server and add into the cache.
1424  *
1425  * \param[in] exp       MDC export
1426  * \param[in] op_data   client MD stack parameters, transfering parameters
1427  *                      between different layers on client MD stack.
1428  * \param[in] cb_op     callback required for ldlm lock enqueue during
1429  *                      read page
1430  * \param[in] hash_offset the hash offset of the page to be read
1431  * \param[in] ppage     the page to be read
1432  *
1433  * retval               = 0 get the page successfully
1434  *                      errno(<0) get the page failed
1435  */
1436 static int mdc_read_page(struct obd_export *exp, struct md_op_data *op_data,
1437                          struct md_callback *cb_op, __u64 hash_offset,
1438                          struct page **ppage)
1439 {
1440         struct lookup_intent    it = { .it_op = IT_READDIR };
1441         struct page             *page;
1442         struct inode            *dir = op_data->op_data;
1443         struct address_space    *mapping;
1444         struct lu_dirpage       *dp;
1445         __u64                   start = 0;
1446         __u64                   end = 0;
1447         struct lustre_handle    lockh;
1448         struct ptlrpc_request   *enq_req = NULL;
1449         struct readpage_param   rp_param;
1450         int rc;
1451
1452         ENTRY;
1453
1454         *ppage = NULL;
1455
1456         LASSERT(dir != NULL);
1457         mapping = dir->i_mapping;
1458
1459         rc = mdc_intent_lock(exp, op_data, &it, &enq_req,
1460                              cb_op->md_blocking_ast, 0);
1461         if (enq_req != NULL)
1462                 ptlrpc_req_finished(enq_req);
1463
1464         if (rc < 0) {
1465                 CERROR("%s: "DFID" lock enqueue fails: rc = %d\n",
1466                        exp->exp_obd->obd_name, PFID(&op_data->op_fid1), rc);
1467                 RETURN(rc);
1468         }
1469
1470         rc = 0;
1471         lockh.cookie = it.it_lock_handle;
1472         mdc_set_lock_data(exp, &lockh, dir, NULL);
1473
1474         rp_param.rp_off = hash_offset;
1475         rp_param.rp_hash64 = op_data->op_cli_flags & CLI_HASH64;
1476         page = mdc_page_locate(mapping, &rp_param.rp_off, &start, &end,
1477                                rp_param.rp_hash64);
1478         if (IS_ERR(page)) {
1479                 CERROR("%s: dir page locate: "DFID" at %llu: rc %ld\n",
1480                        exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
1481                        rp_param.rp_off, PTR_ERR(page));
1482                 GOTO(out_unlock, rc = PTR_ERR(page));
1483         } else if (page != NULL) {
1484                 /*
1485                  * XXX nikita: not entirely correct handling of a corner case:
1486                  * suppose hash chain of entries with hash value HASH crosses
1487                  * border between pages P0 and P1. First both P0 and P1 are
1488                  * cached, seekdir() is called for some entry from the P0 part
1489                  * of the chain. Later P0 goes out of cache. telldir(HASH)
1490                  * happens and finds P1, as it starts with matching hash
1491                  * value. Remaining entries from P0 part of the chain are
1492                  * skipped. (Is that really a bug?)
1493                  *
1494                  * Possible solutions: 0. don't cache P1 is such case, handle
1495                  * it as an "overflow" page. 1. invalidate all pages at
1496                  * once. 2. use HASH|1 as an index for P1.
1497                  */
1498                 GOTO(hash_collision, page);
1499         }
1500
1501         rp_param.rp_exp = exp;
1502         rp_param.rp_mod = op_data;
1503         page = read_cache_page(mapping,
1504                                hash_x_index(rp_param.rp_off,
1505                                             rp_param.rp_hash64),
1506                                mdc_read_page_remote, &rp_param);
1507         if (IS_ERR(page)) {
1508                 CDEBUG(D_INFO, "%s: read cache page: "DFID" at %llu: %ld\n",
1509                        exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
1510                        rp_param.rp_off, PTR_ERR(page));
1511                 GOTO(out_unlock, rc = PTR_ERR(page));
1512         }
1513
1514         wait_on_page_locked(page);
1515         (void)kmap(page);
1516         if (!PageUptodate(page)) {
1517                 CERROR("%s: page not updated: "DFID" at %llu: rc %d\n",
1518                        exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
1519                        rp_param.rp_off, -5);
1520                 goto fail;
1521         }
1522         if (!PageChecked(page))
1523                 SetPageChecked(page);
1524         if (PageError(page)) {
1525                 CERROR("%s: page error: "DFID" at %llu: rc %d\n",
1526                        exp->exp_obd->obd_name, PFID(&op_data->op_fid1),
1527                        rp_param.rp_off, -5);
1528                 goto fail;
1529         }
1530
1531 hash_collision:
1532         dp = page_address(page);
1533         if (BITS_PER_LONG == 32 && rp_param.rp_hash64) {
1534                 start = le64_to_cpu(dp->ldp_hash_start) >> 32;
1535                 end   = le64_to_cpu(dp->ldp_hash_end) >> 32;
1536                 rp_param.rp_off = hash_offset >> 32;
1537         } else {
1538                 start = le64_to_cpu(dp->ldp_hash_start);
1539                 end   = le64_to_cpu(dp->ldp_hash_end);
1540                 rp_param.rp_off = hash_offset;
1541         }
1542         if (end == start) {
1543                 LASSERT(start == rp_param.rp_off);
1544                 CWARN("Page-wide hash collision: %#lx\n", (unsigned long)end);
1545 #if BITS_PER_LONG == 32
1546                 CWARN("Real page-wide hash collision at [%llu %llu] with "
1547                       "hash %llu\n", le64_to_cpu(dp->ldp_hash_start),
1548                       le64_to_cpu(dp->ldp_hash_end), hash_offset);
1549 #endif
1550
1551                 /*
1552                  * Fetch whole overflow chain...
1553                  *
1554                  * XXX not yet.
1555                  */
1556                 goto fail;
1557         }
1558         *ppage = page;
1559 out_unlock:
1560         ldlm_lock_decref(&lockh, it.it_lock_mode);
1561         return rc;
1562 fail:
1563         kunmap(page);
1564         mdc_release_page(page, 1);
1565         rc = -EIO;
1566         goto out_unlock;
1567 }
1568
1569 static int mdc_statfs_interpret(const struct lu_env *env,
1570                                 struct ptlrpc_request *req, void *args, int rc)
1571 {
1572         struct obd_info *oinfo = args;
1573         struct obd_statfs *osfs;
1574
1575         if (!rc) {
1576                 osfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1577                 if (!osfs)
1578                         return -EPROTO;
1579
1580                 oinfo->oi_osfs = osfs;
1581
1582                 CDEBUG(D_CACHE, "blocks=%llu free=%llu avail=%llu "
1583                        "objects=%llu free=%llu state=%x\n",
1584                         osfs->os_blocks, osfs->os_bfree, osfs->os_bavail,
1585                         osfs->os_files, osfs->os_ffree, osfs->os_state);
1586         }
1587
1588         oinfo->oi_cb_up(oinfo, rc);
1589
1590         return rc;
1591 }
1592
1593 static int mdc_statfs_async(struct obd_export *exp,
1594                             struct obd_info *oinfo, time64_t max_age,
1595                             struct ptlrpc_request_set *unused)
1596 {
1597         struct ptlrpc_request *req;
1598         struct obd_info *aa;
1599
1600         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_STATFS,
1601                                         LUSTRE_MDS_VERSION, MDS_STATFS);
1602         if (req == NULL)
1603                 return -ENOMEM;
1604
1605         ptlrpc_request_set_replen(req);
1606         req->rq_interpret_reply = mdc_statfs_interpret;
1607
1608         aa = ptlrpc_req_async_args(aa, req);
1609         *aa = *oinfo;
1610
1611         ptlrpcd_add_req(req);
1612
1613         return 0;
1614 }
1615
1616 static int mdc_statfs(const struct lu_env *env,
1617                       struct obd_export *exp, struct obd_statfs *osfs,
1618                       time64_t max_age, __u32 flags)
1619 {
1620         struct obd_device *obd = class_exp2obd(exp);
1621         struct req_format *fmt;
1622         struct ptlrpc_request *req;
1623         struct obd_statfs *msfs;
1624         struct obd_import *imp = NULL;
1625         int rc;
1626         ENTRY;
1627
1628         /*
1629          * Since the request might also come from lprocfs, so we need
1630          * sync this with client_disconnect_export Bug15684
1631          */
1632         down_read(&obd->u.cli.cl_sem);
1633         if (obd->u.cli.cl_import)
1634                 imp = class_import_get(obd->u.cli.cl_import);
1635         up_read(&obd->u.cli.cl_sem);
1636         if (!imp)
1637                 RETURN(-ENODEV);
1638
1639         fmt = &RQF_MDS_STATFS;
1640         if ((exp_connect_flags2(exp) & OBD_CONNECT2_SUM_STATFS) &&
1641             (flags & OBD_STATFS_SUM))
1642                 fmt = &RQF_MDS_STATFS_NEW;
1643         req = ptlrpc_request_alloc_pack(imp, fmt, LUSTRE_MDS_VERSION,
1644                                         MDS_STATFS);
1645         if (req == NULL)
1646                 GOTO(output, rc = -ENOMEM);
1647
1648         if ((flags & OBD_STATFS_SUM) &&
1649             (exp_connect_flags2(exp) & OBD_CONNECT2_SUM_STATFS)) {
1650                 /* request aggregated states */
1651                 struct mdt_body *body;
1652
1653                 body = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
1654                 if (body == NULL)
1655                         GOTO(out, rc = -EPROTO);
1656                 body->mbo_valid = OBD_MD_FLAGSTATFS;
1657         }
1658
1659         ptlrpc_request_set_replen(req);
1660
1661         if (flags & OBD_STATFS_NODELAY) {
1662                 /* procfs requests not want stay in wait for avoid deadlock */
1663                 req->rq_no_resend = 1;
1664                 req->rq_no_delay = 1;
1665         }
1666
1667         rc = ptlrpc_queue_wait(req);
1668         if (rc) {
1669                 /* check connection error first */
1670                 if (imp->imp_connect_error)
1671                         rc = imp->imp_connect_error;
1672                 GOTO(out, rc);
1673         }
1674
1675         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1676         if (msfs == NULL)
1677                 GOTO(out, rc = -EPROTO);
1678
1679         *osfs = *msfs;
1680         EXIT;
1681 out:
1682         ptlrpc_req_finished(req);
1683 output:
1684         class_import_put(imp);
1685         return rc;
1686 }
1687
1688 static int mdc_ioc_fid2path(struct obd_export *exp, struct getinfo_fid2path *gf)
1689 {
1690         __u32 keylen, vallen;
1691         void *key;
1692         int rc;
1693
1694         if (gf->gf_pathlen > PATH_MAX)
1695                 RETURN(-ENAMETOOLONG);
1696         if (gf->gf_pathlen < 2)
1697                 RETURN(-EOVERFLOW);
1698
1699         /* Key is KEY_FID2PATH + getinfo_fid2path description */
1700         keylen = cfs_size_round(sizeof(KEY_FID2PATH) + sizeof(*gf) +
1701                                 sizeof(struct lu_fid));
1702         OBD_ALLOC(key, keylen);
1703         if (key == NULL)
1704                 RETURN(-ENOMEM);
1705         memcpy(key, KEY_FID2PATH, sizeof(KEY_FID2PATH));
1706         memcpy(key + cfs_size_round(sizeof(KEY_FID2PATH)), gf, sizeof(*gf));
1707         memcpy(key + cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*gf),
1708                gf->gf_u.gf_root_fid, sizeof(struct lu_fid));
1709         CDEBUG(D_IOCTL, "path get "DFID" from %llu #%d\n",
1710                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno);
1711
1712         if (!fid_is_sane(&gf->gf_fid))
1713                 GOTO(out, rc = -EINVAL);
1714
1715         /* Val is struct getinfo_fid2path result plus path */
1716         vallen = sizeof(*gf) + gf->gf_pathlen;
1717
1718         rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf);
1719         if (rc != 0 && rc != -EREMOTE)
1720                 GOTO(out, rc);
1721
1722         if (vallen <= sizeof(*gf))
1723                 GOTO(out, rc = -EPROTO);
1724         if (vallen > sizeof(*gf) + gf->gf_pathlen)
1725                 GOTO(out, rc = -EOVERFLOW);
1726
1727         CDEBUG(D_IOCTL, "path got "DFID" from %llu #%d: %s\n",
1728                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno,
1729                gf->gf_pathlen < 512 ? gf->gf_u.gf_path :
1730                /* only log the last 512 characters of the path */
1731                gf->gf_u.gf_path + gf->gf_pathlen - 512);
1732
1733 out:
1734         OBD_FREE(key, keylen);
1735         return rc;
1736 }
1737
1738 static int mdc_ioc_hsm_progress(struct obd_export *exp,
1739                                 struct hsm_progress_kernel *hpk)
1740 {
1741         struct obd_import               *imp = class_exp2cliimp(exp);
1742         struct hsm_progress_kernel      *req_hpk;
1743         struct ptlrpc_request           *req;
1744         int                              rc;
1745         ENTRY;
1746
1747         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_PROGRESS,
1748                                         LUSTRE_MDS_VERSION, MDS_HSM_PROGRESS);
1749         if (req == NULL)
1750                 GOTO(out, rc = -ENOMEM);
1751
1752         mdc_pack_body(req, NULL, 0, 0, -1, 0);
1753
1754         /* Copy hsm_progress struct */
1755         req_hpk = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_PROGRESS);
1756         if (req_hpk == NULL)
1757                 GOTO(out, rc = -EPROTO);
1758
1759         *req_hpk = *hpk;
1760         req_hpk->hpk_errval = lustre_errno_hton(hpk->hpk_errval);
1761
1762         ptlrpc_request_set_replen(req);
1763
1764         mdc_get_mod_rpc_slot(req, NULL);
1765         rc = ptlrpc_queue_wait(req);
1766         mdc_put_mod_rpc_slot(req, NULL);
1767
1768         GOTO(out, rc);
1769 out:
1770         ptlrpc_req_finished(req);
1771         return rc;
1772 }
1773 /**
1774  * Send hsm_ct_register to MDS
1775  *
1776  * \param[in]   imp             import
1777  * \param[in]   archive_count   if in bitmap format, it is the bitmap,
1778  *                              else it is the count of archive_ids
1779  * \param[in]   archives        if in bitmap format, it is NULL,
1780  *                              else it is archive_id lists
1781  */
1782 static int mdc_ioc_hsm_ct_register(struct obd_import *imp, __u32 archive_count,
1783                                    __u32 *archives)
1784 {
1785         struct ptlrpc_request *req;
1786         __u32 *archive_array;
1787         size_t archives_size;
1788         int rc;
1789         ENTRY;
1790
1791         req = ptlrpc_request_alloc(imp, &RQF_MDS_HSM_CT_REGISTER);
1792         if (req == NULL)
1793                 RETURN(-ENOMEM);
1794
1795         if (archives != NULL)
1796                 archives_size = sizeof(*archive_array) * archive_count;
1797         else
1798                 archives_size = sizeof(archive_count);
1799
1800         req_capsule_set_size(&req->rq_pill, &RMF_MDS_HSM_ARCHIVE,
1801                              RCL_CLIENT, archives_size);
1802
1803         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_CT_REGISTER);
1804         if (rc) {
1805                 ptlrpc_request_free(req);
1806                 RETURN(-ENOMEM);
1807         }
1808
1809         mdc_pack_body(req, NULL, 0, 0, -1, 0);
1810
1811         archive_array = req_capsule_client_get(&req->rq_pill,
1812                                                &RMF_MDS_HSM_ARCHIVE);
1813         if (archive_array == NULL)
1814                 GOTO(out, rc = -EPROTO);
1815
1816         if (archives != NULL)
1817                 memcpy(archive_array, archives, archives_size);
1818         else
1819                 *archive_array = archive_count;
1820
1821         ptlrpc_request_set_replen(req);
1822
1823         rc = mdc_queue_wait(req);
1824         GOTO(out, rc);
1825 out:
1826         ptlrpc_req_finished(req);
1827         return rc;
1828 }
1829
1830 static int mdc_ioc_hsm_current_action(struct obd_export *exp,
1831                                       struct md_op_data *op_data)
1832 {
1833         struct hsm_current_action       *hca = op_data->op_data;
1834         struct hsm_current_action       *req_hca;
1835         struct ptlrpc_request           *req;
1836         int                              rc;
1837         ENTRY;
1838
1839         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1840                                    &RQF_MDS_HSM_ACTION);
1841         if (req == NULL)
1842                 RETURN(-ENOMEM);
1843
1844         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_ACTION);
1845         if (rc) {
1846                 ptlrpc_request_free(req);
1847                 RETURN(rc);
1848         }
1849
1850         mdc_pack_body(req, &op_data->op_fid1, 0, 0,
1851                       op_data->op_suppgids[0], 0);
1852
1853         ptlrpc_request_set_replen(req);
1854
1855         rc = mdc_queue_wait(req);
1856         if (rc)
1857                 GOTO(out, rc);
1858
1859         req_hca = req_capsule_server_get(&req->rq_pill,
1860                                          &RMF_MDS_HSM_CURRENT_ACTION);
1861         if (req_hca == NULL)
1862                 GOTO(out, rc = -EPROTO);
1863
1864         *hca = *req_hca;
1865
1866         EXIT;
1867 out:
1868         ptlrpc_req_finished(req);
1869         return rc;
1870 }
1871
1872 static int mdc_ioc_hsm_ct_unregister(struct obd_import *imp)
1873 {
1874         struct ptlrpc_request   *req;
1875         int                      rc;
1876         ENTRY;
1877
1878         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_UNREGISTER,
1879                                         LUSTRE_MDS_VERSION,
1880                                         MDS_HSM_CT_UNREGISTER);
1881         if (req == NULL)
1882                 GOTO(out, rc = -ENOMEM);
1883
1884         mdc_pack_body(req, NULL, 0, 0, -1, 0);
1885
1886         ptlrpc_request_set_replen(req);
1887
1888         rc = mdc_queue_wait(req);
1889         GOTO(out, rc);
1890 out:
1891         ptlrpc_req_finished(req);
1892         return rc;
1893 }
1894
1895 static int mdc_ioc_hsm_state_get(struct obd_export *exp,
1896                                  struct md_op_data *op_data)
1897 {
1898         struct hsm_user_state   *hus = op_data->op_data;
1899         struct hsm_user_state   *req_hus;
1900         struct ptlrpc_request   *req;
1901         int                      rc;
1902         ENTRY;
1903
1904         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1905                                    &RQF_MDS_HSM_STATE_GET);
1906         if (req == NULL)
1907                 RETURN(-ENOMEM);
1908
1909         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_GET);
1910         if (rc != 0) {
1911                 ptlrpc_request_free(req);
1912                 RETURN(rc);
1913         }
1914
1915         mdc_pack_body(req, &op_data->op_fid1, 0, 0,
1916                       op_data->op_suppgids[0], 0);
1917
1918         ptlrpc_request_set_replen(req);
1919
1920         rc = mdc_queue_wait(req);
1921         if (rc)
1922                 GOTO(out, rc);
1923
1924         req_hus = req_capsule_server_get(&req->rq_pill, &RMF_HSM_USER_STATE);
1925         if (req_hus == NULL)
1926                 GOTO(out, rc = -EPROTO);
1927
1928         *hus = *req_hus;
1929
1930         EXIT;
1931 out:
1932         ptlrpc_req_finished(req);
1933         return rc;
1934 }
1935
1936 static int mdc_ioc_hsm_state_set(struct obd_export *exp,
1937                                  struct md_op_data *op_data)
1938 {
1939         struct hsm_state_set    *hss = op_data->op_data;
1940         struct hsm_state_set    *req_hss;
1941         struct ptlrpc_request   *req;
1942         int                      rc;
1943         ENTRY;
1944
1945         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1946                                    &RQF_MDS_HSM_STATE_SET);
1947         if (req == NULL)
1948                 RETURN(-ENOMEM);
1949
1950         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_SET);
1951         if (rc) {
1952                 ptlrpc_request_free(req);
1953                 RETURN(rc);
1954         }
1955
1956         mdc_pack_body(req, &op_data->op_fid1, 0, 0,
1957                       op_data->op_suppgids[0], 0);
1958
1959         /* Copy states */
1960         req_hss = req_capsule_client_get(&req->rq_pill, &RMF_HSM_STATE_SET);
1961         if (req_hss == NULL)
1962                 GOTO(out, rc = -EPROTO);
1963         *req_hss = *hss;
1964
1965         ptlrpc_request_set_replen(req);
1966
1967         mdc_get_mod_rpc_slot(req, NULL);
1968         rc = ptlrpc_queue_wait(req);
1969         mdc_put_mod_rpc_slot(req, NULL);
1970
1971         GOTO(out, rc);
1972 out:
1973         ptlrpc_req_finished(req);
1974         return rc;
1975 }
1976
1977 static int mdc_ioc_hsm_request(struct obd_export *exp,
1978                                struct hsm_user_request *hur)
1979 {
1980         struct obd_import       *imp = class_exp2cliimp(exp);
1981         struct ptlrpc_request   *req;
1982         struct hsm_request      *req_hr;
1983         struct hsm_user_item    *req_hui;
1984         char                    *req_opaque;
1985         int                      rc;
1986         ENTRY;
1987
1988         req = ptlrpc_request_alloc(imp, &RQF_MDS_HSM_REQUEST);
1989         if (req == NULL)
1990                 GOTO(out, rc = -ENOMEM);
1991
1992         req_capsule_set_size(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM, RCL_CLIENT,
1993                              hur->hur_request.hr_itemcount
1994                              * sizeof(struct hsm_user_item));
1995         req_capsule_set_size(&req->rq_pill, &RMF_GENERIC_DATA, RCL_CLIENT,
1996                              hur->hur_request.hr_data_len);
1997
1998         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_REQUEST);
1999         if (rc) {
2000                 ptlrpc_request_free(req);
2001                 RETURN(rc);
2002         }
2003
2004         mdc_pack_body(req, NULL, 0, 0, -1, 0);
2005
2006         /* Copy hsm_request struct */
2007         req_hr = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_REQUEST);
2008         if (req_hr == NULL)
2009                 GOTO(out, rc = -EPROTO);
2010         *req_hr = hur->hur_request;
2011
2012         /* Copy hsm_user_item structs */
2013         req_hui = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM);
2014         if (req_hui == NULL)
2015                 GOTO(out, rc = -EPROTO);
2016         memcpy(req_hui, hur->hur_user_item,
2017                hur->hur_request.hr_itemcount * sizeof(struct hsm_user_item));
2018
2019         /* Copy opaque field */
2020         req_opaque = req_capsule_client_get(&req->rq_pill, &RMF_GENERIC_DATA);
2021         if (req_opaque == NULL)
2022                 GOTO(out, rc = -EPROTO);
2023         memcpy(req_opaque, hur_data(hur), hur->hur_request.hr_data_len);
2024
2025         ptlrpc_request_set_replen(req);
2026
2027         mdc_get_mod_rpc_slot(req, NULL);
2028         rc = ptlrpc_queue_wait(req);
2029         mdc_put_mod_rpc_slot(req, NULL);
2030
2031         GOTO(out, rc);
2032
2033 out:
2034         ptlrpc_req_finished(req);
2035         return rc;
2036 }
2037
2038 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
2039                                 struct lustre_kernelcomm *lk);
2040
2041 static int mdc_quotactl(struct obd_device *unused, struct obd_export *exp,
2042                         struct obd_quotactl *oqctl)
2043 {
2044         struct ptlrpc_request   *req;
2045         struct obd_quotactl     *oqc;
2046         int                      rc;
2047         ENTRY;
2048
2049         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
2050                                         &RQF_MDS_QUOTACTL, LUSTRE_MDS_VERSION,
2051                                         MDS_QUOTACTL);
2052         if (req == NULL)
2053                 RETURN(-ENOMEM);
2054
2055         oqc = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
2056         *oqc = *oqctl;
2057
2058         ptlrpc_request_set_replen(req);
2059         ptlrpc_at_set_req_timeout(req);
2060
2061         rc = ptlrpc_queue_wait(req);
2062         if (rc)
2063                 CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc);
2064
2065         if (req->rq_repmsg &&
2066             (oqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL))) {
2067                 *oqctl = *oqc;
2068         } else if (!rc) {
2069                 CERROR ("Can't unpack obd_quotactl\n");
2070                 rc = -EPROTO;
2071         }
2072         ptlrpc_req_finished(req);
2073
2074         RETURN(rc);
2075 }
2076
2077 static int mdc_ioc_swap_layouts(struct obd_export *exp,
2078                                 struct md_op_data *op_data)
2079 {
2080         struct list_head cancels = LIST_HEAD_INIT(cancels);
2081         struct ptlrpc_request   *req;
2082         int                      rc, count;
2083         struct mdc_swap_layouts *msl, *payload;
2084         ENTRY;
2085
2086         msl = op_data->op_data;
2087
2088         /* When the MDT will get the MDS_SWAP_LAYOUTS RPC the
2089          * first thing it will do is to cancel the 2 layout
2090          * locks held by this client.
2091          * So the client must cancel its layout locks on the 2 fids
2092          * with the request RPC to avoid extra RPC round trips.
2093          */
2094         count = mdc_resource_get_unused(exp, &op_data->op_fid1, &cancels,
2095                                         LCK_EX, MDS_INODELOCK_LAYOUT |
2096                                         MDS_INODELOCK_XATTR);
2097         count += mdc_resource_get_unused(exp, &op_data->op_fid2, &cancels,
2098                                          LCK_EX, MDS_INODELOCK_LAYOUT |
2099                                          MDS_INODELOCK_XATTR);
2100
2101         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2102                                    &RQF_MDS_SWAP_LAYOUTS);
2103         if (req == NULL) {
2104                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
2105                 RETURN(-ENOMEM);
2106         }
2107
2108         rc = mdc_prep_elc_req(exp, req, MDS_SWAP_LAYOUTS, &cancels, count);
2109         if (rc) {
2110                 ptlrpc_request_free(req);
2111                 RETURN(rc);
2112         }
2113
2114         mdc_swap_layouts_pack(req, op_data);
2115
2116         payload = req_capsule_client_get(&req->rq_pill, &RMF_SWAP_LAYOUTS);
2117         LASSERT(payload);
2118
2119         *payload = *msl;
2120
2121         ptlrpc_request_set_replen(req);
2122
2123         rc = ptlrpc_queue_wait(req);
2124         if (rc)
2125                 GOTO(out, rc);
2126         EXIT;
2127
2128 out:
2129         ptlrpc_req_finished(req);
2130         return rc;
2131 }
2132
2133 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2134                          void *karg, void __user *uarg)
2135 {
2136         struct obd_device *obd = exp->exp_obd;
2137         struct obd_ioctl_data *data = karg;
2138         struct obd_import *imp = obd->u.cli.cl_import;
2139         int rc;
2140         ENTRY;
2141
2142         if (!try_module_get(THIS_MODULE)) {
2143                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
2144                        module_name(THIS_MODULE));
2145                 return -EINVAL;
2146         }
2147         switch (cmd) {
2148         case OBD_IOC_FID2PATH:
2149                 rc = mdc_ioc_fid2path(exp, karg);
2150                 GOTO(out, rc);
2151         case LL_IOC_HSM_CT_START:
2152                 rc = mdc_ioc_hsm_ct_start(exp, karg);
2153                 /* ignore if it was already registered on this MDS. */
2154                 if (rc == -EEXIST)
2155                         rc = 0;
2156                 GOTO(out, rc);
2157         case LL_IOC_HSM_PROGRESS:
2158                 rc = mdc_ioc_hsm_progress(exp, karg);
2159                 GOTO(out, rc);
2160         case LL_IOC_HSM_STATE_GET:
2161                 rc = mdc_ioc_hsm_state_get(exp, karg);
2162                 GOTO(out, rc);
2163         case LL_IOC_HSM_STATE_SET:
2164                 rc = mdc_ioc_hsm_state_set(exp, karg);
2165                 GOTO(out, rc);
2166         case LL_IOC_HSM_ACTION:
2167                 rc = mdc_ioc_hsm_current_action(exp, karg);
2168                 GOTO(out, rc);
2169         case LL_IOC_HSM_REQUEST:
2170                 rc = mdc_ioc_hsm_request(exp, karg);
2171                 GOTO(out, rc);
2172         case OBD_IOC_CLIENT_RECOVER:
2173                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1, 0);
2174                 if (rc < 0)
2175                         GOTO(out, rc);
2176                 GOTO(out, rc = 0);
2177         case IOC_OSC_SET_ACTIVE:
2178                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
2179                 GOTO(out, rc);
2180         /*
2181          * Normally IOC_OBD_STATFS, OBD_IOC_QUOTACTL iocontrol are handled by
2182          * LMV instead of MDC. But when the cluster is upgraded from 1.8,
2183          * there'd be no LMV layer thus we might be called here. Eventually
2184          * this code should be removed.
2185          * bz20731, LU-592.
2186          */
2187         case IOC_OBD_STATFS: {
2188                 struct obd_statfs stat_buf = {0};
2189
2190                 if (*((__u32 *) data->ioc_inlbuf2) != 0)
2191                         GOTO(out, rc = -ENODEV);
2192
2193                 /* copy UUID */
2194                 if (copy_to_user(data->ioc_pbuf2, obd2cli_tgt(obd),
2195                                  min((int)data->ioc_plen2,
2196                                      (int)sizeof(struct obd_uuid))))
2197                         GOTO(out, rc = -EFAULT);
2198
2199                 rc = mdc_statfs(NULL, obd->obd_self_export, &stat_buf,
2200                                 ktime_get_seconds() - OBD_STATFS_CACHE_SECONDS,
2201                                 0);
2202                 if (rc != 0)
2203                         GOTO(out, rc);
2204
2205                 if (copy_to_user(data->ioc_pbuf1, &stat_buf,
2206                                      min((int) data->ioc_plen1,
2207                                          (int) sizeof(stat_buf))))
2208                         GOTO(out, rc = -EFAULT);
2209
2210                 GOTO(out, rc = 0);
2211         }
2212         case OBD_IOC_QUOTACTL: {
2213                 struct if_quotactl *qctl = karg;
2214                 struct obd_quotactl *oqctl;
2215
2216                 OBD_ALLOC_PTR(oqctl);
2217                 if (oqctl == NULL)
2218                         GOTO(out, rc = -ENOMEM);
2219
2220                 QCTL_COPY(oqctl, qctl);
2221                 rc = obd_quotactl(exp, oqctl);
2222                 if (rc == 0) {
2223                         QCTL_COPY(qctl, oqctl);
2224                         qctl->qc_valid = QC_MDTIDX;
2225                         qctl->obd_uuid = obd->u.cli.cl_target_uuid;
2226                 }
2227
2228                 OBD_FREE_PTR(oqctl);
2229                 GOTO(out, rc);
2230         }
2231         case LL_IOC_GET_CONNECT_FLAGS:
2232                 if (copy_to_user(uarg, exp_connect_flags_ptr(exp),
2233                                  sizeof(*exp_connect_flags_ptr(exp))))
2234                         GOTO(out, rc = -EFAULT);
2235
2236                 GOTO(out, rc = 0);
2237         case LL_IOC_LOV_SWAP_LAYOUTS:
2238                 rc = mdc_ioc_swap_layouts(exp, karg);
2239                 GOTO(out, rc);
2240         default:
2241                 CERROR("unrecognised ioctl: cmd = %#x\n", cmd);
2242                 GOTO(out, rc = -ENOTTY);
2243         }
2244 out:
2245         module_put(THIS_MODULE);
2246
2247         return rc;
2248 }
2249
2250 static int mdc_get_info_rpc(struct obd_export *exp,
2251                             u32 keylen, void *key,
2252                             u32 vallen, void *val)
2253 {
2254         struct obd_import      *imp = class_exp2cliimp(exp);
2255         struct ptlrpc_request  *req;
2256         char                   *tmp;
2257         int                     rc = -EINVAL;
2258         ENTRY;
2259
2260         req = ptlrpc_request_alloc(imp, &RQF_MDS_GET_INFO);
2261         if (req == NULL)
2262                 RETURN(-ENOMEM);
2263
2264         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_KEY,
2265                              RCL_CLIENT, keylen);
2266         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VALLEN,
2267                              RCL_CLIENT, sizeof(vallen));
2268
2269         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_INFO);
2270         if (rc) {
2271                 ptlrpc_request_free(req);
2272                 RETURN(rc);
2273         }
2274
2275         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_KEY);
2276         memcpy(tmp, key, keylen);
2277         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_VALLEN);
2278         memcpy(tmp, &vallen, sizeof(vallen));
2279
2280         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VAL,
2281                              RCL_SERVER, vallen);
2282         ptlrpc_request_set_replen(req);
2283
2284         rc = ptlrpc_queue_wait(req);
2285         /* -EREMOTE means the get_info result is partial, and it needs to
2286          * continue on another MDT, see fid2path part in lmv_iocontrol */
2287         if (rc == 0 || rc == -EREMOTE) {
2288                 tmp = req_capsule_server_get(&req->rq_pill, &RMF_GETINFO_VAL);
2289                 memcpy(val, tmp, vallen);
2290                 if (ptlrpc_rep_need_swab(req)) {
2291                         if (KEY_IS(KEY_FID2PATH))
2292                                 lustre_swab_fid2path(val);
2293                 }
2294         }
2295         ptlrpc_req_finished(req);
2296
2297         RETURN(rc);
2298 }
2299
2300 static void lustre_swab_hai(struct hsm_action_item *h)
2301 {
2302         __swab32s(&h->hai_len);
2303         __swab32s(&h->hai_action);
2304         lustre_swab_lu_fid(&h->hai_fid);
2305         lustre_swab_lu_fid(&h->hai_dfid);
2306         __swab64s(&h->hai_cookie);
2307         __swab64s(&h->hai_extent.offset);
2308         __swab64s(&h->hai_extent.length);
2309         __swab64s(&h->hai_gid);
2310 }
2311
2312 static void lustre_swab_hal(struct hsm_action_list *h)
2313 {
2314         struct hsm_action_item  *hai;
2315         __u32                    i;
2316
2317         __swab32s(&h->hal_version);
2318         __swab32s(&h->hal_count);
2319         __swab32s(&h->hal_archive_id);
2320         __swab64s(&h->hal_flags);
2321         hai = hai_first(h);
2322         for (i = 0; i < h->hal_count; i++, hai = hai_next(hai))
2323                 lustre_swab_hai(hai);
2324 }
2325
2326 static void lustre_swab_kuch(struct kuc_hdr *l)
2327 {
2328         __swab16s(&l->kuc_magic);
2329         /* __u8 l->kuc_transport */
2330         __swab16s(&l->kuc_msgtype);
2331         __swab16s(&l->kuc_msglen);
2332 }
2333
2334 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
2335                                 struct lustre_kernelcomm *lk)
2336 {
2337         struct obd_import *imp = class_exp2cliimp(exp);
2338         int rc = 0;
2339
2340         if (lk->lk_group != KUC_GRP_HSM) {
2341                 CERROR("Bad copytool group %d\n", lk->lk_group);
2342                 return -EINVAL;
2343         }
2344
2345         CDEBUG(D_HSM, "CT start r%d w%d u%d g%d f%#x\n", lk->lk_rfd, lk->lk_wfd,
2346                lk->lk_uid, lk->lk_group, lk->lk_flags);
2347
2348         if (lk->lk_flags & LK_FLG_STOP) {
2349                 /* Unregister with the coordinator */
2350                 rc = mdc_ioc_hsm_ct_unregister(imp);
2351         } else {
2352                 __u32 *archives = NULL;
2353
2354                 if ((lk->lk_flags & LK_FLG_DATANR) && lk->lk_data_count > 0)
2355                         archives = lk->lk_data;
2356
2357                 rc = mdc_ioc_hsm_ct_register(imp, lk->lk_data_count, archives);
2358         }
2359
2360         return rc;
2361 }
2362
2363 /**
2364  * Send a message to any listening copytools
2365  * @param val KUC message (kuc_hdr + hsm_action_list)
2366  * @param len total length of message
2367  */
2368 static int mdc_hsm_copytool_send(const struct obd_uuid *uuid,
2369                                  size_t len, void *val)
2370 {
2371         struct kuc_hdr          *lh = (struct kuc_hdr *)val;
2372         struct hsm_action_list  *hal = (struct hsm_action_list *)(lh + 1);
2373         int                      rc;
2374         ENTRY;
2375
2376         if (len < sizeof(*lh) + sizeof(*hal)) {
2377                 CERROR("Short HSM message %zu < %zu\n", len,
2378                        sizeof(*lh) + sizeof(*hal));
2379                 RETURN(-EPROTO);
2380         }
2381         if (lh->kuc_magic == __swab16(KUC_MAGIC)) {
2382                 lustre_swab_kuch(lh);
2383                 lustre_swab_hal(hal);
2384         } else if (lh->kuc_magic != KUC_MAGIC) {
2385                 CERROR("Bad magic %x!=%x\n", lh->kuc_magic, KUC_MAGIC);
2386                 RETURN(-EPROTO);
2387         }
2388
2389         CDEBUG(D_HSM, " Received message mg=%x t=%d m=%d l=%d actions=%d "
2390                "on %s\n",
2391                lh->kuc_magic, lh->kuc_transport, lh->kuc_msgtype,
2392                lh->kuc_msglen, hal->hal_count, hal->hal_fsname);
2393
2394         /* Broadcast to HSM listeners */
2395         rc = libcfs_kkuc_group_put(uuid, KUC_GRP_HSM, lh);
2396
2397         RETURN(rc);
2398 }
2399
2400 /**
2401  * callback function passed to kuc for re-registering each HSM copytool
2402  * running on MDC, after MDT shutdown/recovery.
2403  * @param data copytool registration data
2404  * @param cb_arg callback argument (obd_import)
2405  */
2406 static int mdc_hsm_ct_reregister(void *data, void *cb_arg)
2407 {
2408         struct obd_import *imp = (struct obd_import *)cb_arg;
2409         struct kkuc_ct_data *kcd = data;
2410         __u32 *archives = NULL;
2411         int rc;
2412
2413         if (kcd == NULL ||
2414             (kcd->kcd_magic != KKUC_CT_DATA_ARRAY_MAGIC &&
2415              kcd->kcd_magic != KKUC_CT_DATA_BITMAP_MAGIC))
2416                 return -EPROTO;
2417
2418         if (kcd->kcd_magic == KKUC_CT_DATA_BITMAP_MAGIC) {
2419                 CDEBUG(D_HA, "%s: recover copytool registration to MDT "
2420                        "(archive=%#x)\n", imp->imp_obd->obd_name,
2421                        kcd->kcd_nr_archives);
2422         } else {
2423                 CDEBUG(D_HA, "%s: recover copytool registration to MDT "
2424                        "(archive nr = %u)\n",
2425                        imp->imp_obd->obd_name, kcd->kcd_nr_archives);
2426                 if (kcd->kcd_nr_archives != 0)
2427                         archives = kcd->kcd_archives;
2428         }
2429
2430         rc = mdc_ioc_hsm_ct_register(imp, kcd->kcd_nr_archives, archives);
2431         /* ignore error if the copytool is already registered */
2432         return (rc == -EEXIST) ? 0 : rc;
2433 }
2434
2435 /**
2436  * Re-establish all kuc contexts with MDT
2437  * after MDT shutdown/recovery.
2438  */
2439 static int mdc_kuc_reregister(struct obd_import *imp)
2440 {
2441         /* re-register HSM agents */
2442         return libcfs_kkuc_group_foreach(&imp->imp_obd->obd_uuid, KUC_GRP_HSM,
2443                                          mdc_hsm_ct_reregister, imp);
2444 }
2445
2446 static int mdc_set_info_async(const struct lu_env *env,
2447                               struct obd_export *exp,
2448                               u32 keylen, void *key,
2449                               u32 vallen, void *val,
2450                               struct ptlrpc_request_set *set)
2451 {
2452         struct obd_import       *imp = class_exp2cliimp(exp);
2453         int                      rc;
2454         ENTRY;
2455
2456         if (KEY_IS(KEY_READ_ONLY)) {
2457                 if (vallen != sizeof(int))
2458                         RETURN(-EINVAL);
2459
2460                 spin_lock(&imp->imp_lock);
2461                 if (*((int *)val)) {
2462                         imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY;
2463                         imp->imp_connect_data.ocd_connect_flags |=
2464                                                         OBD_CONNECT_RDONLY;
2465                 } else {
2466                         imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY;
2467                         imp->imp_connect_data.ocd_connect_flags &=
2468                                                         ~OBD_CONNECT_RDONLY;
2469                 }
2470                 spin_unlock(&imp->imp_lock);
2471
2472                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2473                                        keylen, key, vallen, val, set);
2474                 RETURN(rc);
2475         }
2476         if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
2477                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2478                                        keylen, key, vallen, val, set);
2479                 RETURN(rc);
2480         }
2481         if (KEY_IS(KEY_HSM_COPYTOOL_SEND)) {
2482                 rc = mdc_hsm_copytool_send(&imp->imp_obd->obd_uuid, vallen,
2483                                            val);
2484                 RETURN(rc);
2485         }
2486
2487         if (KEY_IS(KEY_DEFAULT_EASIZE)) {
2488                 __u32 *default_easize = val;
2489
2490                 exp->exp_obd->u.cli.cl_default_mds_easize = *default_easize;
2491                 RETURN(0);
2492         }
2493
2494         rc = osc_set_info_async(env, exp, keylen, key, vallen, val, set);
2495         RETURN(rc);
2496 }
2497
2498 static int mdc_get_info(const struct lu_env *env, struct obd_export *exp,
2499                         __u32 keylen, void *key, __u32 *vallen, void *val)
2500 {
2501         int rc = -EINVAL;
2502
2503         if (KEY_IS(KEY_MAX_EASIZE)) {
2504                 __u32 mdsize, *max_easize;
2505
2506                 if (*vallen != sizeof(int))
2507                         RETURN(-EINVAL);
2508                 mdsize = *(__u32 *)val;
2509                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
2510                         exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
2511                 max_easize = val;
2512                 *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
2513                 RETURN(0);
2514         } else if (KEY_IS(KEY_DEFAULT_EASIZE)) {
2515                 __u32 *default_easize;
2516
2517                 if (*vallen != sizeof(int))
2518                         RETURN(-EINVAL);
2519                 default_easize = val;
2520                 *default_easize = exp->exp_obd->u.cli.cl_default_mds_easize;
2521                 RETURN(0);
2522         } else if (KEY_IS(KEY_CONN_DATA)) {
2523                 struct obd_import *imp = class_exp2cliimp(exp);
2524                 struct obd_connect_data *data = val;
2525
2526                 if (*vallen != sizeof(*data))
2527                         RETURN(-EINVAL);
2528
2529                 *data = imp->imp_connect_data;
2530                 RETURN(0);
2531         } else if (KEY_IS(KEY_TGT_COUNT)) {
2532                 *((__u32 *)val) = 1;
2533                 RETURN(0);
2534         }
2535
2536         rc = mdc_get_info_rpc(exp, keylen, key, *vallen, val);
2537
2538         RETURN(rc);
2539 }
2540
2541 static int mdc_fsync(struct obd_export *exp, const struct lu_fid *fid,
2542                      struct ptlrpc_request **request)
2543 {
2544         struct ptlrpc_request *req;
2545         int                    rc;
2546         ENTRY;
2547
2548         *request = NULL;
2549         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_SYNC);
2550         if (req == NULL)
2551                 RETURN(-ENOMEM);
2552
2553         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SYNC);
2554         if (rc) {
2555                 ptlrpc_request_free(req);
2556                 RETURN(rc);
2557         }
2558
2559         mdc_pack_body(req, fid, 0, 0, -1, 0);
2560
2561         ptlrpc_request_set_replen(req);
2562
2563         rc = ptlrpc_queue_wait(req);
2564         if (rc)
2565                 ptlrpc_req_finished(req);
2566         else
2567                 *request = req;
2568         RETURN(rc);
2569 }
2570
2571 struct mdc_rmfid_args {
2572         int *mra_rcs;
2573         int mra_nr;
2574 };
2575
2576 int mdc_rmfid_interpret(const struct lu_env *env, struct ptlrpc_request *req,
2577                           void *args, int rc)
2578 {
2579         struct mdc_rmfid_args *aa;
2580         int *rcs, size;
2581         ENTRY;
2582
2583         if (!rc) {
2584                 aa = ptlrpc_req_async_args(aa, req);
2585
2586                 size = req_capsule_get_size(&req->rq_pill, &RMF_RCS,
2587                                             RCL_SERVER);
2588                 LASSERT(size == sizeof(int) * aa->mra_nr);
2589                 rcs = req_capsule_server_get(&req->rq_pill, &RMF_RCS);
2590                 LASSERT(rcs);
2591                 LASSERT(aa->mra_rcs);
2592                 LASSERT(aa->mra_nr);
2593                 memcpy(aa->mra_rcs, rcs, size);
2594         }
2595
2596         RETURN(rc);
2597 }
2598
2599 static int mdc_rmfid(struct obd_export *exp, struct fid_array *fa,
2600                      int *rcs, struct ptlrpc_request_set *set)
2601 {
2602         struct ptlrpc_request *req;
2603         struct mdc_rmfid_args *aa;
2604         struct mdt_body *b;
2605         struct lu_fid *tmp;
2606         int rc, flen;
2607         ENTRY;
2608
2609         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_RMFID);
2610         if (req == NULL)
2611                 RETURN(-ENOMEM);
2612
2613         flen = fa->fa_nr * sizeof(struct lu_fid);
2614         req_capsule_set_size(&req->rq_pill, &RMF_FID_ARRAY,
2615                              RCL_CLIENT, flen);
2616         req_capsule_set_size(&req->rq_pill, &RMF_FID_ARRAY,
2617                              RCL_SERVER, flen);
2618         req_capsule_set_size(&req->rq_pill, &RMF_RCS,
2619                              RCL_SERVER, fa->fa_nr * sizeof(__u32));
2620         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_RMFID);
2621         if (rc) {
2622                 ptlrpc_request_free(req);
2623                 RETURN(rc);
2624         }
2625         tmp = req_capsule_client_get(&req->rq_pill, &RMF_FID_ARRAY);
2626         memcpy(tmp, fa->fa_fids, flen);
2627
2628         mdc_pack_body(req, NULL, 0, 0, -1, 0);
2629         b = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
2630         b->mbo_ctime = ktime_get_real_seconds();
2631
2632         ptlrpc_request_set_replen(req);
2633
2634         LASSERT(rcs);
2635         aa = ptlrpc_req_async_args(aa, req);
2636         aa->mra_rcs = rcs;
2637         aa->mra_nr = fa->fa_nr;
2638         req->rq_interpret_reply = mdc_rmfid_interpret;
2639
2640         ptlrpc_set_add_req(set, req);
2641         ptlrpc_check_set(NULL, set);
2642
2643         RETURN(rc);
2644 }
2645
2646 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
2647                             enum obd_import_event event)
2648 {
2649         struct client_obd *cli = &obd->u.cli;
2650         int rc = 0;
2651
2652         LASSERT(imp->imp_obd == obd);
2653
2654         switch (event) {
2655         case IMP_EVENT_DISCON:
2656                 spin_lock(&cli->cl_loi_list_lock);
2657                 cli->cl_avail_grant = 0;
2658                 cli->cl_lost_grant = 0;
2659                 spin_unlock(&cli->cl_loi_list_lock);
2660                 break;
2661         case IMP_EVENT_INACTIVE:
2662                 /*
2663                  * Flush current sequence to make client obtain new one
2664                  * from server in case of disconnect/reconnect.
2665                  */
2666                 down_read(&cli->cl_seq_rwsem);
2667                 if (cli->cl_seq)
2668                         seq_client_flush(cli->cl_seq);
2669                 up_read(&cli->cl_seq_rwsem);
2670
2671                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE);
2672                 break;
2673         case IMP_EVENT_INVALIDATE: {
2674                 struct ldlm_namespace *ns = obd->obd_namespace;
2675                 struct lu_env *env;
2676                 __u16 refcheck;
2677
2678                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2679
2680                 env = cl_env_get(&refcheck);
2681                 if (!IS_ERR(env)) {
2682                         /* Reset grants. All pages go to failing rpcs due to
2683                          * the invalid import.
2684                          */
2685                         osc_io_unplug(env, cli, NULL);
2686
2687                         cfs_hash_for_each_nolock(ns->ns_rs_hash,
2688                                                  osc_ldlm_resource_invalidate,
2689                                                  env, 0);
2690                         cl_env_put(env, &refcheck);
2691                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2692                 } else {
2693                         rc = PTR_ERR(env);
2694                 }
2695                 break;
2696         }
2697         case IMP_EVENT_ACTIVE:
2698                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE);
2699                 /* redo the kuc registration after reconnecting */
2700                 if (rc == 0)
2701                         rc = mdc_kuc_reregister(imp);
2702                 break;
2703         case IMP_EVENT_OCD: {
2704                 struct obd_connect_data *ocd = &imp->imp_connect_data;
2705
2706                 if (OCD_HAS_FLAG(ocd, GRANT))
2707                         osc_init_grant(cli, ocd);
2708
2709                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD);
2710                 break;
2711         }
2712         case IMP_EVENT_DEACTIVATE:
2713         case IMP_EVENT_ACTIVATE:
2714                 break;
2715         default:
2716                 CERROR("Unknown import event %x\n", event);
2717                 LBUG();
2718         }
2719         RETURN(rc);
2720 }
2721
2722 int mdc_fid_alloc(const struct lu_env *env, struct obd_export *exp,
2723                   struct lu_fid *fid, struct md_op_data *op_data)
2724 {
2725         struct client_obd *cli = &exp->exp_obd->u.cli;
2726         int rc = -EIO;
2727
2728         ENTRY;
2729
2730         down_read(&cli->cl_seq_rwsem);
2731         if (cli->cl_seq)
2732                 rc = seq_client_alloc_fid(env, cli->cl_seq, fid);
2733         up_read(&cli->cl_seq_rwsem);
2734
2735         RETURN(rc);
2736 }
2737
2738 static struct obd_uuid *mdc_get_uuid(struct obd_export *exp)
2739 {
2740         struct client_obd *cli = &exp->exp_obd->u.cli;
2741         return &cli->cl_target_uuid;
2742 }
2743
2744 /**
2745  * Determine whether the lock can be canceled before replaying it during
2746  * recovery, non zero value will be return if the lock can be canceled,
2747  * or zero returned for not
2748  */
2749 static int mdc_cancel_weight(struct ldlm_lock *lock)
2750 {
2751         if (lock->l_resource->lr_type != LDLM_IBITS)
2752                 RETURN(0);
2753
2754         /* FIXME: if we ever get into a situation where there are too many
2755          * opened files with open locks on a single node, then we really
2756          * should replay these open locks to reget it */
2757         if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
2758                 RETURN(0);
2759
2760         /* Special case for DoM locks, cancel only unused and granted locks */
2761         if (ldlm_has_dom(lock) &&
2762             (lock->l_granted_mode != lock->l_req_mode ||
2763              osc_ldlm_weigh_ast(lock) != 0))
2764                 RETURN(0);
2765
2766         RETURN(1);
2767 }
2768
2769 static int mdc_resource_inode_free(struct ldlm_resource *res)
2770 {
2771         if (res->lr_lvb_inode)
2772                 res->lr_lvb_inode = NULL;
2773
2774         return 0;
2775 }
2776
2777 static struct ldlm_valblock_ops inode_lvbo = {
2778         .lvbo_free = mdc_resource_inode_free
2779 };
2780
2781 static int mdc_llog_init(struct obd_device *obd)
2782 {
2783         struct obd_llog_group   *olg = &obd->obd_olg;
2784         struct llog_ctxt        *ctxt;
2785         int                      rc;
2786
2787         ENTRY;
2788
2789         rc = llog_setup(NULL, obd, olg, LLOG_CHANGELOG_REPL_CTXT, obd,
2790                         &llog_client_ops);
2791         if (rc < 0)
2792                 RETURN(rc);
2793
2794         ctxt = llog_group_get_ctxt(olg, LLOG_CHANGELOG_REPL_CTXT);
2795         llog_initiator_connect(ctxt);
2796         llog_ctxt_put(ctxt);
2797
2798         RETURN(0);
2799 }
2800
2801 static void mdc_llog_finish(struct obd_device *obd)
2802 {
2803         struct llog_ctxt *ctxt;
2804
2805         ENTRY;
2806
2807         ctxt = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
2808         if (ctxt != NULL)
2809                 llog_cleanup(NULL, ctxt);
2810
2811         EXIT;
2812 }
2813
2814 int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
2815 {
2816         int rc;
2817
2818         ENTRY;
2819
2820         rc = osc_setup_common(obd, cfg);
2821         if (rc < 0)
2822                 RETURN(rc);
2823
2824         rc = mdc_tunables_init(obd);
2825         if (rc)
2826                 GOTO(err_osc_cleanup, rc);
2827
2828         obd->u.cli.cl_dom_min_inline_repsize = MDC_DOM_DEF_INLINE_REPSIZE;
2829
2830         ns_register_cancel(obd->obd_namespace, mdc_cancel_weight);
2831
2832         obd->obd_namespace->ns_lvbo = &inode_lvbo;
2833
2834         rc = mdc_llog_init(obd);
2835         if (rc) {
2836                 CERROR("%s: failed to setup llogging subsystems: rc = %d\n",
2837                        obd->obd_name, rc);
2838                 GOTO(err_llog_cleanup, rc);
2839         }
2840
2841         rc = mdc_changelog_cdev_init(obd);
2842         if (rc) {
2843                 CERROR("%s: failed to setup changelog char device: rc = %d\n",
2844                        obd->obd_name, rc);
2845                 GOTO(err_changelog_cleanup, rc);
2846         }
2847
2848         RETURN(rc);
2849
2850 err_changelog_cleanup:
2851         mdc_llog_finish(obd);
2852 err_llog_cleanup:
2853         lprocfs_free_md_stats(obd);
2854         ptlrpc_lprocfs_unregister_obd(obd);
2855 err_osc_cleanup:
2856         osc_cleanup_common(obd);
2857         return rc;
2858 }
2859
2860 /* Initialize the default and maximum LOV EA sizes.  This allows
2861  * us to make MDS RPCs with large enough reply buffers to hold a default
2862  * sized EA without having to calculate this (via a call into the
2863  * LOV + OSCs) each time we make an RPC.  The maximum size is also tracked
2864  * but not used to avoid wastefully vmalloc()'ing large reply buffers when
2865  * a large number of stripes is possible.  If a larger reply buffer is
2866  * required it will be reallocated in the ptlrpc layer due to overflow.
2867  */
2868 static int mdc_init_ea_size(struct obd_export *exp, __u32 easize,
2869                             __u32 def_easize)
2870 {
2871         struct obd_device *obd = exp->exp_obd;
2872         struct client_obd *cli = &obd->u.cli;
2873         ENTRY;
2874
2875         if (cli->cl_max_mds_easize < easize)
2876                 cli->cl_max_mds_easize = easize;
2877
2878         if (cli->cl_default_mds_easize < def_easize)
2879                 cli->cl_default_mds_easize = def_easize;
2880
2881         RETURN(0);
2882 }
2883
2884 static int mdc_precleanup(struct obd_device *obd)
2885 {
2886         ENTRY;
2887
2888         osc_precleanup_common(obd);
2889         mdc_changelog_cdev_finish(obd);
2890
2891         obd_cleanup_client_import(obd);
2892         ptlrpc_lprocfs_unregister_obd(obd);
2893         lprocfs_free_md_stats(obd);
2894         mdc_llog_finish(obd);
2895         RETURN(0);
2896 }
2897
2898 static int mdc_cleanup(struct obd_device *obd)
2899 {
2900         return osc_cleanup_common(obd);
2901 }
2902
2903 static struct obd_ops mdc_obd_ops = {
2904         .o_owner            = THIS_MODULE,
2905         .o_setup            = mdc_setup,
2906         .o_precleanup       = mdc_precleanup,
2907         .o_cleanup          = mdc_cleanup,
2908         .o_add_conn         = client_import_add_conn,
2909         .o_del_conn         = client_import_del_conn,
2910         .o_connect          = client_connect_import,
2911         .o_reconnect        = osc_reconnect,
2912         .o_disconnect       = osc_disconnect,
2913         .o_iocontrol        = mdc_iocontrol,
2914         .o_set_info_async   = mdc_set_info_async,
2915         .o_statfs           = mdc_statfs,
2916         .o_statfs_async     = mdc_statfs_async,
2917         .o_fid_init         = client_fid_init,
2918         .o_fid_fini         = client_fid_fini,
2919         .o_fid_alloc        = mdc_fid_alloc,
2920         .o_import_event     = mdc_import_event,
2921         .o_get_info         = mdc_get_info,
2922         .o_get_uuid         = mdc_get_uuid,
2923         .o_quotactl         = mdc_quotactl,
2924 };
2925
2926 static struct md_ops mdc_md_ops = {
2927         .m_get_root         = mdc_get_root,
2928         .m_null_inode       = mdc_null_inode,
2929         .m_close            = mdc_close,
2930         .m_create           = mdc_create,
2931         .m_enqueue          = mdc_enqueue,
2932         .m_getattr          = mdc_getattr,
2933         .m_getattr_name     = mdc_getattr_name,
2934         .m_intent_lock      = mdc_intent_lock,
2935         .m_link             = mdc_link,
2936         .m_rename           = mdc_rename,
2937         .m_setattr          = mdc_setattr,
2938         .m_setxattr         = mdc_setxattr,
2939         .m_getxattr         = mdc_getxattr,
2940         .m_fsync                = mdc_fsync,
2941         .m_file_resync          = mdc_file_resync,
2942         .m_read_page            = mdc_read_page,
2943         .m_unlink           = mdc_unlink,
2944         .m_cancel_unused    = mdc_cancel_unused,
2945         .m_init_ea_size     = mdc_init_ea_size,
2946         .m_set_lock_data    = mdc_set_lock_data,
2947         .m_lock_match       = mdc_lock_match,
2948         .m_get_lustre_md    = mdc_get_lustre_md,
2949         .m_free_lustre_md   = mdc_free_lustre_md,
2950         .m_set_open_replay_data = mdc_set_open_replay_data,
2951         .m_clear_open_replay_data = mdc_clear_open_replay_data,
2952         .m_intent_getattr_async = mdc_intent_getattr_async,
2953         .m_revalidate_lock      = mdc_revalidate_lock,
2954         .m_rmfid                = mdc_rmfid,
2955 };
2956
2957 static int __init mdc_init(void)
2958 {
2959         return class_register_type(&mdc_obd_ops, &mdc_md_ops, true, NULL,
2960                                    LUSTRE_MDC_NAME, &mdc_device_type);
2961 }
2962
2963 static void __exit mdc_exit(void)
2964 {
2965         class_unregister_type(LUSTRE_MDC_NAME);
2966 }
2967
2968 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2969 MODULE_DESCRIPTION("Lustre Metadata Client");
2970 MODULE_VERSION(LUSTRE_VERSION_STRING);
2971 MODULE_LICENSE("GPL");
2972
2973 module_init(mdc_init);
2974 module_exit(mdc_exit);