lustre/mdc/mdc_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2001-2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  */
  24
  25 #ifndef EXPORT_SYMTAB
  26 # define EXPORT_SYMTAB
  27 #endif
  28 #define DEBUG_SUBSYSTEM S_MDC
  29
  30 #ifdef __KERNEL__
  31 # include <linux/module.h>
  32 # include <linux/pagemap.h>
  33 # include <linux/miscdevice.h>
  34 # include <linux/init.h>
  35 #else
  36 # include <liblustre.h>
  37 #endif
  38
  39 #include <obd_class.h>
  40 #include <lustre_dlm.h>
  41 #include <lprocfs_status.h>
  42 #include <lustre_param.h>
  43 #include "mdc_internal.h"
  44
  45 static quota_interface_t *quota_interface;
  46
  47 #define REQUEST_MINOR 244
  48
  49 static quota_interface_t *quota_interface;
  50 extern quota_interface_t mdc_quota_interface;
  51
  52 static int mdc_cleanup(struct obd_device *obd);
  53
  54 extern int mds_queue_req(struct ptlrpc_request *);
  55 /* Helper that implements most of mdc_getstatus and signal_completed_replay. */
  56 /* XXX this should become mdc_get_info("key"), sending MDS_GET_INFO RPC */
  57 static int send_getstatus(struct obd_import *imp, struct ll_fid *rootfid,
  58                           int level, int msg_flags)
  59 {
  60         struct ptlrpc_request *req;
  61         struct mds_body *body;
  62         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
  63         ENTRY;
  64
  65         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_GETSTATUS, 2, size,
  66                               NULL);
  67         if (!req)
  68                 GOTO(out, rc = -ENOMEM);
  69
  70         req->rq_send_state = level;
  71         ptlrpc_req_set_repsize(req, 2, size);
  72
  73         mdc_pack_req_body(req, REQ_REC_OFF, 0, NULL, 0, 0);
  74         lustre_msg_add_flags(req->rq_reqmsg, msg_flags);
  75         rc = ptlrpc_queue_wait(req);
  76
  77         if (!rc) {
  78                 body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
  79                                           lustre_swab_mds_body);
  80                 if (body == NULL) {
  81                         CERROR ("Can't extract mds_body\n");
  82                         GOTO (out, rc = -EPROTO);
  83                 }
  84
  85                 memcpy(rootfid, &body->fid1, sizeof(*rootfid));
  86
  87                 CDEBUG(D_NET, "root ino="LPU64", last_committed="LPU64
  88                        ", last_xid="LPU64"\n",
  89                        rootfid->id,
  90                        lustre_msg_get_last_committed(req->rq_repmsg),
  91                        lustre_msg_get_last_xid(req->rq_repmsg));
  92         }
  93
  94         EXIT;
  95  out:
  96         ptlrpc_req_finished(req);
  97         return rc;
  98 }
  99
 100 /* This should be mdc_get_info("rootfid") */
 101 int mdc_getstatus(struct obd_export *exp, struct ll_fid *rootfid)
 102 {
 103         return send_getstatus(class_exp2cliimp(exp), rootfid, LUSTRE_IMP_FULL,
 104                               0);
 105 }
 106
 107 static
 108 int mdc_getattr_common(struct obd_export *exp, unsigned int ea_size,
 109                        unsigned int acl_size, struct ptlrpc_request *req)
 110 {
 111         struct obd_device *obddev = class_exp2obd(exp);
 112         struct mds_body *body;
 113         void *eadata;
 114         int size[4] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 115         int bufcount = 2, rc;
 116         ENTRY;
 117
 118         /* request message already built */
 119
 120         if (ea_size != 0) {
 121                 size[bufcount++] = ea_size;
 122                 CDEBUG(D_INODE, "reserved %u bytes for MD/symlink in packet\n",
 123                        ea_size);
 124         }
 125         if (acl_size) {
 126                 size[bufcount++] = acl_size;
 127                 CDEBUG(D_INODE, "reserved %u bytes for ACL\n", acl_size);
 128         }
 129
 130         ptlrpc_req_set_repsize(req, bufcount, size);
 131
 132         mdc_enter_request(&obddev->u.cli);
 133         rc = ptlrpc_queue_wait(req);
 134         mdc_exit_request(&obddev->u.cli);
 135         if (rc != 0)
 136                 RETURN (rc);
 137
 138         body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
 139                                   lustre_swab_mds_body);
 140         if (body == NULL) {
 141                 CERROR ("Can't unpack mds_body\n");
 142                 RETURN (-EPROTO);
 143         }
 144
 145         CDEBUG(D_NET, "mode: %o\n", body->mode);
 146
 147         lustre_set_rep_swabbed(req, REPLY_REC_OFF + 1);
 148         if (body->eadatasize != 0) {
 149                 /* reply indicates presence of eadata; check it's there... */
 150                 eadata = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 1,
 151                                         body->eadatasize);
 152                 if (eadata == NULL) {
 153                         CERROR ("Missing/short eadata\n");
 154                         RETURN (-EPROTO);
 155                 }
 156         }
 157
 158         if (body->valid & OBD_MD_FLMODEASIZE) {
 159                 if (exp->exp_obd->u.cli.cl_max_mds_easize < body->max_mdsize)
 160                         exp->exp_obd->u.cli.cl_max_mds_easize =
 161                                                 body->max_mdsize;
 162                 if (exp->exp_obd->u.cli.cl_max_mds_cookiesize <
 163                                                 body->max_cookiesize)
 164                         exp->exp_obd->u.cli.cl_max_mds_cookiesize =
 165                                                 body->max_cookiesize;
 166         }
 167
 168         RETURN (0);
 169 }
 170
 171 int mdc_getattr(struct obd_export *exp, struct ll_fid *fid,
 172                 obd_valid valid, unsigned int ea_size,
 173                 struct ptlrpc_request **request)
 174 {
 175         struct ptlrpc_request *req;
 176         int size[2] = { sizeof(struct ptlrpc_body), sizeof(struct mds_body) };
 177         int acl_size = 0, rc;
 178         ENTRY;
 179
 180         /* XXX do we need to make another request here?  We just did a getattr
 181          *     to do the lookup in the first place.
 182          */
 183         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 184                               MDS_GETATTR, 2, size, NULL);
 185         if (!req)
 186                 GOTO(out, rc = -ENOMEM);
 187
 188         mdc_pack_req_body(req, REQ_REC_OFF, valid, fid, ea_size,
 189                           MDS_BFLAG_EXT_FLAGS/*request "new" flags(bug 9486)*/);
 190
 191         /* currently only root inode will call us with FLACL */
 192         if (valid & OBD_MD_FLACL)
 193                 acl_size = LUSTRE_POSIX_ACL_MAX_SIZE;
 194
 195         rc = mdc_getattr_common(exp, ea_size, acl_size, req);
 196         if (rc != 0) {
 197                 ptlrpc_req_finished (req);
 198                 req = NULL;
 199         }
 200  out:
 201         *request = req;
 202         RETURN (rc);
 203 }
 204
 205 int mdc_getattr_name(struct obd_export *exp, struct ll_fid *fid,
 206                      const char *filename, int namelen, unsigned long valid,
 207                      unsigned int ea_size, struct ptlrpc_request **request)
 208 {
 209         struct ptlrpc_request *req;
 210         struct mds_body *body;
 211         int rc, size[3] = { sizeof(struct ptlrpc_body), sizeof(*body), namelen};
 212         ENTRY;
 213
 214         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 215                               MDS_GETATTR_NAME, 3, size, NULL);
 216         if (!req)
 217                 GOTO(out, rc = -ENOMEM);
 218
 219         mdc_pack_req_body(req, REQ_REC_OFF, valid, fid, ea_size,
 220                           MDS_BFLAG_EXT_FLAGS/*request "new" flags(bug 9486)*/);
 221
 222         LASSERT(strnlen(filename, namelen) == namelen - 1);
 223         memcpy(lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1, namelen),
 224                filename, namelen);
 225
 226         rc = mdc_getattr_common(exp, ea_size, 0, req);
 227         if (rc != 0) {
 228                 ptlrpc_req_finished (req);
 229                 req = NULL;
 230         }
 231  out:
 232         *request = req;
 233         RETURN(rc);
 234 }
 235
 236 static
 237 int mdc_xattr_common(struct obd_export *exp, struct ll_fid *fid,
 238                      int opcode, obd_valid valid, const char *xattr_name,
 239                      const char *input, int input_size, int output_size,
 240                      int flags, struct ptlrpc_request **request)
 241 {
 242         struct obd_device *obddev = class_exp2obd(exp);
 243         struct ptlrpc_request *req;
 244         int size[4] = { sizeof(struct ptlrpc_body), sizeof(struct mds_body) };
 245         // int size[3] = {sizeof(struct mds_body)}, bufcnt = 1;
 246         int rc, xattr_namelen = 0, bufcnt = 2, offset;
 247         void *tmp;
 248         ENTRY;
 249
 250         if (xattr_name) {
 251                 xattr_namelen = strlen(xattr_name) + 1;
 252                 size[bufcnt++] = xattr_namelen;
 253         }
 254         if (input_size) {
 255                 LASSERT(input);
 256                 size[bufcnt++] = input_size;
 257         }
 258
 259         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 260                               opcode, bufcnt, size, NULL);
 261         if (!req)
 262                 GOTO(out, rc = -ENOMEM);
 263
 264         /* request data */
 265         mdc_pack_req_body(req, REQ_REC_OFF, valid, fid, output_size, flags);
 266
 267         offset = REQ_REC_OFF + 1;
 268
 269         if (xattr_name) {
 270                 tmp = lustre_msg_buf(req->rq_reqmsg, offset++, xattr_namelen);
 271                 memcpy(tmp, xattr_name, xattr_namelen);
 272         }
 273         if (input_size) {
 274                 tmp = lustre_msg_buf(req->rq_reqmsg, offset++, input_size);
 275                 memcpy(tmp, input, input_size);
 276         }
 277
 278         /* reply buffers */
 279         if (opcode == MDS_GETXATTR) {
 280                 size[REPLY_REC_OFF] = sizeof(struct mds_body);
 281                 bufcnt = 2;
 282         } else {
 283                 bufcnt = 1;
 284         }
 285
 286         /* we do this even output_size is 0, because server is doing that */
 287         size[bufcnt++] = output_size;
 288
 289         ptlrpc_req_set_repsize(req, bufcnt, size);
 290
 291         /* make rpc */
 292         if (opcode == MDS_SETXATTR)
 293                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 294         else
 295                 mdc_enter_request(&obddev->u.cli);
 296
 297         rc = ptlrpc_queue_wait(req);
 298
 299         if (opcode == MDS_SETXATTR)
 300                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 301         else
 302                 mdc_exit_request(&obddev->u.cli);
 303
 304         if (rc != 0)
 305                 GOTO(err_out, rc);
 306
 307         if (opcode == MDS_GETXATTR) {
 308                 struct mds_body * body = lustre_swab_repbuf(req, REPLY_REC_OFF,
 309                                           sizeof(*body),
 310                                           lustre_swab_mds_body);
 311                 if (body == NULL) {
 312                         CERROR ("Can't unpack mds_body\n");
 313                         GOTO(err_out, rc = -EPROTO);
 314                 }
 315         }
 316 out:
 317         *request = req;
 318         RETURN (rc);
 319 err_out:
 320         ptlrpc_req_finished(req);
 321         req = NULL;
 322         goto out;
 323 }
 324
 325 int mdc_setxattr(struct obd_export *exp, struct ll_fid *fid,
 326                  obd_valid valid, const char *xattr_name,
 327                  const char *input, int input_size,
 328                  int output_size, int flags,
 329                  struct ptlrpc_request **request)
 330 {
 331         return mdc_xattr_common(exp, fid, MDS_SETXATTR, valid, xattr_name,
 332                                 input, input_size, output_size, flags, request);
 333 }
 334
 335 int mdc_getxattr(struct obd_export *exp, struct ll_fid *fid,
 336                  obd_valid valid, const char *xattr_name,
 337                  const char *input, int input_size,
 338                  int output_size, struct ptlrpc_request **request)
 339 {
 340         return mdc_xattr_common(exp, fid, MDS_GETXATTR, valid, xattr_name,
 341                                 input, input_size, output_size, 0, request);
 342 }
 343
 344 /* This should be called with both the request and the reply still packed. */
 345 void mdc_store_inode_generation(struct ptlrpc_request *req, int reqoff,
 346                                 int repoff)
 347 {
 348         struct mds_rec_create *rec = lustre_msg_buf(req->rq_reqmsg, reqoff,
 349                                                     sizeof(*rec));
 350         struct mds_body *body = lustre_msg_buf(req->rq_repmsg, repoff,
 351                                                sizeof(*body));
 352
 353         LASSERT (rec != NULL);
 354         LASSERT (body != NULL);
 355
 356         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 357         if (body->fid1.id == 0) {
 358                 DEBUG_REQ(D_ERROR, req, "saving replay request with id = 0 "
 359                           "gen = %u", body->fid1.generation);
 360                 LBUG();
 361         }
 362
 363         DEBUG_REQ(D_INODE, req, "storing generation %u for ino "LPU64,
 364                   rec->cr_replayfid.generation, rec->cr_replayfid.id);
 365 }
 366
 367 #ifdef CONFIG_FS_POSIX_ACL
 368 static
 369 int mdc_unpack_acl(struct obd_export *exp, struct ptlrpc_request *req,
 370                    struct lustre_md *md, unsigned int offset)
 371 {
 372         struct mds_body  *body = md->body;
 373         struct posix_acl *acl;
 374         void             *buf;
 375         int               rc;
 376
 377         if (!body->aclsize)
 378                 return 0;
 379
 380         buf = lustre_msg_buf(req->rq_repmsg, offset, body->aclsize);
 381         if (!buf) {
 382                 CERROR("aclsize %u, bufcount %u, bufsize %u\n",
 383                        body->aclsize, lustre_msg_bufcount(req->rq_repmsg),
 384                        (lustre_msg_bufcount(req->rq_repmsg) <= offset) ?
 385                                 -1 : lustre_msg_buflen(req->rq_repmsg, offset));
 386                 return -EPROTO;
 387         }
 388
 389         acl = posix_acl_from_xattr(buf, body->aclsize);
 390         if (IS_ERR(acl)) {
 391                 rc = PTR_ERR(acl);
 392                 CERROR("convert xattr to acl: %d\n", rc);
 393                 return rc;
 394         }
 395
 396         rc = posix_acl_valid(acl);
 397         if (rc) {
 398                 CERROR("validate acl: %d\n", rc);
 399                 posix_acl_release(acl);
 400                 return rc;
 401         }
 402
 403         md->posix_acl = acl;
 404         return 0;
 405 }
 406 #else
 407 #define mdc_unpack_acl(exp, req, md, offset) 0
 408 #endif
 409
 410 int mdc_req2lustre_md(struct ptlrpc_request *req, int offset,
 411                       struct obd_export *exp,
 412                       struct lustre_md *md)
 413 {
 414         int rc = 0;
 415         ENTRY;
 416
 417         LASSERT(md);
 418         memset(md, 0, sizeof(*md));
 419
 420         md->body = lustre_msg_buf(req->rq_repmsg, offset, sizeof (*md->body));
 421         LASSERT (md->body != NULL);
 422         LASSERT(lustre_rep_swabbed(req, offset));
 423         offset++;
 424
 425         if (md->body->valid & OBD_MD_FLEASIZE) {
 426                 int lmmsize;
 427                 struct lov_mds_md *lmm;
 428
 429                 if (!S_ISREG(md->body->mode)) {
 430                         CERROR("OBD_MD_FLEASIZE set, should be a regular file, "
 431                                "but is not\n");
 432                         GOTO(err_out, rc = -EPROTO);
 433                 }
 434
 435                 if (md->body->eadatasize == 0) {
 436                         CERROR ("OBD_MD_FLEASIZE set, but eadatasize 0\n");
 437                         GOTO(err_out, rc = -EPROTO);
 438                 }
 439                 lmmsize = md->body->eadatasize;
 440                 lmm = lustre_msg_buf(req->rq_repmsg, offset, lmmsize);
 441                 if (!lmm) {
 442                         CERROR ("incorrect message: lmm == 0\n");
 443                         GOTO(err_out, rc = -EPROTO);
 444                 }
 445                 LASSERT(lustre_rep_swabbed(req, offset));
 446
 447                 rc = obd_unpackmd(exp, &md->lsm, lmm, lmmsize);
 448                 if (rc < 0)
 449                         GOTO(err_out, rc);
 450
 451                 if (rc < sizeof(*md->lsm)) {
 452                         CERROR ("lsm size too small:  rc < sizeof (*md->lsm) "
 453                                 "(%d < "LPSZ")\n", rc, sizeof(*md->lsm));
 454                         GOTO(err_out, rc = -EPROTO);
 455                 }
 456                 rc = 0;
 457
 458                 offset++;
 459         }
 460
 461         if (md->body->valid & OBD_MD_FLDIREA) {
 462                 if(!S_ISDIR(md->body->mode)) {
 463                         CERROR("OBD_MD_FLDIREA set, should be a directory, but "
 464                                "is not\n");
 465                         GOTO(err_out, rc = -EPROTO);
 466                 }
 467                 offset++;
 468         }
 469
 470         /* for ACL, it's possible that FLACL is set but aclsize is zero.
 471          * only when aclsize != 0 there's an actual segment for ACL in
 472          * reply buffer.
 473          */
 474         if ((md->body->valid & OBD_MD_FLACL) && md->body->aclsize) {
 475                 rc = mdc_unpack_acl(exp, req, md, offset);
 476                 if (rc)
 477                         GOTO(err_out, rc);
 478                 offset++;
 479         }
 480 out:
 481         RETURN(rc);
 482
 483 err_out:
 484         if (md->lsm)
 485                 obd_free_memmd(exp, &md->lsm);
 486         goto out;
 487 }
 488
 489 void mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
 490 {
 491         if (md->lsm)
 492                 obd_free_memmd(exp, &md->lsm);
 493
 494 #ifdef CONFIG_FS_POSIX_ACL
 495         if (md->posix_acl) {
 496                 posix_acl_release(md->posix_acl);
 497                 md->posix_acl = NULL;
 498         }
 499 #endif
 500 }
 501
 502 static void mdc_commit_open(struct ptlrpc_request *req)
 503 {
 504         struct mdc_open_data *mod = req->rq_cb_data;
 505         if (mod == NULL)
 506                 return;
 507
 508         if (mod->mod_close_req != NULL)
 509                 mod->mod_close_req->rq_cb_data = NULL;
 510
 511         if (mod->mod_och != NULL)
 512                 mod->mod_och->och_mod = NULL;
 513
 514         OBD_FREE(mod, sizeof(*mod));
 515         req->rq_cb_data = NULL;
 516 }
 517
 518 static void mdc_replay_open(struct ptlrpc_request *req)
 519 {
 520         struct mdc_open_data *mod = req->rq_cb_data;
 521         struct obd_client_handle *och;
 522         struct ptlrpc_request *close_req;
 523         struct lustre_handle old;
 524         struct mds_body *body;
 525         ENTRY;
 526
 527         body = lustre_swab_repbuf(req, DLM_REPLY_REC_OFF, sizeof(*body),
 528                                   lustre_swab_mds_body);
 529         LASSERT (body != NULL);
 530
 531         if (mod == NULL) {
 532                 DEBUG_REQ(D_ERROR, req,
 533                           "can't properly replay without open data");
 534                 EXIT;
 535                 return;
 536         }
 537
 538         och = mod->mod_och;
 539         if (och != NULL) {
 540                 struct lustre_handle *file_fh;
 541                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 542                 file_fh = &och->och_fh;
 543                 CDEBUG(D_RPCTRACE, "updating handle from "LPX64" to "LPX64"\n",
 544                        file_fh->cookie, body->handle.cookie);
 545                 memcpy(&old, file_fh, sizeof(old));
 546                 memcpy(file_fh, &body->handle, sizeof(*file_fh));
 547         }
 548
 549         close_req = mod->mod_close_req;
 550         if (close_req != NULL) {
 551                 struct mds_body *close_body;
 552                 LASSERT(lustre_msg_get_opc(close_req->rq_reqmsg) == MDS_CLOSE);
 553                 close_body = lustre_msg_buf(close_req->rq_reqmsg, REQ_REC_OFF,
 554                                             sizeof(*close_body));
 555                 if (och != NULL)
 556                         LASSERT(!memcmp(&old, &close_body->handle, sizeof old));
 557                 DEBUG_REQ(D_RPCTRACE, close_req, "updating close with new fh");
 558                 memcpy(&close_body->handle, &body->handle,
 559                        sizeof(close_body->handle));
 560         }
 561
 562         EXIT;
 563 }
 564
 565 void mdc_set_open_replay_data(struct obd_client_handle *och,
 566                               struct ptlrpc_request *open_req)
 567 {
 568         struct mdc_open_data *mod;
 569         struct mds_rec_create *rec = lustre_msg_buf(open_req->rq_reqmsg,
 570                                                     DLM_INTENT_REC_OFF,
 571                                                     sizeof(*rec));
 572         struct mds_body *body = lustre_msg_buf(open_req->rq_repmsg,
 573                                                DLM_REPLY_REC_OFF,
 574                                                sizeof(*body));
 575
 576         /* If request is not eligible for replay, just bail out */
 577         if (!open_req->rq_replay)
 578                 return;
 579
 580         /* incoming message in my byte order (it's been swabbed) */
 581         LASSERT(rec != NULL);
 582         LASSERT(lustre_rep_swabbed(open_req, DLM_REPLY_REC_OFF));
 583         /* outgoing messages always in my byte order */
 584         LASSERT(body != NULL);
 585
 586         if (och) {
 587                 OBD_ALLOC(mod, sizeof(*mod));
 588                 if (mod == NULL) {
 589                         DEBUG_REQ(D_ERROR, open_req, "can't allocate mdc_open_data");
 590                         return;
 591                 }
 592
 593                 spin_lock(&open_req->rq_lock);
 594                 och->och_mod = mod;
 595                 mod->mod_och = och;
 596                 mod->mod_open_req = open_req;
 597                 open_req->rq_cb_data = mod;
 598                 open_req->rq_commit_cb = mdc_commit_open;
 599                 spin_unlock(&open_req->rq_lock);
 600         }
 601
 602         memcpy(&rec->cr_replayfid, &body->fid1, sizeof rec->cr_replayfid);
 603         open_req->rq_replay_cb = mdc_replay_open;
 604         if (body->fid1.id == 0) {
 605                 DEBUG_REQ(D_ERROR, open_req, "saving replay request with "
 606                           "id = 0 gen = %u", body->fid1.generation);
 607                 LBUG();
 608         }
 609
 610         DEBUG_REQ(D_RPCTRACE, open_req, "set up replay data");
 611 }
 612
 613 void mdc_clear_open_replay_data(struct obd_client_handle *och)
 614 {
 615         struct mdc_open_data *mod = och->och_mod;
 616
 617         /* Don't free the structure now (it happens in mdc_commit_open, after
 618          * we're sure we won't need to fix up the close request in the future),
 619          * but make sure that replay doesn't poke at the och, which is about to
 620          * be freed. */
 621         LASSERT(mod != LP_POISON);
 622         if (mod != NULL)
 623                 mod->mod_och = NULL;
 624         och->och_mod = NULL;
 625 }
 626
 627 static void mdc_commit_close(struct ptlrpc_request *req)
 628 {
 629         struct mdc_open_data *mod = req->rq_cb_data;
 630         struct ptlrpc_request *open_req;
 631         struct obd_import *imp = req->rq_import;
 632
 633         DEBUG_REQ(D_RPCTRACE, req, "close req committed");
 634         if (mod == NULL)
 635                 return;
 636
 637         mod->mod_close_req = NULL;
 638         req->rq_cb_data = NULL;
 639         req->rq_commit_cb = NULL;
 640
 641         open_req = mod->mod_open_req;
 642         LASSERT(open_req != NULL);
 643         LASSERT(open_req != LP_POISON);
 644         LASSERT(open_req->rq_type != LI_POISON);
 645
 646         DEBUG_REQ(D_RPCTRACE, open_req, "open req balanced");
 647         LASSERT(open_req->rq_transno != 0);
 648         LASSERT(open_req->rq_import == imp);
 649
 650         /* We no longer want to preserve this for transno-unconditional
 651          * replay. */
 652         spin_lock(&open_req->rq_lock);
 653         open_req->rq_replay = 0;
 654         spin_unlock(&open_req->rq_lock);
 655 }
 656
 657 int mdc_close(struct obd_export *exp, struct obdo *oa,
 658               struct obd_client_handle *och, struct ptlrpc_request **request)
 659 {
 660         struct obd_device *obd = class_exp2obd(exp);
 661         int reqsize[2] = { sizeof(struct ptlrpc_body),
 662                            sizeof(struct mds_body) };
 663         int rc, repsize[4] = { sizeof(struct ptlrpc_body),
 664                                sizeof(struct mds_body),
 665                                obd->u.cli.cl_max_mds_easize,
 666                                obd->u.cli.cl_max_mds_cookiesize };
 667         struct ptlrpc_request *req;
 668         struct mdc_open_data *mod;
 669         ENTRY;
 670
 671         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 672                               MDS_CLOSE, 2, reqsize, NULL);
 673         if (req == NULL)
 674                 GOTO(out, rc = -ENOMEM);
 675
 676         /* To avoid a livelock (bug 7034), we need to send CLOSE RPCs to a
 677          * portal whose threads are not taking any DLM locks and are therefore
 678          * always progressing */
 679         req->rq_request_portal = MDS_READPAGE_PORTAL;
 680         ptlrpc_at_set_req_timeout(req);
 681
 682         /* Ensure that this close's handle is fixed up during replay. */
 683         LASSERT(och != NULL);
 684         LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
 685         mod = och->och_mod;
 686         if (likely(mod != NULL)) {
 687                 if (mod->mod_open_req->rq_type == LI_POISON) {
 688                         CERROR("LBUG POISONED open %p!\n", mod->mod_open_req);
 689                         LBUG();
 690                         ptlrpc_req_finished(req);
 691                         req = NULL;
 692                         GOTO(out, rc = -EIO);
 693                 }
 694                 mod->mod_close_req = req;
 695                 DEBUG_REQ(D_RPCTRACE, mod->mod_open_req, "matched open");
 696         } else {
 697                 CDEBUG(D_RPCTRACE, "couldn't find open req; expecting error\n");
 698         }
 699
 700         mdc_close_pack(req, REQ_REC_OFF, oa, oa->o_valid, och);
 701
 702         ptlrpc_req_set_repsize(req, 4, repsize);
 703         req->rq_commit_cb = mdc_commit_close;
 704         LASSERT(req->rq_cb_data == NULL);
 705         req->rq_cb_data = mod;
 706
 707         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 708         rc = ptlrpc_queue_wait(req);
 709         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
 710
 711         if (req->rq_repmsg == NULL) {
 712                 CDEBUG(D_RPCTRACE, "request failed to send: %p, %d\n", req,
 713                        req->rq_status);
 714                 if (rc == 0)
 715                         rc = req->rq_status ? req->rq_status : -EIO;
 716         } else if (rc == 0) {
 717                 rc = lustre_msg_get_status(req->rq_repmsg);
 718                 if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 719                         DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR, err "
 720                                   "= %d", rc);
 721                         if (rc > 0)
 722                                 rc = -rc;
 723                 } else if (mod == NULL) {
 724                         CERROR("Unexpected: can't find mdc_open_data, but the "
 725                                "close succeeded.  Please tell CFS.\n");
 726                 }
 727                 if (!lustre_swab_repbuf(req, REPLY_REC_OFF,
 728                                         sizeof(struct mds_body),
 729                                         lustre_swab_mds_body)) {
 730                         CERROR("Error unpacking mds_body\n");
 731                         rc = -EPROTO;
 732                 }
 733         }
 734
 735         EXIT;
 736         *request = req;
 737  out:
 738         if (rc != 0 && req && req->rq_commit_cb)
 739                 req->rq_commit_cb(req);
 740
 741         return rc;
 742 }
 743
 744 int mdc_done_writing(struct obd_export *exp, struct obdo *obdo)
 745 {
 746         struct ptlrpc_request *req;
 747         struct mds_body *body;
 748         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 749         ENTRY;
 750
 751         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
 752                               MDS_DONE_WRITING, 2, size, NULL);
 753         if (req == NULL)
 754                 RETURN(-ENOMEM);
 755
 756         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
 757         mdc_pack_fid(&body->fid1, obdo->o_id, 0, obdo->o_mode);
 758         body->size = obdo->o_size;
 759         body->blocks = obdo->o_blocks;
 760         body->flags = obdo->o_flags;
 761         body->valid = obdo->o_valid;
 762 //        memcpy(&body->handle, &och->och_fh, sizeof(body->handle));
 763
 764         ptlrpc_req_set_repsize(req, 2, size);
 765
 766         rc = ptlrpc_queue_wait(req);
 767         ptlrpc_req_finished(req);
 768         RETURN(rc);
 769 }
 770
 771 int mdc_readpage(struct obd_export *exp, struct ll_fid *fid, __u64 offset,
 772                  struct page *page, struct ptlrpc_request **request)
 773 {
 774         struct obd_import *imp = class_exp2cliimp(exp);
 775         struct ptlrpc_request *req = NULL;
 776         struct ptlrpc_bulk_desc *desc = NULL;
 777         struct mds_body *body;
 778         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 779         ENTRY;
 780
 781         CDEBUG(D_INODE, "inode: "LPU64"\n", fid->id);
 782
 783         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE, 2, size,
 784                               NULL);
 785         if (req == NULL)
 786                 GOTO(out, rc = -ENOMEM);
 787
 788         req->rq_request_portal = MDS_READPAGE_PORTAL;
 789         ptlrpc_at_set_req_timeout(req);
 790
 791         desc = ptlrpc_prep_bulk_imp(req, 1, BULK_PUT_SINK, MDS_BULK_PORTAL);
 792         if (desc == NULL)
 793                 GOTO(out, rc = -ENOMEM);
 794         /* NB req now owns desc and will free it when it gets freed */
 795
 796         ptlrpc_prep_bulk_page(desc, page, 0, CFS_PAGE_SIZE);
 797
 798         mdc_readdir_pack(req, REQ_REC_OFF, offset, CFS_PAGE_SIZE, fid);
 799
 800         ptlrpc_req_set_repsize(req, 2, size);
 801         rc = ptlrpc_queue_wait(req);
 802
 803         if (rc == 0) {
 804                 body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
 805                                           lustre_swab_mds_body);
 806                 if (body == NULL) {
 807                         CERROR("Can't unpack mds_body\n");
 808                         GOTO(out, rc = -EPROTO);
 809                 }
 810
 811                 if (req->rq_bulk->bd_nob_transferred != CFS_PAGE_SIZE) {
 812                         CERROR ("Unexpected # bytes transferred: %d"
 813                                 " (%lu expected)\n",
 814                                 req->rq_bulk->bd_nob_transferred,
 815                                 CFS_PAGE_SIZE);
 816                         GOTO (out, rc = -EPROTO);
 817                 }
 818         }
 819
 820         EXIT;
 821  out:
 822         *request = req;
 823         return rc;
 824 }
 825
 826
 827 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 828                          void *karg, void *uarg)
 829 {
 830         struct obd_device *obd = exp->exp_obd;
 831         struct obd_ioctl_data *data = karg;
 832         struct obd_import *imp = obd->u.cli.cl_import;
 833         struct llog_ctxt *ctxt;
 834         int rc;
 835         ENTRY;
 836
 837 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 838         MOD_INC_USE_COUNT;
 839 #else
 840         if (!try_module_get(THIS_MODULE)) {
 841                 CERROR("Can't get module. Is it alive?");
 842                 return -EINVAL;
 843         }
 844 #endif
 845         switch (cmd) {
 846         case OBD_IOC_CLIENT_RECOVER:
 847                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1);
 848                 if (rc < 0)
 849                         GOTO(out, rc);
 850                 GOTO(out, rc = 0);
 851         case IOC_OSC_SET_ACTIVE:
 852                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
 853                 GOTO(out, rc);
 854         case OBD_IOC_PARSE: {
 855                 ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT);
 856                 rc = class_config_parse_llog(ctxt, data->ioc_inlbuf1, NULL);
 857                 llog_ctxt_put(ctxt);
 858                 GOTO(out, rc);
 859         }
 860 #ifdef __KERNEL__
 861         case OBD_IOC_LLOG_INFO:
 862         case OBD_IOC_LLOG_PRINT: {
 863                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
 864                 rc = llog_ioctl(ctxt, cmd, data);
 865                 llog_ctxt_put(ctxt);
 866                 GOTO(out, rc);
 867         }
 868 #endif
 869         case OBD_IOC_POLL_QUOTACHECK:
 870                 rc = lquota_poll_check(quota_interface, exp,
 871                                        (struct if_quotacheck *)karg);
 872                 GOTO(out, rc);
 873         default:
 874                 CERROR("mdc_ioctl(): unrecognised ioctl %#x\n", cmd);
 875                 GOTO(out, rc = -ENOTTY);
 876         }
 877 out:
 878 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 879         MOD_DEC_USE_COUNT;
 880 #else
 881         module_put(THIS_MODULE);
 882 #endif
 883
 884         return rc;
 885 }
 886
 887 int mdc_set_info_async(struct obd_export *exp, obd_count keylen,
 888                        void *key, obd_count vallen, void *val,
 889                        struct ptlrpc_request_set *set)
 890 {
 891         struct obd_import *imp = class_exp2cliimp(exp);
 892         int rc = -EINVAL;
 893
 894         if (KEY_IS(KEY_INIT_RECOV)) {
 895                 if (vallen != sizeof(int))
 896                         RETURN(-EINVAL);
 897                 spin_lock(&imp->imp_lock);
 898                 imp->imp_initial_recov = *(int *)val;
 899                 spin_unlock(&imp->imp_lock);
 900
 901                 CDEBUG(D_HA, "%s: set imp_initial_recov = %d\n",
 902                        exp->exp_obd->obd_name, imp->imp_initial_recov);
 903                 RETURN(0);
 904         }
 905         /* Turn off initial_recov after we try all backup servers once */
 906         if (KEY_IS(KEY_INIT_RECOV_BACKUP)) {
 907                 if (vallen != sizeof(int))
 908                         RETURN(-EINVAL);
 909
 910                 spin_lock(&imp->imp_lock);
 911                 imp->imp_initial_recov_bk = *(int *)val;
 912                 if (imp->imp_initial_recov_bk)
 913                         imp->imp_initial_recov = 1;
 914                 spin_unlock(&imp->imp_lock);
 915
 916                 CDEBUG(D_HA, "%s: set imp_initial_recov_bk = %d\n",
 917                        exp->exp_obd->obd_name, imp->imp_initial_recov_bk);
 918                 RETURN(0);
 919         }
 920         if (KEY_IS("read-only")) {
 921                 struct ptlrpc_request *req;
 922                 int size[3] = { sizeof(struct ptlrpc_body), keylen, vallen };
 923                 char *bufs[3] = { NULL, key, val };
 924
 925                 if (vallen != sizeof(int))
 926                         RETURN(-EINVAL);
 927
 928                 if (*((int *)val)) {
 929                         imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY;
 930                         imp->imp_connect_data.ocd_connect_flags |=
 931                                 OBD_CONNECT_RDONLY;
 932                 } else {
 933                         imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY;
 934                         imp->imp_connect_data.ocd_connect_flags &=
 935                                 ~OBD_CONNECT_RDONLY;
 936                 }
 937
 938                 req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_SET_INFO,
 939                                       3, size, bufs);
 940                 if (req == NULL)
 941                         RETURN(-ENOMEM);
 942
 943                 ptlrpc_req_set_repsize(req, 1, NULL);
 944                 if (set) {
 945                         rc = 0;
 946                         ptlrpc_set_add_req(set, req);
 947                         ptlrpc_check_set(set);
 948                 } else {
 949                         rc = ptlrpc_queue_wait(req);
 950                         ptlrpc_req_finished(req);
 951                 }
 952
 953                 RETURN(rc);
 954         }
 955
 956         RETURN(rc);
 957 }
 958
 959 int mdc_get_info(struct obd_export *exp, __u32 keylen, void *key,
 960                  __u32 *vallen, void *val)
 961 {
 962         int rc = -EINVAL;
 963
 964         if (keylen == strlen("max_easize") &&
 965             memcmp(key, "max_easize", strlen("max_easize")) == 0) {
 966                 int mdsize, *max_easize;
 967
 968                 if (*vallen != sizeof(int))
 969                         RETURN(-EINVAL);
 970                 mdsize = *(int*)val;
 971                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
 972                         exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
 973                 max_easize = val;
 974                 *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
 975                 RETURN(0);
 976         }
 977         RETURN(rc);
 978 }
 979
 980 static int mdc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 981                       __u64 max_age, __u32 flags)
 982 {
 983         struct ptlrpc_request *req;
 984         struct obd_statfs *msfs;
 985         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*msfs) };
 986         ENTRY;
 987
 988         /* We could possibly pass max_age in the request (as an absolute
 989          * timestamp or a "seconds.usec ago") so the target can avoid doing
 990          * extra calls into the filesystem if that isn't necessary (e.g.
 991          * during mount that would help a bit).  Having relative timestamps
 992          * is not so great if request processing is slow, while absolute
 993          * timestamps are not ideal because they need time synchronization. */
 994         req = ptlrpc_prep_req(obd->u.cli.cl_import, LUSTRE_MDS_VERSION,
 995                               MDS_STATFS, 1, NULL, NULL);
 996         if (!req)
 997                 RETURN(-ENOMEM);
 998
 999         ptlrpc_req_set_repsize(req, 2, size);
