drivers/staging/lustre/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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_MDC
  38
  39 # include <linux/module.h>
  40 # include <linux/pagemap.h>
  41 # include <linux/miscdevice.h>
  42 # include <linux/init.h>
  43 # include <linux/utsname.h>
  44
  45 #include "../include/lustre_acl.h"
  46 #include "../include/obd_class.h"
  47 #include "../include/lustre_fid.h"
  48 #include "../include/lprocfs_status.h"
  49 #include "../include/lustre_param.h"
  50 #include "../include/lustre_log.h"
  51
  52 #include "mdc_internal.h"
  53
  54 #define REQUEST_MINOR 244
  55
  56 struct mdc_renew_capa_args {
  57         struct obd_capa *ra_oc;
  58         renew_capa_cb_t  ra_cb;
  59 };
  60
  61 static int mdc_cleanup(struct obd_device *obd);
  62
  63 int mdc_unpack_capa(struct obd_export *exp, struct ptlrpc_request *req,
  64                     const struct req_msg_field *field, struct obd_capa **oc)
  65 {
  66         struct lustre_capa *capa;
  67         struct obd_capa *c;
  68
  69         /* swabbed already in mdc_enqueue */
  70         capa = req_capsule_server_get(&req->rq_pill, field);
  71         if (capa == NULL)
  72                 return -EPROTO;
  73
  74         c = alloc_capa(CAPA_SITE_CLIENT);
  75         if (IS_ERR(c)) {
  76                 CDEBUG(D_INFO, "alloc capa failed!\n");
  77                 return PTR_ERR(c);
  78         } else {
  79                 c->c_capa = *capa;
  80                 *oc = c;
  81                 return 0;
  82         }
  83 }
  84
  85 static inline int mdc_queue_wait(struct ptlrpc_request *req)
  86 {
  87         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
  88         int rc;
  89
  90         /* mdc_enter_request() ensures that this client has no more
  91          * than cl_max_rpcs_in_flight RPCs simultaneously inf light
  92          * against an MDT. */
  93         rc = mdc_enter_request(cli);
  94         if (rc != 0)
  95                 return rc;
  96
  97         rc = ptlrpc_queue_wait(req);
  98         mdc_exit_request(cli);
  99
 100         return rc;
 101 }
 102
 103 /* Helper that implements most of mdc_getstatus and signal_completed_replay. */
 104 /* XXX this should become mdc_get_info("key"), sending MDS_GET_INFO RPC */
 105 static int send_getstatus(struct obd_import *imp, struct lu_fid *rootfid,
 106                           struct obd_capa **pc, int level, int msg_flags)
 107 {
 108         struct ptlrpc_request *req;
 109         struct mdt_body       *body;
 110         int                 rc;
 111
 112         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_GETSTATUS,
 113                                         LUSTRE_MDS_VERSION, MDS_GETSTATUS);
 114         if (req == NULL)
 115                 return -ENOMEM;
 116
 117         mdc_pack_body(req, NULL, NULL, 0, 0, -1, 0);
 118         lustre_msg_add_flags(req->rq_reqmsg, msg_flags);
 119         req->rq_send_state = level;
 120
 121         ptlrpc_request_set_replen(req);
 122
 123         rc = ptlrpc_queue_wait(req);
 124         if (rc)
 125                 GOTO(out, rc);
 126
 127         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 128         if (body == NULL)
 129                 GOTO(out, rc = -EPROTO);
 130
 131         if (body->valid & OBD_MD_FLMDSCAPA) {
 132                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, pc);
 133                 if (rc)
 134                         GOTO(out, rc);
 135         }
 136
 137         *rootfid = body->fid1;
 138         CDEBUG(D_NET,
 139                "root fid="DFID", last_committed=%llu\n",
 140                PFID(rootfid),
 141                lustre_msg_get_last_committed(req->rq_repmsg));
 142 out:
 143         ptlrpc_req_finished(req);
 144         return rc;
 145 }
 146
 147 /* This should be mdc_get_info("rootfid") */
 148 int mdc_getstatus(struct obd_export *exp, struct lu_fid *rootfid,
 149                   struct obd_capa **pc)
 150 {
 151         return send_getstatus(class_exp2cliimp(exp), rootfid, pc,
 152                               LUSTRE_IMP_FULL, 0);
 153 }
 154
 155 /*
 156  * This function now is known to always saying that it will receive 4 buffers
 157  * from server. Even for cases when acl_size and md_size is zero, RPC header
 158  * will contain 4 fields and RPC itself will contain zero size fields. This is
 159  * because mdt_getattr*() _always_ returns 4 fields, but if acl is not needed
 160  * and thus zero, it shrinks it, making zero size. The same story about
 161  * md_size. And this is course of problem when client waits for smaller number
 162  * of fields. This issue will be fixed later when client gets aware of RPC
 163  * layouts.  --umka
 164  */
 165 static int mdc_getattr_common(struct obd_export *exp,
 166                               struct ptlrpc_request *req)
 167 {
 168         struct req_capsule *pill = &req->rq_pill;
 169         struct mdt_body    *body;
 170         void           *eadata;
 171         int              rc;
 172
 173         /* Request message already built. */
 174         rc = ptlrpc_queue_wait(req);
 175         if (rc != 0)
 176                 return rc;
 177
 178         /* sanity check for the reply */
 179         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 180         if (body == NULL)
 181                 return -EPROTO;
 182
 183         CDEBUG(D_NET, "mode: %o\n", body->mode);
 184
 185         if (body->eadatasize != 0) {
 186                 mdc_update_max_ea_from_body(exp, body);
 187
 188                 eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 189                                                       body->eadatasize);
 190                 if (eadata == NULL)
 191                         return -EPROTO;
 192         }
 193
 194         if (body->valid & OBD_MD_FLRMTPERM) {
 195                 struct mdt_remote_perm *perm;
 196
 197                 LASSERT(client_is_remote(exp));
 198                 perm = req_capsule_server_swab_get(pill, &RMF_ACL,
 199                                                 lustre_swab_mdt_remote_perm);
 200                 if (perm == NULL)
 201                         return -EPROTO;
 202         }
 203
 204         if (body->valid & OBD_MD_FLMDSCAPA) {
 205                 struct lustre_capa *capa;
 206                 capa = req_capsule_server_get(pill, &RMF_CAPA1);
 207                 if (capa == NULL)
 208                         return -EPROTO;
 209         }
 210
 211         return 0;
 212 }
 213
 214 int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data,
 215                 struct ptlrpc_request **request)
 216 {
 217         struct ptlrpc_request *req;
 218         int                 rc;
 219
 220         /* Single MDS without an LMV case */
 221         if (op_data->op_flags & MF_GET_MDT_IDX) {
 222                 op_data->op_mds = 0;
 223                 return 0;
 224         }
 225         *request = NULL;
 226         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_GETATTR);
 227         if (req == NULL)
 228                 return -ENOMEM;
 229
 230         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 231
 232         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
 233         if (rc) {
 234                 ptlrpc_request_free(req);
 235                 return rc;
 236         }
 237
 238         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
 239                       op_data->op_valid, op_data->op_mode, -1, 0);
 240
 241         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 242                              op_data->op_mode);
 243         if (op_data->op_valid & OBD_MD_FLRMTPERM) {
 244                 LASSERT(client_is_remote(exp));
 245                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
 246                                      sizeof(struct mdt_remote_perm));
 247         }
 248         ptlrpc_request_set_replen(req);
 249
 250         rc = mdc_getattr_common(exp, req);
 251         if (rc)
 252                 ptlrpc_req_finished(req);
 253         else
 254                 *request = req;
 255         return rc;
 256 }
 257
 258 int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data,
 259                      struct ptlrpc_request **request)
 260 {
 261         struct ptlrpc_request *req;
 262         int                 rc;
 263
 264         *request = NULL;
 265         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 266                                    &RQF_MDS_GETATTR_NAME);
 267         if (req == NULL)
 268                 return -ENOMEM;
 269
 270         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 271         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 272                              op_data->op_namelen + 1);
 273
 274         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR_NAME);
 275         if (rc) {
 276                 ptlrpc_request_free(req);
 277                 return rc;
 278         }
 279
 280         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
 281                       op_data->op_valid, op_data->op_mode,
 282                       op_data->op_suppgids[0], 0);
 283
 284         if (op_data->op_name) {
 285                 char *name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 286                 LASSERT(strnlen(op_data->op_name, op_data->op_namelen) ==
 287                                 op_data->op_namelen);
 288                 memcpy(name, op_data->op_name, op_data->op_namelen);
 289         }
 290
 291         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 292                              op_data->op_mode);
 293         ptlrpc_request_set_replen(req);
 294
 295         rc = mdc_getattr_common(exp, req);
 296         if (rc)
 297                 ptlrpc_req_finished(req);
 298         else
 299                 *request = req;
 300         return rc;
 301 }
 302
 303 static int mdc_is_subdir(struct obd_export *exp,
 304                          const struct lu_fid *pfid,
 305                          const struct lu_fid *cfid,
 306                          struct ptlrpc_request **request)
 307 {
 308         struct ptlrpc_request  *req;
 309         int                  rc;
 310
 311         *request = NULL;
 312         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
 313                                         &RQF_MDS_IS_SUBDIR, LUSTRE_MDS_VERSION,
 314                                         MDS_IS_SUBDIR);
 315         if (req == NULL)
 316                 return -ENOMEM;
 317
 318         mdc_is_subdir_pack(req, pfid, cfid, 0);
 319         ptlrpc_request_set_replen(req);
 320
 321         rc = ptlrpc_queue_wait(req);
 322         if (rc && rc != -EREMOTE)
 323                 ptlrpc_req_finished(req);
 324         else
 325                 *request = req;
 326         return rc;
 327 }
 328
 329 static int mdc_xattr_common(struct obd_export *exp,
 330                             const struct req_format *fmt,
 331                             const struct lu_fid *fid,
 332                             struct obd_capa *oc, int opcode, obd_valid valid,
 333                             const char *xattr_name, const char *input,
 334                             int input_size, int output_size, int flags,
 335                             __u32 suppgid, struct ptlrpc_request **request)
 336 {
 337         struct ptlrpc_request *req;
 338         int   xattr_namelen = 0;
 339         char *tmp;
 340         int   rc;
 341
 342         *request = NULL;
 343         req = ptlrpc_request_alloc(class_exp2cliimp(exp), fmt);
 344         if (req == NULL)
 345                 return -ENOMEM;
 346
 347         mdc_set_capa_size(req, &RMF_CAPA1, oc);
 348         if (xattr_name) {
 349                 xattr_namelen = strlen(xattr_name) + 1;
 350                 req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
 351                                      xattr_namelen);
 352         }
 353         if (input_size) {
 354                 LASSERT(input);
 355                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
 356                                      input_size);
 357         }
 358
 359         /* Flush local XATTR locks to get rid of a possible cancel RPC */
 360         if (opcode == MDS_REINT && fid_is_sane(fid) &&
 361             exp->exp_connect_data.ocd_ibits_known & MDS_INODELOCK_XATTR) {
 362                 LIST_HEAD(cancels);
 363                 int count;
 364
 365                 /* Without that packing would fail */
 366                 if (input_size == 0)
 367                         req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
 368                                              RCL_CLIENT, 0);
 369
 370                 count = mdc_resource_get_unused(exp, fid,
 371                                                 &cancels, LCK_EX,
 372                                                 MDS_INODELOCK_XATTR);
 373
 374                 rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
 375                 if (rc) {
 376                         ptlrpc_request_free(req);
 377                         return rc;
 378                 }
 379         } else {
 380                 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, opcode);
 381                 if (rc) {
 382                         ptlrpc_request_free(req);
 383                         return rc;
 384                 }
 385         }
 386
 387         if (opcode == MDS_REINT) {
 388                 struct mdt_rec_setxattr *rec;
 389
 390                 CLASSERT(sizeof(struct mdt_rec_setxattr) ==
 391                          sizeof(struct mdt_rec_reint));
 392                 rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
 393                 rec->sx_opcode = REINT_SETXATTR;
 394                 rec->sx_fsuid  = from_kuid(&init_user_ns, current_fsuid());
 395                 rec->sx_fsgid  = from_kgid(&init_user_ns, current_fsgid());
 396                 rec->sx_cap    = cfs_curproc_cap_pack();
 397                 rec->sx_suppgid1 = suppgid;
 398                 rec->sx_suppgid2 = -1;
 399                 rec->sx_fid    = *fid;
 400                 rec->sx_valid  = valid | OBD_MD_FLCTIME;
 401                 rec->sx_time   = get_seconds();
 402                 rec->sx_size   = output_size;
 403                 rec->sx_flags  = flags;
 404
 405                 mdc_pack_capa(req, &RMF_CAPA1, oc);
 406         } else {
 407                 mdc_pack_body(req, fid, oc, valid, output_size, suppgid, flags);
 408         }
 409
 410         if (xattr_name) {
 411                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 412                 memcpy(tmp, xattr_name, xattr_namelen);
 413         }
 414         if (input_size) {
 415                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_EADATA);
 416                 memcpy(tmp, input, input_size);
 417         }
 418
 419         if (req_capsule_has_field(&req->rq_pill, &RMF_EADATA, RCL_SERVER))
 420                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
 421                                      RCL_SERVER, output_size);
 422         ptlrpc_request_set_replen(req);
 423
 424         /* make rpc */
 425         if (opcode == MDS_REINT)
 426                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 427
 428         rc = ptlrpc_queue_wait(req);
 429
 430         if (opcode == MDS_REINT)
 431                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 432
 433         if (rc)
 434                 ptlrpc_req_finished(req);
 435         else
 436                 *request = req;
 437         return rc;
 438 }
 439
 440 int mdc_setxattr(struct obd_export *exp, const struct lu_fid *fid,
 441                  struct obd_capa *oc, obd_valid valid, const char *xattr_name,
 442                  const char *input, int input_size, int output_size,
 443                  int flags, __u32 suppgid, struct ptlrpc_request **request)
 444 {
 445         return mdc_xattr_common(exp, &RQF_MDS_REINT_SETXATTR,
 446                                 fid, oc, MDS_REINT, valid, xattr_name,
 447                                 input, input_size, output_size, flags,
 448                                 suppgid, request);
 449 }
 450
 451 int mdc_getxattr(struct obd_export *exp, const struct lu_fid *fid,
 452                  struct obd_capa *oc, obd_valid valid, const char *xattr_name,
 453                  const char *input, int input_size, int output_size,
 454                  int flags, struct ptlrpc_request **request)
 455 {
 456         return mdc_xattr_common(exp, &RQF_MDS_GETXATTR,
 457                                 fid, oc, MDS_GETXATTR, valid, xattr_name,
 458                                 input, input_size, output_size, flags,
 459                                 -1, request);
 460 }
 461
 462 #ifdef CONFIG_FS_POSIX_ACL
 463 static int mdc_unpack_acl(struct ptlrpc_request *req, struct lustre_md *md)
 464 {
 465         struct req_capsule     *pill = &req->rq_pill;
 466         struct mdt_body *body = md->body;
 467         struct posix_acl       *acl;
 468         void               *buf;
 469         int                  rc;
 470
 471         if (!body->aclsize)
 472                 return 0;
 473
 474         buf = req_capsule_server_sized_get(pill, &RMF_ACL, body->aclsize);
 475
 476         if (!buf)
 477                 return -EPROTO;
 478
 479         acl = posix_acl_from_xattr(&init_user_ns, buf, body->aclsize);
 480         if (IS_ERR(acl)) {
 481                 rc = PTR_ERR(acl);
 482                 CERROR("convert xattr to acl: %d\n", rc);
 483                 return rc;
 484         }
 485
 486         rc = posix_acl_valid(acl);
 487         if (rc) {
 488                 CERROR("validate acl: %d\n", rc);
 489                 posix_acl_release(acl);
 490                 return rc;
 491         }
 492
 493         md->posix_acl = acl;
 494         return 0;
 495 }
 496 #else
 497 #define mdc_unpack_acl(req, md) 0
 498 #endif
 499
 500 int mdc_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,
 501                       struct obd_export *dt_exp, struct obd_export *md_exp,
 502                       struct lustre_md *md)
 503 {
 504         struct req_capsule *pill = &req->rq_pill;
 505         int rc;
 506
 507         LASSERT(md);
 508         memset(md, 0, sizeof(*md));
 509
 510         md->body = req_capsule_server_get(pill, &RMF_MDT_BODY);
 511         LASSERT(md->body != NULL);
 512
 513         if (md->body->valid & OBD_MD_FLEASIZE) {
 514                 int lmmsize;
 515                 struct lov_mds_md *lmm;
 516
 517                 if (!S_ISREG(md->body->mode)) {
 518                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, should be a "
 519                                "regular file, but is not\n");
 520                         GOTO(out, rc = -EPROTO);
 521                 }
 522
 523                 if (md->body->eadatasize == 0) {
 524                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, "
 525                                "but eadatasize 0\n");
 526                         GOTO(out, rc = -EPROTO);
 527                 }
 528                 lmmsize = md->body->eadatasize;
 529                 lmm = req_capsule_server_sized_get(pill, &RMF_MDT_MD, lmmsize);
 530                 if (!lmm)
 531                         GOTO(out, rc = -EPROTO);
 532
 533                 rc = obd_unpackmd(dt_exp, &md->lsm, lmm, lmmsize);
 534                 if (rc < 0)
 535                         GOTO(out, rc);
 536
 537                 if (rc < sizeof(*md->lsm)) {
 538                         CDEBUG(D_INFO, "lsm size too small: "
 539                                "rc < sizeof (*md->lsm) (%d < %d)\n",
 540                                rc, (int)sizeof(*md->lsm));
 541                         GOTO(out, rc = -EPROTO);
 542                 }
 543
 544         } else if (md->body->valid & OBD_MD_FLDIREA) {
 545                 int lmvsize;
 546                 struct lov_mds_md *lmv;
 547
 548                 if (!S_ISDIR(md->body->mode)) {
 549                         CDEBUG(D_INFO, "OBD_MD_FLDIREA set, should be a "
 550                                "directory, but is not\n");
 551                         GOTO(out, rc = -EPROTO);
 552                 }
 553
 554                 if (md->body->eadatasize == 0) {
 555                         CDEBUG(D_INFO, "OBD_MD_FLDIREA is set, "
 556                                "but eadatasize 0\n");
 557                         return -EPROTO;
 558                 }
 559                 if (md->body->valid & OBD_MD_MEA) {
 560                         lmvsize = md->body->eadatasize;
 561                         lmv = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
 562                                                            lmvsize);
 563                         if (!lmv)
 564                                 GOTO(out, rc = -EPROTO);
 565
 566                         rc = obd_unpackmd(md_exp, (void *)&md->mea, lmv,
 567                                           lmvsize);
 568                         if (rc < 0)
 569                                 GOTO(out, rc);
 570
 571                         if (rc < sizeof(*md->mea)) {
 572                                 CDEBUG(D_INFO, "size too small:  "
 573                                        "rc < sizeof(*md->mea) (%d < %d)\n",
 574                                         rc, (int)sizeof(*md->mea));
 575                                 GOTO(out, rc = -EPROTO);
 576                         }
 577                 }
 578         }
 579         rc = 0;
 580
 581         if (md->body->valid & OBD_MD_FLRMTPERM) {
 582                 /* remote permission */
 583                 LASSERT(client_is_remote(exp));
 584                 md->remote_perm = req_capsule_server_swab_get(pill, &RMF_ACL,
 585                                                 lustre_swab_mdt_remote_perm);
 586                 if (!md->remote_perm)
 587                         GOTO(out, rc = -EPROTO);
 588         }
 589         else if (md->body->valid & OBD_MD_FLACL) {
 590                 /* for ACL, it's possible that FLACL is set but aclsize is zero.
 591                  * only when aclsize != 0 there's an actual segment for ACL
 592                  * in reply buffer.
 593                  */
 594                 if (md->body->aclsize) {
 595                         rc = mdc_unpack_acl(req, md);
 596                         if (rc)
 597                                 GOTO(out, rc);
 598 #ifdef CONFIG_FS_POSIX_ACL
 599                 } else {
 600                         md->posix_acl = NULL;
 601 #endif
 602                 }
 603         }
 604         if (md->body->valid & OBD_MD_FLMDSCAPA) {
 605                 struct obd_capa *oc = NULL;
 606
 607                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, &oc);
 608                 if (rc)
 609                         GOTO(out, rc);
 610                 md->mds_capa = oc;
 611         }
 612
 613         if (md->body->valid & OBD_MD_FLOSSCAPA) {
 614                 struct obd_capa *oc = NULL;
 615
 616                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA2, &oc);
 617                 if (rc)
 618                         GOTO(out, rc);
 619                 md->oss_capa = oc;
 620         }
 621
 622 out:
 623         if (rc) {
 624                 if (md->oss_capa) {
 625                         capa_put(md->oss_capa);
 626                         md->oss_capa = NULL;
 627                 }
 628                 if (md->mds_capa) {
 629                         capa_put(md->mds_capa);
 630                         md->mds_capa = NULL;
 631                 }
 632 #ifdef CONFIG_FS_POSIX_ACL
 633                 posix_acl_release(md->posix_acl);
 634 #endif
 635                 if (md->lsm)
 636                         obd_free_memmd(dt_exp, &md->lsm);
 637         }
 638         return rc;
 639 }
 640
 641 int mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
 642 {
 643         return 0;
 644 }
 645
 646 /**
 647  * Handles both OPEN and SETATTR RPCs for OPEN-CLOSE and SETATTR-DONE_WRITING
 648  * RPC chains.
 649  */
 650 void mdc_replay_open(struct ptlrpc_request *req)
 651 {
 652         struct md_open_data *mod = req->rq_cb_data;
 653         struct ptlrpc_request *close_req;
 654         struct obd_client_handle *och;
 655         struct lustre_handle old;
 656         struct mdt_body *body;
 657
 658         if (mod == NULL) {
 659                 DEBUG_REQ(D_ERROR, req,
 660                           "Can't properly replay without open data.");
 661                 return;
 662         }
 663
 664         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 665         LASSERT(body != NULL);
 666
 667         och = mod->mod_och;
 668         if (och != NULL) {
 669                 struct lustre_handle *file_fh;
 670
 671                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 672
 673                 file_fh = &och->och_fh;
 674                 CDEBUG(D_HA, "updating handle from %#llx to %#llx\n",
 675                        file_fh->cookie, body->handle.cookie);
 676                 old = *file_fh;
 677                 *file_fh = body->handle;
 678         }
 679         close_req = mod->mod_close_req;
 680         if (close_req != NULL) {
 681                 __u32 opc = lustre_msg_get_opc(close_req->rq_reqmsg);
 682                 struct mdt_ioepoch *epoch;
 683
 684                 LASSERT(opc == MDS_CLOSE || opc == MDS_DONE_WRITING);
 685                 epoch = req_capsule_client_get(&close_req->rq_pill,
 686                                                &RMF_MDT_EPOCH);
 687                 LASSERT(epoch);
 688
 689                 if (och != NULL)
 690                         LASSERT(!memcmp(&old, &epoch->handle, sizeof(old)));
 691                 DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
 692                 epoch->handle = body->handle;
 693         }
 694 }
 695
 696 void mdc_commit_open(struct ptlrpc_request *req)
 697 {
 698         struct md_open_data *mod = req->rq_cb_data;
 699         if (mod == NULL)
 700                 return;
 701
 702         /**
 703          * No need to touch md_open_data::mod_och, it holds a reference on
 704          * \var mod and will zero references to each other, \var mod will be
 705          * freed after that when md_open_data::mod_och will put the reference.
 706          */
 707
 708         /**
 709          * Do not let open request to disappear as it still may be needed
 710          * for close rpc to happen (it may happen on evict only, otherwise
 711          * ptlrpc_request::rq_replay does not let mdc_commit_open() to be
 712          * called), just mark this rpc as committed to distinguish these 2
 713          * cases, see mdc_close() for details. The open request reference will
 714          * be put along with freeing \var mod.
 715          */
 716         ptlrpc_request_addref(req);
 717         spin_lock(&req->rq_lock);
 718         req->rq_committed = 1;
 719         spin_unlock(&req->rq_lock);
 720         req->rq_cb_data = NULL;
 721         obd_mod_put(mod);
 722 }
 723
 724 int mdc_set_open_replay_data(struct obd_export *exp,
 725                              struct obd_client_handle *och,
 726                              struct lookup_intent *it)
 727 {
 728         struct md_open_data   *mod;
 729         struct mdt_rec_create *rec;
 730         struct mdt_body       *body;
 731         struct ptlrpc_request *open_req = it->d.lustre.it_data;
 732         struct obd_import     *imp = open_req->rq_import;
 733
 734         if (!open_req->rq_replay)
 735                 return 0;
 736
 737         rec = req_capsule_client_get(&open_req->rq_pill, &RMF_REC_REINT);
 738         body = req_capsule_server_get(&open_req->rq_pill, &RMF_MDT_BODY);
 739         LASSERT(rec != NULL);
 740         /* Incoming message in my byte order (it's been swabbed). */
 741         /* Outgoing messages always in my byte order. */
 742         LASSERT(body != NULL);
 743
 744         /* Only if the import is replayable, we set replay_open data */
 745         if (och && imp->imp_replayable) {
 746                 mod = obd_mod_alloc();
 747                 if (mod == NULL) {
 748                         DEBUG_REQ(D_ERROR, open_req,
 749                                   "Can't allocate md_open_data");
 750                         return 0;
 751                 }
 752
 753                 /**
 754                  * Take a reference on \var mod, to be freed on mdc_close().
 755                  * It protects \var mod from being freed on eviction (commit
 756                  * callback is called despite rq_replay flag).
 757                  * Another reference for \var och.
 758                  */
 759                 obd_mod_get(mod);
 760                 obd_mod_get(mod);
 761
 762                 spin_lock(&open_req->rq_lock);
 763                 och->och_mod = mod;
 764                 mod->mod_och = och;
 765                 mod->mod_is_create = it_disposition(it, DISP_OPEN_CREATE) ||
 766                                      it_disposition(it, DISP_OPEN_STRIPE);
 767                 mod->mod_open_req = open_req;
 768                 open_req->rq_cb_data = mod;
 769                 open_req->rq_commit_cb = mdc_commit_open;
 770                 spin_unlock(&open_req->rq_lock);
 771         }
 772
 773         rec->cr_fid2 = body->fid1;
 774         rec->cr_ioepoch = body->ioepoch;
 775         rec->cr_old_handle.cookie = body->handle.cookie;
 776         open_req->rq_replay_cb = mdc_replay_open;
 777         if (!fid_is_sane(&body->fid1)) {
 778                 DEBUG_REQ(D_ERROR, open_req, "Saving replay request with "
 779                           "insane fid");
 780                 LBUG();
 781         }
 782
 783         DEBUG_REQ(D_RPCTRACE, open_req, "Set up open replay data");
 784         return 0;
 785 }
 786
 787 static void mdc_free_open(struct md_open_data *mod)
 788 {
 789         int committed = 0;
 790
 791         if (mod->mod_is_create == 0 &&
 792             imp_connect_disp_stripe(mod->mod_open_req->rq_import))
 793                 committed = 1;
 794
 795         LASSERT(mod->mod_open_req->rq_replay == 0);
 796
 797         DEBUG_REQ(D_RPCTRACE, mod->mod_open_req, "free open request\n");
 798
 799         ptlrpc_request_committed(mod->mod_open_req, committed);
 800         if (mod->mod_close_req)
 801                 ptlrpc_request_committed(mod->mod_close_req, committed);
 802 }
 803
 804 int mdc_clear_open_replay_data(struct obd_export *exp,
 805                                struct obd_client_handle *och)
 806 {
 807         struct md_open_data *mod = och->och_mod;
 808
 809         /**
 810          * It is possible to not have \var mod in a case of eviction between
 811          * lookup and ll_file_open().
 812          **/
 813         if (mod == NULL)
 814                 return 0;
 815
 816         LASSERT(mod != LP_POISON);
 817         LASSERT(mod->mod_open_req != NULL);
 818         mdc_free_open(mod);
 819
 820         mod->mod_och = NULL;
 821         och->och_mod = NULL;
 822         obd_mod_put(mod);
 823
 824         return 0;
 825 }
 826
 827 /* Prepares the request for the replay by the given reply */
 828 static void mdc_close_handle_reply(struct ptlrpc_request *req,
 829                                    struct md_op_data *op_data, int rc) {
 830         struct mdt_body  *repbody;
 831         struct mdt_ioepoch *epoch;
 832
 833         if (req && rc == -EAGAIN) {
 834                 repbody = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 835                 epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
 836
 837                 epoch->flags |= MF_SOM_AU;
 838                 if (repbody->valid & OBD_MD_FLGETATTRLOCK)
 839                         op_data->op_flags |= MF_GETATTR_LOCK;
 840         }
 841 }
 842
 843 int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
 844               struct md_open_data *mod, struct ptlrpc_request **request)
 845 {
 846         struct obd_device     *obd = class_exp2obd(exp);
 847         struct ptlrpc_request *req;
 848         struct req_format     *req_fmt;
 849         int                    rc;
 850         int                    saved_rc = 0;
 851
 852
 853         req_fmt = &RQF_MDS_CLOSE;
 854         if (op_data->op_bias & MDS_HSM_RELEASE) {
 855                 req_fmt = &RQF_MDS_RELEASE_CLOSE;
 856
 857                 /* allocate a FID for volatile file */
 858                 rc = mdc_fid_alloc(exp, &op_data->op_fid2, op_data);
 859                 if (rc < 0) {
 860                         CERROR("%s: "DFID" failed to allocate FID: %d\n",
 861                                obd->obd_name, PFID(&op_data->op_fid1), rc);
 862                         /* save the errcode and proceed to close */
 863                         saved_rc = rc;
 864                 }
 865         }
 866
 867         *request = NULL;
 868         req = ptlrpc_request_alloc(class_exp2cliimp(exp), req_fmt);
 869         if (req == NULL)
 870                 return -ENOMEM;
 871
 872         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 873
 874         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_CLOSE);
 875         if (rc) {
 876                 ptlrpc_request_free(req);
 877                 return rc;
 878         }
 879
 880         /* To avoid a livelock (bug 7034), we need to send CLOSE RPCs to a
 881          * portal whose threads are not taking any DLM locks and are therefore
 882          * always progressing */
 883         req->rq_request_portal = MDS_READPAGE_PORTAL;
 884         ptlrpc_at_set_req_timeout(req);
 885
 886         /* Ensure that this close's handle is fixed up during replay. */
 887         if (likely(mod != NULL)) {
 888                 LASSERTF(mod->mod_open_req != NULL &&
 889                          mod->mod_open_req->rq_type != LI_POISON,
 890                          "POISONED open %p!\n", mod->mod_open_req);
 891
 892                 mod->mod_close_req = req;
 893
 894                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched open");
 895                 /* We no longer want to preserve this open for replay even
 896                  * though the open was committed. b=3632, b=3633 */
 897                 spin_lock(&mod->mod_open_req->rq_lock);
 898                 mod->mod_open_req->rq_replay = 0;
 899                 spin_unlock(&mod->mod_open_req->rq_lock);
 900         } else {
 901                  CDEBUG(D_HA,
 902                         "couldn't find open req; expecting close error\n");
 903         }
 904
 905         mdc_close_pack(req, op_data);
 906
 907         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
 908                              obd->u.cli.cl_default_mds_easize);
 909         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
 910                              obd->u.cli.cl_default_mds_cookiesize);
 911
 912         ptlrpc_request_set_replen(req);
 913
 914         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 915         rc = ptlrpc_queue_wait(req);
 916         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 917
 918         if (req->rq_repmsg == NULL) {
 919                 CDEBUG(D_RPCTRACE, "request failed to send: %p, %d\n", req,
 920                        req->rq_status);
 921                 if (rc == 0)
 922                         rc = req->rq_status ?: -EIO;
 923         } else if (rc == 0 || rc == -EAGAIN) {
 924                 struct mdt_body *body;
 925
 926                 rc = lustre_msg_get_status(req->rq_repmsg);
 927                 if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 928                         DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR, err "
 929                                   "= %d", rc);
 930                         if (rc > 0)
 931                                 rc = -rc;
 932                 }
 933                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 934                 if (body == NULL)
 935                         rc = -EPROTO;
 936         } else if (rc == -ESTALE) {
 937                 /**
 938                  * it can be allowed error after 3633 if open was committed and
 939                  * server failed before close was sent. Let's check if mod
 940                  * exists and return no error in that case
 941                  */
 942                 if (mod) {
 943                         DEBUG_REQ(D_HA, req, "Reset ESTALE = %d", rc);
 944                         LASSERT(mod->mod_open_req != NULL);
 945                         if (mod->mod_open_req->rq_committed)
 946                                 rc = 0;
 947                 }
 948         }
 949
 950         if (mod) {
 951                 if (rc != 0)
 952                         mod->mod_close_req = NULL;
 953                 /* Since now, mod is accessed through open_req only,
 954                  * thus close req does not keep a reference on mod anymore. */
 955                 obd_mod_put(mod);
 956         }
 957         *request = req;
 958         mdc_close_handle_reply(req, op_data, rc);
 959         return rc < 0 ? rc : saved_rc;
 960 }
 961
 962 int mdc_done_writing(struct obd_export *exp, struct md_op_data *op_data,
 963                      struct md_open_data *mod)
 964 {
 965         struct obd_device     *obd = class_exp2obd(exp);
 966         struct ptlrpc_request *req;
 967         int                 rc;
 968
 969         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 970                                    &RQF_MDS_DONE_WRITING);
 971         if (req == NULL)
 972                 return -ENOMEM;
 973
 974         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 975         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_DONE_WRITING);
 976         if (rc) {
 977                 ptlrpc_request_free(req);
 978                 return rc;
 979         }
 980
 981         if (mod != NULL) {
 982                 LASSERTF(mod->mod_open_req != NULL &&
 983                          mod->mod_open_req->rq_type != LI_POISON,
 984                          "POISONED setattr %p!\n", mod->mod_open_req);
 985
 986                 mod->mod_close_req = req;
 987                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched setattr");
 988                 /* We no longer want to preserve this setattr for replay even
 989                  * though the open was committed. b=3632, b=3633 */
 990                 spin_lock(&mod->mod_open_req->rq_lock);
 991                 mod->mod_open_req->rq_replay = 0;
 992                 spin_unlock(&mod->mod_open_req->rq_lock);
 993         }
 994
 995         mdc_close_pack(req, op_data);
 996         ptlrpc_request_set_replen(req);
 997
 998         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 999         rc = ptlrpc_queue_wait(req);