1000
1001         if (flags & OBD_STATFS_NODELAY) {
1002                 /* procfs requests not want stay in wait for avoid deadlock */
1003                 req->rq_no_resend = 1;
1004                 req->rq_no_delay = 1;
1005         }
1006
1007         rc = ptlrpc_queue_wait(req);
1008
1009         if (rc)
1010                 GOTO(out, rc);
1011
1012         msfs = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*msfs),
1013                                   lustre_swab_obd_statfs);
1014         if (msfs == NULL) {
1015                 CERROR("Can't unpack obd_statfs\n");
1016                 GOTO(out, rc = -EPROTO);
1017         }
1018
1019         memcpy(osfs, msfs, sizeof(*msfs));
1020         EXIT;
1021 out:
1022         ptlrpc_req_finished(req);
1023
1024         return rc;
1025 }
1026
1027 static int mdc_pin(struct obd_export *exp, obd_id ino, __u32 gen, int type,
1028                    struct obd_client_handle *handle, int flag)
1029 {
1030         struct ptlrpc_request *req;
1031         struct mds_body *body;
1032         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
1033         ENTRY;
1034
1035         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1036                               MDS_PIN, 2, size, NULL);
1037         if (req == NULL)
1038                 RETURN(-ENOMEM);
1039
1040         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
1041         mdc_pack_fid(&body->fid1, ino, gen, type);
1042         body->flags = flag;
1043
1044         ptlrpc_req_set_repsize(req, 2, size);
1045
1046         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1047         rc = ptlrpc_queue_wait(req);
1048         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1049         if (rc) {
1050                 CERROR("pin failed: %d\n", rc);
1051                 ptlrpc_req_finished(req);
1052                 RETURN(rc);
1053         }
1054
1055         body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
1056                                   lustre_swab_mds_body);
1057         if (body == NULL) {
1058                 ptlrpc_req_finished(req);
1059                 RETURN(rc);
1060         }
1061
1062         memcpy(&handle->och_fh, &body->handle, sizeof(body->handle));
1063         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
1064
1065         OBD_ALLOC(handle->och_mod, sizeof(*handle->och_mod));
1066         if (handle->och_mod == NULL) {
1067                 DEBUG_REQ(D_ERROR, req, "can't allocate mdc_open_data");
1068                 RETURN(-ENOMEM);
1069         }
1070         handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
1071
1072         RETURN(rc);
1073 }
1074
1075 static int mdc_unpin(struct obd_export *exp,
1076                      struct obd_client_handle *handle, int flag)
1077 {
1078         struct ptlrpc_request *req;
1079         struct mds_body *body;
1080         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
1081         ENTRY;
1082
1083         if (handle->och_magic != OBD_CLIENT_HANDLE_MAGIC)
1084                 RETURN(0);
1085
1086         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1087                               MDS_CLOSE, 2, size, NULL);
1088         if (req == NULL)
1089                 RETURN(-ENOMEM);
1090
1091         body = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*body));
1092         memcpy(&body->handle, &handle->och_fh, sizeof(body->handle));
1093         body->flags = flag;
1094
1095         ptlrpc_req_set_repsize(req, 1, NULL);
1096         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1097         rc = ptlrpc_queue_wait(req);
1098         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
1099
1100         if (rc != 0)
1101                 CERROR("unpin failed: %d\n", rc);
1102
1103         ptlrpc_req_finished(req);
1104         ptlrpc_req_finished(handle->och_mod->mod_open_req);
1105         OBD_FREE(handle->och_mod, sizeof(*handle->och_mod));
1106         RETURN(rc);
1107 }
1108
1109 int mdc_sync(struct obd_export *exp, struct ll_fid *fid,
1110              struct ptlrpc_request **request)
1111 {
1112         struct ptlrpc_request *req;
1113         int size[2] = { sizeof(struct ptlrpc_body), sizeof(struct mds_body) };
1114         int rc;
1115         ENTRY;
1116
1117         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
1118                               MDS_SYNC, 2, size, NULL);
1119         if (!req)
1120                 RETURN(rc = -ENOMEM);
1121
1122         mdc_pack_req_body(req, REQ_REC_OFF, 0, fid, 0, 0);
1123
1124         ptlrpc_req_set_repsize(req, 2, size);
1125
1126         rc = ptlrpc_queue_wait(req);
1127         if (rc || request == NULL)
1128                 ptlrpc_req_finished(req);
1129         else
1130                 *request = req;
1131
1132         RETURN(rc);
1133 }
1134
1135 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
1136                             enum obd_import_event event)
1137 {
1138         int rc = 0;
1139
1140         LASSERT(imp->imp_obd == obd);
1141
1142         switch (event) {
1143         case IMP_EVENT_DISCON: {
1144                 ptlrpc_import_setasync(imp, -obd->obd_namespace->ns_max_unused);
1145                 break;
1146         }
1147         case IMP_EVENT_INACTIVE: {
1148                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
1149                 break;
1150         }
1151         case IMP_EVENT_INVALIDATE: {
1152                 struct ldlm_namespace *ns = obd->obd_namespace;
1153
1154                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
1155
1156                 break;
1157         }
1158         case IMP_EVENT_ACTIVE: {
1159                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
1160                 break;
1161         }
1162         case IMP_EVENT_OCD:
1163                 ptlrpc_import_setasync(imp, obd->obd_namespace->ns_max_unused);
1164                 break;
1165
1166         default:
1167                 CERROR("Unknown import event %x\n", event);
1168                 LBUG();
1169         }
1170         RETURN(rc);
1171 }
1172
1173 static int mdc_setup(struct obd_device *obd, obd_count len, void *buf)
1174 {
1175         struct client_obd *cli = &obd->u.cli;
1176         struct lprocfs_static_vars lvars = { 0 };
1177         int rc;
1178         ENTRY;
1179
1180         OBD_ALLOC(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1181         if (!cli->cl_rpc_lock)
1182                 RETURN(-ENOMEM);
1183         mdc_init_rpc_lock(cli->cl_rpc_lock);
1184
1185         ptlrpcd_addref();
1186
1187         OBD_ALLOC(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1188         if (!cli->cl_setattr_lock)
1189                 GOTO(err_rpc_lock, rc = -ENOMEM);
1190         mdc_init_rpc_lock(cli->cl_setattr_lock);
1191
1192         OBD_ALLOC(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1193         if (!cli->cl_close_lock)
1194                 GOTO(err_setattr_lock, rc = -ENOMEM);
1195         mdc_init_rpc_lock(cli->cl_close_lock);
1196
1197         rc = client_obd_setup(obd, len, buf);
1198         if (rc)
1199                 GOTO(err_close_lock, rc);
1200         lprocfs_mdc_init_vars(&lvars);
1201         if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0)
1202                 ptlrpc_lprocfs_register_obd(obd);
1203
1204         rc = obd_llog_init(obd, obd, 0, NULL, NULL);
1205         if (rc) {
1206                 mdc_cleanup(obd);
1207                 CERROR("failed to setup llogging subsystems\n");
1208         }
1209
1210         RETURN(rc);
1211
1212 err_close_lock:
1213         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1214 err_setattr_lock:
1215         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1216 err_rpc_lock:
1217         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1218         ptlrpcd_decref();
1219         RETURN(rc);
1220 }
1221
1222 /* Initialize the default and maximum LOV EA and cookie sizes.  This allows
1223  * us to make MDS RPCs with large enough reply buffers to hold the
1224  * maximum-sized (= maximum striped) EA and cookie without having to
1225  * calculate this (via a call into the LOV + OSCs) each time we make an RPC. */
1226 int mdc_init_ea_size(struct obd_export *mdc_exp, struct obd_export *lov_exp)
1227 {
1228         struct obd_device *obd = mdc_exp->exp_obd;
1229         struct client_obd *cli = &obd->u.cli;
1230         struct lov_stripe_md lsm = { .lsm_magic = LOV_MAGIC };
1231         struct lov_desc desc;
1232         __u32 valsize = sizeof(desc);
1233         __u32 stripes;
1234         int rc, size;
1235         ENTRY;
1236
1237         rc = obd_get_info(lov_exp, strlen(KEY_LOVDESC) + 1, KEY_LOVDESC,
1238                           &valsize, &desc);
1239         if (rc)
1240                 RETURN(rc);
1241
1242         stripes = min(desc.ld_tgt_count, (__u32)LOV_MAX_STRIPE_COUNT);
1243         lsm.lsm_stripe_count = stripes;
1244         size = obd_size_diskmd(lov_exp, &lsm);
1245
1246         if (cli->cl_max_mds_easize < size)
1247                 cli->cl_max_mds_easize = size;
1248
1249         lsm.lsm_stripe_count = desc.ld_default_stripe_count;
1250         size = obd_size_diskmd(lov_exp, &lsm);
1251
1252         if (cli->cl_default_mds_easize < size)
1253                 cli->cl_default_mds_easize = size;
1254
1255         size = stripes * sizeof(struct llog_cookie);
1256         if (cli->cl_max_mds_cookiesize < size)
1257                 cli->cl_max_mds_cookiesize = size;
1258
1259         CDEBUG(D_HA, "updating max_mdsize/max_cookiesize: %d/%d\n",
1260                cli->cl_max_mds_easize, cli->cl_max_mds_cookiesize);
1261
1262         RETURN(0);
1263 }
1264
1265 static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
1266 {
1267         int rc = 0;
1268         ENTRY;
1269
1270         switch (stage) {
1271         case OBD_CLEANUP_EARLY:
1272         case OBD_CLEANUP_EXPORTS:
1273                 /* If we set up but never connected, the
1274                    client import will not have been cleaned. */
1275                 if (obd->u.cli.cl_import) {
1276                         struct obd_import *imp;
1277                         imp = obd->u.cli.cl_import;
1278                         CERROR("client import never connected\n");
1279                         ptlrpc_invalidate_import(imp);
1280                         ptlrpc_free_rq_pool(imp->imp_rq_pool);
1281                         class_destroy_import(imp);
1282                         obd->u.cli.cl_import = NULL;
1283                 }
1284                 break;
1285         case OBD_CLEANUP_SELF_EXP:
1286                 rc = obd_llog_finish(obd, 0);
1287                 if (rc != 0)
1288                         CERROR("failed to cleanup llogging subsystems\n");
1289         case OBD_CLEANUP_OBD:
1290                 break;
1291         }
1292         RETURN(rc);
1293 }
1294
1295 static int mdc_cleanup(struct obd_device *obd)
1296 {
1297         struct client_obd *cli = &obd->u.cli;
1298
1299         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
1300         OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
1301         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
1302
1303         ptlrpc_lprocfs_unregister_obd(obd);
1304         lprocfs_obd_cleanup(obd);
1305         ptlrpcd_decref();
1306
1307         return client_obd_cleanup(obd);
1308 }
1309
1310
1311 static int mdc_llog_init(struct obd_device *obd, struct obd_device *tgt,
1312                          int count, struct llog_catid *logid,
1313                          struct obd_uuid *uuid)
1314 {
1315         struct llog_ctxt *ctxt;
1316         int rc;
1317         ENTRY;
1318
1319         rc = llog_setup(obd, LLOG_CONFIG_REPL_CTXT, tgt, 0, NULL,
1320                         &llog_client_ops);
1321         if (rc == 0) {
1322                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
1323                 ctxt->loc_imp = obd->u.cli.cl_import;
1324                 llog_ctxt_put(ctxt);
1325         }
1326
1327         rc = llog_setup(obd, LLOG_LOVEA_REPL_CTXT, tgt, 0, NULL,
1328                        &llog_client_ops);
1329         if (rc == 0) {
1330                 ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
1331                 ctxt->loc_imp = obd->u.cli.cl_import;
1332                 llog_ctxt_put(ctxt);
1333         }
1334
1335         RETURN(rc);
1336 }
1337
1338 static int mdc_llog_finish(struct obd_device *obd, int count)
1339 {
1340         int rc;
1341         ENTRY;
1342
1343         rc = llog_cleanup(llog_get_context(obd, LLOG_LOVEA_REPL_CTXT));
1344         if (rc) {
1345                 CERROR("can not cleanup LLOG_CONFIG_REPL_CTXT rc %d\n", rc);
1346         }
1347         rc = llog_cleanup(llog_get_context(obd, LLOG_CONFIG_REPL_CTXT));
1348         RETURN(rc);
1349 }
1350
1351 static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf)
1352 {
1353         struct lustre_cfg *lcfg = buf;
1354         struct lprocfs_static_vars lvars = { 0 };
1355         int rc = 0;
1356
1357         lprocfs_mdc_init_vars(&lvars);
1358
1359         rc = class_process_proc_param(PARAM_MDC, lvars.obd_vars, lcfg, obd);
1360         return(rc);
1361 }
1362
1363 struct obd_ops mdc_obd_ops = {
1364         .o_owner        = THIS_MODULE,
1365         .o_setup        = mdc_setup,
1366         .o_precleanup   = mdc_precleanup,
1367         .o_cleanup      = mdc_cleanup,
1368         .o_add_conn     = client_import_add_conn,
1369         .o_del_conn     = client_import_del_conn,
1370         .o_connect      = client_connect_import,
1371         .o_disconnect   = client_disconnect_export,
1372         .o_iocontrol    = mdc_iocontrol,
1373         .o_set_info_async = mdc_set_info_async,
1374         .o_get_info     = mdc_get_info,
1375         .o_statfs       = mdc_statfs,
1376         .o_pin          = mdc_pin,
1377         .o_unpin        = mdc_unpin,
1378         .o_import_event = mdc_import_event,
1379         .o_llog_init    = mdc_llog_init,
1380         .o_llog_finish  = mdc_llog_finish,
1381         .o_process_config = mdc_process_config,
1382 };
1383
1384 int __init mdc_init(void)
1385 {
1386         int rc;
1387         struct lprocfs_static_vars lvars = { 0 };
1388         lprocfs_mdc_init_vars(&lvars);
1389         request_module("lquota");
1390         quota_interface = PORTAL_SYMBOL_GET(mdc_quota_interface);
1391         init_obd_quota_ops(quota_interface, &mdc_obd_ops);
1392
1393         rc = class_register_type(&mdc_obd_ops, lvars.module_vars,
1394                                  LUSTRE_MDC_NAME);
1395         if (rc && quota_interface)
1396                 PORTAL_SYMBOL_PUT(mdc_quota_interface);
1397
1398         RETURN(rc);
1399 }
1400
1401 #ifdef __KERNEL__
1402 static void /*__exit*/ mdc_exit(void)
1403 {
1404         if (quota_interface)
1405                 PORTAL_SYMBOL_PUT(mdc_quota_interface);
1406
1407         class_unregister_type(LUSTRE_MDC_NAME);
1408 }
1409
1410 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1411 MODULE_DESCRIPTION("Lustre Metadata Client");
1412 MODULE_LICENSE("GPL");
1413
1414 EXPORT_SYMBOL(mdc_req2lustre_md);
1415 EXPORT_SYMBOL(mdc_free_lustre_md);
1416 EXPORT_SYMBOL(mdc_change_cbdata);
1417 EXPORT_SYMBOL(mdc_getstatus);
1418 EXPORT_SYMBOL(mdc_getattr);
1419 EXPORT_SYMBOL(mdc_getattr_name);
1420 EXPORT_SYMBOL(mdc_create);
1421 EXPORT_SYMBOL(mdc_unlink);
1422 EXPORT_SYMBOL(mdc_rename);
1423 EXPORT_SYMBOL(mdc_link);
1424 EXPORT_SYMBOL(mdc_readpage);
1425 EXPORT_SYMBOL(mdc_setattr);
1426 EXPORT_SYMBOL(mdc_close);
1427 EXPORT_SYMBOL(mdc_done_writing);
1428 EXPORT_SYMBOL(mdc_sync);
1429 EXPORT_SYMBOL(mdc_set_open_replay_data);
1430 EXPORT_SYMBOL(mdc_clear_open_replay_data);
1431 EXPORT_SYMBOL(mdc_store_inode_generation);
1432 EXPORT_SYMBOL(mdc_init_ea_size);
1433 EXPORT_SYMBOL(mdc_getxattr);
1434 EXPORT_SYMBOL(mdc_setxattr);
1435
1436 module_init(mdc_init);
1437 module_exit(mdc_exit);
1438 #endif