1000         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
1001
1002         if (rc == -ESTALE) {
1003                 /**
1004                  * it can be allowed error after 3633 if open or setattr were
1005                  * committed and server failed before close was sent.
1006                  * Let's check if mod exists and return no error in that case
1007                  */
1008                 if (mod) {
1009                         LASSERT(mod->mod_open_req != NULL);
1010                         if (mod->mod_open_req->rq_committed)
1011                                 rc = 0;
1012                 }
1013         }
1014
1015         if (mod) {
1016                 if (rc != 0)
1017                         mod->mod_close_req = NULL;
1018                 LASSERT(mod->mod_open_req != NULL);
1019                 mdc_free_open(mod);
1020
1021                 /* Since now, mod is accessed through setattr req only,
1022                  * thus DW req does not keep a reference on mod anymore. */
1023                 obd_mod_put(mod);
1024         }
1025
1026         mdc_close_handle_reply(req, op_data, rc);
1027         ptlrpc_req_finished(req);
1028         return rc;
1029 }
1030
1031
1032 int mdc_readpage(struct obd_export *exp, struct md_op_data *op_data,
1033                  struct page **pages, 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
1043         *request = NULL;
1044         init_waitqueue_head(&waitq);
1045
1046 restart_bulk:
1047         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_READPAGE);
1048         if (req == NULL)
1049                 return -ENOMEM;
1050
1051         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1052
1053         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_READPAGE);
1054         if (rc) {
1055                 ptlrpc_request_free(req);
1056                 return rc;
1057         }
1058
1059         req->rq_request_portal = MDS_READPAGE_PORTAL;
1060         ptlrpc_at_set_req_timeout(req);
1061
1062         desc = ptlrpc_prep_bulk_imp(req, op_data->op_npages, 1, BULK_PUT_SINK,
1063                                     MDS_BULK_PORTAL);
1064         if (desc == NULL) {
1065                 ptlrpc_request_free(req);
1066                 return -ENOMEM;
1067         }
1068
1069         /* NB req now owns desc and will free it when it gets freed */
1070         for (i = 0; i < op_data->op_npages; i++)
1071                 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1072
1073         mdc_readdir_pack(req, op_data->op_offset,
1074                          PAGE_CACHE_SIZE * op_data->op_npages,
1075                          &op_data->op_fid1, op_data->op_capa1);
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("too many resend retries, returning error\n");
1087                         return -EIO;
1088                 }
1089                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends),
1090                                        NULL, NULL, NULL);
1091                 l_wait_event(waitq, 0, &lwi);
1092
1093                 goto restart_bulk;
1094         }
1095
1096         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1097                                           req->rq_bulk->bd_nob_transferred);
1098         if (rc < 0) {
1099                 ptlrpc_req_finished(req);
1100                 return rc;
1101         }
1102
1103         if (req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK) {
1104                 CERROR("Unexpected # bytes transferred: %d (%ld expected)\n",
1105                         req->rq_bulk->bd_nob_transferred,
1106                         PAGE_CACHE_SIZE * op_data->op_npages);
1107                 ptlrpc_req_finished(req);
1108                 return -EPROTO;
1109         }
1110
1111         *request = req;
1112         return 0;
1113 }
1114
1115 static int mdc_statfs(const struct lu_env *env,
1116                       struct obd_export *exp, struct obd_statfs *osfs,
1117                       __u64 max_age, __u32 flags)
1118 {
1119         struct obd_device     *obd = class_exp2obd(exp);
1120         struct ptlrpc_request *req;
1121         struct obd_statfs     *msfs;
1122         struct obd_import     *imp = NULL;
1123         int                 rc;
1124
1125         /*
1126          * Since the request might also come from lprocfs, so we need
1127          * sync this with client_disconnect_export Bug15684
1128          */
1129         down_read(&obd->u.cli.cl_sem);
1130         if (obd->u.cli.cl_import)
1131                 imp = class_import_get(obd->u.cli.cl_import);
1132         up_read(&obd->u.cli.cl_sem);
1133         if (!imp)
1134                 return -ENODEV;
1135
1136         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_STATFS,
1137                                         LUSTRE_MDS_VERSION, MDS_STATFS);
1138         if (req == NULL)
1139                 GOTO(output, rc = -ENOMEM);
1140
1141         ptlrpc_request_set_replen(req);
1142
1143         if (flags & OBD_STATFS_NODELAY) {
1144                 /* procfs requests not want stay in wait for avoid deadlock */
1145                 req->rq_no_resend = 1;
1146                 req->rq_no_delay = 1;
1147         }
1148
1149         rc = ptlrpc_queue_wait(req);
1150         if (rc) {
1151                 /* check connection error first */
1152                 if (imp->imp_connect_error)
1153                         rc = imp->imp_connect_error;
1154                 GOTO(out, rc);
1155         }
1156
1157         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1158         if (msfs == NULL)
1159                 GOTO(out, rc = -EPROTO);
1160
1161         *osfs = *msfs;
1162 out:
1163         ptlrpc_req_finished(req);
1164 output:
1165         class_import_put(imp);
1166         return rc;
1167 }
1168
1169 static int mdc_ioc_fid2path(struct obd_export *exp, struct getinfo_fid2path *gf)
1170 {
1171         __u32 keylen, vallen;
1172         void *key;
1173         int rc;
1174
1175         if (gf->gf_pathlen > PATH_MAX)
1176                 return -ENAMETOOLONG;
1177         if (gf->gf_pathlen < 2)
1178                 return -EOVERFLOW;
1179
1180         /* Key is KEY_FID2PATH + getinfo_fid2path description */
1181         keylen = cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*gf);
1182         OBD_ALLOC(key, keylen);
1183         if (key == NULL)
1184                 return -ENOMEM;
1185         memcpy(key, KEY_FID2PATH, sizeof(KEY_FID2PATH));
1186         memcpy(key + cfs_size_round(sizeof(KEY_FID2PATH)), gf, sizeof(*gf));
1187
1188         CDEBUG(D_IOCTL, "path get "DFID" from %llu #%d\n",
1189                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno);
1190
1191         if (!fid_is_sane(&gf->gf_fid))
1192                 GOTO(out, rc = -EINVAL);
1193
1194         /* Val is struct getinfo_fid2path result plus path */
1195         vallen = sizeof(*gf) + gf->gf_pathlen;
1196
1197         rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf, NULL);
1198         if (rc != 0 && rc != -EREMOTE)
1199                 GOTO(out, rc);
1200
1201         if (vallen <= sizeof(*gf))
1202                 GOTO(out, rc = -EPROTO);
1203         else if (vallen > sizeof(*gf) + gf->gf_pathlen)
1204                 GOTO(out, rc = -EOVERFLOW);
1205
1206         CDEBUG(D_IOCTL, "path get "DFID" from %llu #%d\n%s\n",
1207                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno, gf->gf_path);
1208
1209 out:
1210         OBD_FREE(key, keylen);
1211         return rc;
1212 }
1213
1214 static int mdc_ioc_hsm_progress(struct obd_export *exp,
1215                                 struct hsm_progress_kernel *hpk)
1216 {
1217         struct obd_import               *imp = class_exp2cliimp(exp);
1218         struct hsm_progress_kernel      *req_hpk;
1219         struct ptlrpc_request           *req;
1220         int                              rc;
1221
1222         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_PROGRESS,
1223                                         LUSTRE_MDS_VERSION, MDS_HSM_PROGRESS);
1224         if (req == NULL)
1225                 GOTO(out, rc = -ENOMEM);
1226
1227         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1228
1229         /* Copy hsm_progress struct */
1230         req_hpk = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_PROGRESS);
1231         if (req_hpk == NULL)
1232                 GOTO(out, rc = -EPROTO);
1233
1234         *req_hpk = *hpk;
1235         req_hpk->hpk_errval = lustre_errno_hton(hpk->hpk_errval);
1236
1237         ptlrpc_request_set_replen(req);
1238
1239         rc = mdc_queue_wait(req);
1240         GOTO(out, rc);
1241 out:
1242         ptlrpc_req_finished(req);
1243         return rc;
1244 }
1245
1246 static int mdc_ioc_hsm_ct_register(struct obd_import *imp, __u32 archives)
1247 {
1248         __u32                   *archive_mask;
1249         struct ptlrpc_request   *req;
1250         int                      rc;
1251
1252         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_REGISTER,
1253                                         LUSTRE_MDS_VERSION,
1254                                         MDS_HSM_CT_REGISTER);
1255         if (req == NULL)
1256                 GOTO(out, rc = -ENOMEM);
1257
1258         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1259
1260         /* Copy hsm_progress struct */
1261         archive_mask = req_capsule_client_get(&req->rq_pill,
1262                                               &RMF_MDS_HSM_ARCHIVE);
1263         if (archive_mask == NULL)
1264                 GOTO(out, rc = -EPROTO);
1265
1266         *archive_mask = archives;
1267
1268         ptlrpc_request_set_replen(req);
1269
1270         rc = mdc_queue_wait(req);
1271         GOTO(out, rc);
1272 out:
1273         ptlrpc_req_finished(req);
1274         return rc;
1275 }
1276
1277 static int mdc_ioc_hsm_current_action(struct obd_export *exp,
1278                                       struct md_op_data *op_data)
1279 {
1280         struct hsm_current_action       *hca = op_data->op_data;
1281         struct hsm_current_action       *req_hca;
1282         struct ptlrpc_request           *req;
1283         int                              rc;
1284
1285         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1286                                    &RQF_MDS_HSM_ACTION);
1287         if (req == NULL)
1288                 return -ENOMEM;
1289
1290         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1291
1292         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_ACTION);
1293         if (rc) {
1294                 ptlrpc_request_free(req);
1295                 return rc;
1296         }
1297
1298         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1299                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1300
1301         ptlrpc_request_set_replen(req);
1302
1303         rc = mdc_queue_wait(req);
1304         if (rc)
1305                 GOTO(out, rc);
1306
1307         req_hca = req_capsule_server_get(&req->rq_pill,
1308                                          &RMF_MDS_HSM_CURRENT_ACTION);
1309         if (req_hca == NULL)
1310                 GOTO(out, rc = -EPROTO);
1311
1312         *hca = *req_hca;
1313
1314 out:
1315         ptlrpc_req_finished(req);
1316         return rc;
1317 }
1318
1319 static int mdc_ioc_hsm_ct_unregister(struct obd_import *imp)
1320 {
1321         struct ptlrpc_request   *req;
1322         int                      rc;
1323
1324         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_UNREGISTER,
1325                                         LUSTRE_MDS_VERSION,
1326                                         MDS_HSM_CT_UNREGISTER);
1327         if (req == NULL)
1328                 GOTO(out, rc = -ENOMEM);
1329
1330         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1331
1332         ptlrpc_request_set_replen(req);
1333
1334         rc = mdc_queue_wait(req);
1335         GOTO(out, rc);
1336 out:
1337         ptlrpc_req_finished(req);
1338         return rc;
1339 }
1340
1341 static int mdc_ioc_hsm_state_get(struct obd_export *exp,
1342                                  struct md_op_data *op_data)
1343 {
1344         struct hsm_user_state   *hus = op_data->op_data;
1345         struct hsm_user_state   *req_hus;
1346         struct ptlrpc_request   *req;
1347         int                      rc;
1348
1349         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1350                                    &RQF_MDS_HSM_STATE_GET);
1351         if (req == NULL)
1352                 return -ENOMEM;
1353
1354         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1355
1356         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_GET);
1357         if (rc != 0) {
1358                 ptlrpc_request_free(req);
1359                 return rc;
1360         }
1361
1362         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1363                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1364
1365         ptlrpc_request_set_replen(req);
1366
1367         rc = mdc_queue_wait(req);
1368         if (rc)
1369                 GOTO(out, rc);
1370
1371         req_hus = req_capsule_server_get(&req->rq_pill, &RMF_HSM_USER_STATE);
1372         if (req_hus == NULL)
1373                 GOTO(out, rc = -EPROTO);
1374
1375         *hus = *req_hus;
1376
1377 out:
1378         ptlrpc_req_finished(req);
1379         return rc;
1380 }
1381
1382 static int mdc_ioc_hsm_state_set(struct obd_export *exp,
1383                                  struct md_op_data *op_data)
1384 {
1385         struct hsm_state_set    *hss = op_data->op_data;
1386         struct hsm_state_set    *req_hss;
1387         struct ptlrpc_request   *req;
1388         int                      rc;
1389
1390         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1391                                    &RQF_MDS_HSM_STATE_SET);
1392         if (req == NULL)
1393                 return -ENOMEM;
1394
1395         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1396
1397         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_SET);
1398         if (rc) {
1399                 ptlrpc_request_free(req);
1400                 return rc;
1401         }
1402
1403         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1404                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1405
1406         /* Copy states */
1407         req_hss = req_capsule_client_get(&req->rq_pill, &RMF_HSM_STATE_SET);
1408         if (req_hss == NULL)
1409                 GOTO(out, rc = -EPROTO);
1410         *req_hss = *hss;
1411
1412         ptlrpc_request_set_replen(req);
1413
1414         rc = mdc_queue_wait(req);
1415         GOTO(out, rc);
1416
1417 out:
1418         ptlrpc_req_finished(req);
1419         return rc;
1420 }
1421
1422 static int mdc_ioc_hsm_request(struct obd_export *exp,
1423                                struct hsm_user_request *hur)
1424 {
1425         struct obd_import       *imp = class_exp2cliimp(exp);
1426         struct ptlrpc_request   *req;
1427         struct hsm_request      *req_hr;
1428         struct hsm_user_item    *req_hui;
1429         char                    *req_opaque;
1430         int                      rc;
1431
1432         req = ptlrpc_request_alloc(imp, &RQF_MDS_HSM_REQUEST);
1433         if (req == NULL)
1434                 GOTO(out, rc = -ENOMEM);
1435
1436         req_capsule_set_size(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM, RCL_CLIENT,
1437                              hur->hur_request.hr_itemcount
1438                              * sizeof(struct hsm_user_item));
1439         req_capsule_set_size(&req->rq_pill, &RMF_GENERIC_DATA, RCL_CLIENT,
1440                              hur->hur_request.hr_data_len);
1441
1442         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_REQUEST);
1443         if (rc) {
1444                 ptlrpc_request_free(req);
1445                 return rc;
1446         }
1447
1448         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1449
1450         /* Copy hsm_request struct */
1451         req_hr = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_REQUEST);
1452         if (req_hr == NULL)
1453                 GOTO(out, rc = -EPROTO);
1454         *req_hr = hur->hur_request;
1455
1456         /* Copy hsm_user_item structs */
1457         req_hui = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM);
1458         if (req_hui == NULL)
1459                 GOTO(out, rc = -EPROTO);
1460         memcpy(req_hui, hur->hur_user_item,
1461                hur->hur_request.hr_itemcount * sizeof(struct hsm_user_item));
1462
1463         /* Copy opaque field */
1464         req_opaque = req_capsule_client_get(&req->rq_pill, &RMF_GENERIC_DATA);
1465         if (req_opaque == NULL)
1466                 GOTO(out, rc = -EPROTO);
1467         memcpy(req_opaque, hur_data(hur), hur->hur_request.hr_data_len);
1468
1469         ptlrpc_request_set_replen(req);
1470
1471         rc = mdc_queue_wait(req);
1472         GOTO(out, rc);
1473
1474 out:
1475         ptlrpc_req_finished(req);
1476         return rc;
1477 }
1478
1479 static struct kuc_hdr *changelog_kuc_hdr(char *buf, int len, int flags)
1480 {
1481         struct kuc_hdr *lh = (struct kuc_hdr *)buf;
1482
1483         LASSERT(len <= KUC_CHANGELOG_MSG_MAXSIZE);
1484
1485         lh->kuc_magic = KUC_MAGIC;
1486         lh->kuc_transport = KUC_TRANSPORT_CHANGELOG;
1487         lh->kuc_flags = flags;
1488         lh->kuc_msgtype = CL_RECORD;
1489         lh->kuc_msglen = len;
1490         return lh;
1491 }
1492
1493 #define D_CHANGELOG 0
1494
1495 struct changelog_show {
1496         __u64           cs_startrec;
1497         __u32           cs_flags;
1498         struct file     *cs_fp;
1499         char            *cs_buf;
1500         struct obd_device *cs_obd;
1501 };
1502
1503 static int changelog_kkuc_cb(const struct lu_env *env, struct llog_handle *llh,
1504                              struct llog_rec_hdr *hdr, void *data)
1505 {
1506         struct changelog_show *cs = data;
1507         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
1508         struct kuc_hdr *lh;
1509         int len, rc;
1510
1511         if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
1512                 rc = -EINVAL;
1513                 CERROR("%s: not a changelog rec %x/%d: rc = %d\n",
1514                        cs->cs_obd->obd_name, rec->cr_hdr.lrh_type,
1515                        rec->cr.cr_type, rc);
1516                 return rc;
1517         }
1518
1519         if (rec->cr.cr_index < cs->cs_startrec) {
1520                 /* Skip entries earlier than what we are interested in */
1521                 CDEBUG(D_CHANGELOG, "rec=%llu start=%llu\n",
1522                        rec->cr.cr_index, cs->cs_startrec);
1523                 return 0;
1524         }
1525
1526         CDEBUG(D_CHANGELOG, "%llu %02d%-5s %llu 0x%x t="DFID" p="DFID
1527                 " %.*s\n", rec->cr.cr_index, rec->cr.cr_type,
1528                 changelog_type2str(rec->cr.cr_type), rec->cr.cr_time,
1529                 rec->cr.cr_flags & CLF_FLAGMASK,
1530                 PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
1531                 rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
1532
1533         len = sizeof(*lh) + changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
1534
1535         /* Set up the message */
1536         lh = changelog_kuc_hdr(cs->cs_buf, len, cs->cs_flags);
1537         memcpy(lh + 1, &rec->cr, len - sizeof(*lh));
1538
1539         rc = libcfs_kkuc_msg_put(cs->cs_fp, lh);
1540         CDEBUG(D_CHANGELOG, "kucmsg fp %p len %d rc %d\n", cs->cs_fp, len, rc);
1541
1542         return rc;
1543 }
1544
1545 static int mdc_changelog_send_thread(void *csdata)
1546 {
1547         struct changelog_show *cs = csdata;
1548         struct llog_ctxt *ctxt = NULL;
1549         struct llog_handle *llh = NULL;
1550         struct kuc_hdr *kuch;
1551         int rc;
1552
1553         CDEBUG(D_CHANGELOG, "changelog to fp=%p start %llu\n",
1554                cs->cs_fp, cs->cs_startrec);
1555
1556         OBD_ALLOC(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
1557         if (cs->cs_buf == NULL)
1558                 GOTO(out, rc = -ENOMEM);
1559
1560         /* Set up the remote catalog handle */
1561         ctxt = llog_get_context(cs->cs_obd, LLOG_CHANGELOG_REPL_CTXT);
1562         if (ctxt == NULL)
1563                 GOTO(out, rc = -ENOENT);
1564         rc = llog_open(NULL, ctxt, &llh, NULL, CHANGELOG_CATALOG,
1565                        LLOG_OPEN_EXISTS);
1566         if (rc) {
1567                 CERROR("%s: fail to open changelog catalog: rc = %d\n",
1568                        cs->cs_obd->obd_name, rc);
1569                 GOTO(out, rc);
1570         }
1571         rc = llog_init_handle(NULL, llh, LLOG_F_IS_CAT, NULL);
1572         if (rc) {
1573                 CERROR("llog_init_handle failed %d\n", rc);
1574                 GOTO(out, rc);
1575         }
1576
1577         rc = llog_cat_process(NULL, llh, changelog_kkuc_cb, cs, 0, 0);
1578
1579         /* Send EOF no matter what our result */
1580         kuch = changelog_kuc_hdr(cs->cs_buf, sizeof(*kuch), cs->cs_flags);
1581         if (kuch) {
1582                 kuch->kuc_msgtype = CL_EOF;
1583                 libcfs_kkuc_msg_put(cs->cs_fp, kuch);
1584         }
1585
1586 out:
1587         fput(cs->cs_fp);
1588         if (llh)
1589                 llog_cat_close(NULL, llh);
1590         if (ctxt)
1591                 llog_ctxt_put(ctxt);
1592         if (cs->cs_buf)
1593                 OBD_FREE(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
1594         OBD_FREE_PTR(cs);
1595         return rc;
1596 }
1597
1598 static int mdc_ioc_changelog_send(struct obd_device *obd,
1599                                   struct ioc_changelog *icc)
1600 {
1601         struct changelog_show *cs;
1602         int rc;
1603
1604         /* Freed in mdc_changelog_send_thread */
1605         OBD_ALLOC_PTR(cs);
1606         if (!cs)
1607                 return -ENOMEM;
1608
1609         cs->cs_obd = obd;
1610         cs->cs_startrec = icc->icc_recno;
1611         /* matching fput in mdc_changelog_send_thread */
1612         cs->cs_fp = fget(icc->icc_id);
1613         cs->cs_flags = icc->icc_flags;
1614
1615         /*
1616          * New thread because we should return to user app before
1617          * writing into our pipe
1618          */
1619         rc = PTR_ERR(kthread_run(mdc_changelog_send_thread, cs,
1620                                  "mdc_clg_send_thread"));
1621         if (!IS_ERR_VALUE(rc)) {
1622                 CDEBUG(D_CHANGELOG, "start changelog thread\n");
1623                 return 0;
1624         }
1625
1626         CERROR("Failed to start changelog thread: %d\n", rc);
1627         OBD_FREE_PTR(cs);
1628         return rc;
1629 }
1630
1631 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
1632                                 struct lustre_kernelcomm *lk);
1633
1634 static int mdc_quotacheck(struct obd_device *unused, struct obd_export *exp,
1635                           struct obd_quotactl *oqctl)
1636 {
1637         struct client_obd       *cli = &exp->exp_obd->u.cli;
1638         struct ptlrpc_request   *req;
1639         struct obd_quotactl     *body;
1640         int                   rc;
1641
1642         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1643                                         &RQF_MDS_QUOTACHECK, LUSTRE_MDS_VERSION,
1644                                         MDS_QUOTACHECK);
1645         if (req == NULL)
1646                 return -ENOMEM;
1647
1648         body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1649         *body = *oqctl;
1650
1651         ptlrpc_request_set_replen(req);
1652
1653         /* the next poll will find -ENODATA, that means quotacheck is
1654          * going on */
1655         cli->cl_qchk_stat = -ENODATA;
1656         rc = ptlrpc_queue_wait(req);
1657         if (rc)
1658                 cli->cl_qchk_stat = rc;
1659         ptlrpc_req_finished(req);
1660         return rc;
1661 }
1662
1663 static int mdc_quota_poll_check(struct obd_export *exp,
1664                                 struct if_quotacheck *qchk)
1665 {
1666         struct client_obd *cli = &exp->exp_obd->u.cli;
1667         int rc;
1668
1669         qchk->obd_uuid = cli->cl_target_uuid;
1670         memcpy(qchk->obd_type, LUSTRE_MDS_NAME, strlen(LUSTRE_MDS_NAME));
1671
1672         rc = cli->cl_qchk_stat;
1673         /* the client is not the previous one */
1674         if (rc == CL_NOT_QUOTACHECKED)
1675                 rc = -EINTR;
1676         return rc;
1677 }
1678
1679 static int mdc_quotactl(struct obd_device *unused, struct obd_export *exp,
1680                         struct obd_quotactl *oqctl)
1681 {
1682         struct ptlrpc_request   *req;
1683         struct obd_quotactl     *oqc;
1684         int                   rc;
1685
1686         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1687                                         &RQF_MDS_QUOTACTL, LUSTRE_MDS_VERSION,
1688                                         MDS_QUOTACTL);
1689         if (req == NULL)
1690                 return -ENOMEM;
1691
1692         oqc = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1693         *oqc = *oqctl;
1694
1695         ptlrpc_request_set_replen(req);
1696         ptlrpc_at_set_req_timeout(req);
1697         req->rq_no_resend = 1;
1698
1699         rc = ptlrpc_queue_wait(req);
1700         if (rc)
1701                 CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc);
1702
1703         if (req->rq_repmsg) {
1704                 oqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1705                 if (oqc) {
1706                         *oqctl = *oqc;
1707                 } else if (!rc) {
1708                         CERROR("Can't unpack obd_quotactl\n");
1709                         rc = -EPROTO;
1710                 }
1711         } else if (!rc) {
1712                 CERROR("Can't unpack obd_quotactl\n");
1713                 rc = -EPROTO;
1714         }
1715         ptlrpc_req_finished(req);
1716
1717         return rc;
1718 }
1719
1720 static int mdc_ioc_swap_layouts(struct obd_export *exp,
1721                                 struct md_op_data *op_data)
1722 {
1723         LIST_HEAD(cancels);
1724         struct ptlrpc_request   *req;
1725         int                      rc, count;
1726         struct mdc_swap_layouts *msl, *payload;
1727
1728         msl = op_data->op_data;
1729
1730         /* When the MDT will get the MDS_SWAP_LAYOUTS RPC the
1731          * first thing it will do is to cancel the 2 layout
1732          * locks hold by this client.
1733          * So the client must cancel its layout locks on the 2 fids
1734          * with the request RPC to avoid extra RPC round trips
1735          */
1736         count = mdc_resource_get_unused(exp, &op_data->op_fid1, &cancels,
1737                                         LCK_CR, MDS_INODELOCK_LAYOUT);
1738         count += mdc_resource_get_unused(exp, &op_data->op_fid2, &cancels,
1739                                          LCK_CR, MDS_INODELOCK_LAYOUT);
1740
1741         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1742                                    &RQF_MDS_SWAP_LAYOUTS);
1743         if (req == NULL) {
1744                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
1745                 return -ENOMEM;
1746         }
1747
1748         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1749         mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa2);
1750
1751         rc = mdc_prep_elc_req(exp, req, MDS_SWAP_LAYOUTS, &cancels, count);
1752         if (rc) {
1753                 ptlrpc_request_free(req);
1754                 return rc;
1755         }
1756
1757         mdc_swap_layouts_pack(req, op_data);
1758
1759         payload = req_capsule_client_get(&req->rq_pill, &RMF_SWAP_LAYOUTS);
1760         LASSERT(payload);
1761
1762         *payload = *msl;
1763
1764         ptlrpc_request_set_replen(req);
1765
1766         rc = ptlrpc_queue_wait(req);
1767         if (rc)
1768                 GOTO(out, rc);
1769
1770 out:
1771         ptlrpc_req_finished(req);
1772         return rc;
1773 }
1774
1775 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1776                          void *karg, void *uarg)
1777 {
1778         struct obd_device *obd = exp->exp_obd;
1779         struct obd_ioctl_data *data = karg;
1780         struct obd_import *imp = obd->u.cli.cl_import;
1781         struct llog_ctxt *ctxt;
1782         int rc;
1783
1784         if (!try_module_get(THIS_MODULE)) {
1785                 CERROR("Can't get module. Is it alive?");
1786                 return -EINVAL;
1787         }
1788         switch (cmd) {
1789         case OBD_IOC_CHANGELOG_SEND:
1790                 rc = mdc_ioc_changelog_send(obd, karg);
1791                 GOTO(out, rc);
1792         case OBD_IOC_CHANGELOG_CLEAR: {
1793                 struct ioc_changelog *icc = karg;
1794                 struct changelog_setinfo cs =
1795                         {.cs_recno = icc->icc_recno, .cs_id = icc->icc_id};
1796                 rc = obd_set_info_async(NULL, exp, strlen(KEY_CHANGELOG_CLEAR),
1797                                         KEY_CHANGELOG_CLEAR, sizeof(cs), &cs,
1798                                         NULL);
1799                 GOTO(out, rc);
1800         }
1801         case OBD_IOC_FID2PATH:
1802                 rc = mdc_ioc_fid2path(exp, karg);
1803                 GOTO(out, rc);
1804         case LL_IOC_HSM_CT_START:
1805                 rc = mdc_ioc_hsm_ct_start(exp, karg);
1806                 /* ignore if it was already registered on this MDS. */
1807                 if (rc == -EEXIST)
1808                         rc = 0;
1809                 GOTO(out, rc);
1810         case LL_IOC_HSM_PROGRESS:
1811                 rc = mdc_ioc_hsm_progress(exp, karg);
1812                 GOTO(out, rc);
1813         case LL_IOC_HSM_STATE_GET:
1814                 rc = mdc_ioc_hsm_state_get(exp, karg);
1815                 GOTO(out, rc);
1816         case LL_IOC_HSM_STATE_SET:
1817                 rc = mdc_ioc_hsm_state_set(exp, karg);
1818                 GOTO(out, rc);
1819         case LL_IOC_HSM_ACTION:
1820                 rc = mdc_ioc_hsm_current_action(exp, karg);
1821                 GOTO(out, rc);
1822         case LL_IOC_HSM_REQUEST:
1823                 rc = mdc_ioc_hsm_request(exp, karg);
1824                 GOTO(out, rc);
1825         case OBD_IOC_CLIENT_RECOVER:
1826                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1, 0);
1827                 if (rc < 0)
1828                         GOTO(out, rc);
1829                 GOTO(out, rc = 0);
1830         case IOC_OSC_SET_ACTIVE:
1831                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
1832                 GOTO(out, rc);
1833         case OBD_IOC_PARSE: {
1834                 ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT);
1835                 rc = class_config_parse_llog(NULL, ctxt, data->ioc_inlbuf1,
1836                                              NULL);
1837                 llog_ctxt_put(ctxt);
1838                 GOTO(out, rc);
1839         }
1840         case OBD_IOC_LLOG_INFO:
1841         case OBD_IOC_LLOG_PRINT: {
1842                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
1843                 rc = llog_ioctl(NULL, ctxt, cmd, data);
1844                 llog_ctxt_put(ctxt);
1845                 GOTO(out, rc);
1846         }
1847         case OBD_IOC_POLL_QUOTACHECK:
1848                 rc = mdc_quota_poll_check(exp, (struct if_quotacheck *)karg);
1849                 GOTO(out, rc);
1850         case OBD_IOC_PING_TARGET:
1851                 rc = ptlrpc_obd_ping(obd);
1852                 GOTO(out, rc);
1853         /*
1854          * Normally IOC_OBD_STATFS, OBD_IOC_QUOTACTL iocontrol are handled by
1855          * LMV instead of MDC. But when the cluster is upgraded from 1.8,
1856          * there'd be no LMV layer thus we might be called here. Eventually
1857          * this code should be removed.
1858          * bz20731, LU-592.
1859          */
1860         case IOC_OBD_STATFS: {
1861                 struct obd_statfs stat_buf = {0};
1862
1863                 if (*((__u32 *) data->ioc_inlbuf2) != 0)
1864                         GOTO(out, rc = -ENODEV);
1865
1866                 /* copy UUID */
1867                 if (copy_to_user(data->ioc_pbuf2, obd2cli_tgt(obd),
1868                                      min((int) data->ioc_plen2,
1869                                          (int) sizeof(struct obd_uuid))))
1870                         GOTO(out, rc = -EFAULT);
1871
1872                 rc = mdc_statfs(NULL, obd->obd_self_export, &stat_buf,
1873                                 cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS),
1874                                 0);
1875                 if (rc != 0)
1876                         GOTO(out, rc);
1877
1878                 if (copy_to_user(data->ioc_pbuf1, &stat_buf,
1879                                      min((int) data->ioc_plen1,
1880                                          (int) sizeof(stat_buf))))
1881                         GOTO(out, rc = -EFAULT);
1882
1883                 GOTO(out, rc = 0);
1884         }
1885         case OBD_IOC_QUOTACTL: {
1886                 struct if_quotactl *qctl = karg;
1887                 struct obd_quotactl *oqctl;
1888
1889                 OBD_ALLOC_PTR(oqctl);
1890                 if (oqctl == NULL)
1891                         GOTO(out, rc = -ENOMEM);
1892
1893                 QCTL_COPY(oqctl, qctl);
1894                 rc = obd_quotactl(exp, oqctl);
1895                 if (rc == 0) {
1896                         QCTL_COPY(qctl, oqctl);
1897                         qctl->qc_valid = QC_MDTIDX;
1898                         qctl->obd_uuid = obd->u.cli.cl_target_uuid;
1899                 }
1900
1901                 OBD_FREE_PTR(oqctl);
1902                 GOTO(out, rc);
1903         }
1904         case LL_IOC_GET_CONNECT_FLAGS:
1905                 if (copy_to_user(uarg, exp_connect_flags_ptr(exp),
1906                                  sizeof(*exp_connect_flags_ptr(exp))))
1907                         GOTO(out, rc = -EFAULT);
1908
1909                 GOTO(out, rc = 0);
1910         case LL_IOC_LOV_SWAP_LAYOUTS:
1911                 rc = mdc_ioc_swap_layouts(exp, karg);
1912                 GOTO(out, rc);
1913         default:
1914                 CERROR("unrecognised ioctl: cmd = %#x\n", cmd);
1915                 GOTO(out, rc = -ENOTTY);
1916         }
1917 out:
1918         module_put(THIS_MODULE);
1919
1920         return rc;
1921 }
1922
1923 int mdc_get_info_rpc(struct obd_export *exp,
1924                      obd_count keylen, void *key,
1925                      int vallen, void *val)
1926 {
1927         struct obd_import      *imp = class_exp2cliimp(exp);
1928         struct ptlrpc_request  *req;
1929         char               *tmp;
1930         int                  rc = -EINVAL;
1931
1932         req = ptlrpc_request_alloc(imp, &RQF_MDS_GET_INFO);
1933         if (req == NULL)
1934                 return -ENOMEM;
1935
1936         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_KEY,
1937                              RCL_CLIENT, keylen);
1938         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VALLEN,
1939                              RCL_CLIENT, sizeof(__u32));
1940
1941         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_INFO);
1942         if (rc) {
1943                 ptlrpc_request_free(req);
1944                 return rc;
1945         }
1946
1947         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_KEY);
1948         memcpy(tmp, key, keylen);
1949         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_VALLEN);
1950         memcpy(tmp, &vallen, sizeof(__u32));
1951
1952         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VAL,
1953                              RCL_SERVER, vallen);
1954         ptlrpc_request_set_replen(req);
1955
1956         rc = ptlrpc_queue_wait(req);
1957         /* -EREMOTE means the get_info result is partial, and it needs to
1958          * continue on another MDT, see fid2path part in lmv_iocontrol */
1959         if (rc == 0 || rc == -EREMOTE) {
1960                 tmp = req_capsule_server_get(&req->rq_pill, &RMF_GETINFO_VAL);
1961                 memcpy(val, tmp, vallen);
1962                 if (ptlrpc_rep_need_swab(req)) {
1963                         if (KEY_IS(KEY_FID2PATH))
1964                                 lustre_swab_fid2path(val);
1965                 }
1966         }
1967         ptlrpc_req_finished(req);
1968
1969         return rc;
1970 }
1971
1972 static void lustre_swab_hai(struct hsm_action_item *h)
1973 {
1974         __swab32s(&h->hai_len);
1975         __swab32s(&h->hai_action);
1976         lustre_swab_lu_fid(&h->hai_fid);
1977         lustre_swab_lu_fid(&h->hai_dfid);
1978         __swab64s(&h->hai_cookie);
1979         __swab64s(&h->hai_extent.offset);
1980         __swab64s(&h->hai_extent.length);
1981         __swab64s(&h->hai_gid);
1982 }
1983
1984 static void lustre_swab_hal(struct hsm_action_list *h)
1985 {
1986         struct hsm_action_item  *hai;
1987         int                      i;
1988
1989         __swab32s(&h->hal_version);
1990         __swab32s(&h->hal_count);
1991         __swab32s(&h->hal_archive_id);
1992         __swab64s(&h->hal_flags);
1993         hai = hai_zero(h);
1994         for (i = 0; i < h->hal_count; i++, hai = hai_next(hai))
1995                 lustre_swab_hai(hai);
1996 }
1997
1998 static void lustre_swab_kuch(struct kuc_hdr *l)
1999 {
2000         __swab16s(&l->kuc_magic);
2001         /* __u8 l->kuc_transport */
2002         __swab16s(&l->kuc_msgtype);
2003         __swab16s(&l->kuc_msglen);
2004 }
2005
2006 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
2007                                 struct lustre_kernelcomm *lk)
2008 {
2009         struct obd_import  *imp = class_exp2cliimp(exp);
2010         __u32               archive = lk->lk_data;
2011         int                 rc = 0;
2012
2013         if (lk->lk_group != KUC_GRP_HSM) {
2014                 CERROR("Bad copytool group %d\n", lk->lk_group);
2015                 return -EINVAL;
2016         }
2017
2018         CDEBUG(D_HSM, "CT start r%d w%d u%d g%d f%#x\n", lk->lk_rfd, lk->lk_wfd,
2019                lk->lk_uid, lk->lk_group, lk->lk_flags);
2020
2021         if (lk->lk_flags & LK_FLG_STOP) {
2022                 /* Unregister with the coordinator */
2023                 rc = mdc_ioc_hsm_ct_unregister(imp);
2024         } else {
2025                 rc = mdc_ioc_hsm_ct_register(imp, archive);
2026         }
2027
2028         return rc;
2029 }
2030
2031 /**
2032  * Send a message to any listening copytools
2033  * @param val KUC message (kuc_hdr + hsm_action_list)
2034  * @param len total length of message
2035  */
2036 static int mdc_hsm_copytool_send(int len, void *val)
2037 {
2038         struct kuc_hdr          *lh = (struct kuc_hdr *)val;
2039         struct hsm_action_list  *hal = (struct hsm_action_list *)(lh + 1);
2040         int                      rc;
2041
2042         if (len < sizeof(*lh) + sizeof(*hal)) {
2043                 CERROR("Short HSM message %d < %d\n", len,
2044                        (int) (sizeof(*lh) + sizeof(*hal)));
2045                 return -EPROTO;
2046         }
2047         if (lh->kuc_magic == __swab16(KUC_MAGIC)) {
2048                 lustre_swab_kuch(lh);
2049                 lustre_swab_hal(hal);
2050         } else if (lh->kuc_magic != KUC_MAGIC) {
2051                 CERROR("Bad magic %x!=%x\n", lh->kuc_magic, KUC_MAGIC);
2052                 return -EPROTO;
2053         }
2054
2055         CDEBUG(D_HSM, " Received message mg=%x t=%d m=%d l=%d actions=%d "
2056                "on %s\n",
2057                lh->kuc_magic, lh->kuc_transport, lh->kuc_msgtype,
2058                lh->kuc_msglen, hal->hal_count, hal->hal_fsname);
2059
2060         /* Broadcast to HSM listeners */
2061         rc = libcfs_kkuc_group_put(KUC_GRP_HSM, lh);
2062
2063         return rc;
2064 }
2065
2066 /**
2067  * callback function passed to kuc for re-registering each HSM copytool
2068  * running on MDC, after MDT shutdown/recovery.
2069  * @param data archive id served by the copytool
2070  * @param cb_arg callback argument (obd_import)
2071  */
2072 static int mdc_hsm_ct_reregister(__u32 data, void *cb_arg)
2073 {
2074         struct obd_import       *imp = (struct obd_import *)cb_arg;
2075         __u32                    archive = data;
2076         int                      rc;
2077
2078         CDEBUG(D_HA, "recover copytool registration to MDT (archive=%#x)\n",
2079                archive);
2080         rc = mdc_ioc_hsm_ct_register(imp, archive);
2081
2082         /* ignore error if the copytool is already registered */
2083         return ((rc != 0) && (rc != -EEXIST)) ? rc : 0;
2084 }
2085
2086 /**
2087  * Re-establish all kuc contexts with MDT
2088  * after MDT shutdown/recovery.
2089  */
2090 static int mdc_kuc_reregister(struct obd_import *imp)
2091 {
2092         /* re-register HSM agents */
2093         return libcfs_kkuc_group_foreach(KUC_GRP_HSM, mdc_hsm_ct_reregister,
2094                                          (void *)imp);
2095 }
2096
2097 int mdc_set_info_async(const struct lu_env *env,
2098                        struct obd_export *exp,
2099                        obd_count keylen, void *key,
2100                        obd_count vallen, void *val,
2101                        struct ptlrpc_request_set *set)
2102 {
2103         struct obd_import       *imp = class_exp2cliimp(exp);
2104         int                      rc;
2105
2106         if (KEY_IS(KEY_READ_ONLY)) {
2107                 if (vallen != sizeof(int))
2108                         return -EINVAL;
2109
2110                 spin_lock(&imp->imp_lock);
2111                 if (*((int *)val)) {
2112                         imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY;
2113                         imp->imp_connect_data.ocd_connect_flags |=
2114                                                         OBD_CONNECT_RDONLY;
2115                 } else {
2116                         imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY;
2117                         imp->imp_connect_data.ocd_connect_flags &=
2118                                                         ~OBD_CONNECT_RDONLY;
2119                 }
2120                 spin_unlock(&imp->imp_lock);
2121
2122                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2123                                        keylen, key, vallen, val, set);
2124                 return rc;
2125         }
2126         if (KEY_IS(KEY_SPTLRPC_CONF)) {
2127                 sptlrpc_conf_client_adapt(exp->exp_obd);
2128                 return 0;
2129         }
2130         if (KEY_IS(KEY_FLUSH_CTX)) {
2131                 sptlrpc_import_flush_my_ctx(imp);
2132                 return 0;
2133         }
2134         if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
2135                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2136                                        keylen, key, vallen, val, set);
2137                 return rc;
2138         }
2139         if (KEY_IS(KEY_HSM_COPYTOOL_SEND)) {
2140                 rc = mdc_hsm_copytool_send(vallen, val);
2141                 return rc;
2142         }
2143
2144         CERROR("Unknown key %s\n", (char *)key);
2145         return -EINVAL;
2146 }
2147
2148 int mdc_get_info(const struct lu_env *env, struct obd_export *exp,
2149                  __u32 keylen, void *key, __u32 *vallen, void *val,
2150                  struct lov_stripe_md *lsm)
2151 {
2152         int rc = -EINVAL;
2153
2154         if (KEY_IS(KEY_MAX_EASIZE)) {
2155                 int mdsize, *max_easize;
2156
2157                 if (*vallen != sizeof(int))
2158                         return -EINVAL;
2159                 mdsize = *(int *)val;
2160                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
2161                         exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
2162                 max_easize = val;
2163                 *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
2164                 return 0;
2165         } else if (KEY_IS(KEY_DEFAULT_EASIZE)) {
2166                 int *default_easize;
2167
2168                 if (*vallen != sizeof(int))
2169                         return -EINVAL;
2170                 default_easize = val;
2171                 *default_easize = exp->exp_obd->u.cli.cl_default_mds_easize;
2172                 return 0;
2173         } else if (KEY_IS(KEY_MAX_COOKIESIZE)) {
2174                 int mdsize, *max_cookiesize;
2175
2176                 if (*vallen != sizeof(int))
2177                         return -EINVAL;
2178                 mdsize = *(int *)val;
2179                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_cookiesize)
2180                         exp->exp_obd->u.cli.cl_max_mds_cookiesize = mdsize;
2181                 max_cookiesize = val;
2182                 *max_cookiesize = exp->exp_obd->u.cli.cl_max_mds_cookiesize;
2183                 return 0;
2184         } else if (KEY_IS(KEY_DEFAULT_COOKIESIZE)) {
2185                 int *default_cookiesize;
2186
2187                 if (*vallen != sizeof(int))
2188                         return -EINVAL;
2189                 default_cookiesize = val;
2190                 *default_cookiesize =
2191                         exp->exp_obd->u.cli.cl_default_mds_cookiesize;
2192                 return 0;
2193         } else if (KEY_IS(KEY_CONN_DATA)) {
2194                 struct obd_import *imp = class_exp2cliimp(exp);
2195                 struct obd_connect_data *data = val;
2196
2197                 if (*vallen != sizeof(*data))
2198                         return -EINVAL;
2199
2200                 *data = imp->imp_connect_data;
2201                 return 0;
2202         } else if (KEY_IS(KEY_TGT_COUNT)) {
2203                 *((int *)val) = 1;
2204                 return 0;
2205         }
2206
2207         rc = mdc_get_info_rpc(exp, keylen, key, *vallen, val);
2208
2209         return rc;
2210 }
2211
2212 static int mdc_pin(struct obd_export *exp, const struct lu_fid *fid,
2213                    struct obd_capa *oc, struct obd_client_handle *handle,
2214                    int flags)
2215 {
2216         struct ptlrpc_request *req;
2217         struct mdt_body       *body;
2218         int                 rc;
2219
2220         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_PIN);
2221         if (req == NULL)
2222                 return -ENOMEM;
2223
2224         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2225
2226         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_PIN);
2227         if (rc) {
2228                 ptlrpc_request_free(req);
2229                 return rc;
2230         }
2231
2232         mdc_pack_body(req, fid, oc, 0, 0, -1, flags);
2233
2234         ptlrpc_request_set_replen(req);
2235
2236         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2237         rc = ptlrpc_queue_wait(req);
2238         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2239         if (rc) {
2240                 CERROR("Pin failed: %d\n", rc);
2241                 GOTO(err_out, rc);
2242         }
2243
2244         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2245         if (body == NULL)
2246                 GOTO(err_out, rc = -EPROTO);
2247
2248         handle->och_fh = body->handle;
2249         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
2250
2251         handle->och_mod = obd_mod_alloc();
2252         if (handle->och_mod == NULL) {
2253                 DEBUG_REQ(D_ERROR, req, "can't allocate md_open_data");
2254                 GOTO(err_out, rc = -ENOMEM);
2255         }
2256         handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
2257
2258         return 0;
2259
2260 err_out:
2261         ptlrpc_req_finished(req);
2262         return rc;
2263 }
2264
2265 static int mdc_unpin(struct obd_export *exp, struct obd_client_handle *handle,
2266                      int flag)
2267 {
2268         struct ptlrpc_request *req;
2269         struct mdt_body       *body;
2270         int                 rc;
2271
2272         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_UNPIN,
2273                                         LUSTRE_MDS_VERSION, MDS_UNPIN);
2274         if (req == NULL)
2275                 return -ENOMEM;
2276
2277         body = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
2278         body->handle = handle->och_fh;
2279         body->flags = flag;
2280
2281         ptlrpc_request_set_replen(req);
2282
2283         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2284         rc = ptlrpc_queue_wait(req);
2285         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2286
2287         if (rc != 0)
2288                 CERROR("Unpin failed: %d\n", rc);
2289
2290         ptlrpc_req_finished(req);
2291         ptlrpc_req_finished(handle->och_mod->mod_open_req);
2292
2293         obd_mod_put(handle->och_mod);
2294         return rc;
2295 }
2296
2297 int mdc_sync(struct obd_export *exp, const struct lu_fid *fid,
2298              struct obd_capa *oc, struct ptlrpc_request **request)
2299 {
2300         struct ptlrpc_request *req;
2301         int                 rc;
2302
2303         *request = NULL;
2304         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_SYNC);
2305         if (req == NULL)
2306                 return -ENOMEM;
2307
2308         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2309
2310         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SYNC);
2311         if (rc) {
2312                 ptlrpc_request_free(req);
2313                 return rc;
2314         }
2315
2316         mdc_pack_body(req, fid, oc, 0, 0, -1, 0);
2317
2318         ptlrpc_request_set_replen(req);
2319
2320         rc = ptlrpc_queue_wait(req);
2321         if (rc)
2322                 ptlrpc_req_finished(req);
2323         else
2324                 *request = req;
2325         return rc;
2326 }
2327
2328 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
2329                             enum obd_import_event event)
2330 {
2331         int rc = 0;
2332
2333         LASSERT(imp->imp_obd == obd);
2334
2335         switch (event) {
2336         case IMP_EVENT_DISCON: {
2337 #if 0
2338                 /* XXX Pass event up to OBDs stack. used only for FLD now */
2339                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DISCON, NULL);
2340 #endif
2341                 break;
2342         }
2343         case IMP_EVENT_INACTIVE: {
2344                 struct client_obd *cli = &obd->u.cli;
2345                 /*
2346                  * Flush current sequence to make client obtain new one
2347                  * from server in case of disconnect/reconnect.
2348                  */
2349                 if (cli->cl_seq != NULL)
2350                         seq_client_flush(cli->cl_seq);
2351
2352                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
2353                 break;
2354         }
2355         case IMP_EVENT_INVALIDATE: {
2356                 struct ldlm_namespace *ns = obd->obd_namespace;
2357
2358                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2359
2360                 break;
2361         }
2362         case IMP_EVENT_ACTIVE:
2363                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
2364                 /* redo the kuc registration after reconnecting */
2365                 if (rc == 0)
2366                         rc = mdc_kuc_reregister(imp);
2367                 break;
2368         case IMP_EVENT_OCD:
2369                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
2370                 break;
2371         case IMP_EVENT_DEACTIVATE:
2372         case IMP_EVENT_ACTIVATE:
2373                 break;
2374         default:
2375                 CERROR("Unknown import event %x\n", event);
2376                 LBUG();
2377         }
2378         return rc;
2379 }
2380
2381 int mdc_fid_alloc(struct obd_export *exp, struct lu_fid *fid,
2382                   struct md_op_data *op_data)
2383 {
2384         struct client_obd *cli = &exp->exp_obd->u.cli;
2385         struct lu_client_seq *seq = cli->cl_seq;
2386
2387         return seq_client_alloc_fid(NULL, seq, fid);
2388 }
2389
2390 struct obd_uuid *mdc_get_uuid(struct obd_export *exp)
2391 {
2392         struct client_obd *cli = &exp->exp_obd->u.cli;
2393         return &cli->cl_target_uuid;
2394 }
2395
2396 /**
2397  * Determine whether the lock can be canceled before replaying it during
2398  * recovery, non zero value will be return if the lock can be canceled,
2399  * or zero returned for not
2400  */
2401 static int mdc_cancel_for_recovery(struct ldlm_lock *lock)
2402 {
2403         if (lock->l_resource->lr_type != LDLM_IBITS)
2404                 return 0;
2405
2406         /* FIXME: if we ever get into a situation where there are too many
2407          * opened files with open locks on a single node, then we really
2408          * should replay these open locks to reget it */
2409         if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
2410                 return 0;
2411
2412         return 1;
2413 }
2414
2415 static int mdc_resource_inode_free(struct ldlm_resource *res)
2416 {
2417         if (res->lr_lvb_inode)
2418                 res->lr_lvb_inode = NULL;
2419
2420         return 0;
2421 }
2422
2423 struct ldlm_valblock_ops inode_lvbo = {
2424         .lvbo_free = mdc_resource_inode_free,
2425 };
2426
2427 static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
2428 {
2429         struct client_obd *cli = &obd->u.cli;
2430         struct lprocfs_static_vars lvars = { NULL };
2431         int rc;
2432
2433         OBD_ALLOC(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
2434         if (!cli->cl_rpc_lock)
2435                 return -ENOMEM;
2436         mdc_init_rpc_lock(cli->cl_rpc_lock);
2437
2438         ptlrpcd_addref();
2439
2440         OBD_ALLOC(cli->cl_close_lock, sizeof(*cli->cl_close_lock));
2441         if (!cli->cl_close_lock)
2442                 GOTO(err_rpc_lock, rc = -ENOMEM);
2443         mdc_init_rpc_lock(cli->cl_close_lock);
2444
2445         rc = client_obd_setup(obd, cfg);
2446         if (rc)
2447                 GOTO(err_close_lock, rc);
2448         lprocfs_mdc_init_vars(&lvars);
2449         lprocfs_obd_setup(obd, lvars.obd_vars);
2450         sptlrpc_lprocfs_cliobd_attach(obd);
2451         ptlrpc_lprocfs_register_obd(obd);
2452
2453         ns_register_cancel(obd->obd_namespace, mdc_cancel_for_recovery);
2454
2455         obd->obd_namespace->ns_lvbo = &inode_lvbo;
2456
2457         rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
2458         if (rc) {
2459                 mdc_cleanup(obd);
2460                 CERROR("failed to setup llogging subsystems\n");
2461         }
2462
2463         return rc;
2464
2465 err_close_lock:
2466         OBD_FREE(cli->cl_close_lock, sizeof(*cli->cl_close_lock));
2467 err_rpc_lock:
2468         OBD_FREE(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
2469         ptlrpcd_decref();
2470         return rc;
2471 }
2472
2473 /* Initialize the default and maximum LOV EA and cookie sizes.  This allows
2474  * us to make MDS RPCs with large enough reply buffers to hold a default
2475  * sized EA and cookie without having to calculate this (via a call into the
2476  * LOV + OSCs) each time we make an RPC.  The maximum size is also tracked
2477  * but not used to avoid wastefully vmalloc()'ing large reply buffers when
2478  * a large number of stripes is possible.  If a larger reply buffer is
2479  * required it will be reallocated in the ptlrpc layer due to overflow.
2480  */
2481 static int mdc_init_ea_size(struct obd_export *exp, int easize,
2482                             int def_easize, int cookiesize, int def_cookiesize)
2483 {
2484         struct obd_device *obd = exp->exp_obd;
2485         struct client_obd *cli = &obd->u.cli;
2486
2487         if (cli->cl_max_mds_easize < easize)
2488                 cli->cl_max_mds_easize = easize;
2489
2490         if (cli->cl_default_mds_easize < def_easize)
2491                 cli->cl_default_mds_easize = def_easize;
2492
2493         if (cli->cl_max_mds_cookiesize < cookiesize)
2494                 cli->cl_max_mds_cookiesize = cookiesize;
2495
2496         if (cli->cl_default_mds_cookiesize < def_cookiesize)
2497                 cli->cl_default_mds_cookiesize = def_cookiesize;
2498
2499         return 0;
2500 }
2501
2502 static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
2503 {
2504         int rc = 0;
2505
2506         switch (stage) {
2507         case OBD_CLEANUP_EARLY:
2508                 break;
2509         case OBD_CLEANUP_EXPORTS:
2510                 /* Failsafe, ok if racy */
2511                 if (obd->obd_type->typ_refcnt <= 1)
2512                         libcfs_kkuc_group_rem(0, KUC_GRP_HSM);
2513
2514                 obd_cleanup_client_import(obd);
2515                 ptlrpc_lprocfs_unregister_obd(obd);
2516                 lprocfs_obd_cleanup(obd);
2517
2518                 rc = obd_llog_finish(obd, 0);
2519                 if (rc != 0)
2520                         CERROR("failed to cleanup llogging subsystems\n");
2521                 break;
2522         }
2523         return rc;
2524 }
2525
2526 static int mdc_cleanup(struct obd_device *obd)
2527 {
2528         struct client_obd *cli = &obd->u.cli;
2529
2530         OBD_FREE(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
2531         OBD_FREE(cli->cl_close_lock, sizeof(*cli->cl_close_lock));
2532
2533         ptlrpcd_decref();
2534
2535         return client_obd_cleanup(obd);
2536 }
2537
2538
2539 static int mdc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
2540                          struct obd_device *tgt, int *index)
2541 {
2542         struct llog_ctxt        *ctxt;
2543         int                      rc;
2544
2545         LASSERT(olg == &obd->obd_olg);
2546
2547         rc = llog_setup(NULL, obd, olg, LLOG_CHANGELOG_REPL_CTXT, tgt,
2548                         &llog_client_ops);
2549         if (rc)
2550                 return rc;
2551
2552         ctxt = llog_group_get_ctxt(olg, LLOG_CHANGELOG_REPL_CTXT);
2553         llog_initiator_connect(ctxt);
2554         llog_ctxt_put(ctxt);
2555
2556         return 0;
2557 }
2558
2559 static int mdc_llog_finish(struct obd_device *obd, int count)
2560 {
2561         struct llog_ctxt *ctxt;
2562
2563         ctxt = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
2564         if (ctxt)
2565                 llog_cleanup(NULL, ctxt);
2566
2567         return 0;
2568 }
2569
2570 static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf)
2571 {
2572         struct lustre_cfg *lcfg = buf;
2573         struct lprocfs_static_vars lvars = { NULL };
2574         int rc = 0;
2575
2576         lprocfs_mdc_init_vars(&lvars);
2577         switch (lcfg->lcfg_command) {
2578         default:
2579                 rc = class_process_proc_param(PARAM_MDC, lvars.obd_vars,
2580                                               lcfg, obd);
2581                 if (rc > 0)
2582                         rc = 0;
2583                 break;
2584         }
2585         return(rc);
2586 }
2587
2588
2589 /* get remote permission for current user on fid */
2590 int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid,
2591                         struct obd_capa *oc, __u32 suppgid,
2592                         struct ptlrpc_request **request)
2593 {
2594         struct ptlrpc_request  *req;
2595         int                 rc;
2596
2597         LASSERT(client_is_remote(exp));
2598
2599         *request = NULL;
2600         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_GETATTR);
2601         if (req == NULL)
2602                 return -ENOMEM;
2603
2604         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2605
2606         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
2607         if (rc) {
2608                 ptlrpc_request_free(req);
2609                 return rc;
2610         }
2611
2612         mdc_pack_body(req, fid, oc, OBD_MD_FLRMTPERM, 0, suppgid, 0);
2613
2614         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
2615                              sizeof(struct mdt_remote_perm));
2616
2617         ptlrpc_request_set_replen(req);
2618
2619         rc = ptlrpc_queue_wait(req);
2620         if (rc)
2621                 ptlrpc_req_finished(req);
2622         else
2623                 *request = req;
2624         return rc;
2625 }
2626
2627 static int mdc_interpret_renew_capa(const struct lu_env *env,
2628                                     struct ptlrpc_request *req, void *args,
2629                                     int status)
2630 {
2631         struct mdc_renew_capa_args *ra = args;
2632         struct mdt_body *body = NULL;
2633         struct lustre_capa *capa;
2634
2635         if (status)
2636                 GOTO(out, capa = ERR_PTR(status));
2637
2638         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2639         if (body == NULL)
2640                 GOTO(out, capa = ERR_PTR(-EFAULT));
2641
2642         if ((body->valid & OBD_MD_FLOSSCAPA) == 0)
2643                 GOTO(out, capa = ERR_PTR(-ENOENT));
2644
2645         capa = req_capsule_server_get(&req->rq_pill, &RMF_CAPA2);
2646         if (!capa)
2647                 GOTO(out, capa = ERR_PTR(-EFAULT));
2648 out:
2649         ra->ra_cb(ra->ra_oc, capa);
2650         return 0;
2651 }
2652
2653 static int mdc_renew_capa(struct obd_export *exp, struct obd_capa *oc,
2654                           renew_capa_cb_t cb)
2655 {
2656         struct ptlrpc_request *req;
2657         struct mdc_renew_capa_args *ra;
2658
2659         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_GETATTR,
2660                                         LUSTRE_MDS_VERSION, MDS_GETATTR);
2661         if (req == NULL)
2662                 return -ENOMEM;
2663
2664         /* NB, OBD_MD_FLOSSCAPA is set here, but it doesn't necessarily mean the
2665          * capa to renew is oss capa.
2666          */
2667         mdc_pack_body(req, &oc->c_capa.lc_fid, oc, OBD_MD_FLOSSCAPA, 0, -1, 0);
2668         ptlrpc_request_set_replen(req);
2669
2670         CLASSERT(sizeof(*ra) <= sizeof(req->rq_async_args));
2671         ra = ptlrpc_req_async_args(req);
2672         ra->ra_oc = oc;
2673         ra->ra_cb = cb;
2674         req->rq_interpret_reply = mdc_interpret_renew_capa;
2675         ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
2676         return 0;
2677 }
2678
2679 struct obd_ops mdc_obd_ops = {
2680         .o_owner            = THIS_MODULE,
2681         .o_setup            = mdc_setup,
2682         .o_precleanup       = mdc_precleanup,
2683         .o_cleanup        = mdc_cleanup,
2684         .o_add_conn      = client_import_add_conn,
2685         .o_del_conn      = client_import_del_conn,
2686         .o_connect          = client_connect_import,
2687         .o_disconnect       = client_disconnect_export,
2688         .o_iocontrol    = mdc_iocontrol,
2689         .o_set_info_async   = mdc_set_info_async,
2690         .o_statfs          = mdc_statfs,
2691         .o_pin        = mdc_pin,
2692         .o_unpin            = mdc_unpin,
2693         .o_fid_init         = client_fid_init,
2694         .o_fid_fini         = client_fid_fini,
2695         .o_fid_alloc    = mdc_fid_alloc,
2696         .o_import_event     = mdc_import_event,
2697         .o_llog_init    = mdc_llog_init,
2698         .o_llog_finish      = mdc_llog_finish,
2699         .o_get_info      = mdc_get_info,
2700         .o_process_config   = mdc_process_config,
2701         .o_get_uuid      = mdc_get_uuid,
2702         .o_quotactl      = mdc_quotactl,
2703         .o_quotacheck       = mdc_quotacheck
2704 };
2705
2706 struct md_ops mdc_md_ops = {
2707         .m_getstatus    = mdc_getstatus,
2708         .m_null_inode       = mdc_null_inode,
2709         .m_find_cbdata      = mdc_find_cbdata,
2710         .m_close            = mdc_close,
2711         .m_create          = mdc_create,
2712         .m_done_writing     = mdc_done_writing,
2713         .m_enqueue        = mdc_enqueue,
2714         .m_getattr        = mdc_getattr,
2715         .m_getattr_name     = mdc_getattr_name,
2716         .m_intent_lock      = mdc_intent_lock,
2717         .m_link      = mdc_link,
2718         .m_is_subdir    = mdc_is_subdir,
2719         .m_rename          = mdc_rename,
2720         .m_setattr        = mdc_setattr,
2721         .m_setxattr      = mdc_setxattr,
2722         .m_getxattr      = mdc_getxattr,
2723         .m_sync      = mdc_sync,
2724         .m_readpage      = mdc_readpage,
2725         .m_unlink          = mdc_unlink,
2726         .m_cancel_unused    = mdc_cancel_unused,
2727         .m_init_ea_size     = mdc_init_ea_size,
2728         .m_set_lock_data    = mdc_set_lock_data,
2729         .m_lock_match       = mdc_lock_match,
2730         .m_get_lustre_md    = mdc_get_lustre_md,
2731         .m_free_lustre_md   = mdc_free_lustre_md,
2732         .m_set_open_replay_data = mdc_set_open_replay_data,
2733         .m_clear_open_replay_data = mdc_clear_open_replay_data,
2734         .m_renew_capa       = mdc_renew_capa,
2735         .m_unpack_capa      = mdc_unpack_capa,
2736         .m_get_remote_perm  = mdc_get_remote_perm,
2737         .m_intent_getattr_async = mdc_intent_getattr_async,
2738         .m_revalidate_lock      = mdc_revalidate_lock
2739 };
2740
2741 int __init mdc_init(void)
2742 {
2743         int rc;
2744         struct lprocfs_static_vars lvars = { NULL };
2745         lprocfs_mdc_init_vars(&lvars);
2746
2747         rc = class_register_type(&mdc_obd_ops, &mdc_md_ops, lvars.module_vars,
2748                                  LUSTRE_MDC_NAME, NULL);
2749         return rc;
2750 }
2751
2752 static void /*__exit*/ mdc_exit(void)
2753 {
2754         class_unregister_type(LUSTRE_MDC_NAME);
2755 }
2756
2757 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
2758 MODULE_DESCRIPTION("Lustre Metadata Client");
2759 MODULE_LICENSE("GPL");
2760
2761 module_init(mdc_init);
2762 module_exit(mdc_exit);